airflow UI failed to read correct worker host in logs after multiple retries.

[ranmx]

Apache Airflow version

2.10.3

If "Other Airflow 2 version" selected, which one?

No response

What happened?

I have a latest (2.10.3) airflow installed in k8s.
Executors are celeryExecutors.
I have two workers deployed.

• airflow-worker-0
• airflow-worker-1

I find out that if a task failed and retry, the log address host part will be overwritten by the last attemp.
Thus if the task retired on different workers, the log on machines different from the last worker will be unavaliable.
But if I go to the work node, I can find the log in the local path.

This attemp failed to get the log. Note that this actually run on airflow-worker-1 and logs are avaliable on local path

airflow-worker-1 local path

This is because the host is over written by the last attempt:

I think the code here in airflow/utils/log/file_task_handler.py:461 should be changed as it uses the last task_instance for all the attempts.

 def read(self, task_instance, try_number=None, metadata=None):
        """
        Read logs of given task instance from local machine.

        :param task_instance: task instance object
        :param try_number: task instance try_number to read logs from. If None
                           it returns all logs separated by try_number
        :param metadata: log metadata, can be used for steaming log reading and auto-tailing.
        :return: a list of listed tuples which order log string by host
        """
        # Task instance increments its try number when it starts to run.
        # So the log for a particular task try will only show up when
        # try number gets incremented in DB, i.e logs produced the time
        # after cli run and before try_number + 1 in DB will not be displayed.
        if try_number is None:
            next_try = task_instance.next_try_number
            try_numbers = list(range(1, next_try))
        elif try_number < 1:
            logs = [
                [("default_host", f"Error fetching the logs. Try number {try_number} is invalid.")],
            ]
            return logs, [{"end_of_log": True}]
        else:
            try_numbers = [try_number]

        logs = [""] * len(try_numbers)
        metadata_array = [{}] * len(try_numbers)

        # subclasses implement _read and may not have log_type, which was added recently
        for i, try_number_element in enumerate(try_numbers):
            log, out_metadata = self._read(task_instance, try_number_element, metadata)
            # es_task_handler return logs grouped by host. wrap other handler returning log string
            # with default/ empty host so that UI can render the response in the same way
            logs[i] = log if self._read_grouped_logs() else [(task_instance.hostname, log)]
            metadata_array[i] = out_metadata

        return logs, metadata_array

What you think should happen instead?

Each attempt should has its own host.

How to reproduce

• latest (2.10.3) airflow installed in k8s.
• Executors are celeryExecutors.
• Multiple workers deployed.
• Task failed and retied on different wokers

Operating System

Debian GNU/Linux 12 (bookworm)

Versions of Apache Airflow Providers

No response

Deployment

Docker-Compose

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[potiuk]

This can be easily solved (and is recommended now) by having a shared volume where you keep logs. We are not likely to address it because we are completely changing it for Airflow 3 - and logs will be retrieved via Task API server, not via workers most likely, so we can convert that to a discussion now.

[ranmx]

Thank you for your reply. Hopefully this information could be added to the documents.

[potiuk]

Sure. Feel free to add it where you think is appropiate. You can become a contributor to Airflow like more than 3000 others. It's super easy - just click "suggest a change on this page" button in the documentation website and Pull Request will be open for you - and you can add it there and submit PR. It's as easy as commenting on issue here - and you can not only comment but also actively contribute to the free software you use. Least you can do to thank thousands of people that work in mostly their free time to create Airflow. Can we coun t on your help there @ranmx ?