This is an automated email from the ASF dual-hosted git repository.
shahar1 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 0f2827e36e5 Fix EcsRunTaskOperator deferred logs read from the wrong
region (#70474)
0f2827e36e5 is described below
commit 0f2827e36e53ae32779f480913dfc95fbf9c24c9
Author: Jyun-An Chen <[email protected]>
AuthorDate: Mon Jul 27 20:48:47 2026 +0800
Fix EcsRunTaskOperator deferred logs read from the wrong region (#70474)
TaskDoneTrigger received a single region_name and used it to build both the
ECS waiter client and the CloudWatch client, so a task definition shipping
its logs to a different region got no logs forwarded while the task was
deferred -- only repeated "didn't exist (yet)" messages for the whole run.
The non-deferrable path and the post-deferral fetch already resolve
awslogs_region; this brings the in-deferral forwarding in line with them.
closes: #70465
---
.../airflow/providers/amazon/aws/operators/ecs.py | 1 +
.../airflow/providers/amazon/aws/triggers/ecs.py | 11 +++-
.../tests/unit/amazon/aws/operators/test_ecs.py | 20 +++++++
.../tests/unit/amazon/aws/triggers/test_ecs.py | 68 ++++++++++++++++++----
4 files changed, 89 insertions(+), 11 deletions(-)
diff --git a/providers/amazon/src/airflow/providers/amazon/aws/operators/ecs.py
b/providers/amazon/src/airflow/providers/amazon/aws/operators/ecs.py
index 3bdc3e63cc5..89fc67b6d04 100644
--- a/providers/amazon/src/airflow/providers/amazon/aws/operators/ecs.py
+++ b/providers/amazon/src/airflow/providers/amazon/aws/operators/ecs.py
@@ -554,6 +554,7 @@ class EcsRunTaskOperator(EcsBaseOperator):
waiter_max_attempts=self.waiter_max_attempts,
aws_conn_id=self.aws_conn_id,
region_name=self.region_name,
+ log_region_name=self.resolve_awslogs_region(),
log_group=self.awslogs_group,
log_stream=self._get_logs_stream_name(),
verify=self.verify,
diff --git a/providers/amazon/src/airflow/providers/amazon/aws/triggers/ecs.py
b/providers/amazon/src/airflow/providers/amazon/aws/triggers/ecs.py
index 39896960cfa..630a2c1de55 100644
--- a/providers/amazon/src/airflow/providers/amazon/aws/triggers/ecs.py
+++ b/providers/amazon/src/airflow/providers/amazon/aws/triggers/ecs.py
@@ -128,6 +128,9 @@ class TaskDoneTrigger(BaseTrigger):
Will fail after that many unsuccessful attempts.
:param aws_conn_id: The Airflow connection used for AWS credentials.
:param region_name: The AWS region where the cluster is located. Used to
build the hook.
+ :param log_region_name: The AWS region where the CloudWatch logs are
stored. Defaults to
+ ``region_name`` when not set, which is correct unless the task
definition ships its logs
+ to another region.
:param verify: Whether or not to verify SSL certificates. Used to build
the hook.
:param botocore_config: Configuration dictionary for the botocore client.
Used to build the hook.
:param region: (deprecated) use ``region_name`` instead.
@@ -146,6 +149,7 @@ class TaskDoneTrigger(BaseTrigger):
verify: bool | str | None = None,
botocore_config: dict | None = None,
region: str | None = None,
+ log_region_name: str | None = None,
):
if region is not None:
warnings.warn(
@@ -167,6 +171,7 @@ class TaskDoneTrigger(BaseTrigger):
self.log_group = log_group
self.log_stream = log_stream
+ self.log_region_name = log_region_name
def serialize(self) -> tuple[str, dict[str, Any]]:
return (
@@ -178,6 +183,7 @@ class TaskDoneTrigger(BaseTrigger):
"waiter_max_attempts": self.waiter_max_attempts,
"aws_conn_id": self.aws_conn_id,
"region_name": self.region_name,
+ "log_region_name": self.log_region_name,
"log_group": self.log_group,
"log_stream": self.log_stream,
"verify": self.verify,
@@ -186,6 +192,9 @@ class TaskDoneTrigger(BaseTrigger):
)
async def run(self) -> AsyncIterator[TriggerEvent]:
+ # Triggers serialized before ``log_region_name`` existed deserialize
without it, so an
+ # unset value keeps reading logs from the cluster region as before.
+ log_region_name = self.log_region_name if self.log_region_name is not
None else self.region_name
async with (
await EcsHook(
aws_conn_id=self.aws_conn_id,
@@ -195,7 +204,7 @@ class TaskDoneTrigger(BaseTrigger):
).get_async_conn() as ecs_client,
await AwsLogsHook(
aws_conn_id=self.aws_conn_id,
- region_name=self.region_name,
+ region_name=log_region_name,
verify=self.verify,
config=self.botocore_config,
).get_async_conn() as logs_client,
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 346c29efbaf..8a28069fd25 100644
--- a/providers/amazon/tests/unit/amazon/aws/operators/test_ecs.py
+++ b/providers/amazon/tests/unit/amazon/aws/operators/test_ecs.py
@@ -846,6 +846,26 @@ class TestEcsRunTaskOperator(EcsBaseTestCase):
assert isinstance(deferred.value.trigger, TaskDoneTrigger)
assert deferred.value.trigger.task_arn ==
f"arn:aws:ecs:us-east-1:012345678910:task/{TASK_ID}"
+ @mock.patch.object(EcsRunTaskOperator, "client")
+ def test_with_defer_passes_awslogs_region_to_trigger(self, client_mock):
+ self.set_up_operator(
+ awslogs_group="awslogs-group",
+ awslogs_region="logs-region",
+ awslogs_stream_prefix="prefix",
+ region_name="task-region",
+ deferrable=True,
+ )
+ client_mock.run_task.return_value = RESPONSE_WITHOUT_FAILURES
+
+ mock_ti = mock.MagicMock()
+ mock_context = {"ti": mock_ti, "task_instance": mock_ti}
+
+ with pytest.raises(TaskDeferred) as deferred:
+ self.ecs.execute(mock_context)
+
+ assert deferred.value.trigger.region_name == "task-region"
+ assert deferred.value.trigger.log_region_name == "logs-region"
+
@mock.patch.object(EcsRunTaskOperator, "client", new_callable=PropertyMock)
def test_execute_complete(self, client_mock):
event = {"status": "success", "task_arn": "my_arn", "cluster":
"test_cluster"}
diff --git a/providers/amazon/tests/unit/amazon/aws/triggers/test_ecs.py
b/providers/amazon/tests/unit/amazon/aws/triggers/test_ecs.py
index a46fc053c38..2a38af826ea 100644
--- a/providers/amazon/tests/unit/amazon/aws/triggers/test_ecs.py
+++ b/providers/amazon/tests/unit/amazon/aws/triggers/test_ecs.py
@@ -35,6 +35,15 @@ if TYPE_CHECKING:
from airflow.triggers.base import TriggerEvent
+def _make_async_hook(client):
+ ctx = mock.MagicMock()
+ ctx.__aenter__ = AsyncMock(return_value=client)
+ ctx.__aexit__ = AsyncMock(return_value=False)
+ instance = mock.MagicMock()
+ instance.get_async_conn = AsyncMock(return_value=ctx)
+ return instance
+
+
class TestTaskDoneTrigger:
def test_deprecated_region_alias(self):
with pytest.warns(AirflowProviderDeprecationWarning, match="region"):
@@ -73,28 +82,35 @@ class TestTaskDoneTrigger:
"waiter_max_attempts": 10,
"aws_conn_id": "my_conn",
"region_name": "eu-west-1",
+ "log_region_name": None,
"log_group": "lg",
"log_stream": "ls",
"verify": False,
"botocore_config": {"read_timeout": 7},
}
+ def test_serialize_keeps_log_region_name_as_passed(self):
+ trigger = TaskDoneTrigger(
+ cluster="cluster",
+ task_arn="task_arn",
+ waiter_delay=5,
+ waiter_max_attempts=10,
+ aws_conn_id="my_conn",
+ region_name="ap-northeast-1",
+ log_region_name="us-east-1",
+ )
+ _, kwargs = trigger.serialize()
+ assert kwargs["log_region_name"] == "us-east-1"
+ assert kwargs["region_name"] == "ap-northeast-1"
+
@pytest.mark.asyncio
@mock.patch("airflow.providers.amazon.aws.triggers.ecs.AwsLogsHook")
@mock.patch("airflow.providers.amazon.aws.triggers.ecs.EcsHook")
async def test_run_builds_hooks_with_generic_params(self, ecs_hook_cls,
logs_hook_cls):
- def make_hook(client):
- ctx = mock.MagicMock()
- ctx.__aenter__ = AsyncMock(return_value=client)
- ctx.__aexit__ = AsyncMock(return_value=False)
- instance = mock.MagicMock()
- instance.get_async_conn = AsyncMock(return_value=ctx)
- return instance
-
ecs_client = mock.MagicMock()
ecs_client.get_waiter().wait = AsyncMock()
- ecs_hook_cls.return_value = make_hook(ecs_client)
- logs_hook_cls.return_value = make_hook(mock.MagicMock())
+ ecs_hook_cls.return_value = _make_async_hook(ecs_client)
+ logs_hook_cls.return_value = _make_async_hook(mock.MagicMock())
trigger = TaskDoneTrigger(
cluster="cluster",
@@ -117,6 +133,38 @@ class TestTaskDoneTrigger:
ecs_hook_cls.assert_called_once_with(**expected)
logs_hook_cls.assert_called_once_with(**expected)
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
+ ("log_region_name", "expected_logs_hook_region"),
+ [
+ pytest.param(None, "ap-northeast-1",
id="falls-back-to-cluster-region"),
+ pytest.param("us-east-1", "us-east-1",
id="uses-explicit-log-region"),
+ ],
+ )
+ @mock.patch("airflow.providers.amazon.aws.triggers.ecs.AwsLogsHook")
+ @mock.patch("airflow.providers.amazon.aws.triggers.ecs.EcsHook")
+ async def test_run_reads_logs_from_log_region(
+ self, ecs_hook_cls, logs_hook_cls, log_region_name,
expected_logs_hook_region
+ ):
+ ecs_client = mock.MagicMock()
+ ecs_client.get_waiter().wait = AsyncMock()
+ ecs_hook_cls.return_value = _make_async_hook(ecs_client)
+ logs_hook_cls.return_value = _make_async_hook(mock.MagicMock())
+
+ trigger = TaskDoneTrigger(
+ cluster="cluster",
+ task_arn="task_arn",
+ waiter_delay=0,
+ waiter_max_attempts=10,
+ aws_conn_id="my_conn",
+ region_name="ap-northeast-1",
+ log_region_name=log_region_name,
+ )
+ await trigger.run().asend(None)
+
+ assert ecs_hook_cls.call_args.kwargs["region_name"] == "ap-northeast-1"
+ assert logs_hook_cls.call_args.kwargs["region_name"] ==
expected_logs_hook_region
+
@pytest.mark.asyncio
@mock.patch.object(EcsHook, "get_async_conn")
# this mock is only necessary to avoid a "No module named 'aiobotocore'"
error in the LatestBoto CI step
