You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "fpiped (via GitHub)" <gi...@apache.org> on 2026/09/17 12:26:01 UTC
[I] Allow observing container log events from EcsRunTaskOperator and BatchOperator [airflow]
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: commits-unsubscribe@airflow.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Allow observing container log events from EcsRunTaskOperator and BatchOperator [airflow]
Posted by "namanjain24-sudo (via GitHub)" <gi...@apache.org>.
namanjain24-sudo commented on issue #73285:
URL: https://github.com/apache/airflow/issues/73285#issuecomment-5761337628
Some observations from the current code that may help when choosing the shape (no preference between the options from my side):
- In deferrable mode `AwsTaskLogFetcher` is not what forwards the logs for ECS: `TaskDoneTrigger._forward_logs` polls CloudWatch itself inside the triggerer, and a callable cannot be handed to a trigger since its kwargs are stored serialized. For Batch, `execute_complete` builds a fetcher and reads the last messages once after the job has ended (`get_last_log_messages(10000)`), so a callback there would receive everything at once instead of while the job runs. An `on_log_event` parameter would therefore only behave as described with `deferrable=False`, unless it takes a serializable form such as an import path, or the limitation is documented.
- `AwsTaskLogFetcher.run()` calls `self.logger.log(...)` in a loop without exception handling. An exception raised by a user callback would end that background thread without failing the task, and log fetching would silently stop, so the callback call would need to be wrapped and its errors logged.
---
Drafted-by: a Gen-AI coding assistant (no human review before posting)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Allow observing container log events from EcsRunTaskOperator and BatchOperator [airflow]
Posted by "fpiped (via GitHub)" <gi...@apache.org>.
fpiped commented on issue #73285:
URL: https://github.com/apache/airflow/issues/73285#issuecomment-5714714805
Thanks @AbarnaaSree, good to hear it is useful to you as well.
I opened this intending to implement it myself: I have the repo set up locally and an open PR touching the same file (#73211), and I was waiting for a maintainer to weigh in on the API shape before writing any code, since this adds public API to two widely used operators.
That said, I would rather work on it together than in parallel. My suggestion: wait for a maintainer to pick the shape (the `on_log_event` parameter, or a `KubernetesPodOperatorCallback`-style protocol), then split it, for example one of us takes the fetcher and operator wiring and the other the tests and documentation. If you prefer to drive the PR once the shape is settled, that is fine with me too and I will help and review.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Allow observing container log events from EcsRunTaskOperator and BatchOperator [airflow]
Posted by "AbarnaaSree (via GitHub)" <gi...@apache.org>.
AbarnaaSree commented on issue #73285:
URL: https://github.com/apache/airflow/issues/73285#issuecomment-5714673812
Hi @fpiped, this looks like a useful addition, especially for downstream integrations such as Astronomer Cosmos.
I’d be interested in working on this issue. Before I start, would you be okay with me taking this up and submitting a PR?
I’m also happy to follow whichever API shape you and the maintainers prefer, whether that’s the proposed `on_log_event` callback or an alternative based on the existing provider patterns.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Allow observing container log events from EcsRunTaskOperator and BatchOperator [airflow]
Posted by "AbarnaaSree (via GitHub)" <gi...@apache.org>.
AbarnaaSree commented on issue #73285:
URL: https://github.com/apache/airflow/issues/73285#issuecomment-5714741141
Thanks @fpiped, that works for me.
I’m happy to wait for the maintainer’s direction on the API shape and then split the work with you. Once the approach is settled, I can take the tests and documentation, while you handle the fetcher/operator wiring.
Please feel free to tag me when there’s an update from the maintainers, and I’ll pick up the corresponding part.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org