You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by js...@apache.org on 2026/04/18 13:46:50 UTC
(airflow) branch main updated: Fix KubernetesPodOperator to forward pod log levels to Airflow task logs (#64829)
This is an automated email from the ASF dual-hosted git repository.
jscheffl pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 277f01c280a Fix KubernetesPodOperator to forward pod log levels to Airflow task logs (#64829)
277f01c280a is described below
commit 277f01c280aecd35faef4669f024fdae05870fbb
Author: VishwasPatel <67...@users.noreply.github.com>
AuthorDate: Sat Apr 18 06:46:42 2026 -0700
Fix KubernetesPodOperator to forward pod log levels to Airflow task logs (#64829)
* Fix KubernetesPodOperator to forward pod log levels to Airflow task logs
Previously all pod container output was logged at INFO level regardless
of the original log level. Now ERROR, WARNING, DEBUG, CRITICAL prefixes
are detected and forwarded at the correct Python logging level.
closes: #64820
* Fix _log_message to use log.info() for INFO level messages
Use self.log.info() for INFO-level messages and self.log.log(level, ...)
only for non-INFO levels, restoring backward compatibility with existing
tests that mock log.info directly.
* Address review feedback: simplify log dispatch and add level detection to async path
- Consolidate if/else in _log_message to a single self.log.log(level, msg) call
- Apply _parse_log_level() in AsyncPodManager.fetch_container_logs_before_current_sec()
so deferrable/async task logs are also forwarded at the correct level instead of always INFO
- Update test to assert self.log.log() instead of self.log.info() for async path
* Fix test assertions for log level detection
- Add missing assert_called_once() in test_log_message_defaults_to_info_for_plain_lines
for consistency with test_log_message_uses_detected_log_level
- Use _parse_log_level() to derive expected level in async test assertions
instead of hardcoding logging.INFO, matching what production code does
* Fix system test assertions broken by log.log() change
test_volume_mount and test_log_output_configurations were asserting
log.info() calls, but _log_message now always uses self.log.log(level, ...)
Update both tests to assert log.log() calls instead.
---
.../test_kubernetes_pod_operator.py | 8 ++--
.../providers/cncf/kubernetes/utils/pod_manager.py | 43 +++++++++++++++---
.../unit/cncf/kubernetes/utils/test_pod_manager.py | 53 ++++++++++++++++++++--
3 files changed, 90 insertions(+), 14 deletions(-)
diff --git a/kubernetes-tests/tests/kubernetes_tests/test_kubernetes_pod_operator.py b/kubernetes-tests/tests/kubernetes_tests/test_kubernetes_pod_operator.py
index 652ee0ad954..b9c4ae9e653 100644
--- a/kubernetes-tests/tests/kubernetes_tests/test_kubernetes_pod_operator.py
+++ b/kubernetes-tests/tests/kubernetes_tests/test_kubernetes_pod_operator.py
@@ -554,7 +554,7 @@ class TestKubernetesPodOperatorSystem:
)
context = create_context(k)
k.execute(context=context)
- mock_logger.info.assert_any_call("%s", StringContainingId("retrieved from mount"))
+ mock_logger.log.assert_any_call(logging.INFO, StringContainingId("retrieved from mount"))
actual_pod = self.api_client.sanitize_for_serialization(k.pod)
self.expected_pod["spec"]["containers"][0]["args"] = args
self.expected_pod["spec"]["containers"][0]["volumeMounts"] = [
@@ -1480,7 +1480,7 @@ class TestKubernetesPodOperatorSystem:
)
# Test the _log_message method directly
- with mock.patch.object(k.pod_manager.log, "info") as mock_info:
+ with mock.patch.object(k.pod_manager.log, "log") as mock_log:
k.pod_manager._log_message(
message=marker,
container_name="base",
@@ -1489,8 +1489,8 @@ class TestKubernetesPodOperatorSystem:
)
# Check that the message was logged with the expected format
- mock_info.assert_called_once()
- logged_message = mock_info.call_args[0][1] # Second argument is the message
+ mock_log.assert_called_once()
+ logged_message = mock_log.call_args[0][1] # Second argument is the message
assert expected_log_message_check(marker, logged_message)
diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/pod_manager.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/pod_manager.py
index 3cc0229ea61..0ece2b978b0 100644
--- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/pod_manager.py
+++ b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/pod_manager.py
@@ -21,7 +21,9 @@ from __future__ import annotations
import asyncio
import enum
import json
+import logging
import math
+import re
import time
from collections.abc import Callable, Generator, Iterable
from contextlib import closing
@@ -74,6 +76,33 @@ Sentinel for no xcom result.
:meta private:
"""
+_POD_LOG_LEVEL_PATTERN = re.compile(
+ r"^\s*(?:\[)?(DEBUG|INFO|WARNING|WARN|ERROR|CRITICAL|FATAL)(?:\])?\s*[:\-]?\s*",
+ re.IGNORECASE,
+)
+_POD_LOG_LEVEL_MAP: dict[str, int] = {
+ "DEBUG": logging.DEBUG,
+ "INFO": logging.INFO,
+ "WARNING": logging.WARNING,
+ "WARN": logging.WARNING,
+ "ERROR": logging.ERROR,
+ "CRITICAL": logging.CRITICAL,
+ "FATAL": logging.CRITICAL,
+}
+
+
+def _parse_log_level(message: str) -> int:
+ """
+ Detect the Python logging level from a pod log line's prefix.
+
+ Recognises common formats: ``ERROR:``, ``[ERROR]``, ``WARNING -``, etc.
+ Returns ``logging.INFO`` when no known prefix is found (backwards-compatible).
+ """
+ match = _POD_LOG_LEVEL_PATTERN.match(message)
+ if match:
+ return _POD_LOG_LEVEL_MAP.get(match.group(1).upper(), logging.INFO)
+ return logging.INFO
+
class XComRetrievalError(AirflowException):
"""When not possible to get xcom."""
@@ -436,18 +465,18 @@ class PodManager(LoggingMixin):
container_name_log_prefix_enabled: bool,
log_formatter: Callable[[str, str], str] | None,
) -> None:
- """Log a message with appropriate formatting."""
+ """Log a message at the level detected from its prefix, with appropriate formatting."""
if is_log_group_marker(message):
print(message)
else:
+ level = _parse_log_level(message)
if log_formatter:
formatted_message = log_formatter(container_name, message)
- self.log.info("%s", formatted_message)
else:
- log_message = (
+ formatted_message = (
f"[{container_name}] {message}" if container_name_log_prefix_enabled else message
)
- self.log.info("%s", log_message)
+ self.log.log(level, formatted_message)
def fetch_container_logs(
self,
@@ -1180,7 +1209,8 @@ class AsyncPodManager(LoggingMixin):
if is_log_group_marker(message_to_log):
print(message_to_log)
else:
- self.log.info("[%s] %s", container_name, message_to_log)
+ level = _parse_log_level(message_to_log)
+ self.log.log(level, "[%s] %s", container_name, message_to_log)
message_to_log = message
elif message_to_log: # continuation of the previous log line
message_to_log = f"{message_to_log}\n{message}"
@@ -1190,5 +1220,6 @@ class AsyncPodManager(LoggingMixin):
if is_log_group_marker(message_to_log):
print(message_to_log)
else:
- self.log.info("[%s] %s", container_name, message_to_log)
+ level = _parse_log_level(message_to_log)
+ self.log.log(level, "[%s] %s", container_name, message_to_log)
return now # Return the current time as the last log time to ensure logs from the current second are read in the next fetch.
diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_pod_manager.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_pod_manager.py
index c48a6608992..155c73fc7a6 100644
--- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_pod_manager.py
+++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_pod_manager.py
@@ -36,6 +36,7 @@ from airflow.providers.cncf.kubernetes.utils.pod_manager import (
PodManager,
PodPhase,
XComRetrievalError,
+ _parse_log_level,
log_pod_event,
parse_log_line,
)
@@ -153,6 +154,48 @@ class TestPodManager:
callbacks=[MockKubernetesPodOperatorCallback],
)
+ @pytest.mark.parametrize(
+ ("message", "expected_level"),
+ [
+ ("ERROR: something went wrong", logging.ERROR),
+ ("WARNING: low disk space", logging.WARNING),
+ ("WARN: deprecated usage", logging.WARNING),
+ ("DEBUG: entering function", logging.DEBUG),
+ ("CRITICAL: system failure", logging.CRITICAL),
+ ("FATAL: unrecoverable error", logging.CRITICAL),
+ ("INFO: starting up", logging.INFO),
+ ("[ERROR] bracketed prefix", logging.ERROR),
+ ("plain log line with no level", logging.INFO),
+ ("", logging.INFO),
+ ],
+ )
+ def test_parse_log_level(self, message, expected_level):
+ assert _parse_log_level(message) == expected_level
+
+ def test_log_message_uses_detected_log_level(self):
+ """_log_message should forward ERROR lines at ERROR level, not INFO."""
+ with mock.patch.object(self.pod_manager.log, "log") as mock_log:
+ self.pod_manager._log_message(
+ message="ERROR: something failed",
+ container_name="base",
+ container_name_log_prefix_enabled=True,
+ log_formatter=None,
+ )
+ mock_log.assert_called_once()
+ assert mock_log.call_args[0][0] == logging.ERROR
+
+ def test_log_message_defaults_to_info_for_plain_lines(self):
+ """_log_message should use INFO for lines without a known level prefix."""
+ with mock.patch.object(self.pod_manager.log, "log") as mock_log:
+ self.pod_manager._log_message(
+ message="Starting application",
+ container_name="base",
+ container_name_log_prefix_enabled=True,
+ log_formatter=None,
+ )
+ mock_log.assert_called_once()
+ assert mock_log.call_args[0][0] == logging.INFO
+
def test_read_pod_logs_successfully_returns_logs(self):
mock.sentinel.metadata = mock.MagicMock()
self.mock_kube_client.read_namespaced_pod_log.return_value = mock.sentinel.logs
@@ -1503,17 +1546,19 @@ class TestAsyncPodManager:
async_hook=mock_async_hook,
callbacks=[],
)
- with mock.patch.object(async_pod_manager.log, "info") as mock_log_info:
+ with mock.patch.object(async_pod_manager.log, "log") as mock_log:
result = await async_pod_manager.fetch_container_logs_before_current_sec(
pod=pod, container_name=container_name, since_time=since_time
)
assert result == now
for expected in expected_log_messages:
- mock_log_info.assert_any_call("[%s] %s", container_name, expected)
+ mock_log.assert_any_call(_parse_log_level(expected), "[%s] %s", container_name, expected)
for not_expected in not_expected_log_messages:
- unexpected_call = mock.call("[%s] %s", container_name, not_expected)
- assert unexpected_call not in mock_log_info.mock_calls
+ unexpected_call = mock.call(
+ _parse_log_level(not_expected), "[%s] %s", container_name, not_expected
+ )
+ assert unexpected_call not in mock_log.mock_calls
@pytest.mark.asyncio
async def test_fetch_container_logs_before_current_sec_error_handling(self):