Fix missing traceback in Elasticsearch and OpenSearch task logs

closes: #63736

Author: Subham-KRLX

airflow-core/newsfragments/63739.bugfix.rst

@@ -0,0 +1 @@
+Fix missing traceback in Elasticsearch and OpenSearch task logs

providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py

@@ -79,7 +79,46 @@
 # not exist, the task handler should use the log_id_template attribute instead.
 USE_PER_RUN_LOG_ID = hasattr(DagRun, "get_log_template")
 
-TASK_LOG_FIELDS = ["timestamp", "event", "level", "chan", "logger"]
+TASK_LOG_FIELDS = ["timestamp", "event", "level", "chan", "logger", "error_detail"]
+
+
+def _format_error_detail(error_detail: Any) -> str | None:
+    """Render the structured ``error_detail`` written by the Airflow 3 supervisor as a traceback string."""
+    if not error_detail:
+        return None
+    if not isinstance(error_detail, list):
+        return str(error_detail)
+
+    lines: list[str] = ["Traceback (most recent call last):"]
+    for exc_info in error_detail:
+        if not isinstance(exc_info, dict):
+            lines.append(str(exc_info))
+            continue
+        if exc_info.get("is_cause"):
+            lines.append("\nThe above exception was the direct cause of the following exception:\n")
+            lines.append("Traceback (most recent call last):")
+        for frame in exc_info.get("frames", []):
+            lines.append(
+                f'  File "{frame.get("filename", "<unknown>")}", line {frame.get("lineno", "?")}, in {frame.get("name", "<unknown>")}'
+            )
+        exc_type = exc_info.get("exc_type", "")
+        exc_value = exc_info.get("exc_value", "")
+        if exc_type:
+            lines.append(f"{exc_type}: {exc_value}" if exc_value else exc_type)
+    return "\n".join(lines)
+
+
+def _build_log_fields(hit_dict: dict[str, Any]) -> dict[str, Any]:
+    """Filter an ES hit to ``TASK_LOG_FIELDS``, formatting ``error_detail`` as a plain-text traceback."""
+    fields = {k: v for k, v in hit_dict.items() if k.lower() in TASK_LOG_FIELDS}
+    if "error_detail" in fields:
+        formatted = _format_error_detail(fields["error_detail"])
+        if formatted:
+            fields["error_detail"] = formatted
+        else:
+            fields.pop("error_detail")
+    return fields
+
 
 VALID_ES_CONFIG_KEYS = set(inspect.signature(elasticsearch.Elasticsearch.__init__).parameters.keys())
 # Remove `self` from the valid set of kwargs
