Fix remote processor injection happening before dictConfig runs in configure_logging - #66633
Fix remote processor injection happening before dictConfig runs in configure_logging#66633korex-f wants to merge 2 commits into
Conversation
|
I can confirm that this works in our ECS environment. Thank you! |
|
You're very much welcome. |
Good point, I have added tests to test_cloudwatch_task_handler.py covering the self-healing handler property (verifies a new handler is created when shutting_down=True) and the dynamic self.handler access in the processors closure. |
o-nikolas
left a comment
There was a problem hiding this comment.
Looks reasonable to me now, but I still think @ashb or @amoghrajesh should have a look
|
@korex-f — There are 2 unresolved review threads on this PR from @o-nikolas. Could you either push a fix or reply in each thread explaining why the feedback doesn't apply? Once you believe the feedback is addressed, mark the thread as resolved so the reviewer isn't re-pinged needlessly. Thanks! Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you. |
seanghaeli
left a comment
There was a problem hiding this comment.
Verified end-to-end this fixes remote logging. Thanks @korex-f this issue has been floating around for a while.
904f1ea to
ff723d1
Compare
1fanwang
left a comment
There was a problem hiding this comment.
Thanks for addressing the comments, lgtm!
ff723d1 to
93a67a7
Compare
|
@korex-f — A reviewer (@ashb) has requested changes on this PR, so I've removed the Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you. |
|
Thanks for the review, @ashb. I have addressed the requested changes, pushed the fixes, and re-requested review. Please do take another look when you have the chance. |
|
Hey, running into this exact issue on Airflow 3.2.2 with KubernetesExecutor — task pods emit WatchtowerWarning: Received message after logging system shutdown and logs never make it to CloudWatch. Any idea on ETA for merge and which provider version will ship the fix? Also is there any interim workaround while we wait? Happy to test if there's a pre-release available. |
There was a problem hiding this comment.
Thanks for fixing this and I verify the root case as well. The overall direction LGTM.
However, this PR touches both Task-SDK and provider.
IIRC, the current convention is separating PRs of Task-SDK and provider. So we need to make this PR Task-SDK only and raise another one for provider or vice versa.
Any idea on ETA for merge and which provider version will ship the fix? Also is there any interim workaround while we wait?
If this PR doesn't make it to the 3.3 release (since it change the Task-SDK). Then I will merge the #68779 which solve the exact issue in different direction with provider-only changes (and I verify on k8s executor setup) -> the next AWS provider version should solve this problem.
If this PR catches 3.3 release -> using Airflow 3.3 plus the next AWS provider version could solve.
|
Thanks for sharing #68779 @jason810496, having read through it, I agree it's the cleaner approach to the provider side. The _closed flag correctly distinguishes dictConfig mid-task closures from genuine teardown, which is something my approach didn't handle. Given this, I'd like to scope this PR down to task-sdk only:
That means reverting all provider changes from this PR and letting #68779 carry those. Does that scope make sense @ashb? Happy to also drop the close() protocol addition from RemoteLogIO per your comment, since #68779 handles the provider side without needing it. |
jason810496
left a comment
There was a problem hiding this comment.
Happy to also drop the close() protocol addition from RemoteLogIO per your comment
Yes, I think we should make this PR targeting task-sdk changes only. (more specific, only change task-sdk/src/airflow/sdk/execution_time/supervisor.py and task-sdk/src/airflow/sdk/log.py)
|
Addressed all outstanding review comments:
|
|
Can the provider related changes be split into a dedicated PR? |
Done! |
|
@jason810496 and @ferruzzi, does this look good to approve/merge for y'all? @ashb it looks like your feedback was addressed, but you "requested changes" which makes me think you saw something very serious, are you happy with this PR now? |
You may not have pushed that, there are still provider changes in the current diff: Edit to add: the whitespace change in supervisor.py that you said was done is also not there. |
921eb9d to
f6d0f20
Compare
Sorry for the confusion, the revert was there but spread across multiple commits, making it look like provider changes were still present in the per-commit view. The net diff against upstream/main now only touches task-sdk/src/airflow/sdk/execution_time/supervisor.py and task-sdk/src/airflow/sdk/log.py. No provider files are modified. |
jason810496
left a comment
There was a problem hiding this comment.
Here're final nits that would appreciate be addressed before merge, thanks.
| def send_msg( | ||
| self, | ||
| msg: BaseModel | None, | ||
| request_id: int, | ||
| error: ErrorResponse | None = None, | ||
| **dump_opts, | ||
| self, msg: BaseModel | None, request_id: int, error: ErrorResponse | None = None, **dump_opts | ||
| ): |
There was a problem hiding this comment.
Would prefer not change this part.
| if (remote_handler := load_remote_log_handler()) is not None: | ||
| _close_remote_log_handler(remote_handler) | ||
|
|
||
| if log_file_descriptor is not None: | ||
| with contextlib.suppress(Exception): | ||
| log_file_descriptor.close() |
There was a problem hiding this comment.
Should we switch the order of closing log_file_descriptor and the remote log handler? IIUC, closing the file first then flush the renaming logs to remote IO makes more sense to me.
| def _close_remote_log_handler(handler: RemoteLogIO) -> None: | ||
| """ | ||
| Close the remote log handler explicitly after all task log messages have been drained from the subprocess pipe. | ||
|
|
||
| Called after process.wait() returns, before process exit triggers | ||
| logging.shutdown(). This ensures the remote handler's internal batch | ||
| queue is flushed before the process tears down. For example, the AWS | ||
| CloudWatch logger will emit: | ||
|
|
||
| WatchtowerWarning: "Received message after logging system shutdown" | ||
|
|
||
| if this is not done before process exit. | ||
| """ | ||
| try: | ||
| if hasattr(handler, "close"): | ||
| handler.close() | ||
| except Exception: | ||
| log.warning("Failed to close remote log handler", exc_info=True) |
There was a problem hiding this comment.
| def _close_remote_log_handler(handler: RemoteLogIO) -> None: | |
| """ | |
| Close the remote log handler explicitly after all task log messages have been drained from the subprocess pipe. | |
| Called after process.wait() returns, before process exit triggers | |
| logging.shutdown(). This ensures the remote handler's internal batch | |
| queue is flushed before the process tears down. For example, the AWS | |
| CloudWatch logger will emit: | |
| WatchtowerWarning: "Received message after logging system shutdown" | |
| if this is not done before process exit. | |
| """ | |
| try: | |
| if hasattr(handler, "close"): | |
| handler.close() | |
| except Exception: | |
| log.warning("Failed to close remote log handler", exc_info=True) |
|
|
||
| # TODO: Use logging providers to handle the chunked upload for us etc. | ||
| if (remote_handler := load_remote_log_handler()) is not None: | ||
| _close_remote_log_handler(remote_handler) |
There was a problem hiding this comment.
The scope and the purpose of the function is small and clear enough, we don't need the additional function for this. I'd like to collapse it back.
| _close_remote_log_handler(remote_handler) | |
| try: | |
| if hasattr(handler, "close"): | |
| handler.close() | |
| except Exception: | |
| log.warning("Failed to close remote log handler", exc_info=True) |
|
Picking up on @o-nikolas's merge question -- I think this needs a second look at scope before it goes in, because a fair bit of it landed via other PRs while this was open:
Given that, the supervisor.py side here (explicit @korex-f -- is there a failure mode left that the three merged PRs don't cover? If the log.py reorder is the remaining value, it may be worth narrowing this to just that (with a test); otherwise it looks mostly superseded now. Not trying to discount the work here, just want to make sure we're not re-landing something. |
Thanks for the detailed audit @kaxil — really helpful context. You're right that #68779, #67935, and #68370 together cover the provider-layer symptom. Looking at what's left: log.py reorder (Bug 1): supervisor.py What would you prefer? |
jason810496
left a comment
There was a problem hiding this comment.
Thank you for the update.
is there a failure mode left that the three merged PRs don't cover?
Yes log.py reorder (Bug 1): getattr(remote, "processors") was called before dictConfig ran, which meant dictConfig's _clearExistingHandlers closed the just-built watchtower handler before any task log was emitted. statement is correct.
I'd like to keep the only the log.py changes as that is the key point to resolve the unexpected .close on logging handler. The supervisors.py change is more "nice to have" but it might introduce further regression. Let's keep the change minimal if possible, thanks.
Thanks for the clarity. Reverting all supervisor.py changes now, this PR will only contain the log.py reorder (Bug 1). supervisor.py can be revisited in a follow-up if needed. |
jason810496
left a comment
There was a problem hiding this comment.
Hi @korex-f,
Thanks for follow-up on this.
We need to do the following items before merge.
- Add unit test to exercise the patch itself
- Fix the CI failure
- Rephrase the PR title and the description
Thanks.
Added a regression test in test_log.py that verifies load_remote_log_handler() is called only after the inner configure_logging() (dictConfig) returns. The CI failure in test_inject_parent_job_info_with_resume_on_retry is unrelated, it's a Glue operator test making a real network call that times out, last touched in #64513. Happy to re-trigger CI to confirm it's a flake. |
…nfigure_logging In configure_logging(), getattr(remote, 'processors') was called before dictConfig() ran. dictConfig() calls _clearExistingHandlers() which closes every handler in logging._handlerList — including the remote handler built moments earlier. This caused CloudWatch/Watchtower logs to be silently dropped when using the Task SDK with remote logging on ECS/Kubernetes workers. Fix: move the remote processor injection to after dictConfig() has run via a second structlog.configure() call. Also gate on not sending_to_supervisor to avoid unmasked events from the task subprocess, and add a None default to getattr() to handle third-party RemoteLogIO objects. Provider-side fix: apache#68779 Closes apache#66475
608b35f to
ec31091
Compare
…nfigure_logging In configure_logging(), getattr(remote, 'processors') was called before dictConfig() ran. dictConfig() calls _clearExistingHandlers() which closes every handler in logging._handlerList — including the remote handler built moments earlier. This caused CloudWatch/Watchtower logs to be silently dropped when using the Task SDK with remote logging on ECS/Kubernetes workers. Fix: move the remote processor injection to after dictConfig() has run via a second structlog.configure() call. Also gate on not sending_to_supervisor to avoid unmasked events from the task subprocess, and add a None default to getattr() to handle third-party RemoteLogIO objects. Closes apache#66475
jason810496
left a comment
There was a problem hiding this comment.
#70938 - fix the OpenAI provider CI failure. Rebasing on top of the latest main should fix. Thanks.
jason810496
left a comment
There was a problem hiding this comment.
LGTM overall now, thanks.
| # Accessing remote.processors triggers creation of the remote handler | ||
| # via a cached_property. The configure_logging() call below runs dictConfig() internally, |
There was a problem hiding this comment.
We don't need to mention the cached_property workaround anymore. Since this PR address the root cause.
| # Accessing remote.processors triggers creation of the remote handler | |
| # via a cached_property. The configure_logging() call below runs dictConfig() internally, | |
| # The configure_logging() call below runs dictConfig() internally, |
| assert events[0]["log_level"] == "warning" | ||
| assert events[0]["ti_id"] == str(ti.id) | ||
| assert events[0]["exc_info"] is boom |
There was a problem hiding this comment.
nit: How about asserting with events[0] == {"log_level": ..., "ti_id": ...} style?
| assert events[0]["log_level"] == "warning" | ||
| assert events[0]["ti_id"] == str(ti.id) | ||
| assert events[0]["log_relative_path"] == relative.as_posix() | ||
| assert events[0]["exc_info"] is boom |
Problem
In
configure_logging(),getattr(remote, "processors")was called beforedictConfig()ran.dictConfig()calls_clearExistingHandlers()which closes every handler registered inlogging._handlerList, including the watchtower handler built moments earlier. This caused CloudWatch/Watchtower logs to be silently dropped when using the Task SDK with remote logging.Fix
Move the remote processor injection to after
dictConfig()has run. The handler is now built into a clean logging state and cannot be killed prematurely.Also gate the injection on
not sending_to_supervisorto avoid unmasked events from the task subprocess, and add aNonedefault togetattrto handle third-partyRemoteLogIOobjects missing theprocessorsattribute.Related
Provider-side fix: #68779
closes: #66475
Important
🛠️ Maintainer triage note for @korex-f · by
@potiuk· 2026-07-02 17:46 UTCSome review feedback from
@jason810496is waiting on you:@jason810496need a reply or a fix.The ball is in your court — you've been assigned to this PR. Reply or push a fix in each thread, then mark them resolved.
Automated triage — may be imperfect; a maintainer takes the next look.