Deferrable tasks with remote logging don't push triggerer logs to AWS CloudWatch

[maxprychynenko-buildzoom]

Under which category would you file this issue?

Task SDK

Apache Airflow version

3.3.0

What happened and how to reproduce it?

Issue Description

In Airflow 3.3.0 (and 3.1.7 as well) when one is using deferrable tasks with remote logging configured to send logs to AWS CloudWatch logs which are produced by triggers in the triggerer process are not being sent to AWS CloudWatch. Even though logs produced by non-deferrable tasks and triggerer process itself ("# triggers currently running", etc) are being pushed to AWS CloudWatch as expected.

Issue Analysis

I don't know much about internals of Airflow, but this is what my AI assistant and I were able to identify while debugging this problem:

The SDK's configure_logging() accesses CloudWatchRemoteLogIO.processors to build the structlog processor chain, and then calls the shared configure_logging() which runs logging.config.dictConfig(). The problem is that accessing .processors eagerly creates a watchtower.CloudWatchLogHandler (which, as a logging.Handler subclass, auto-registers in logging._handlerList), and the dictConfig() call that follows closes every handler in that list via _clearExistingHandlers(). This sets watchtower's shutting_down = True, so all subsequent calls to handler.handle() from the CloudWatch structlog processor are silently dropped. The net effect is that trigger logs are written to local files but never reach CloudWatch — and since delete_local_logs is typically enabled, the local files are deleted on trigger completion, losing the logs permanently.

Steps To Reproduce

I've created a docker compose simulation that can be used to reproduce the problem.

Prerequisites: the simulation expects one to have an AWS account with privileges needed to write logs to a CloudWatch group.

File: ./Dockerfile

FROM apache/airflow:3.3.0-python3.13

USER airflow

RUN pip install --no-cache-dir \
    apache-airflow-providers-amazon==9.31.0 \
    apache-airflow-providers-standard==1.16.0

File: ./docker-compose.yml

x-airflow-common:
  &airflow-common
  build: .
  environment:
    &airflow-common-env
    AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres:5432/airflow
    AIRFLOW__CORE__EXECUTOR: LocalExecutor
    AIRFLOW__CORE__FERNET_KEY: 'olFLXR4nU_6WBvIsSDPIDCKWjh3OCWSMfWjt6XU0VfM='
    AIRFLOW__CORE__EXECUTION_API_SERVER_URL: http://airflow-webserver:8080/execution/
    AIRFLOW__CORE__SIMPLE_AUTH_MANAGER_ALL_ADMINS: 'True'
    AIRFLOW__API__SECRET_KEY: reproduction-secret-key-not-for-production
    AIRFLOW__API_AUTH__JWT_SECRET: reproduction-jwt-secret-not-for-production
    AIRFLOW__API_AUTH__JWT_ALGORITHM: HS512
    AIRFLOW__LOGGING__REMOTE_LOGGING: 'True'
    AIRFLOW__LOGGING__REMOTE_BASE_LOG_FOLDER: 'cloudwatch://arn:aws:logs:${AWS_REGION}:${AWS_ACCOUNT_ID}:log-group:${AWS_LOG_GROUP_NAME}'
    AIRFLOW__LOGGING__REMOTE_LOG_CONN_ID: aws_default
    AIRFLOW__LOGGING__DELETE_LOCAL_LOGS: 'True'
    AIRFLOW__LOGGING__LOGGING_LEVEL: INFO
    AWS_DEFAULT_REGION: ${AWS_REGION}
    AWS_REGION: ${AWS_REGION}
    AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID}
    AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY}
    AWS_SESSION_TOKEN: ${AWS_SESSION_TOKEN:-}
    PYTHONPATH: '/opt/airflow/config:/opt/airflow/dags'
    PYTHONUNBUFFERED: '1'
  volumes:
    - ./dags:/opt/airflow/dags
    - ./config:/opt/airflow/config
  depends_on:
    postgres:
      condition: service_healthy

