[v3-2-test] Fix: Restore live stdout logging for Elasticsearch log forwarding (#64067)

airflow-core/newsfragments/64067.bugfix.rst

@@ -0,0 +1 @@
+Restore live stdout logging for Elasticsearch in Airflow 3 by correctly configuring the handler in ``airflow_local_settings.py`` and forwarding task logs to stdout in ``LocalExecutor``.

airflow-core/src/airflow/config_templates/airflow_local_settings.py

@@ -291,6 +291,27 @@ def _default_conn_name_from(mod_path, hook_name):
         ELASTICSEARCH_HOST_FIELD: str = conf.get_mandatory_value("elasticsearch", "HOST_FIELD")
         ELASTICSEARCH_OFFSET_FIELD: str = conf.get_mandatory_value("elasticsearch", "OFFSET_FIELD")
         ELASTICSEARCH_LOG_ID_TEMPLATE: str = conf.get_mandatory_value("elasticsearch", "LOG_ID_TEMPLATE")
+        ELASTICSEARCH_END_OF_LOG_MARK: str = conf.get_mandatory_value("elasticsearch", "END_OF_LOG_MARK")
+        ELASTICSEARCH_FRONTEND: str = conf.get_mandatory_value("elasticsearch", "FRONTEND")
+        ELASTICSEARCH_JSON_FIELDS: str = conf.get_mandatory_value("elasticsearch", "JSON_FIELDS")
+
+        ELASTICSEARCH_REMOTE_HANDLERS: dict[str, dict[str, str | bool | None]] = {
+            "task": {
+                "class": "airflow.providers.elasticsearch.log.es_task_handler.ElasticsearchTaskHandler",
+                "formatter": "airflow",
+                "base_log_folder": BASE_LOG_FOLDER,
+                "end_of_log_mark": ELASTICSEARCH_END_OF_LOG_MARK,
+                "host": ELASTICSEARCH_HOST,
+                "frontend": ELASTICSEARCH_FRONTEND,
+                "write_stdout": ELASTICSEARCH_WRITE_STDOUT,
+                "write_to_es": ELASTICSEARCH_WRITE_TO_ES,
+                "json_format": ELASTICSEARCH_JSON_FORMAT,
+                "json_fields": ELASTICSEARCH_JSON_FIELDS,
+                "host_field": ELASTICSEARCH_HOST_FIELD,
+                "offset_field": ELASTICSEARCH_OFFSET_FIELD,
+            },
+        }
+        DEFAULT_LOGGING_CONFIG["handlers"].update(ELASTICSEARCH_REMOTE_HANDLERS)
 
         REMOTE_TASK_LOG = ElasticsearchRemoteLogIO(
             host=ELASTICSEARCH_HOST,

airflow-core/src/airflow/executors/local_executor.py

@@ -149,6 +149,7 @@ def _execute_work(log: Logger, workload: workloads.ExecuteTask, team_conf) -> No
         token=workload.token,
         server=team_conf.get("core", "execution_api_server_url", fallback=default_execution_api_server),
         log_path=workload.log_path,
+        subprocess_logs_to_stdout=True,
     )
 
 

airflow-core/tests/unit/executors/test_local_executor.py

@@ -268,7 +268,7 @@ def test_execution_api_server_url_config(self, mock_supervise, conf_values, expe
 
         with conf_vars(conf_values):
             team_conf = ExecutorConf(team_name=None)
-            _execute_work(log=mock.ANY, workload=mock.MagicMock(), team_conf=team_conf)
+            _execute_work(log=mock.MagicMock(), workload=mock.MagicMock(), team_conf=team_conf)
 
             mock_supervise.assert_called_with(
                 ti=mock.ANY,
@@ -277,6 +277,7 @@ def test_execution_api_server_url_config(self, mock_supervise, conf_values, expe
                 token=mock.ANY,
                 server=expected_server,
                 log_path=mock.ANY,
+                subprocess_logs_to_stdout=True,
             )
 
     @mock.patch("airflow.sdk.execution_time.supervisor.supervise")
@@ -303,7 +304,7 @@ def test_team_and_global_config_isolation(self, mock_supervise):
             with conf_vars(config_overrides):
                 # Test team-specific config
                 team_conf = ExecutorConf(team_name=team_name)
-                _execute_work(log=mock.ANY, workload=mock.MagicMock(), team_conf=team_conf)
+                _execute_work(log=mock.MagicMock(), workload=mock.MagicMock(), team_conf=team_conf)
 
                 # Verify team-specific server URL was used
                 assert mock_supervise.call_count == 1
@@ -314,7 +315,7 @@ def test_team_and_global_config_isolation(self, mock_supervise):
 
                 # Test global config (no team)
                 global_conf = ExecutorConf(team_name=None)
-                _execute_work(log=mock.ANY, workload=mock.MagicMock(), team_conf=global_conf)
+                _execute_work(log=mock.MagicMock(), workload=mock.MagicMock(), team_conf=global_conf)
 
                 # Verify default server URL was used
                 assert mock_supervise.call_count == 1