fpiped opened a new issue, #73210: URL: https://github.com/apache/airflow/issues/73210
### Under which category would you file this issue? Providers ### Apache Airflow version 3.1.8 (the code below is unchanged on `main`, commit `49674b776d`) ### What happened and how to reproduce it? `AwsTaskLogFetcher` (used by `EcsRunTaskOperator` and, through `BatchClientHook.wait_for_job`, by `BatchOperator`) can exit without ever reading the log events its container wrote at the very end, so those lines never reach the Airflow task log. The thread loop checks the stop flag at the top, then sleeps, then fetches ([`task_log_fetcher.py`](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/utils/task_log_fetcher.py)): ```python def run(self) -> None: continuation_token = AwsLogsHook.ContinuationToken() while not self.is_stopped(): time.sleep(self.fetch_interval.total_seconds()) log_events = self._get_log_events(continuation_token) ... ``` and the callers stop the fetcher as soon as the task or job has ended: ```python # EcsRunTaskOperator.execute try: self._wait_for_task_ended() finally: self.task_log_fetcher.stop() self.task_log_fetcher.join() ``` ```python # BatchClientHook.wait_for_job finally: if batch_log_fetcher: batch_log_fetcher.stop() batch_log_fetcher.join() ``` Usually the thread is sleeping when `stop()` arrives, so it wakes up, performs one more fetch and only then leaves the loop: nothing is lost. But when `stop()` arrives while the thread is inside a fetch, or in the gap between a fetch finishing and the next top-of-loop check, the loop condition is already false and the thread exits **with no read after the stop**. Everything the container wrote at the end (and everything CloudWatch ingested during that last fetch) is silently dropped from the task log. The window is small with the default `awslogs_fetch_interval` of 30 seconds, since a fetch is a small fraction of each cycle, but it is hit regularly with the short intervals used when the logs are meant to be followed while the task runs. Reproduction, driving the released `AwsTaskLogFetcher` (`apache-airflow-providers-amazon==9.36.0`) with a fake `_get_log_events` that makes the last event readable during the final fetch, and calling `stop()` at that moment, as the operators do: ```python def scenario(mod): seen, delivered = [], set() logger = mock.MagicMock(); logger.log.side_effect = lambda lvl, msg: seen.append(msg) f = mod.AwsTaskLogFetcher(log_group="g", log_stream_name="s", fetch_interval=timedelta(seconds=0.3), logger=logger) start, in_fetch = time.monotonic(), threading.Event() def fake_get_log_events(token=None): now = time.monotonic() - start out = [{"timestamp": ts, "message": m} for at, ts, m in ((0.0, 1617400267123, "early line"), (0.45, 1617400467789, "LAST LINE")) if now >= at and m not in delivered] for e in out: delivered.add(e["message"]) in_fetch.set() time.sleep(0.4) # a fetch paginates, it takes time return iter(out) f._get_log_events = fake_get_log_events t = threading.Thread(target=f.run); t.start() in_fetch.wait(5); time.sleep(0.1) f.stop() # stop() lands while the thread is inside a fetch t.join(timeout=10) return seen ``` ``` provider 9.36.0 forwarded 1: ['[2021-04-02 21:51:07,123] early line'] ``` `LAST LINE` was readable in the log stream and is never forwarded. ### What you think should happen instead? The fetcher should perform one read **after** it has been stopped, so the events written between the last fetch and the end of the task are forwarded before the thread exits. The continuation token already guarantees that this extra read returns only events not yet seen, and `AwsLogsHook.get_log_events` returns as soon as the stream is exhausted, so the cost is one additional `get_log_events` call at thread exit. ### Operating System any ### Versions of Apache Airflow Providers apache-airflow-providers-amazon 9.36.0, and `main` at `49674b776d` ### Deployment Other ### Deployment details Reproduced against the released provider; the code path is identical on `main`. ### Anything else? Both `EcsRunTaskOperator` (with `awslogs_group` / `awslogs_stream_prefix` set) and `BatchOperator` (with `awslogs_enabled=True`) are affected, since both stop the fetcher in a `finally` block as soon as the task or job ends. ### Are you willing to submit 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]