services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U airflow"]
      interval: 10s
      timeout: 5s
      retries: 5

  airflow-init:
    <<: *airflow-common
    entrypoint: /bin/bash
    command:
      - -c
      - |
        set -ex
        airflow db migrate
        airflow connections create-default-connections
        echo "Airflow init complete."
    environment:
      <<: *airflow-common-env
      TASK_NAME: init

  airflow-scheduler:
    <<: *airflow-common
    command: airflow scheduler
    environment:
      <<: *airflow-common-env
      TASK_NAME: scheduler
    depends_on:
      airflow-init:
        condition: service_completed_successfully

  airflow-webserver:
    <<: *airflow-common
    command: airflow api-server
    ports:
      - "8080:8080"
    environment:
      <<: *airflow-common-env
      TASK_NAME: webserver
    depends_on:
      airflow-init:
        condition: service_completed_successfully

  airflow-dag-processor:
    <<: *airflow-common
    command: airflow dag-processor
    environment:
      <<: *airflow-common-env
      TASK_NAME: dag-processor
    depends_on:
      airflow-init:
        condition: service_completed_successfully

  airflow-triggerer:
    <<: *airflow-common
    command: airflow triggerer
    environment:
      <<: *airflow-common-env
      TASK_NAME: triggerer
    depends_on:
      airflow-init:
        condition: service_completed_successfully

File: ./dags/test_triggerer_logging.py

from datetime import datetime, timedelta

from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator
from airflow.providers.standard.sensors.time_delta import TimeDeltaSensor


with DAG(
    dag_id="test_triggerer_logging",
    schedule=None,
    start_date=datetime(2020, 1, 1),
    catchup=False
) as dag:

    before = PythonOperator(
        task_id="before_deferral",
        python_callable=lambda: print("Pre-deferral task log."))

    # Logs produced by this task will be missing in CloudWatch
    test_triggerer = TimeDeltaSensor(
        task_id="test_triggerer",
        delta=timedelta(seconds=120),
        deferrable=True)

    after = PythonOperator(
        task_id="after_deferral",
        python_callable=lambda: print("Post-deferral task log."))

    before >> test_triggerer >> after

File: ./config/airflow_local_settings.py

# Custom settings

File: ./.env

# AWS region and account ID
AWS_REGION=change_me
AWS_ACCOUNT_ID=change_me

# AWS credentials — ECS injects these via task role; Docker needs them explicitly.
# Populate via: aws configure export-credentials --profile change_me --format env
AWS_ACCESS_KEY_ID=change_me
AWS_SECRET_ACCESS_KEY=change_me
AWS_SESSION_TOKEN=change_me
AWS_LOG_GROUP_NAME=change_me

Trigger execution of the test_triggerer_logging DAG. Logs produced by before_deferral and after_deferral tasks, as well as non-deferrable part of test_triggerer tasks will be sent to the configured AWS CloudWatch group. But the trigger logs won't.

Getting into the triggerer container via docker compose exec airflow-triggerer bash command and searching for trigger logs via find /opt/airflow/logs -name "*.trigger.*" -ls command while the deferred task is running one can see that "trigger" logs are being written to the file system, but not pushed to AWS CloudWatch.

What you think should happen instead?

Logs produced by deferrable triggers are being pushed to AWS CloudWatch similar to logs produced by non-deferrable tasks.

Operating System

Docker with apache/airflow:3.3.0-python3.13 image

Deployment

Other

Apache Airflow Provider(s)

No response

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==9.31.0
apache-airflow-providers-standard==1.16.0

Official Helm Chart version

Not Applicable

Kubernetes Version

No response

Helm Chart configuration

No response

Docker Image customizations

See the supplied Dockerfile and docker-compose.yml in the "how to reproduce it" section.

Anything else?

I was able to find a workaround that works via monkey-patching. Even though it seems to work, it is pretty fragile as it can break at any point if Airflow decides to change something in future versions.

Put the following code into the ./config/airflow_local_settings.py file and re-start Docker Compose:

import os


def _patch_dictconfig_for_watchtower():
    import logging
    import logging.config
    import watchtower

    _original_dictConfig = logging.config.dictConfig

    def _dictConfig_hide_watchtower(config):
        saved = []
        remaining = []
        for ref in logging._handlerList:
            h = ref()
            if h is not None and isinstance(h, watchtower.CloudWatchLogHandler):
                saved.append(ref)
            else:
                remaining.append(ref)

        logging._handlerList[:] = remaining
        try:
            _original_dictConfig(config)
        finally:
            logging._handlerList.extend(saved)

    logging.config.dictConfig = _dictConfig_hide_watchtower


if os.environ.get("TASK_NAME") == "triggerer":
    _patch_dictconfig_for_watchtower()

The idea of this patch is that it temporarily excludes the CloudWatchLogHandler from the list of logging handlers before logging.config.dictConfig function is called, which prevents it from destroying the handler.

When triggering the test DAG with this path trigger logs are getting pushed to AWS CloudWatch as expected.

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

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[jd878-gif]

I'd like to work on this — is it still open? I'll start by reproducing it locally.

[maxprychynenko-buildzoom]

@jd878-gif it looks like it is open, as far as I can tell. Thank you for considering working on it. Although, I'm not really sure whether somebody is already working on it or not.

I've checked open PRs and it looks like this issue may be somewhat similar to these ones:

• Fix remote processor injection happening before dictConfig runs in configure_logging #66633
• Fix CloudWatch remote logging for ephemeral lifecycle executor #68779

At least, based on function names mentioned in the descriptions. Maybe, it is the same problem, but observed in a different use case.

@jason810496 @korex-f @ferruzzi @o-nikolas @seanghaeli it looks like you've been working on something similar recently. Can you take a look whether it is the same issue or something different, please?

[jd878-gif]

@jd878-gif it looks like it is open, as far as I can tell. Thank you for considering working on it. Although, I'm not really sure whether somebody is already working on it or not.

I've checked open PRs and it looks like this issue may be somewhat similar to these ones:

• Fix remote processor injection happening before dictConfig runs in configure_logging #66633
• Fix CloudWatch remote logging for ephemeral lifecycle executor #68779

At least, based on function names mentioned in the descriptions. Maybe, it is the same problem, but observed in a different use case.

@jason810496 @korex-f @ferruzzi @o-nikolas @seanghaeli it looks like you've been working on something similar recently. Can you take a look whether it is the same issue or something different, please?

Thanks for the detailed root-cause writeup. I traced this through the codebase and it matches exactly: configure_logging() in the Task SDK (task-sdk/src/airflow/sdk/log.py) accesses .processors on the remote log handler — which lazily builds the watchtower.CloudWatchLogHandler — before calling the shared configure_logging(), which runs logging.config.dictConfig() and closes every handler in logging._handlerList, including that one.

This was already fixed in #68779 ("Fix CloudWatch remote logging for ephemeral lifecycle executor"), merged into main and released in apache-airflow-providers-amazon==9.32.0. The fix makes CloudWatchRemoteLogIO.handler self-heal by rebuilding the watchtower handler if it detects shutting_down=True, rather than trying to prevent dictConfig from touching it. There's a regression test covering this exact scenario: test_log_message_after_handler_closed_by_dictconfig.

I confirmed this directly using your own repro DAG and docker-compose setup, bumping only the provider pin to apache-airflow-providers-amazon==9.32.0. The trigger's CloudWatch log stream (.../task_id=test_triggerer/attempt=1.log.trigger.1.log) now contains the full trigger lifecycle that was previously missing entirely:

trigger ... starting
trigger starting
109 seconds remaining; sleeping 10 seconds
...
yielding event with payload DateTime(...)
Trigger fired event
trigger completed

[seanghaeli]

As you noted @jd878-gif , #68779 ships the fix for this, so I think the issue can be closed.

[jd878-gif]

As you noted @jd878-gif , #68779 ships the fix for this, so I think the issue can be closed.

Sounds good, thanks for confirming!

[maxprychynenko-buildzoom]

I've just re-tested it and also confirm that trigger logs are being pushed to AWS CloudWatch as expected with both apache-airflow-providers-amazon version 9.32.0 and 9.33.0.

It is kind of funny that I started digging into this problem the same day 9.32.0 got released :-)

Thank you, @jason810496 and everybody who worked on the fix.

[jason810496]

Good new. Nice to heard that!