Separate configurations for colorized and json logs in Task SDK / Celery Executor

task-sdk/src/airflow/sdk/log.py

@@ -140,7 +140,7 @@ def emit(self, record: logging.LogRecord):
 
 
 @cache
-def logging_processors(enable_pretty_log: bool, mask_secrets: bool = True):
+def logging_processors(enable_pretty_log: bool, mask_secrets: bool = True, enable_colors: bool = True):
     if enable_pretty_log:
         timestamper = structlog.processors.MaybeTimeStamper(fmt="%Y-%m-%d %H:%M:%S.%f")
     else:
@@ -176,20 +176,32 @@ def logging_processors(enable_pretty_log: bool, mask_secrets: bool = True):
     )
 
     if enable_pretty_log:
-        rich_exc_formatter = structlog.dev.RichTracebackFormatter(
-            # These values are picked somewhat arbitrarily to produce useful-but-compact tracebacks. If
-            # we ever need to change these then they should be configurable.
-            extra_lines=0,
-            max_frames=30,
-            indent_guides=False,
-            suppress=suppress,
-        )
-        my_styles = structlog.dev.ConsoleRenderer.get_default_level_styles()
-        my_styles["debug"] = structlog.dev.CYAN
+        if enable_colors:
+            rich_exc_formatter = structlog.dev.RichTracebackFormatter(
+                # These values are picked somewhat arbitrarily to produce useful-but-compact tracebacks. If
+                # we ever need to change these then they should be configurable.
+                extra_lines=0,
+                max_frames=30,
+                indent_guides=False,
+                suppress=suppress,
+            )
+            my_styles = structlog.dev.ConsoleRenderer.get_default_level_styles()
+            my_styles["debug"] = structlog.dev.CYAN
 
-        console = structlog.dev.ConsoleRenderer(
-            exception_formatter=rich_exc_formatter, level_styles=my_styles
-        )
+            console = structlog.dev.ConsoleRenderer(
+                exception_formatter=rich_exc_formatter, level_styles=my_styles
+            )
+        else:
+            # Create a console renderer without colors - use the same RichTracebackFormatter
+            # but rely on ConsoleRenderer(colors=False) to disable colors
+            rich_exc_formatter = structlog.dev.RichTracebackFormatter(
+                suppress=suppress,
+                show_locals=False,
+            )
+            console = structlog.dev.ConsoleRenderer(
+                colors=False,
+                exception_formatter=rich_exc_formatter,
+            )
         processors.append(console)
         return processors, {
             "timestamper": timestamper,
@@ -252,6 +264,7 @@ def configure_logging(
     output: BinaryIO | TextIO | None = None,
     cache_logger_on_first_use: bool = True,
     sending_to_supervisor: bool = False,
+    enable_colors: bool | None = None,
 ):
     """Set up struct logging and stdlib logging config."""
     if log_level == "DEFAULT":
@@ -261,13 +274,24 @@ def configure_logging(
 
             log_level = conf.get("logging", "logging_level", fallback="INFO")
 
+    # If enable_colors is not explicitly set, read from configuration
+    if enable_colors is None:
+        if "airflow.configuration" in sys.modules:
+            from airflow.configuration import conf
+
+            enable_colors = conf.getboolean("logging", "colored_console_log", fallback=True)
+        else:
+            enable_colors = True
+
     lvl = structlog.stdlib.NAME_TO_LEVEL[log_level.lower()]
 
     if enable_pretty_log:
         formatter = "colored"
     else:
         formatter = "plain"
-    processors, named = logging_processors(enable_pretty_log, mask_secrets=not sending_to_supervisor)
+    processors, named = logging_processors(
+        enable_pretty_log, mask_secrets=not sending_to_supervisor, enable_colors=enable_colors
+    )
     timestamper = named["timestamper"]
 
     pre_chain: list[structlog.typing.Processor] = [

task-sdk/tests/task_sdk/log/test_log.py

@@ -147,3 +147,73 @@ def test_logs_are_masked(captured_logs):
         "try_number=1, map_index=-1, hostname=None, context_carrier=None)",
         "timestamp": "2025-03-25T05:13:27.073918Z",
     }
+
+
+def test_logging_processors_with_colors():
+    """Test that logging_processors creates colored console renderer when enable_colors=True."""
+    from airflow.sdk.log import logging_processors
+
+    _, named = logging_processors(enable_pretty_log=True, enable_colors=True)
+    assert "console" in named
+    console_renderer = named["console"]
+    assert hasattr(console_renderer, "_styles")
+
+
+def test_logging_processors_without_colors():
+    """Test that logging_processors creates non-colored console renderer when enable_colors=False."""
+    from airflow.sdk.log import logging_processors
+
+    _, named = logging_processors(enable_pretty_log=True, enable_colors=False)
+    assert "console" in named
+    console_renderer = named["console"]
+    assert hasattr(console_renderer, "_styles")
+    assert console_renderer._styles.__name__ == "_PlainStyles"
+
+
+def test_logging_processors_json_format():
+    """Test that logging_processors creates JSON renderer when enable_pretty_log=False."""
+    from airflow.sdk.log import logging_processors
+
+    _, named = logging_processors(enable_pretty_log=False, enable_colors=True)
+    assert "console" not in named
+    assert "json" in named
+
+
+def test_configure_logging_respects_colored_console_log_config():
+    """Test that configure_logging respects the colored_console_log configuration."""
+    from airflow.sdk.log import configure_logging, reset_logging
+
+    mock_conf = mock.MagicMock()
+    mock_conf.get.return_value = "INFO"
+    mock_conf.getboolean.return_value = False  # colored_console_log = False
+    mock_config_module = mock.MagicMock()
+    mock_config_module.conf = mock_conf
+    with mock.patch.dict("sys.modules", {"airflow.configuration": mock_config_module}):
+        reset_logging()
+        configure_logging(enable_pretty_log=True)
+        mock_conf.getboolean.assert_called_with("logging", "colored_console_log", fallback=True)
+
+
+def test_configure_logging_explicit_enable_colors():
+    """Test that configure_logging respects explicit enable_colors parameter."""
+    from airflow.sdk.log import configure_logging, reset_logging
+
+    mock_conf = mock.MagicMock()
+    mock_conf.get.return_value = "INFO"
+    mock_conf.getboolean.return_value = True  # colored_console_log = True
+
+    with mock.patch("airflow.sdk.log.sys.modules", {"airflow.configuration": mock.MagicMock()}):
+        with mock.patch("airflow.configuration.conf", mock_conf):
+            reset_logging()
+            # Explicitly disable colors despite config saying True
+            configure_logging(enable_pretty_log=True, enable_colors=False)
+            mock_conf.getboolean.assert_not_called()
+
+
+def test_configure_logging_no_airflow_config():
+    """Test that configure_logging works when airflow.configuration is not available."""
+    from airflow.sdk.log import configure_logging, reset_logging
+
+    with mock.patch("airflow.sdk.log.sys.modules", {}):
+        reset_logging()
+        configure_logging(enable_pretty_log=True)