This is an automated email from the ASF dual-hosted git repository.
o-nikolas 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 5e903931b2a Enable DMS task logging in example_dms system test (#76336)
5e903931b2a is described below
commit 5e903931b2ae2c9b55e4a7768d03f9af748a66b8
Author: Ramit Kataria <[email protected]>
AuthorDate: Tue Nov 3 15:30:51 2026 -0700
Enable DMS task logging in example_dms system test (#76336)
The example_dms system test intermittently fails in await_task_start
because DMS stops the replication task a few seconds after starting it,
before it reaches the running state. The StopReason added to the sensor
failure message in #75240 only said "Stop Reason NORMAL", with no
failure message, so the DMS API does not explain the stop. The DMS task
logs are the next place to look.
Copying the task logs into the Airflow task log puts them next to the
rest of the system test output, and deleting the log group afterwards
keeps the hourly runs from leaving one behind each time.
---
.../amazon/tests/system/amazon/aws/example_dms.py | 31 ++++++++++++++++++++++
1 file changed, 31 insertions(+)
diff --git a/providers/amazon/tests/system/amazon/aws/example_dms.py
b/providers/amazon/tests/system/amazon/aws/example_dms.py
index 6fe5109af62..1a5deeb85dc 100644
--- a/providers/amazon/tests/system/amazon/aws/example_dms.py
+++ b/providers/amazon/tests/system/amazon/aws/example_dms.py
@@ -349,6 +349,23 @@ def delete_dms_assets(
)
+@task(trigger_rule=TriggerRule.ALL_DONE)
+def print_and_delete_dms_task_logs(log_group_name: str):
+ """
+ Copy the DMS task logs into this task's log, then delete their CloudWatch
log group.
+
+ Runs after the replication instance is deleted, so DMS has finished
shipping the task logs.
+ """
+ logs_client = boto3.client("logs")
+ try:
+ for page in
logs_client.get_paginator("filter_log_events").paginate(logGroupName=log_group_name):
+ for event in page["events"]:
+ print(event["message"])
+ logs_client.delete_log_group(logGroupName=log_group_name)
+ except logs_client.exceptions.ResourceNotFoundException:
+ print(f"Log group {log_group_name} does not exist, DMS wrote no task
logs.")
+
+
@task(trigger_rule=TriggerRule.ALL_DONE)
def delete_security_group(
security_group_id: str, security_group_name: str, existing_security_group:
str | None = None
@@ -462,6 +479,19 @@ with DAG(
},
)
# [END howto_operator_dms_create_task]
+ # Kept out of the docs snippet; print_and_delete_dms_task_logs collects
these logs in teardown.
+ create_task.create_task_kwargs["ReplicationTaskSettings"] = json.dumps(
+ {
+
**json.loads(create_task.create_task_kwargs["ReplicationTaskSettings"]),
+ "Logging": {
+ "EnableLogging": True,
+ "LogComponents": [
+ {"Id": "SOURCE_CAPTURE", "Severity":
"LOGGER_SEVERITY_DEBUG"},
+ {"Id": "TASK_MANAGER", "Severity":
"LOGGER_SEVERITY_DEBUG"},
+ ],
+ },
+ }
+ )
task_arn = cast("str", create_task.output)
@@ -603,6 +633,7 @@ with DAG(
# TEST TEARDOWN
delete_task,
delete_assets,
+
print_and_delete_dms_task_logs(f"dms-tasks-{dms_replication_instance_name}"),
delete_db_instance,
delete_parameter_group,
delete_security_group(create_sg, security_group_name, security_group),
