Add support for loading logging configuration in configure_logging function (#56340)

[nailo2c]

closes: #56340

Why

The user configured a custom logging handler in Airflow by following this doc.

It works in a direct Python test, but it does not work when setting logging_config_class.

How

It lacks the load_logging_config step in the configure_logging function; adding it resolves this issue.

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

Line 96 in 67c95c3

def configure_logging(

What

Step 1: Setup

Here are the settings I used to reproduce this issue locally:

files/airflow-breeze-config/environment_variables.env (Click to expand)

AIRFLOW__LOGGING__LOGGING_CONFIG_CLASS=log_config.LOGGING_CONFIG
PYTHONPATH=/files/config${PYTHONPATH:+:$PYTHONPATH}

files/config/clickhouse_logging.py (Click to expand)

import json
import logging
import os
from pathlib import Path
from threading import Lock
from typing import Any

DEFAULT_LOG_PATH = os.path.join(
    os.environ.get("AIRFLOW_HOME", "/opt/airflow"),
    "logs",
    "clickhouse_handler.log",
)


class ClickHouseHTTPHandler(logging.Handler):
    def __init__(self, log_path: str | None = None) -> None:
        super().__init__()
        self.log_path = log_path or DEFAULT_LOG_PATH
        Path(self.log_path).parent.mkdir(parents=True, exist_ok=True)
        self._lock = Lock()

    def emit(self, record: logging.LogRecord) -> None:
        try:
            payload: dict[str, Any] = {
                "ts": record.created,
                "logger": record.name,
                "level": record.levelname,
                "message": self.format(record),
            }
            line = json.dumps(payload, ensure_ascii=True)
            with self._lock:
                with open(self.log_path, "a", encoding="utf-8") as handle:
                    handle.write(line + "\n")
        except Exception:
            self.handleError(record)

files/config/log_config.py (Click to expand)

import os
from copy import deepcopy

from airflow.config_templates.airflow_local_settings import DEFAULT_LOGGING_CONFIG

LOGGING_CONFIG = deepcopy(DEFAULT_LOGGING_CONFIG)
REMOTE_TASK_LOG = None        # to prevent triggerer and dag-processor crash
DEFAULT_REMOTE_CONN_ID = None # to prevent triggerer and dag-processor crash

LOGGING_CONFIG["handlers"]["clickhouse"] = {
    "class": "clickhouse_logging.ClickHouseHTTPHandler",
    "log_path": os.environ.get(
        "AIRFLOW__CLICKHOUSE_LOG_PATH",
        os.path.join(os.environ.get("AIRFLOW_HOME", "/opt/airflow"), "logs", "clickhouse_handler.log"),
    ),
    "level": "INFO",
}

task_handlers = list(LOGGING_CONFIG["loggers"]["airflow.task"]["handlers"])
if "clickhouse" not in task_handlers:
    task_handlers.append("clickhouse")
LOGGING_CONFIG["loggers"]["airflow.task"]["handlers"] = task_handlers

files/dags/repro_logging_config_class.py (Click to expand)

import logging

import pendulum

from airflow import DAG
from airflow.operators.python import PythonOperator


def _describe_handlers(logger: logging.Logger) -> list[str]:
    descriptions = []
    for handler in logger.handlers:
        name = getattr(handler, "name", None)
        level = logging.getLevelName(handler.level)
        descriptions.append(
            f"{handler.__class__.__module__}.{handler.__class__.__name__}(name={name}, level={level})"
        )
    return descriptions


def emit_logs() -> None:
    logger = logging.getLogger("airflow.task")
    logger.info("airflow.task handlers: %s", _describe_handlers(logger))
    logger.info("airflow.task propagate=%s parent=%s", logger.propagate, getattr(logger.parent, "name", None))
    logger.info("stdlib logger message from repro task")
    print("stdout message from repro task")


with DAG(
    dag_id="repro_logging_config_class",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    schedule=None,
    catchup=False,
    tags=["repro"],
) as dag:
    PythonOperator(task_id="emit_logs", python_callable=emit_logs)

Step 2: Before the fix - Reproducing the issue

First, make sure the Airflow config has the correct value:

[Breeze:3.10.19] root@e09e71f55eb8:/opt/airflow/logs$ airflow config get-value logging logging_config_class
log_config.LOGGING_CONFIG

I ran this command in the container, and it wrote a line to logs/clickhouse_handler.log as expected:

PYTHONPATH=/files/config python - <<'PY'
import logging
from clickhouse_logging import ClickHouseHTTPHandler

logger = logging.getLogger("airflow.task")
logger.setLevel(logging.INFO)
logger.addHandler(ClickHouseHTTPHandler())
logger.info("direct python handler test")
PY

logs/clickhouse_handler.log

{"ts": 1768007319.4109492, "logger": "airflow.task", "level": "INFO", "message": "direct python handler test"}

However, when I triggered the DAG, nothing was written to the log file, and the UI logs showed that the handler had not been loaded from the config.

Step 3: After the fix

After the fix, it works as expected and the config is loaded:

It also wrote logs to logs/clickhouse_handler.log:

{"ts": 1768084859.17813, "logger": "airflow.task", "level": "INFO", "message": "airflow.task handlers: ['airflow.utils.log.file_task_handler.FileTaskHandler(name=task, level=NOTSET)', 'clickhouse_logging.ClickHouseHTTPHandler(name=clickhouse, level=INFO)']"}
{"ts": 1768084859.1806972, "logger": "airflow.task", "level": "INFO", "message": "airflow.task propagate=True parent=airflow"}
{"ts": 1768084859.1817467, "logger": "airflow.task", "level": "INFO", "message": "stdlib logger message from repro task"}

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)
  Generated-by: GPT5.2 following the guidelines

[ashb]

The doc is out of date, and this change will result in half the logs not going via this processor.

In Airflow 3 the mechanism to configure logging has changed, and things should be implemented as a structlog processor, and user control of the full logging config is not possible anymore. Sorry.

Yes this is a problem that needs addressing, but the approach in this PR is not correct so I'm going to close this PR

[nailo2c]

Thanks for the review, and sorry my approach didn't match the current logging architecture in Airflow 3. This is also a good opportunity for me to learn.

To align with the architecture, you mentioned using a structlog processor. Does that typically sit within a RemoteLogIO implementation? (e.g. CloudWatchRemoteLogIO)

Appreciate the guidance, this review helps a lot :)