seanghaeli opened a new pull request, #76631:
URL: https://github.com/apache/airflow/pull/76631
closes: #70317
The triggerer writes a deferred task's trigger lines to their own CloudWatch
stream, `<task log>.trigger.<triggerer job id>.log`, but
`CloudWatchRemoteLogIO.stream` read only the task log's exact stream. While the
task is deferred the lines come live from the triggerer. Once it finishes they
vanish from the UI, even though they are in CloudWatch. This fixes it on main
by itself and needs no other PR.
- `stream` now reads every stream the path is a prefix of, as the S3, GCS
and WASB readers read every object under it.
- While the task is deferred, it skips the current triggerer's stream, which
the task log reader already reads live, so no line shows twice.
- If listing the streams fails, a task log falls back to its exact stream,
as before.
- A path ending in `/` with no stream under it has no logs, instead of a "No
log stream found" notice as its one line. If it can't be listed, `stream`
raises, as the S3 reader does, rather than reading it as a stream name.
- Listing needs `logs:DescribeLogStreams` on the log group, which CloudWatch
remote logging didn't need before. The CloudWatch logging docs now say so.
Without it, task logs read their exact stream as on main, with a note in the
source details, and a finished task's trigger lines stay missing.
How to review: `CloudWatchRemoteLogIO.stream` and `_list_log_stream_names`
in `providers/amazon/aws/log/cloudwatch_task_handler.py`.
Breeze on Postgres with CloudWatch on moto. Two tasks: one defers once
(`deferred_once`), and one is still deferred when its triggerer stops, then
finishes on a second triggerer (`deferred_across_restart`). The triggerer runs
on its own disk with `delete_local_logs=True`, and on the API server's disk.
These are the trigger lines in the task's log while it is deferred and once it
is done. On main, the same steps give 0 lines once done, in both modes:
```text
# after this PR
################ MODE cw_split1_delTrue: remote=cw triggerer_own_disk=1
delete_local_logs=True
[task log, p1_deferred] TRIGGER-LINE lines=14 distinct=14 duplicates=0
triggerer_pids=['pid=73'] TASK-LINE lines=1
[task log, p1_done] TRIGGER-LINE lines=23 distinct=23 duplicates=0
triggerer_pids=['pid=73'] TASK-LINE lines=2
cloudwatch stream
dag_id=deferred_once/run_id=cw_split1_delTrue_short/task_id=defer_and_log/attempt=1.log:
events=18 TRIGGER-LINE=0
cloudwatch stream
dag_id=deferred_once/run_id=cw_split1_delTrue_short/task_id=defer_and_log/attempt=1.log.trigger.3.log:
events=26 TRIGGER-LINE=23
[task log, p2_owner_stopped] TRIGGER-LINE lines=11 distinct=11
duplicates=0 triggerer_pids=['pid=73'] TASK-LINE lines=1
[task log, p2_done] TRIGGER-LINE lines=46 distinct=46 duplicates=0
triggerer_pids=['pid=376', 'pid=73'] TASK-LINE lines=2
cloudwatch stream
dag_id=deferred_across_restart/run_id=cw_split1_delTrue_long/task_id=defer_and_log/attempt=1.log:
events=18 TRIGGER-LINE=0
cloudwatch stream
dag_id=deferred_across_restart/run_id=cw_split1_delTrue_long/task_id=defer_and_log/attempt=1.log.trigger.3.log:
events=13 TRIGGER-LINE=11
cloudwatch stream
dag_id=deferred_across_restart/run_id=cw_split1_delTrue_long/task_id=defer_and_log/attempt=1.log.trigger.4.log:
events=38 TRIGGER-LINE=35
################ MODE cw_split0_delFalse: remote=cw triggerer_own_disk=0
delete_local_logs=False
[task log, p1_deferred] TRIGGER-LINE lines=14 distinct=14 duplicates=0
triggerer_pids=['pid=570'] TASK-LINE lines=2
[task log, p1_done] TRIGGER-LINE lines=23 distinct=23 duplicates=0
triggerer_pids=['pid=570'] TASK-LINE lines=2
cloudwatch stream
dag_id=deferred_once/run_id=cw_split0_delFalse_short/task_id=defer_and_log/attempt=1.log:
events=18 TRIGGER-LINE=0
cloudwatch stream
dag_id=deferred_once/run_id=cw_split0_delFalse_short/task_id=defer_and_log/attempt=1.log.trigger.6.log:
events=26 TRIGGER-LINE=23
[task log, p2_owner_stopped] TRIGGER-LINE lines=11 distinct=11
duplicates=0 triggerer_pids=['pid=570'] TASK-LINE lines=2
[task log, p2_done] TRIGGER-LINE lines=46 distinct=46 duplicates=0
triggerer_pids=['pid=570', 'pid=871'] TASK-LINE lines=2
cloudwatch stream
dag_id=deferred_across_restart/run_id=cw_split0_delFalse_long/task_id=defer_and_log/attempt=1.log:
events=18 TRIGGER-LINE=0
cloudwatch stream
dag_id=deferred_across_restart/run_id=cw_split0_delFalse_long/task_id=defer_and_log/attempt=1.log.trigger.6.log:
events=13 TRIGGER-LINE=11
cloudwatch stream
dag_id=deferred_across_restart/run_id=cw_split0_delFalse_long/task_id=defer_and_log/attempt=1.log.trigger.8.log:
events=38 TRIGGER-LINE=35
```
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR,
please
change below checkbox to `[X]` followed by the name of the tool, uncomment
the "Generated-by".
-->
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Opus 5.5) following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
