From ae5782cd2d884f111cc4ef8efb3dab23752a1ba5 Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Tue, 25 Apr 2023 03:55:09 -0700
Subject: [PATCH] TaskLogReader should check both "airflow.task" and root
 logger for handler.

When task is running, handlers are moved from airflow.task to root.  Under normal circumstances this is a non-issue because the reader is instantiated in a webserver but there are certain contrived scenarios such as test dags which instantiate the reader in the context of a running task.
---
 airflow/utils/log/log_reader.py | 16 ++++++++++++----
 1 file changed, 12 insertions(+), 4 deletions(-)

diff --git a/airflow/utils/log/log_reader.py b/airflow/utils/log/log_reader.py
index 5cc8b9377e186..63529e71e2924 100644
--- a/airflow/utils/log/log_reader.py
+++ b/airflow/utils/log/log_reader.py
@@ -97,11 +97,19 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
 
     @cached_property
     def log_handler(self):
-        """Log handler, which is configured to read logs."""
-        logger = logging.getLogger("airflow.task")
+        """Get the log handler which is configured to read logs."""
         task_log_reader = conf.get("logging", "task_log_reader")
-        handler = next((handler for handler in logger.handlers if handler.name == task_log_reader), None)
-        return handler
+
+        def handlers():
+            """
+            Yield all handlers first from airflow.task logger then root logger.
+
+            Depending on whether we're in a running task, it could be in either of these locations.
+            """
+            yield from logging.getLogger("airflow.task").handlers
+            yield from logging.getLogger().handlers
+
+        return next((h for h in handlers() if h.name == task_log_reader), None)
 
     @property
     def supports_read(self):