Allow FileTaskHandler get running tasks logs not only from default executor

[insomnes]

Description

FileTaskHandler should have the ability to get logs not only from the default executor, but from one configured on the task instance

Use case/motivation

With the introduction of multiple executors flow of the FileTaskHandler log reading of the running task was broken if the task was run by a non-default executor.

For example, usage of "LocalExecutor,KubernetesExecutor" leads to an inability of online logs reading in web interface.

I was able to overcome it with a custom executor workaround, but I belive that this would be a reasonable feature to expect with multiple executors.

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

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

[insomnes]

I am talking about this line:

        if ti.state == TaskInstanceState.RUNNING:
            response = self._executor_get_task_log(ti, try_number)

...

    @cached_property
    def _executor_get_task_log(self) -> Callable[[TaskInstance, int], tuple[list[str], list[str]]]:
        """This cached property avoids loading executor repeatedly."""
        executor = ExecutorLoader.get_default_executor()
        return executor.get_task_log

I'm not very familiar with ExecutorLoader and maybe the real culprit is @cached_property usage, but this needs some rework for multiple executors support

[jason810496]

I've been working on refactoring the log reading process recently, also focusing on FileTaskHandler.
I believe this can be achieved by reading directly from ti.executor instead of using ExecutorLoader to determine the correct executor method.