This is an automated email from the ASF dual-hosted git repository.
josh-fell pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new ac263314c1f Surface dbt Cloud failure details in Airflow task logs
(#70171)
ac263314c1f is described below
commit ac263314c1f766117a7dbfbef9766868e83380fe
Author: Senior Data Engineer <[email protected]>
AuthorDate: Mon Sep 14 07:36:34 2026 -0600
Surface dbt Cloud failure details in Airflow task logs (#70171)
* Surface dbt Cloud failure details in Airflow task logs
Log run status, status messages, and failed run steps when a dbt Cloud
job run reaches ERROR or CANCELLED in the hook, operator, and sensor.
Closes #46923
* fix(dbt-cloud): repair CI after failure-detail logging
- Mock log_job_run_failure_details in deferrable operator failure test
- Expect extra get_job_run call when logging ERROR/CANCELLED runs
- Fix sensor test account_id assertion; add newsfragment; ruff format
* fix(dbt-cloud): rename newsfragment to match PR number 70171
* Remove newsfragment per maintainer review
Simple provider improvements do not require newsfragments in this repo.
* fix(dbt-cloud): drop redundant fallback on failure logging helper
log_job_run_failure_details only needs account_id when it calls get_job_run,
which already applies @fallback_to_default_account.
* test(dbt-cloud): expect raw account_id in failure logging helper
get_job_run applies @fallback_to_default_account;
log_job_run_failure_details passes account_id through unchanged.
---
.../src/airflow/providers/dbt/cloud/hooks/dbt.py | 56 +++++++++++++++
.../airflow/providers/dbt/cloud/operators/dbt.py | 3 +
.../src/airflow/providers/dbt/cloud/sensors/dbt.py | 2 +
.../cloud/tests/unit/dbt/cloud/hooks/test_dbt.py | 81 ++++++++++++++++++++++
.../tests/unit/dbt/cloud/operators/test_dbt.py | 11 ++-
.../cloud/tests/unit/dbt/cloud/sensors/test_dbt.py | 7 +-
6 files changed, 157 insertions(+), 3 deletions(-)
diff --git a/providers/dbt/cloud/src/airflow/providers/dbt/cloud/hooks/dbt.py
b/providers/dbt/cloud/src/airflow/providers/dbt/cloud/hooks/dbt.py
index 719b5181bfe..40129cf41b4 100644
--- a/providers/dbt/cloud/src/airflow/providers/dbt/cloud/hooks/dbt.py
+++ b/providers/dbt/cloud/src/airflow/providers/dbt/cloud/hooks/dbt.py
@@ -825,6 +825,61 @@ class DbtCloudHook(HttpHook):
return job_run_status
+ @staticmethod
+ def _format_run_step_failure(step: dict[str, Any]) -> str:
+ details = [f"step {step.get('index')}: {step.get('name',
'<unknown>')}"]
+ if status_humanized := step.get("status_humanized"):
+ details.append(f"status={status_humanized}")
+ elif status := step.get("status"):
+ details.append(f"status_code={status}")
+ for field in ("status_message", "log", "logs", "debug_logs"):
+ if value := step.get(field):
+ details.append(f"{field}={value}")
+ return " | ".join(details)
+
+ def log_job_run_failure_details(
+ self,
+ run_id: int,
+ account_id: int | None = None,
+ job_run: dict[str, Any] | None = None,
+ ) -> None:
+ """
+ Log dbt Cloud run failure context in Airflow task logs.
+
+ Fetches run metadata (including run steps) when ``job_run`` is not
provided.
+ """
+ if job_run is None:
+ job_run = self.get_job_run(
+ run_id=run_id,
+ account_id=account_id,
+ include_related=["run_steps"],
+ ).json()["data"]
+
+ run_status = job_run.get("status")
+ if run_status is not None:
+ try:
+ status_name = DbtCloudJobRunStatus(run_status).name
+ except ValueError:
+ status_name = str(run_status)
+ self.log.error("dbt Cloud job run %s ended with status %s.",
run_id, status_name)
+
+ for field in ("status_message", "status_humanized", "status_msg"):
+ if message := job_run.get(field):
+ self.log.error("dbt Cloud job run %s: %s", run_id, message)
+ break
+
+ run_steps = job_run.get("run_steps") or []
+ error_steps = [step for step in run_steps if step.get("status") ==
DbtCloudJobRunStatus.ERROR.value]
+
+ if error_steps:
+ for step in error_steps:
+ self.log.error("dbt Cloud failed step — %s",
self._format_run_step_failure(step))
+ elif run_steps:
+ self.log.error(
+ "dbt Cloud last run step — %s",
+ self._format_run_step_failure(run_steps[-1]),
+ )
+
def wait_for_job_run_status(
self,
run_id: int,
@@ -863,6 +918,7 @@ class DbtCloudHook(HttpHook):
# Reached terminal failure before expected state.
if DbtCloudJobRunStatus.is_terminal(job_run_status):
+ self.log_job_run_failure_details(run_id=run_id,
account_id=account_id)
raise DbtCloudJobRunException(
f"Job run {run_id} reached terminal status "
f"{DbtCloudJobRunStatus(job_run_status).name} "
diff --git
a/providers/dbt/cloud/src/airflow/providers/dbt/cloud/operators/dbt.py
b/providers/dbt/cloud/src/airflow/providers/dbt/cloud/operators/dbt.py
index b38d9a0f3cd..23948af8702 100644
--- a/providers/dbt/cloud/src/airflow/providers/dbt/cloud/operators/dbt.py
+++ b/providers/dbt/cloud/src/airflow/providers/dbt/cloud/operators/dbt.py
@@ -209,6 +209,7 @@ class DbtCloudRunJobOperator(BaseOperator):
DbtCloudJobRunStatus.CANCELLED.value,
DbtCloudJobRunStatus.ERROR.value,
):
+ self.hook.log_job_run_failure_details(run_id=self.run_id,
account_id=self.account_id)
raise DbtCloudJobRunException(f"Job run {self.run_id} has failed
or has been cancelled.")
return None
@@ -287,8 +288,10 @@ class DbtCloudRunJobOperator(BaseOperator):
"""Execute when the trigger fires - returns immediately."""
self.run_id = event["run_id"]
if event["status"] == "cancelled":
+ self.hook.log_job_run_failure_details(run_id=int(self.run_id),
account_id=self.account_id)
raise DbtCloudJobRunException(f"Job run {self.run_id} has been
cancelled.")
if event["status"] == "error":
+ self.hook.log_job_run_failure_details(run_id=int(self.run_id),
account_id=self.account_id)
raise DbtCloudJobRunException(f"Job run {self.run_id} has failed.")
# Enforce execution_timeout semantics in deferrable mode by cancelling
the job.
diff --git a/providers/dbt/cloud/src/airflow/providers/dbt/cloud/sensors/dbt.py
b/providers/dbt/cloud/src/airflow/providers/dbt/cloud/sensors/dbt.py
index 1ef13ab61f7..2b40dd8724f 100644
--- a/providers/dbt/cloud/src/airflow/providers/dbt/cloud/sensors/dbt.py
+++ b/providers/dbt/cloud/src/airflow/providers/dbt/cloud/sensors/dbt.py
@@ -79,10 +79,12 @@ class DbtCloudJobRunSensor(BaseSensorOperator):
job_run_status = self.hook.get_job_run_status(run_id=self.run_id,
account_id=self.account_id)
if job_run_status == DbtCloudJobRunStatus.ERROR.value:
+ self.hook.log_job_run_failure_details(run_id=self.run_id,
account_id=self.account_id)
message = f"Job run {self.run_id} has failed."
raise DbtCloudJobRunException(message)
if job_run_status == DbtCloudJobRunStatus.CANCELLED.value:
+ self.hook.log_job_run_failure_details(run_id=self.run_id,
account_id=self.account_id)
message = f"Job run {self.run_id} has been cancelled."
raise DbtCloudJobRunException(message)
diff --git a/providers/dbt/cloud/tests/unit/dbt/cloud/hooks/test_dbt.py
b/providers/dbt/cloud/tests/unit/dbt/cloud/hooks/test_dbt.py
index 951e8505496..1c573d8a36a 100644
--- a/providers/dbt/cloud/tests/unit/dbt/cloud/hooks/test_dbt.py
+++ b/providers/dbt/cloud/tests/unit/dbt/cloud/hooks/test_dbt.py
@@ -1080,6 +1080,81 @@ class TestDbtCloudHook:
)
hook._paginate.assert_not_called()
+ def test_format_run_step_failure(self):
+ step = {
+ "index": 2,
+ "name": "Run dbt",
+ "status": DbtCloudJobRunStatus.ERROR.value,
+ "status_humanized": "Error",
+ "status_message": "dbt run failed",
+ }
+ formatted = DbtCloudHook._format_run_step_failure(step)
+ assert "step 2: Run dbt" in formatted
+ assert "status=Error" in formatted
+ assert "status_message=dbt run failed" in formatted
+
+ @pytest.mark.parametrize(
+ argnames=("conn_id", "account_id"),
+ argvalues=[(ACCOUNT_ID_CONN, None), (NO_ACCOUNT_ID_CONN, ACCOUNT_ID)],
+ ids=["default_account", "explicit_account"],
+ )
+ def test_log_job_run_failure_details_with_error_steps(self, conn_id,
account_id):
+ hook = DbtCloudHook(conn_id)
+ job_run = {
+ "status": DbtCloudJobRunStatus.ERROR.value,
+ "status_message": "Run failed",
+ "run_steps": [
+ {"index": 1, "name": "Clone git repo", "status": 10,
"status_humanized": "Success"},
+ {
+ "index": 2,
+ "name": "Run dbt",
+ "status": DbtCloudJobRunStatus.ERROR.value,
+ "status_humanized": "Error",
+ "status_message": "Compilation Error",
+ },
+ ],
+ }
+
+ with patch.object(hook.log, "error") as mock_log_error:
+ hook.log_job_run_failure_details(run_id=RUN_ID,
account_id=account_id, job_run=job_run)
+
+ mock_log_error.assert_any_call("dbt Cloud job run %s ended with status
%s.", RUN_ID, "ERROR")
+ mock_log_error.assert_any_call("dbt Cloud job run %s: %s", RUN_ID,
"Run failed")
+ mock_log_error.assert_any_call(
+ "dbt Cloud failed step — %s",
+ DbtCloudHook._format_run_step_failure(job_run["run_steps"][1]),
+ )
+
+ @pytest.mark.parametrize(
+ argnames=("conn_id", "account_id"),
+ argvalues=[(ACCOUNT_ID_CONN, None), (NO_ACCOUNT_ID_CONN, ACCOUNT_ID)],
+ ids=["default_account", "explicit_account"],
+ )
+ @patch.object(DbtCloudHook, "get_job_run")
+ def test_log_job_run_failure_details_fetches_run_steps(self,
mock_get_job_run, conn_id, account_id):
+ hook = DbtCloudHook(conn_id)
+ job_run = {
+ "status": DbtCloudJobRunStatus.ERROR.value,
+ "run_steps": [
+ {
+ "index": 1,
+ "name": "Run dbt",
+ "status": DbtCloudJobRunStatus.ERROR.value,
+ "status_humanized": "Error",
+ }
+ ],
+ }
+ mock_get_job_run.return_value.json.return_value = {"data": job_run}
+
+ with patch.object(hook.log, "error"):
+ hook.log_job_run_failure_details(run_id=RUN_ID,
account_id=account_id)
+
+ mock_get_job_run.assert_called_once_with(
+ run_id=RUN_ID,
+ account_id=account_id,
+ include_related=["run_steps"],
+ )
+
wait_for_job_run_status_test_args = [
(DbtCloudJobRunStatus.SUCCESS.value,
DbtCloudJobRunStatus.SUCCESS.value, True),
(DbtCloudJobRunStatus.ERROR.value, DbtCloudJobRunStatus.SUCCESS.value,
"exception"),
@@ -1117,15 +1192,21 @@ class TestDbtCloudHook:
with (
patch.object(DbtCloudHook, "get_job_run_status") as
mock_job_run_status,
+ patch.object(DbtCloudHook, "log_job_run_failure_details") as
mock_log_failure_details,
patch("airflow.providers.dbt.cloud.hooks.dbt.time.sleep",
side_effect=fake_sleep),
):
mock_job_run_status.return_value = job_run_status
if expected_output not in ("timeout", "exception"):
assert hook.wait_for_job_run_status(**config) ==
expected_output
+ mock_log_failure_details.assert_not_called()
else:
with pytest.raises(DbtCloudJobRunException):
hook.wait_for_job_run_status(**config)
+ if expected_output == "exception":
+
mock_log_failure_details.assert_called_once_with(run_id=RUN_ID, account_id=None)
+ else:
+ mock_log_failure_details.assert_not_called()
@pytest.mark.parametrize(
argnames=("conn_id", "account_id"),
diff --git a/providers/dbt/cloud/tests/unit/dbt/cloud/operators/test_dbt.py
b/providers/dbt/cloud/tests/unit/dbt/cloud/operators/test_dbt.py
index 67d3fc33100..134ad6b6ecf 100644
--- a/providers/dbt/cloud/tests/unit/dbt/cloud/operators/test_dbt.py
+++ b/providers/dbt/cloud/tests/unit/dbt/cloud/operators/test_dbt.py
@@ -158,6 +158,7 @@ class TestDbtCloudRunJobOperator:
"airflow.providers.dbt.cloud.hooks.dbt.DbtCloudHook.get_job_run_status",
return_value=DbtCloudJobRunStatus.ERROR.value,
)
+
@patch("airflow.providers.dbt.cloud.hooks.dbt.DbtCloudHook.log_job_run_failure_details")
@patch("airflow.providers.dbt.cloud.operators.dbt.DbtCloudRunJobOperator.defer")
@patch("airflow.providers.dbt.cloud.hooks.dbt.DbtCloudHook.get_connection")
@patch(
@@ -165,7 +166,7 @@ class TestDbtCloudRunJobOperator:
return_value=mock_response_json(DEFAULT_ACCOUNT_JOB_RUN_RESPONSE),
)
def test_execute_failed_before_getting_deferred(
- self, mock_trigger_job_run, mock_dbt_hook, mock_defer,
mock_job_run_status
+ self, mock_trigger_job_run, mock_dbt_hook, mock_defer,
mock_log_failure_details, mock_job_run_status
):
dbt_op = DbtCloudRunJobOperator(
dbt_cloud_conn_id=ACCOUNT_ID_CONN,
@@ -558,7 +559,13 @@ class TestDbtCloudRunJobOperator:
)
if job_run_status in DbtCloudJobRunStatus.TERMINAL_STATUSES.value:
- assert mock_get_job_run.call_count == 1
+ if job_run_status in (
+ DbtCloudJobRunStatus.ERROR.value,
+ DbtCloudJobRunStatus.CANCELLED.value,
+ ):
+ assert mock_get_job_run.call_count == 2
+ else:
+ assert mock_get_job_run.call_count == 1
else:
# When the job run status is not in a terminal status or
"Success", the operator will
# continue to call ``get_job_run()`` until a ``timeout``
number of seconds has passed
diff --git a/providers/dbt/cloud/tests/unit/dbt/cloud/sensors/test_dbt.py
b/providers/dbt/cloud/tests/unit/dbt/cloud/sensors/test_dbt.py
index bc3a8e542fb..9435fad5d2b 100644
--- a/providers/dbt/cloud/tests/unit/dbt/cloud/sensors/test_dbt.py
+++ b/providers/dbt/cloud/tests/unit/dbt/cloud/sensors/test_dbt.py
@@ -87,8 +87,11 @@ class TestDbtCloudJobRunSensor:
(30, "exception"), # CANCELLED
],
)
+ @patch.object(DbtCloudHook, "log_job_run_failure_details")
@patch.object(DbtCloudHook, "get_job_run_status")
- def test_poke_with_exception(self, mock_job_run_status, job_run_status,
expected_poke_result):
+ def test_poke_with_exception(
+ self, mock_job_run_status, mock_log_failure_details, job_run_status,
expected_poke_result
+ ):
mock_job_run_status.return_value = job_run_status
# The sensor should fail if the job run status is 20 (aka Error) or 30
(aka Cancelled).
@@ -100,6 +103,8 @@ class TestDbtCloudJobRunSensor:
with pytest.raises(DbtCloudJobRunException, match=error_message):
self.sensor.poke({})
+ mock_log_failure_details.assert_called_once_with(run_id=RUN_ID,
account_id=ACCOUNT_ID)
+
@mock.patch("airflow.providers.dbt.cloud.sensors.dbt.DbtCloudHook")
@mock.patch("airflow.providers.dbt.cloud.sensors.dbt.DbtCloudJobRunSensor.defer")
def test_dbt_cloud_job_run_sensor_finish_before_deferred(self, mock_defer,
mock_hook):
