Skip to content

Rework SUPERVISOR_COMMS class to have log stored as a attribute #66694

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Prab-27 wants to merge 1 commit into from
Closed

Rework SUPERVISOR_COMMS class to have log stored as a attribute #66694

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions task-sdk/src/airflow/sdk/execution_time/task_runner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2085,7 +2085,7 @@ def main():
log = structlog.get_logger(logger_name="task")

global SUPERVISOR_COMMS
SUPERVISOR_COMMS = CommsDecoder[ToTask, ToSupervisor](log=log)
SUPERVISOR_COMMS = CommsDecoder[ToTask, ToSupervisor]()

@kaxil kaxil Jun 15, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This isn't quite a no-op. The local log two lines up is structlog.get_logger(logger_name="task"), but the class default is factory=structlog.get_logger, which builds an unnamed logger. CommsDecoder.log is used in _from_frame for self.log.exception("Unable to decode message"), so after this change that decode-error log loses its logger=task binding (the logger_name processor reads logger_name off the bound context). It's a minor, error-path-only regression, but it does mean the cleanup changes behavior rather than just dropping a redundant arg.

Worth noting #44615 asked to give the class a log attribute with a default so callers don't have to pass one, and that already landed in #51699. Here main() does have a meaningful named logger on hand, and callback_supervisor.py still passes log=get_logger(logger_name="callback_runner") for the same reason. Keeping log=log at this site preserves the task name and matches that pattern.

@Prab-27 Prab-27 Jun 23, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes I also confused for this - now got it ! Thanks @kaxil !

I have checked on my side that the issue requirements are satisfied
Would you like me to give any other tasks to do in this or I'll close this PR and will close this issue as completed
Would love to hear your thoughts @uranusjr !

@uranusjr uranusjr Jul 1, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It can be functools.partial(structlog.get_logger, "logger name") here (with the logger name being the CommsDecoder’s import path)

@uranusjr uranusjr Jul 1, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But yes let’s keep the explicit log parameters.


stats.initialize(
factory=stats_utils.get_stats_factory(),
Expand Down Expand Up @@ -2168,7 +2168,7 @@ def reinit_supervisor_comms() -> None:

fd = int(os.environ.get("__AIRFLOW_SUPERVISOR_FD", "0"))

SUPERVISOR_COMMS = CommsDecoder[ToTask, ToSupervisor](log=log, socket=socket.socket(fileno=fd))
SUPERVISOR_COMMS = CommsDecoder[ToTask, ToSupervisor](socket=socket.socket(fileno=fd))

logs = SUPERVISOR_COMMS.send(ResendLoggingFD())
if isinstance(logs, SentFDs):
Expand Down
Loading