This is an automated email from the ASF dual-hosted git repository.
vincbeck 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 b7b69d70389 Skip or fail ECS tasks correctly when their CloudWatch log
stream is missing (#74730)
b7b69d70389 is described below
commit b7b69d70389ecca02c6ed36215e2b1a39d30e49c
Author: Tom Ron <[email protected]>
AuthorDate: Tue Oct 20 14:56:43 2026 +0300
Skip or fail ECS tasks correctly when their CloudWatch log stream is
missing (#74730)
When an ECS task's container exits non-zero, EcsRunTaskOperator reads the
last
CloudWatch log lines to build its error message. If the log stream did not
exist, CloudWatch's ResourceNotFoundException escaped instead of the
operator's
own exception, so skip_on_exit_code was ignored and tasks that should be
skipped failed. A missing stream now counts as no log lines, as it already
does
in the log fetcher's polling loop.
Co-authored-by: Tom <[email protected]>
---
.../providers/amazon/aws/utils/task_log_fetcher.py | 24 ++++++++++++++++------
.../tests/unit/amazon/aws/operators/test_ecs.py | 22 ++++++++++++++++++++
.../unit/amazon/aws/utils/test_task_log_fetcher.py | 8 ++++++++
3 files changed, 48 insertions(+), 6 deletions(-)
diff --git
a/providers/amazon/src/airflow/providers/amazon/aws/utils/task_log_fetcher.py
b/providers/amazon/src/airflow/providers/amazon/aws/utils/task_log_fetcher.py
index a2a7a79f385..1ffdd857c8c 100644
---
a/providers/amazon/src/airflow/providers/amazon/aws/utils/task_log_fetcher.py
+++
b/providers/amazon/src/airflow/providers/amazon/aws/utils/task_log_fetcher.py
@@ -158,13 +158,25 @@ class AwsTaskLogFetcher(Thread):
NOTE: some restrictions apply:
- if logs are too old, the response will be empty
- the max number of messages we can retrieve is constrained by
cloudwatch limits (10,000).
+ - if the log stream does not exist, an empty list is returned.
"""
- response = self.hook.conn.get_log_events(
- logGroupName=self.log_group,
- logStreamName=self.log_stream_name,
- startFromHead=False,
- limit=number_messages,
- )
+ try:
+ response = self.hook.conn.get_log_events(
+ logGroupName=self.log_group,
+ logStreamName=self.log_stream_name,
+ startFromHead=False,
+ limit=number_messages,
+ )
+ except ClientError as error:
+ if error.response["Error"]["Code"] != "ResourceNotFoundException":
+ raise
+ self.logger.warning(
+ "Cannot find log stream, check that the log group and stream
are correct: "
+ "group: %s\tstream: %s",
+ self.log_group,
+ self.log_stream_name,
+ )
+ return []
return [log["message"] for log in response["events"]]
def get_last_log_message(self) -> str | None:
diff --git a/providers/amazon/tests/unit/amazon/aws/operators/test_ecs.py
b/providers/amazon/tests/unit/amazon/aws/operators/test_ecs.py
index dcb46ffa520..b53ca45d9c6 100644
--- a/providers/amazon/tests/unit/amazon/aws/operators/test_ecs.py
+++ b/providers/amazon/tests/unit/amazon/aws/operators/test_ecs.py
@@ -18,6 +18,7 @@
from __future__ import annotations
from copy import deepcopy
+from datetime import timedelta
from unittest import mock
from unittest.mock import MagicMock, PropertyMock
@@ -27,6 +28,7 @@ from botocore.exceptions import ClientError, WaiterError
from airflow.providers.amazon.aws.exceptions import EcsOperatorError,
EcsTaskFailToStart
from airflow.providers.amazon.aws.hooks.ecs import EcsClusterStates, EcsHook
+from airflow.providers.amazon.aws.hooks.logs import AwsLogsHook
from airflow.providers.amazon.aws.operators.ecs import (
EcsBaseOperator,
EcsCreateClusterOperator,
@@ -631,6 +633,26 @@ class TestEcsRunTaskOperator(EcsBaseTestCase):
with pytest.raises(AirflowSkipException, match="This task is not in
success state"):
self.ecs._check_success_task()
+ @mock.patch.object(AwsLogsHook, "conn")
+ @mock.patch.object(EcsBaseOperator, "client")
+ def test_check_success_task_skip_exception_with_missing_log_stream(self,
client_mock, logs_conn_mock):
+ self.ecs.arn = "arn"
+ self.ecs.skip_on_exit_code = [2]
+ self.ecs.task_log_fetcher = AwsTaskLogFetcher(
+ log_group="group",
+ log_stream_name="stream",
+ fetch_interval=timedelta(seconds=1),
+ logger=self.ecs.log,
+ )
+ logs_conn_mock.get_log_events.side_effect = ClientError(
+ {"Error": {"Code": "ResourceNotFoundException"}}, "GetLogEvents"
+ )
+ client_mock.describe_tasks.return_value = {
+ "tasks": [{"containers": [{"name": "container-name", "lastStatus":
"STOPPED", "exitCode": 2}]}]
+ }
+ with pytest.raises(AirflowSkipException, match="This task is not in
success state"):
+ self.ecs._check_success_task()
+
@mock.patch.object(EcsBaseOperator, "client")
def
test_check_success_task_unmatched_exit_code_raises_airflow_exception(self,
client_mock):
"""Exit codes not in skip_on_exit_code raise AirflowException."""
diff --git
a/providers/amazon/tests/unit/amazon/aws/utils/test_task_log_fetcher.py
b/providers/amazon/tests/unit/amazon/aws/utils/test_task_log_fetcher.py
index 6125d0bc3cf..1c2a6f49392 100644
--- a/providers/amazon/tests/unit/amazon/aws/utils/test_task_log_fetcher.py
+++ b/providers/amazon/tests/unit/amazon/aws/utils/test_task_log_fetcher.py
@@ -174,6 +174,14 @@ class TestAwsTaskLogFetcher:
def test_get_last_log_messages_with_no_log_events(self, mock_conn):
assert self.log_fetcher.get_last_log_messages(2) == []
+ @mock.patch.object(AwsLogsHook, "conn")
+ def test_get_last_log_messages_with_missing_log_stream(self,
log_conn_mock):
+ log_conn_mock.get_log_events.side_effect = ClientError(
+ {"Error": {"Code": "ResourceNotFoundException"}}, "GetLogEvents"
+ )
+ assert self.log_fetcher.get_last_log_messages(2) == []
+ self.logger_mock.warning.assert_called_once()
+
@mock.patch(
"airflow.providers.amazon.aws.hooks.logs.AwsLogsHook.get_log_events",
side_effect=(
