bujjibabukatta opened a new pull request, #70635:
URL: https://github.com/apache/airflow/pull/70635
Title
fix(amazon): include deferred-task trigger logs when reading from CloudWatch
(#70317)
Body
Closes #70317.
Root cause
CloudWatchRemoteLogIO.stream() (and the legacy
CloudwatchTaskHandler._read_remote_logs()) only ever read the task's own
CloudWatch stream. A deferred task's triggerer logs are written to a separate
stream, <task log stream>.trigger.<triggerer job id>.log
(FileTaskHandler.add_triggerer_suffix). While the task is DEFERRED, the UI
tails those logs live from the triggerer over HTTP, which is why they're
visible during the deferral — but once the task finishes, that live path goes
away and nothing ever looked in the separate CloudWatch stream. The local-file
backend doesn't have this problem because FileTaskHandler._read_from_local()
uses glob(worker_log_path.name + "*"), which picks up attempt=1.log and
attempt=1.log.trigger.*.log for free. CloudWatch has no glob equivalent, so the
trigger stream has to be discovered explicitly.
Fix
Added AwsLogsHook.describe_log_streams(): a synchronous, paginated wrapper
around DescribeLogStreams with a prefix filter (mirrors the existing async
variant's ResourceNotFoundException → [] handling for a not-yet-existing log
group).
CloudWatchRemoteLogIO.stream() now looks up any <stream>.trigger.*.log
streams via that prefix scan and reads each one the same way it reads the base
stream, merging them into the returned log groups. Skipped while the task is
actively DEFERRED, since the live HTTP path already covers that state and the
extra API call would just be repeated on every UI poll for a potentially
long-running deferral.
Multiple trigger streams (a task deferred, resumed, and deferred again —
even by different triggerer processes) are sorted numerically by job id, not
lexicographically — job id 7 must sort before job id 200. (Caught this via test
— an initial plain-string sort got it backwards.)
Applied the identical fix to the legacy
CloudwatchTaskHandler._read_remote_logs() path (still registered in
provider.yaml for older-style log-handler configs). Also removed a pre-existing
dead call there (self.io.read(...), whose result was immediately discarded and
redone manually right below it — wasting a CloudWatch API call on every read).
Trigger-stream discovery failures (e.g. a permissions issue) are reported as
an extra message rather than raising, consistent with the existing error
handling on the base stream.
Backward compatibility
When no trigger streams exist (the common, non-deferred case), behavior and
API call count are unchanged — verified via test that the discovery call only
fires when needed and returns [] cleanly for a non-existent/empty-prefix case.
Testing
12 new tests added to test_cloudwatch_task_handler.py, covering:
base+trigger merge, correct numeric ordering across multiple trigger streams,
the DEFERRED-state skip, graceful handling of a DescribeLogStreams failure, the
legacy handler path, correct colon-replacement in the discovery prefix, and
hook-level pagination (tested with 54+ streams) and missing-log-group handling.
All 42 tests in the file pass (33 existing + 12 new — some existing tests
exercise overlapping paths).
ruff check, ruff format --check, and mypy all pass clean on the three
changed files.
Trade-off worth flagging for reviewers
This adds one DescribeLogStreams API call per log read for tasks that are
not currently DEFERRED (previously: zero). For a task that was never deferred,
this returns an empty list almost instantly, but it is still an extra
round-trip. Happy to gate this further (e.g. only when try_number matches a
TaskInstanceHistory row known to have gone through DEFERRED) if that trade-off
is a concern — went with the simpler, always-check approach first since it's
the correct and easy default and mirrors what the issue reporter's own tested
workaround did.
AI Disclosure
This contribution used AI assistance.
Model(s) used: Claude
AI was used for:
Helping write this PR description.
Running the test suite locally to make sure everything passes before
submitting.
Reviewing the implementation from a performance perspective.
The implementation and code changes were developed and validated by me.
--
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]
