d-hervas opened a new pull request, #73004:
URL: https://github.com/apache/airflow/pull/73004
Structured Triggerer logging reused one queued Watchtower handler across all
CloudWatch stream paths. Watchtower retains queue, sequence-token, and worker
state for every path until that handler closes, while Airflow keeps the remote
I/O object alive for the Triggerer process.
This change owns one Watchtower handler per active structured-log path and
closes/removes that handler when `upload(path)` signals completion. Other
active paths remain live, a later reuse of the same path creates a fresh
handler, and records racing with close cannot reopen the path during teardown.
The legacy task-handler path keeps its current behavior.
### Expected behavior
When `upload(path)` marks one trigger stream complete, all
queued records for that stream are delivered and its Watchtower queue,
worker,
and sequence-token state are released. Other active streams must remain
usable,
and later reuse of the same path must create fresh state.
### Failure mode
A long-lived Triggerer can process many distinct trigger log paths,
including sequentially. With Watchtower's default queued delivery, the first
record for each path creates a queue, sequence-token entry, and worker thread.
Airflow calls `upload(path)` when the trigger finishes, but the current
implementation only flushes the shared handler. `flush()` delivers pending
records without stopping the worker or removing that stream's state. Since the
shared handler remains reachable for the lifetime of the Triggerer, retained
state grows with the number of completed streams. Eventually the Triggerer's
RSS or thread count can reach its container limit and the process is terminated
with `SIGKILL`/OOM.
The retention can be reproduced without AWS credentials or API calls using
Watchtower 3.4.0. This mirrors the current provider behavior of changing
`log_stream_name` on one queued handler:
```python
import logging
from itertools import count
import watchtower
class FakeLogsClient:
def __init__(self):
self.tokens = count(1)
def put_log_events(self, **kwargs):
return {"nextSequenceToken": str(next(self.tokens))}
handler = watchtower.CloudWatchLogHandler(
log_group_name="reproducer",
log_stream_name="unused",
boto3_client=FakeLogsClient(),
create_log_group=False,
create_log_stream=False,
use_queues=True,
send_interval=60,
)
for job_id in range(300):
handler.log_stream_name = (
f"dag/task/attempt=1.log.trigger.{job_id}.log"
)
record = logging.LogRecord(
"trigger",
logging.INFO,
"repro.py",
1,
"event",
(),
None,
)
handler.handle(record)
handler.flush()
print("queues:", len(handler.queues))
print("sequence tokens:", len(handler.sequence_tokens))
print("thread references:", len(handler.threads))
print("live workers:", sum(thread.is_alive() for thread in handler.threads))
handler.close()
```
Output:
```text
queues: 300
sequence tokens: 300
thread references: 300
live workers: 300
```
Calling `flush()` therefore does not bound retained state as streams
complete.
Calling `close()` stops all workers, but the current shared-handler design
cannot do that for one completed stream without also closing other active
streams.
This change gives each active path an independently closable handler.
### Related work
#70635 adds read-side discovery of deferred-task trigger log streams so they
remain visible after deferral. This PR addresses the separate write-side
lifecycle of those streams and releases Watchtower's retained queues,
sequence-token state, and worker threads when each stream completes.
### Validation
All 39 CloudWatch task-handler tests pass on Python 3.10.21 and Python
3.14.7. New tests cover active-stream isolation, repeated completion,
exact-path reuse, 100 sequential processor-driven paths with no retained
handlers, construction and close/log races, and final shutdown. Ruff, the
Airflow Python 3.10 CI image's pinned mypy for both changed files, the full
pre-commit `prek` stage, and `git diff --check` also pass.
---
##### Was generative AI tooling used to co-author this PR?
- [x] Yes: OpenAI Codex
--
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]
