Task SDK cannot upload remote logs due to missing handlers

[ianbuss]

Apache Airflow version

3.0.0

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

No response

What happened?

The configure_logging method in the task sdk overwrites the configured Airflow loggers from airflow.logging_config, which makes it look like the airflow.task logger has no handlers. This prevents the _upload_logs function from being able to update logs.

A potential fix is to cache the configured handler for airflow.task before the task sdk supervise method and use that cached handler. A longer term fix is to not use the derived handlers but to read the logging configuration directly in the supervise method to instantiate the correct log upload functionality.

What you think should happen instead?

The task sdk superviser should be able to use the configured remote loggers.

How to reproduce

Run any executor with logging.remote_logging=True and a configured remote logging configuration.

Operating System

Debian GNU/Linux

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

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

[ianbuss]

While identifying the issue I applied the following "hack" in https://github.com/astronomer/airflow/blob/eebdf4e9fbbe9badd1f93dbaf1c9a0d1d2aa357c/task-sdk/src/airflow/sdk/log.py

diff --git i/task-sdk/src/airflow/sdk/log.py w/task-sdk/src/airflow/sdk/log.py
index 3da1d1e1f7..0eb938e4fc 100644
--- i/task-sdk/src/airflow/sdk/log.py
+++ w/task-sdk/src/airflow/sdk/log.py
@@ -481,6 +481,9 @@ def upload_to_remote(logger: FilteringBoundLogger):
     base_log_folder = conf.get("logging", "base_log_folder")
     relative_path = Path(fname).relative_to(base_log_folder)

+    from airflow.logging_config import configure_logging
+    configure_logging()
+
     handler = TaskLogReader().log_handler
     if not isinstance(handler, FileTaskHandler):
         logger.warning(
@@ -493,7 +496,9 @@ def upload_to_remote(logger: FilteringBoundLogger):
     # set_context() which opens a real FH again. (And worse, in some cases it _truncates_ the file too). This
     # is just for the first Airflow 3 betas, but we will re-write a better remote log interface that isn't
     # tied to being a logging Handler.
+    handler.log_meta_dict = {}
     handler.log_relative_path = relative_path.as_posix()  # type: ignore[attr-defined]
     handler.upload_on_close = True  # type: ignore[attr-defined]

     handler.close()

[ashb]

On a related note, the CloudWatch log handler needs a different interface, as it wants to upload log messages as it happens.