Allow observing container log events from EcsRunTaskOperator and BatchOperator #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:

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 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) 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

• AwsTaskLogFetcher can exit without reading the log events written at the end of an ECS task or Batch job #73210 / Read the AWS task logs once more after the fetcher is stopped #73211, the fetcher's final read, same component.

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct