Fix: Restore live stdout logging for Elasticsearch in Airflow 3

[Subham-KRLX]

This PR restores the ElasticsearchTaskHandler in DEFAULT_LOGGING_CONFIG within airflow_local_settings.py, resolving the regression where task logs were no longer visible in stdout after upgrading to Airflow 3.While ElasticsearchRemoteLogIO
was previously added for remote retrieval, the handler itself was omitted from the configuration block. This caused Airflow to fall back to the generic FileTaskHandler, which does not support the write_stdout configuration. Restoring the handler ensures that logs are correctly sent to stdout(for Fluentd/container log shippers) when configured.

This aligns the Elasticsearch provider's logging configuration with the patterns used by Opensearch and Stackdriver.

closes: #63960
closes: #49863
closes: #54501

Was generative AI tooling used to co-author this PR?
Yes — Gemini (Code Research and PR Description)

[eladkal]

Just to clarify - this PR solves #54501 and suppress #63960 ?

[eladkal]

cc @Owen-CH-Leung for review as the expert :)

[Subham-KRLX]

Confirming that this PR correctly resolves the root cause for #49863, #54501, and #63960. By restoring the ElasticsearchTaskHandler in airflow_local_settings.py, we ensure that write_stdout is once again respected across all executors.

[eladkal]

@Subham-KRLX can you share screenshots from elastic confirming this works?

[Subham-KRLX]

@Subham-KRLX can you share screenshots from elastic confirming this works?

@eladkal I am currently in a remote environment and can't provide a live Kibana dashboard screenshot however can I share the actual JSON stdout logs and the passing unit test results from a local run? this will confirm that the ElasticsearchTaskHandler is correctly restored in the configuration and is emitting logs to stdout as intended.

[Subham-KRLX]

@Subham-KRLX can you share screenshots from elastic confirming this works?

@eladkal I am currently in a remote environment and can't provide a live Kibana dashboard screenshot however can I share the actual JSON stdout logs and the passing unit test results from a local run? this will confirm that the ElasticsearchTaskHandler is correctly restored in the configuration and is emitting logs to stdout as intended.

{
"asctime": "2026-03-26T13:16:19.650+0530",
"levelname": "INFO",
"message": "end_of_log",
"offset": 1774511179650571008,
"dag_id": "dag_for_testing_es_task_handler",
"task_id": "task_for_testing_es_log_handler",
"logical_date": "2016_01_01T00_00_00_000000",
"try_number": "1",
"log_id": "dag_for_testing_es_task_handler-task_for_testing_es_log_handler-run_for_testing_es_log_handler--1-1"
}

The test suite also confirms the core fix with 60 passing tests:

providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py [100%]
======= 60 passed in 5.90s =======

This ensures that ElasticsearchTaskHandler is once again respected in the logging configuration template resolving the regression where stdout logs were no longer visible.

[Owen-CH-Leung]

LGTM. Thanks!

[taranlu-houzz]

Hi @Subham-KRLX, I just wanted to double check that this PR will fully address the issue of logs not showing up in container stdout when using LocalExecutor, which is what I was trying to fix via: #63960.

This is a Claude summary of what was tried locally:

  Attempted to apply this fix (#64067) locally via airflowLocalSettings (Helm ConfigMap for /opt/airflow/config/airflow_local_settings.py) on Airflow 3.1.6 + LocalExecutor.

  Two issues encountered:

  1. The PR branch imports ElasticsearchRemoteLogIO which doesn't exist in apache-airflow-providers-elasticsearch==6.4.2 (bundled with 3.1.6) — immediate ImportError at
  startup.
  2. After working around that by using the 3.1.6-native equivalent (setting remote_logging=True, elasticsearch.host, write_stdout=True, write_to_es=False), Airflow
  started and tasks ran successfully — but task logs still did not appear in container stdout.

  Root cause: write_stdout=True makes the task subprocess write to its stdout, but with LocalExecutor the supervisor connects subprocess stdout to an internal pipe and
  doesn't forward it to the container's stdout unless subprocess_logs_to_stdout=True is passed to supervise(). The containerized executor sets this flag; LocalExecutor
  does not. This fix alone is not sufficient for LocalExecutor users who need task logs on stdout.

  Workaround that does work: patching supervise() directly via airflowLocalSettings:

  import functools
  import airflow.sdk.execution_time.supervisor as _sup

  _orig_supervise = _sup.supervise

  @functools.wraps(_orig_supervise)
  def _patched_supervise(*args, **kwargs):
      kwargs.setdefault("subprocess_logs_to_stdout", True)
      return _orig_supervise(*args, **kwargs)

  _sup.supervise = _patched_supervise

  The proper fix would be for LocalExecutor._execute_work to pass subprocess_logs_to_stdout=True to supervise(), or for a config option to be exposed for it.

[Subham-KRLX]

@taranlu-houzz You are absolutely right about the LocalExecutor log forwarding I am updating this PR to pass subprocess_logs_to_stdout=True to the supervisor in LocalExecutor which will ensure task logs are correctly forwarded to the container's stdout in Airflow 3.

[github-actions]

Backport successfully created: v3-2-test

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
✅	v3-2-test