This is an automated email from the ASF dual-hosted git repository.
dabla 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 71a19df58c5 Write container logs to the task log in deferrable
AzureContainerInstancesOperator (#76330)
71a19df58c5 is described below
commit 71a19df58c50f12d5ad937a9c4a6c7f8373cc6b8
Author: Ankit Raj <[email protected]>
AuthorDate: Fri Nov 6 00:57:12 2026 +0530
Write container logs to the task log in deferrable
AzureContainerInstancesOperator (#76330)
* Write container logs to the task log in deferrable
AzureContainerInstancesOperator
* Fail the task when logs needed for XCom cannot be read
---
.../azure/operators/container_instances.py | 26 +++-
.../azure/operators/test_container_instances.py | 143 +++++++++++++++++++++
2 files changed, 163 insertions(+), 6 deletions(-)
diff --git
a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/container_instances.py
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/container_instances.py
index a196282502f..0b7df5bc821 100644
---
a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/container_instances.py
+++
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/container_instances.py
@@ -467,6 +467,10 @@ class AzureContainerInstancesOperator(BaseOperator):
exit_code: int = event.get("exit_code", 1)
+ # The trigger does not stream the container output, so write it to the
task log here,
+ # before the container group may be removed.
+ logs = self._log_container_output()
+
if event["status"] == "error":
if self.remove_on_error:
self.on_kill()
@@ -479,13 +483,11 @@ class AzureContainerInstancesOperator(BaseOperator):
try:
if self.xcom_all is not None:
- logs = self.hook.get_logs(self.resource_group, self.name)
if logs is None:
- context["ti"].xcom_push(key="logs", value=[])
- elif self.xcom_all:
- context["ti"].xcom_push(key="logs", value=logs)
- else:
- context["ti"].xcom_push(key="logs", value=logs[-1:])
+ # The first fetch failed; fetch again so the error fails
the task
+ # instead of pushing empty logs to XCom.
+ logs = self.hook.get_logs(self.resource_group, self.name)
+ context["ti"].xcom_push(key="logs", value=logs if
self.xcom_all else logs[-1:])
self.log.info("Container had exit code: %s", exit_code)
return exit_code
@@ -563,6 +565,18 @@ class AzureContainerInstancesOperator(BaseOperator):
time.sleep(1)
+ def _log_container_output(self) -> list | None:
+ """Write the container group's logs to the task log and return them."""
+ try:
+ logs = self.hook.get_logs(self.resource_group, self.name)
+ except Exception as e:
+ self.log.warning(
+ "Failed to fetch logs for container group %s/%s: %s",
self.resource_group, self.name, e
+ )
+ return None
+ self._log_last(logs, None)
+ return logs
+
def _log_last(self, logs: list | None, last_line_logged: Any) -> Any |
None:
if logs:
# determine the last line which was logged before
diff --git
a/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_container_instances.py
b/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_container_instances.py
index d21070de0c3..99ba304e3b4 100644
---
a/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_container_instances.py
+++
b/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_container_instances.py
@@ -799,6 +799,149 @@ class TestACIOperator:
)
assert aci_mock.return_value.delete.call_count == 1
+ @mock.patch(
+
"airflow.providers.microsoft.azure.operators.container_instances.AzureContainerInstanceHook",
+ autospec=True,
+ )
+ def
test_execute_complete_error_event_logs_container_output_before_removal(self,
aci_mock, caplog):
+ """On error, the container output is written to the task log before
the container group is removed."""
+ calls = []
+ aci_mock.return_value.get_logs.side_effect = lambda *args, **kwargs: (
+ calls.append("get_logs") or ["the real error\n"]
+ )
+ aci_mock.return_value.delete.side_effect = lambda *args, **kwargs:
calls.append("delete")
+
+ aci = AzureContainerInstancesOperator(
+ ci_conn_id="azure_default",
+ registry_conn_id=None,
+ resource_group="resource-group",
+ name="container-name",
+ image="container-image",
+ region="region",
+ task_id="task",
+ deferrable=True,
+ remove_on_error=True,
+ )
+ with caplog.at_level("INFO"), pytest.raises(RuntimeError,
match="Container group failed"):
+ aci.execute_complete(
+ context=None,
+ event={
+ "status": "error",
+ "exit_code": 1,
+ "detail_status": "Failed",
+ "message": "Container group failed with exit code 1",
+ "resource_group": "resource-group",
+ "name": "container-name",
+ },
+ )
+
+ assert "the real error" in caplog.text
+ assert calls == ["get_logs", "delete"]
+
+ @mock.patch(
+
"airflow.providers.microsoft.azure.operators.container_instances.AzureContainerInstanceHook",
+ autospec=True,
+ )
+ def test_execute_complete_success_logs_container_output(self, aci_mock,
caplog):
+ """On success, the container output is written to the task log."""
+ aci_mock.return_value.get_logs.return_value = ["hello from the
container\n"]
+
+ aci = AzureContainerInstancesOperator(
+ ci_conn_id="azure_default",
+ registry_conn_id=None,
+ resource_group="resource-group",
+ name="container-name",
+ image="container-image",
+ region="region",
+ task_id="task",
+ deferrable=True,
+ )
+ with caplog.at_level("INFO"):
+ aci.execute_complete(
+ context=None,
+ event={
+ "status": "success",
+ "exit_code": 0,
+ "detail_status": "Completed",
+ "resource_group": "resource-group",
+ "name": "container-name",
+ },
+ )
+
+ assert "hello from the container" in caplog.text
+
aci_mock.return_value.get_logs.assert_called_once_with("resource-group",
"container-name")
+
+ @mock.patch(
+
"airflow.providers.microsoft.azure.operators.container_instances.AzureContainerInstanceHook",
+ autospec=True,
+ )
+ def test_execute_complete_log_fetch_failure_keeps_container_error(self,
aci_mock, caplog):
+ """A failure while fetching logs does not replace the container's own
error."""
+ aci_mock.return_value.get_logs.side_effect = Exception("cannot read
logs")
+
+ aci = AzureContainerInstancesOperator(
+ ci_conn_id="azure_default",
+ registry_conn_id=None,
+ resource_group="resource-group",
+ name="container-name",
+ image="container-image",
+ region="region",
+ task_id="task",
+ deferrable=True,
+ remove_on_error=False,
+ )
+ with caplog.at_level("WARNING"), pytest.raises(RuntimeError,
match="Container group failed"):
+ aci.execute_complete(
+ context=None,
+ event={
+ "status": "error",
+ "exit_code": 1,
+ "detail_status": "Failed",
+ "message": "Container group failed with exit code 1",
+ "resource_group": "resource-group",
+ "name": "container-name",
+ },
+ )
+
+ assert "Failed to fetch logs for container group
resource-group/container-name" in caplog.text
+
+ @pytest.mark.parametrize("xcom_all", [True, False])
+ @mock.patch(
+
"airflow.providers.microsoft.azure.operators.container_instances.AzureContainerInstanceHook",
+ autospec=True,
+ )
+ def
test_execute_complete_success_log_fetch_failure_fails_task_when_xcom_needs_logs(
+ self, aci_mock, xcom_all
+ ):
+ """On success, if XCom needs the logs and they cannot be read, the
task fails instead of pushing []."""
+ aci_mock.return_value.get_logs.side_effect = Exception("cannot read
logs")
+
+ aci = AzureContainerInstancesOperator(
+ ci_conn_id="azure_default",
+ registry_conn_id=None,
+ resource_group="resource-group",
+ name="container-name",
+ image="container-image",
+ region="region",
+ task_id="task",
+ deferrable=True,
+ xcom_all=xcom_all,
+ )
+ context = Context(ti=XcomMock())
+ with pytest.raises(Exception, match="cannot read logs"):
+ aci.execute_complete(
+ context=context,
+ event={
+ "status": "success",
+ "exit_code": 0,
+ "detail_status": "Completed",
+ "resource_group": "resource-group",
+ "name": "container-name",
+ },
+ )
+
+ assert "logs" not in context["ti"].values
+
@mock.patch(
"airflow.providers.microsoft.azure.operators.container_instances.AzureContainerInstanceHook",
autospec=True,