@@ -356,9 +395,7 @@ def _read(
 
                 # Flatten all hits, filter to only desired fields, and construct StructuredLogMessage objects
                 message = header + [
-                    StructuredLogMessage(
-                        **{k: v for k, v in hit.to_dict().items() if k.lower() in TASK_LOG_FIELDS}
-                    )
+                    StructuredLogMessage(**_build_log_fields(hit.to_dict()))
                     for hits in logs_by_host.values()
                     for hit in hits
                 ]
@@ -668,7 +705,7 @@ def read(self, _relative_path: str, ti: RuntimeTI) -> tuple[LogSourceInfo, LogMe
         # Structured log messages
         for hits in logs_by_host.values():
             for hit in hits:
-                filtered = {k: v for k, v in hit.to_dict().items() if k.lower() in TASK_LOG_FIELDS}
+                filtered = _build_log_fields(hit.to_dict())
                 message.append(json.dumps(filtered))
 
         return header, message

providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py

@@ -1060,3 +1060,190 @@ def test_read_with_missing_log(self, mocked_count, ti):
         assert log_source_info == []
         assert f"*** Log {log_id} not found in Elasticsearch" in log_messages[0]
         mocked_count.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# Tests for the error_detail helpers (issue #63736)
+# ---------------------------------------------------------------------------
+
+
+class TestFormatErrorDetail:
+    """Unit tests for _format_error_detail."""
+
+    def test_returns_none_for_empty(self):
+        from airflow.providers.elasticsearch.log.es_task_handler import _format_error_detail
+
+        assert _format_error_detail(None) is None
+        assert _format_error_detail([]) is None
+
+    def test_returns_string_for_non_list(self):
+        from airflow.providers.elasticsearch.log.es_task_handler import _format_error_detail
+
+        assert _format_error_detail("raw string") == "raw string"
+
+    def test_formats_single_exception(self):
+        from airflow.providers.elasticsearch.log.es_task_handler import _format_error_detail
+
+        error_detail = [
+            {
+                "is_cause": False,
+                "frames": [
+                    {"filename": "/app/task.py", "lineno": 13, "name": "log_and_raise"},
+                ],
+                "exc_type": "RuntimeError",
+                "exc_value": "Something went wrong.",
+                "exceptions": [],
+                "is_group": False,
+            }
+        ]
+        result = _format_error_detail(error_detail)
+        assert result is not None
+        assert "Traceback (most recent call last):" in result
+        assert 'File "/app/task.py", line 13, in log_and_raise' in result
+        assert "RuntimeError: Something went wrong." in result
+
+    def test_formats_chained_exceptions(self):
+        from airflow.providers.elasticsearch.log.es_task_handler import _format_error_detail
+
+        error_detail = [
+            {
+                "is_cause": True,
+                "frames": [{"filename": "/a.py", "lineno": 1, "name": "foo"}],
+                "exc_type": "ValueError",
+                "exc_value": "original",
+                "exceptions": [],
+            },
+            {
+                "is_cause": False,
+                "frames": [{"filename": "/b.py", "lineno": 2, "name": "bar"}],
+                "exc_type": "RuntimeError",
+                "exc_value": "wrapped",
+                "exceptions": [],
+            },
+        ]
+        result = _format_error_detail(error_detail)
+        assert result is not None
+        assert "direct cause" in result
+        assert "ValueError: original" in result
+        assert "RuntimeError: wrapped" in result
+
+    def test_exc_type_without_value(self):
+        from airflow.providers.elasticsearch.log.es_task_handler import _format_error_detail
+
+        error_detail = [
+            {
+                "is_cause": False,
+                "frames": [],
+                "exc_type": "StopIteration",
+                "exc_value": "",
+            }
+        ]
+        result = _format_error_detail(error_detail)
+        assert result is not None
+        assert result.endswith("StopIteration")
+
+    def test_non_dict_items_are_stringified(self):
+        from airflow.providers.elasticsearch.log.es_task_handler import _format_error_detail
+
+        result = _format_error_detail(["unexpected string item"])
+        assert result is not None
+        assert "unexpected string item" in result
+
+
+class TestBuildStructuredLogFields:
+    """Unit tests for _build_log_fields."""
+
+    def test_filters_to_allowed_fields(self):
+        from airflow.providers.elasticsearch.log.es_task_handler import _build_log_fields
+
+        hit = {"event": "hello", "level": "info", "unknown_field": "should be dropped"}
+        result = _build_log_fields(hit)
+        assert "event" in result
+        assert "level" in result
+        assert "unknown_field" not in result
+
+    def test_error_detail_formatted_as_string(self):
+        from airflow.providers.elasticsearch.log.es_task_handler import _build_log_fields
+
+        hit = {
+            "event": "Task failed with exception",
+            "error_detail": [
+                {
+                    "is_cause": False,
+                    "frames": [{"filename": "/dag.py", "lineno": 10, "name": "run"}],
+                    "exc_type": "RuntimeError",
+                    "exc_value": "Woopsie.",
+                }
+            ],
+        }
+        result = _build_log_fields(hit)
+        assert isinstance(result["error_detail"], str)
+        assert "RuntimeError: Woopsie." in result["error_detail"]
+        assert 'File "/dag.py", line 10, in run' in result["error_detail"]
+
+    def test_error_detail_dropped_when_empty(self):
+        from airflow.providers.elasticsearch.log.es_task_handler import _build_log_fields
+
+        hit = {"event": "msg", "error_detail": []}
+        result = _build_log_fields(hit)
+        assert "error_detail" not in result
+
+    @pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="StructuredLogMessage only exists in Airflow 3+")
+    @elasticmock
+    def test_read_includes_error_detail_in_structured_message(self):
+        """End-to-end: a hit with error_detail should surface it in the returned StructuredLogMessage."""
+        from airflow.providers.elasticsearch.log.es_task_handler import ElasticsearchTaskHandler
+
+        local_log_location = "local/log/location"
+        handler = ElasticsearchTaskHandler(
+            base_log_folder=local_log_location,
+            end_of_log_mark="end_of_log\n",
+            write_stdout=False,
+            json_format=False,
+            json_fields="asctime,filename,lineno,levelname,message,exc_text",
+        )
+
+        es = elasticsearch.Elasticsearch("http://localhost:9200")
+        log_id = "test_dag-test_task-test_run--1-1"
+        body = {
+            "event": "Task failed with exception",
+            "log_id": log_id,
+            "offset": 1,
+            "error_detail": [
+                {
+                    "is_cause": False,
+                    "frames": [
+                        {"filename": "/opt/airflow/dags/fail.py", "lineno": 13, "name": "log_and_raise"}
+                    ],
+                    "exc_type": "RuntimeError",
+                    "exc_value": "Woopsie. Something went wrong.",
+                }
+            ],
+        }
+        es.index(index="test_index", doc_type="log", body=body, id=1)
+
+        # Patch the IO layer to return our fake document
+        mock_hit_dict = body.copy()
+
+        from airflow.providers.elasticsearch.log.es_response import ElasticSearchResponse, Hit
+
+        mock_hit = Hit({"_source": mock_hit_dict})
+        mock_response = mock.MagicMock(spec=ElasticSearchResponse)
+        mock_response.hits = [mock_hit]
+        mock_response.__iter__ = mock.Mock(return_value=iter([mock_hit]))
+        mock_response.__bool__ = mock.Mock(return_value=True)
+        mock_response.__getitem__ = mock.Mock(return_value=mock_hit)
+
+        with mock.patch.object(handler.io, "_es_read", return_value=mock_response):
+            with mock.patch.object(handler.io, "_group_logs_by_host", return_value={"localhost": [mock_hit]}):
+                # Build StructuredLogMessages
+                from airflow.providers.elasticsearch.log.es_task_handler import _build_log_fields
+                from airflow.utils.log.file_task_handler import StructuredLogMessage
+
+                fields = _build_log_fields(mock_hit.to_dict())
+                msg = StructuredLogMessage(**fields)
+
+                assert msg.event == "Task failed with exception"
+                assert hasattr(msg, "error_detail")
+                assert "RuntimeError: Woopsie. Something went wrong." in msg.error_detail
+                assert 'File "/opt/airflow/dags/fail.py", line 13, in log_and_raise' in msg.error_detail

providers/opensearch/src/airflow/providers/opensearch/log/os_task_handler.py

@@ -58,7 +58,45 @@
 
 USE_PER_RUN_LOG_ID = hasattr(DagRun, "get_log_template")
 LOG_LINE_DEFAULTS = {"exc_text": "", "stack_info": ""}
-TASK_LOG_FIELDS = ["timestamp", "event", "level", "chan", "logger"]
+TASK_LOG_FIELDS = ["timestamp", "event", "level", "chan", "logger", "error_detail"]
+
+
+def _format_error_detail(error_detail: Any) -> str | None:
+    """Render the structured ``error_detail`` written by the Airflow 3 supervisor as a traceback string."""
+    if not error_detail:
+        return None
+    if not isinstance(error_detail, list):
+        return str(error_detail)
+
+    lines: list[str] = ["Traceback (most recent call last):"]
+    for exc_info in error_detail:
+        if not isinstance(exc_info, dict):
+            lines.append(str(exc_info))
+            continue
+        if exc_info.get("is_cause"):
+            lines.append("\nThe above exception was the direct cause of the following exception:\n")
+            lines.append("Traceback (most recent call last):")
+        for frame in exc_info.get("frames", []):
+            lines.append(
+                f'  File "{frame.get("filename", "<unknown>")}", line {frame.get("lineno", "?")}, in {frame.get("name", "<unknown>")}'
+            )
+        exc_type = exc_info.get("exc_type", "")
+        exc_value = exc_info.get("exc_value", "")
+        if exc_type:
+            lines.append(f"{exc_type}: {exc_value}" if exc_value else exc_type)
+    return "\n".join(lines)
+
+
+def _build_log_fields(hit_dict: dict[str, Any]) -> dict[str, Any]:
+    """Filter an OpenSearch hit to ``TASK_LOG_FIELDS``, formatting ``error_detail`` as a plain-text traceback."""
+    fields = {k: v for k, v in hit_dict.items() if k.lower() in TASK_LOG_FIELDS}
+    if "error_detail" in fields:
+        formatted = _format_error_detail(fields["error_detail"])
+        if formatted:
+            fields["error_detail"] = formatted
+        else:
+            fields.pop("error_detail")
+    return fields
 
 
 def getattr_nested(obj, item, default):
@@ -416,9 +454,7 @@ def concat_logs(hits: list[Hit]):
 
                 # Flatten all hits, filter to only desired fields, and construct StructuredLogMessage objects
                 message = header + [
-                    StructuredLogMessage(
-                        **{k: v for k, v in hit.to_dict().items() if k.lower() in TASK_LOG_FIELDS}
-                    )
+                    StructuredLogMessage(**_build_log_fields(hit.to_dict()))
                     for hits in logs_by_host.values()
                     for hit in hits
                 ]