fpiped opened a new issue, #73285:
URL: https://github.com/apache/airflow/issues/73285
### Description
`EcsRunTaskOperator` and `BatchOperator` stream their container's CloudWatch
logs through `AwsTaskLogFetcher`, which forwards every event to the operator's
logger. There is no supported way to *observe* those events programmatically:
the fetcher is built inside `EcsRunTaskOperator._get_task_log_fetcher()` /
`BatchOperator._get_batch_log_fetcher()` with `logger=self.log` hard-coded, and
both are private.
I would like to add an optional per-event hook, so an operator subclass or a
framework built on these operators can react to what the container prints while
it runs, without overriding private methods.
Proposed shape, additive and defaulting to today's behaviour:
```python
EcsRunTaskOperator(
...,
on_log_event=my_callable, # Callable[[dict], None] | None = None
)
```
forwarded to `AwsTaskLogFetcher`, which would call it for each CloudWatch
event it already reads (the raw `{"timestamp": ..., "message": ...}` dict),
next to the existing `self.logger.log(...)`. Nothing changes when it is not
set. The same parameter on `BatchOperator` would cover the Batch path, since
both build the same fetcher.
Alternatives I considered:
- A callback protocol class, as `KubernetesPodOperatorCallback` does for
`KubernetesPodOperator` (`on_pod_creation`, `on_pod_completion`, ...,
`progress_callback` per log line). That is the established precedent in the
providers, and a natural fit if you would rather have one extensible surface
than a single parameter; it is also a much larger API to commit to.
- Letting the caller pass the `logger` the fetcher receives. Smallest
change, but it overloads "logger" with "event sink" and gives the callback no
typed access to the event.
I am happy to implement whichever shape you prefer, or to drop the idea if
you would rather keep the surface closed.
### Use case/motivation
dbt containers emit structured JSON events on stdout (`dbt build
--log-format json`), one per model as it finishes.
[astronomer-cosmos](https://github.com/astronomer/astronomer-cosmos) turns
those events into per-node Airflow task statuses: one container runs the whole
dbt project, and one sensor per model waits for its event. That already exists
for Kubernetes, built on `KubernetesPodOperator`'s `progress_callback`.
The ECS equivalent
([astronomer-cosmos#3000](https://github.com/astronomer/astronomer-cosmos/pull/3000))
has no such hook, so it currently overrides three private methods of
`EcsRunTaskOperator` — `_get_task_log_fetcher`, `_wait_for_task_ended` and
`_after_execution` — which is fragile for a downstream project. `on_log_event`
would remove the first of the three; the other two exist because
`AwsTaskLogFetcher` is a background thread, so a callback there can only
collect events while the operator thread does the work that needs the task
context.
More generally, anything that wants progress, metrics or structured events
out of an ECS or Batch container (rather than just having them printed to the
task log) has no supported way to get them today.
### Related issues
- #73210 / #73211, the fetcher's final read, same component.
### Are you willing to submit a PR?
- [X] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
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]
