You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/12/18 08:52:28 UTC
[GitHub] [airflow] dstandish opened a new pull request, #28440: Propagate logs to stdout when in k8s executor pod
dstandish opened a new pull request, #28440:
URL: https://github.com/apache/airflow/pull/28440
This is necessary because file task handler reads from pod logs (i.e. stdout) when pod is running.
Previously we were not propagating task logs, presumably to avoid duplicating entries, because we had copied the handler to root. However, if we just remove the handler from task, we can safely enable propagation, since there won't be multiple task handlers floating around.
--
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
[GitHub] [airflow] dstandish commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1055110103
##########
airflow/cli/commands/task_command.py:
##########
@@ -280,38 +281,80 @@ def _extract_external_executor_id(args) -> str | None:
@contextmanager
-def _capture_task_logs(ti: TaskInstance) -> Generator[None, None, None]:
+def _move_task_handlers_to_root(ti: TaskInstance) -> Generator[None, None, None]:
"""
- Manage logging context for a task run.
-
- - Replace the root logger configuration with the airflow.task configuration
- so we can capture logs from any custom loggers used in the task.
-
- - Redirect stdout and stderr to the task instance log, as INFO and WARNING
- level messages, respectively.
+ Move handlers for task logging to root logger.
+ We want anything logged during task run to be propagated to task log handlers.
+ If running in a k8s executor pod, also keep the stream handler on root logger
+ so that logs are still emitted to stdout.
"""
modify = not settings.DONOT_MODIFY_HANDLERS
+ did_modify = False
+ is_k8s_executor_pod = os.environ.get("AIRFLOW_IS_K8S_EXECUTOR_POD")
if modify:
- root_logger, task_logger = logging.getLogger(), logging.getLogger("airflow.task")
+ root_logger = logging.getLogger()
+ root_level = root_logger.level
+ task_logger = ti.log
+ task_propagate = task_logger.propagate
+ task_handlers = task_logger.handlers.copy()
+
+ # if there are no task handlers, then we should not do anything
+ # because either the handlers were already moved by the LocalTaskJob
+ # invocation of task_run (which wraps the --raw invocation), or
+ # user is doing something custom / unexpected
+ if task_handlers:
Review Comment:
was just focused on getting this to work up to this point without worrying too much about quality :)
--
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
[GitHub] [airflow] dstandish commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1060445184
##########
airflow/utils/log/file_task_handler.py:
##########
@@ -62,19 +62,29 @@ def __init__(self, base_log_folder: str, filename_template: str | None = None):
# handler, not the one that calls super()__init__.
stacklevel=(2 if type(self) == FileTaskHandler else 3),
)
+ self.maintain_propagate: bool = False
def set_context(self, ti: TaskInstance) -> None | SetContextPropagate:
"""
Provide task_instance context to airflow task handler.
+ Generally speaking returns None. But if attr `maintain_propagate` has
Review Comment:
Not mocking and not replacing entire logging config with a fake one is how I discovered we were creating more connections unnecessarily
--
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
[GitHub] [airflow] dstandish commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1060415545
##########
airflow/cli/commands/task_command.py:
##########
@@ -281,38 +283,62 @@ def _extract_external_executor_id(args) -> str | None:
@contextmanager
-def _capture_task_logs(ti: TaskInstance) -> Generator[None, None, None]:
+def _move_task_handlers_to_root(ti: TaskInstance) -> Generator[None, None, None]:
"""
- Manage logging context for a task run.
+ Move handlers for task logging to root logger.
- - Replace the root logger configuration with the airflow.task configuration
- so we can capture logs from any custom loggers used in the task.
+ We want anything logged during task run to be propagated to task log handlers.
+ If running in a k8s executor pod, also keep the stream handler on root logger
+ so that logs are still emitted to stdout.
+ """
+ # nothing to do
+ if not ti.log.handlers or settings.DONOT_MODIFY_HANDLERS:
+ yield
+ return
+
+ def get_console_handler(logger):
+ for h in logger.handlers:
+ if h.name == "console":
+ return h
+
+ def ensure_handler(logger, handler):
+ if not handler:
+ return
+ if handler not in logger.handlers:
+ logger.addHandler(handler)
+
+ # Move task handlers to root and reset task logger and restore original logger settings after exit.
+ # If k8s executor, we need to ensure that root logger has a console handler, so that
+ # task logs propagate to stdout (this is how webserver retrieves them while task is running).
+ root_logger = logging.getLogger()
+ console_handler = get_console_handler(root_logger)
Review Comment:
done
--
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
[GitHub] [airflow] uranusjr commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
uranusjr commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1055103624
##########
airflow/cli/commands/task_command.py:
##########
@@ -280,38 +281,80 @@ def _extract_external_executor_id(args) -> str | None:
@contextmanager
-def _capture_task_logs(ti: TaskInstance) -> Generator[None, None, None]:
+def _move_task_handlers_to_root(ti: TaskInstance) -> Generator[None, None, None]:
"""
- Manage logging context for a task run.
-
- - Replace the root logger configuration with the airflow.task configuration
- so we can capture logs from any custom loggers used in the task.
-
- - Redirect stdout and stderr to the task instance log, as INFO and WARNING
- level messages, respectively.
+ Move handlers for task logging to root logger.
+ We want anything logged during task run to be propagated to task log handlers.
+ If running in a k8s executor pod, also keep the stream handler on root logger
+ so that logs are still emitted to stdout.
"""
modify = not settings.DONOT_MODIFY_HANDLERS
+ did_modify = False
+ is_k8s_executor_pod = os.environ.get("AIRFLOW_IS_K8S_EXECUTOR_POD")
if modify:
- root_logger, task_logger = logging.getLogger(), logging.getLogger("airflow.task")
+ root_logger = logging.getLogger()
+ root_level = root_logger.level
+ task_logger = ti.log
+ task_propagate = task_logger.propagate
+ task_handlers = task_logger.handlers.copy()
+
+ # if there are no task handlers, then we should not do anything
+ # because either the handlers were already moved by the LocalTaskJob
+ # invocation of task_run (which wraps the --raw invocation), or
+ # user is doing something custom / unexpected
+ if task_handlers:
Review Comment:
I think it’d be good enough if we simply do a
```python
if not task_handlers:
yield
return
```
and get rid of the `did_modify` flag? This also allows everything after this to be unindented one level.
(I’m not even sure if the extra `yield` is necessary; probably?)
--
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
[GitHub] [airflow] malthe commented on pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
malthe commented on PR #28440:
URL: https://github.com/apache/airflow/pull/28440#issuecomment-1367271199
@dstandish I think an abstraction point in the executor framework would be better rather than the `if`-statements in this changeset. That is, we should make the executor pluggable in terms of supporting this logging capability.
In terms of actual implementation, rather than adjusting log handlers at runtime, perhaps an internal pub/sub mechanism could be devised.
--
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
[GitHub] [airflow] dstandish merged pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish merged PR #28440:
URL: https://github.com/apache/airflow/pull/28440
--
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
[GitHub] [airflow] dstandish commented on pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on PR #28440:
URL: https://github.com/apache/airflow/pull/28440#issuecomment-1368185546
Ok @malthe i have simplified the code, if you would like to have another look:
https://github.com/apache/airflow/blob/0ade033eaa9feb52aecbe12d1b91131a27a80b4b/airflow/cli/commands/task_command.py#L322-L326
--
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
[GitHub] [airflow] dstandish commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1060411097
##########
airflow/utils/log/file_task_handler.py:
##########
@@ -62,19 +62,29 @@ def __init__(self, base_log_folder: str, filename_template: str | None = None):
# handler, not the one that calls super()__init__.
stacklevel=(2 if type(self) == FileTaskHandler else 3),
)
+ self.maintain_propagate: bool = False
def set_context(self, ti: TaskInstance) -> None | SetContextPropagate:
"""
Provide task_instance context to airflow task handler.
+ Generally speaking returns None. But if attr `maintain_propagate` has
Review Comment:
i've updated it to mark it as `:meta private:` to indicate should not be used by users / not part of public interface / can be removed at any time.
--
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
[GitHub] [airflow] dstandish commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1064886465
##########
tests/cli/commands/test_task_command.py:
##########
@@ -69,6 +72,15 @@ def move_back(old_path, new_path):
shutil.move(new_path, old_path)
+@pytest.fixture()
+def freeze_time():
+ timestamp = "2022-06-10T12:02:44+00:00"
+ freezer = time_machine.travel(timestamp, tick=False)
+ freezer.start()
+ yield
+ freezer.stop()
Review Comment:
yup, removed
--
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
[GitHub] [airflow] dstandish commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1055109812
##########
airflow/cli/commands/task_command.py:
##########
@@ -280,38 +281,80 @@ def _extract_external_executor_id(args) -> str | None:
@contextmanager
-def _capture_task_logs(ti: TaskInstance) -> Generator[None, None, None]:
+def _move_task_handlers_to_root(ti: TaskInstance) -> Generator[None, None, None]:
"""
- Manage logging context for a task run.
-
- - Replace the root logger configuration with the airflow.task configuration
- so we can capture logs from any custom loggers used in the task.
-
- - Redirect stdout and stderr to the task instance log, as INFO and WARNING
- level messages, respectively.
+ Move handlers for task logging to root logger.
+ We want anything logged during task run to be propagated to task log handlers.
+ If running in a k8s executor pod, also keep the stream handler on root logger
+ so that logs are still emitted to stdout.
"""
modify = not settings.DONOT_MODIFY_HANDLERS
+ did_modify = False
+ is_k8s_executor_pod = os.environ.get("AIRFLOW_IS_K8S_EXECUTOR_POD")
if modify:
- root_logger, task_logger = logging.getLogger(), logging.getLogger("airflow.task")
+ root_logger = logging.getLogger()
+ root_level = root_logger.level
+ task_logger = ti.log
+ task_propagate = task_logger.propagate
+ task_handlers = task_logger.handlers.copy()
+
+ # if there are no task handlers, then we should not do anything
+ # because either the handlers were already moved by the LocalTaskJob
+ # invocation of task_run (which wraps the --raw invocation), or
+ # user is doing something custom / unexpected
+ if task_handlers:
Review Comment:
yeah that should work thanks
--
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
[GitHub] [airflow] dstandish commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1061155406
##########
airflow/utils/log/file_task_handler.py:
##########
@@ -62,19 +62,29 @@ def __init__(self, base_log_folder: str, filename_template: str | None = None):
# handler, not the one that calls super()__init__.
stacklevel=(2 if type(self) == FileTaskHandler else 3),
)
+ self.maintain_propagate: bool = False
def set_context(self, ti: TaskInstance) -> None | SetContextPropagate:
"""
Provide task_instance context to airflow task handler.
+ Generally speaking returns None. But if attr `maintain_propagate` has
+ been set to propagate, then returns sentinel MAINTAIN_PROPAGATE. This
+ has the effect of overriding the default behavior to set `propagate`
+ to False whenever set_context is called. At time of writing, this
+ functionality is only used in unit testing.
+
:param ti: task instance object
"""
- local_loc = self._init_file(ti)
- self.handler = NonCachingFileHandler(local_loc, encoding="utf-8")
- if self.formatter:
- self.handler.setFormatter(self.formatter)
- self.handler.setLevel(self.level)
- return None
+ from airflow.models.taskinstance import TaskInstance
+
+ if isinstance(ti, TaskInstance):
+ local_loc = self._init_file(ti)
+ self.handler = NonCachingFileHandler(local_loc, encoding="utf-8")
+ if self.formatter:
+ self.handler.setFormatter(self.formatter)
+ self.handler.setLevel(self.level)
Review Comment:
well... i guess there are degrees of hurt...
anyway, it was needed because something in existing test was mocking a TI and the actual methods don't like the mock obj
and since i restored normal logging config, now the mocked ti is hitting handler. anyway, mocked this method away so non issue now. removed the if, and tests should go green again.
--
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
[GitHub] [airflow] malthe commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
malthe commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1060327281
##########
airflow/cli/commands/task_command.py:
##########
@@ -280,38 +281,85 @@ def _extract_external_executor_id(args) -> str | None:
@contextmanager
-def _capture_task_logs(ti: TaskInstance) -> Generator[None, None, None]:
+def _move_task_handlers_to_root(ti: TaskInstance) -> Generator[None, None, None]:
"""
- Manage logging context for a task run.
+ Move handlers for task logging to root logger.
- - Replace the root logger configuration with the airflow.task configuration
- so we can capture logs from any custom loggers used in the task.
+ We want anything logged during task run to be propagated to task log handlers.
+ If running in a k8s executor pod, also keep the stream handler on root logger
+ so that logs are still emitted to stdout.
+ """
+ # if there are no task handlers, then we should not do anything
+ # because either the handlers were already moved by the LocalTaskJob
+ # invocation of task_run (which wraps the --raw invocation), or
+ # user is doing something custom / unexpected
+ if not ti.log.handlers or settings.DONOT_MODIFY_HANDLERS:
+ yield
+ return
+ is_k8s_executor_pod = os.environ.get("AIRFLOW_IS_K8S_EXECUTOR_POD")
+ root_logger = logging.getLogger()
+ root_level = root_logger.level
+ task_logger = ti.log
+ task_propagate = task_logger.propagate
+ task_handlers = task_logger.handlers.copy()
+
+ # these are already copied to root logger, so remove and propagate
+ # (this ensures they are not handled more than they need to be)
+ # we'll add them back later
+ for h in task_logger.handlers[:]:
+ task_logger.removeHandler(h)
+ task_logger.propagate = True
+ root_logger.setLevel(task_logger.level)
+ root_handlers = root_logger.handlers.copy()
+ root_logger.handlers[:] = task_handlers
+
+ # task log handler reads from k8s pod logs when pod still running
+ # so we need to keep the console handler and set it to not respect
+ # redirection because we read live pod logging from container stdout
+ if is_k8s_executor_pod:
+ for h in root_handlers:
+ if isinstance(h, RedirectStdHandler):
+ root_logger.addHandler(h)
+ # "redirect stdout" context managers work by changing what lives at
+ # sys.stdout. If respect_redirection is True, then we reference whatever
+ # "sys.stdout" is at runtime. If False, this handler will store the value
+ # of sys.stdout at handler *init* time, and stream to *that* instead of whatever
+ # the current value is.
+ h.respect_redirection = False
+ break
+ try:
+ yield
+ finally:
+ task_logger.propagate = task_propagate
+ root_logger.setLevel(root_level)
+ root_logger.handlers[:] = root_handlers
+ task_logger.handlers[:] = task_handlers
+ if is_k8s_executor_pod:
+ for h in root_handlers:
+ if isinstance(h, RedirectStdHandler):
+ h.respect_redirection = True
- - Redirect stdout and stderr to the task instance log, as INFO and WARNING
- level messages, respectively.
+@contextmanager
+def _redirect_stdout_to_ti_log(ti: TaskInstance) -> Generator[None, None, None]:
"""
- modify = not settings.DONOT_MODIFY_HANDLERS
- if modify:
- root_logger, task_logger = logging.getLogger(), logging.getLogger("airflow.task")
+ Redirect stdout to ti logger.
- orig_level = root_logger.level
- root_logger.setLevel(task_logger.level)
- orig_handlers = root_logger.handlers.copy()
- root_logger.handlers[:] = task_logger.handlers
Review Comment:
I didn't quite understand what "stdout redirection" means – is that within Python or is that something happening outside of Python (in a shell script perhaps) – ?
When I think about this problem, it seems that there is an orthogonal concern in the logging setup of Airflow which is whether or not the task logger (during execution) should be emitted to the stdout stream.
That's something you could want in any situation, not just K8S.
--
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
[GitHub] [airflow] dstandish commented on pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on PR #28440:
URL: https://github.com/apache/airflow/pull/28440#issuecomment-1362457897
one thing i should add too though is....
with this change, we might actually be able to remove the maintain propagate logic entirely. indeed we might be able to remove the recursive logger walking / propagate management in set_context as well.
the reason is, since we move FTH to root (rather than copy) propagation doesn't matter, because there's nothing you can propagate to beyond root. it mattered when FTH was both at `airflow.task` and root because you could log twice.
we wanted it at task because that's what `self.log` expected. and we wanted it at root so it could capture anything logged during task run (i.e. modules not children of airflow task). but because we had it in both places, we had to be careful about propagation. so this avoids that problem.
--
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
[GitHub] [airflow] dstandish commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1060391953
##########
airflow/cli/commands/task_command.py:
##########
@@ -280,38 +281,85 @@ def _extract_external_executor_id(args) -> str | None:
@contextmanager
-def _capture_task_logs(ti: TaskInstance) -> Generator[None, None, None]:
+def _move_task_handlers_to_root(ti: TaskInstance) -> Generator[None, None, None]:
"""
- Manage logging context for a task run.
+ Move handlers for task logging to root logger.
- - Replace the root logger configuration with the airflow.task configuration
- so we can capture logs from any custom loggers used in the task.
+ We want anything logged during task run to be propagated to task log handlers.
+ If running in a k8s executor pod, also keep the stream handler on root logger
+ so that logs are still emitted to stdout.
+ """
+ # if there are no task handlers, then we should not do anything
+ # because either the handlers were already moved by the LocalTaskJob
+ # invocation of task_run (which wraps the --raw invocation), or
+ # user is doing something custom / unexpected
+ if not ti.log.handlers or settings.DONOT_MODIFY_HANDLERS:
+ yield
+ return
+ is_k8s_executor_pod = os.environ.get("AIRFLOW_IS_K8S_EXECUTOR_POD")
+ root_logger = logging.getLogger()
+ root_level = root_logger.level
+ task_logger = ti.log
+ task_propagate = task_logger.propagate
+ task_handlers = task_logger.handlers.copy()
+
+ # these are already copied to root logger, so remove and propagate
+ # (this ensures they are not handled more than they need to be)
+ # we'll add them back later
+ for h in task_logger.handlers[:]:
+ task_logger.removeHandler(h)
+ task_logger.propagate = True
+ root_logger.setLevel(task_logger.level)
+ root_handlers = root_logger.handlers.copy()
+ root_logger.handlers[:] = task_handlers
+
+ # task log handler reads from k8s pod logs when pod still running
+ # so we need to keep the console handler and set it to not respect
+ # redirection because we read live pod logging from container stdout
+ if is_k8s_executor_pod:
+ for h in root_handlers:
+ if isinstance(h, RedirectStdHandler):
+ root_logger.addHandler(h)
+ # "redirect stdout" context managers work by changing what lives at
+ # sys.stdout. If respect_redirection is True, then we reference whatever
+ # "sys.stdout" is at runtime. If False, this handler will store the value
+ # of sys.stdout at handler *init* time, and stream to *that* instead of whatever
+ # the current value is.
+ h.respect_redirection = False
+ break
+ try:
+ yield
+ finally:
+ task_logger.propagate = task_propagate
+ root_logger.setLevel(root_level)
+ root_logger.handlers[:] = root_handlers
+ task_logger.handlers[:] = task_handlers
+ if is_k8s_executor_pod:
+ for h in root_handlers:
+ if isinstance(h, RedirectStdHandler):
+ h.respect_redirection = True
- - Redirect stdout and stderr to the task instance log, as INFO and WARNING
- level messages, respectively.
+@contextmanager
+def _redirect_stdout_to_ti_log(ti: TaskInstance) -> Generator[None, None, None]:
"""
- modify = not settings.DONOT_MODIFY_HANDLERS
- if modify:
- root_logger, task_logger = logging.getLogger(), logging.getLogger("airflow.task")
+ Redirect stdout to ti logger.
- orig_level = root_logger.level
- root_logger.setLevel(task_logger.level)
- orig_handlers = root_logger.handlers.copy()
- root_logger.handlers[:] = task_logger.handlers
Review Comment:
> I didn't quite understand what "stdout redirection" means – is that within Python or is that something happening outside of Python (in a shell script perhaps) – ?
i am not sure what comment you refer to but, what this means is there are stdlib helpers to take things that would ordinarily go to stdout (e.g. print) and send them to some other stream.
see `from contextlib import redirect_stdout`. it just monkey patches `sys.stdout` temporarily.
we use them (for better or worse) to redirect stdout to task logs.
our log read logic when task is running is usually, read from flask log server. most task logging is redirected to log (and therefore file) and our "console" handler respects this redirection. for obvious reasons -- the celery worker log, or local executor log would get unreasonably chatty. but in k8s executor context, there is no log server on the worker, and there is no problem with keeping the stdout, and importantly our log read logic _assumes_ everything will be forwarded to stdout -- and this is what's broken that i am here fixing.
--
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
[GitHub] [airflow] malthe commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
malthe commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1060322687
##########
airflow/cli/commands/task_command.py:
##########
@@ -281,38 +283,62 @@ def _extract_external_executor_id(args) -> str | None:
@contextmanager
-def _capture_task_logs(ti: TaskInstance) -> Generator[None, None, None]:
+def _move_task_handlers_to_root(ti: TaskInstance) -> Generator[None, None, None]:
"""
- Manage logging context for a task run.
+ Move handlers for task logging to root logger.
- - Replace the root logger configuration with the airflow.task configuration
- so we can capture logs from any custom loggers used in the task.
+ We want anything logged during task run to be propagated to task log handlers.
+ If running in a k8s executor pod, also keep the stream handler on root logger
+ so that logs are still emitted to stdout.
+ """
+ # nothing to do
+ if not ti.log.handlers or settings.DONOT_MODIFY_HANDLERS:
+ yield
+ return
+
+ def get_console_handler(logger):
+ for h in logger.handlers:
+ if h.name == "console":
+ return h
+
+ def ensure_handler(logger, handler):
+ if not handler:
+ return
+ if handler not in logger.handlers:
+ logger.addHandler(handler)
+
+ # Move task handlers to root and reset task logger and restore original logger settings after exit.
+ # If k8s executor, we need to ensure that root logger has a console handler, so that
+ # task logs propagate to stdout (this is how webserver retrieves them while task is running).
+ root_logger = logging.getLogger()
+ console_handler = get_console_handler(root_logger)
+ with LoggerMutationHelper(root_logger), LoggerMutationHelper(ti.log) as task_helper:
+ task_helper.move(root_logger)
+ if IS_K8S_EXECUTOR_POD:
Review Comment:
I think we ought to have enough abstraction in the executor interface that we don't need this `if`-clause here.
--
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
[GitHub] [airflow] malthe commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
malthe commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1060324041
##########
airflow/utils/log/file_task_handler.py:
##########
@@ -92,16 +102,17 @@ def _render_filename(self, ti: TaskInstance, try_number: int) -> str:
with create_session() as session:
dag_run = ti.get_dagrun(session=session)
template = dag_run.get_log_template(session=session).filename
- str_tpl, jinja_tpl = parse_template_string(template)
+ str_tpl, jinja_tpl = parse_template_string(template)
Review Comment:
Why is this parsed within the session context now?
--
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
[GitHub] [airflow] dstandish commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1060327384
##########
airflow/cli/commands/task_command.py:
##########
@@ -281,38 +283,62 @@ def _extract_external_executor_id(args) -> str | None:
@contextmanager
-def _capture_task_logs(ti: TaskInstance) -> Generator[None, None, None]:
+def _move_task_handlers_to_root(ti: TaskInstance) -> Generator[None, None, None]:
"""
- Manage logging context for a task run.
+ Move handlers for task logging to root logger.
- - Replace the root logger configuration with the airflow.task configuration
- so we can capture logs from any custom loggers used in the task.
+ We want anything logged during task run to be propagated to task log handlers.
+ If running in a k8s executor pod, also keep the stream handler on root logger
+ so that logs are still emitted to stdout.
+ """
+ # nothing to do
+ if not ti.log.handlers or settings.DONOT_MODIFY_HANDLERS:
+ yield
+ return
+
+ def get_console_handler(logger):
+ for h in logger.handlers:
+ if h.name == "console":
+ return h
+
+ def ensure_handler(logger, handler):
+ if not handler:
+ return
+ if handler not in logger.handlers:
+ logger.addHandler(handler)
+
+ # Move task handlers to root and reset task logger and restore original logger settings after exit.
+ # If k8s executor, we need to ensure that root logger has a console handler, so that
+ # task logs propagate to stdout (this is how webserver retrieves them while task is running).
+ root_logger = logging.getLogger()
+ console_handler = get_console_handler(root_logger)
+ with LoggerMutationHelper(root_logger), LoggerMutationHelper(ti.log) as task_helper:
+ task_helper.move(root_logger)
+ if IS_K8S_EXECUTOR_POD:
Review Comment:
a noble goal. but i would say out of scope for this PR. there already is this env var. and if executor provides an interface we can use here, we can later update to use it.
--
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
[GitHub] [airflow] uranusjr commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
uranusjr commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1060273027
##########
airflow/utils/log/file_task_handler.py:
##########
@@ -62,19 +62,29 @@ def __init__(self, base_log_folder: str, filename_template: str | None = None):
# handler, not the one that calls super()__init__.
stacklevel=(2 if type(self) == FileTaskHandler else 3),
)
+ self.maintain_propagate: bool = False
def set_context(self, ti: TaskInstance) -> None | SetContextPropagate:
"""
Provide task_instance context to airflow task handler.
+ Generally speaking returns None. But if attr `maintain_propagate` has
+ been set to propagate, then returns sentinel MAINTAIN_PROPAGATE. This
+ has the effect of overriding the default behavior to set `propagate`
+ to False whenever set_context is called. At time of writing, this
+ functionality is only used in unit testing.
+
:param ti: task instance object
"""
- local_loc = self._init_file(ti)
- self.handler = NonCachingFileHandler(local_loc, encoding="utf-8")
- if self.formatter:
- self.handler.setFormatter(self.formatter)
- self.handler.setLevel(self.level)
- return None
+ from airflow.models.taskinstance import TaskInstance
+
+ if isinstance(ti, TaskInstance):
+ local_loc = self._init_file(ti)
+ self.handler = NonCachingFileHandler(local_loc, encoding="utf-8")
+ if self.formatter:
+ self.handler.setFormatter(self.formatter)
+ self.handler.setLevel(self.level)
Review Comment:
When is this not true…?
--
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
[GitHub] [airflow] malthe commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
malthe commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1060323659
##########
airflow/utils/log/file_task_handler.py:
##########
@@ -62,19 +62,29 @@ def __init__(self, base_log_folder: str, filename_template: str | None = None):
# handler, not the one that calls super()__init__.
stacklevel=(2 if type(self) == FileTaskHandler else 3),
)
+ self.maintain_propagate: bool = False
def set_context(self, ti: TaskInstance) -> None | SetContextPropagate:
"""
Provide task_instance context to airflow task handler.
+ Generally speaking returns None. But if attr `maintain_propagate` has
Review Comment:
It might be possible to use mocking instead to avoid having functionality that's used only in unit testing.
--
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
[GitHub] [airflow] dstandish commented on pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on PR #28440:
URL: https://github.com/apache/airflow/pull/28440#issuecomment-1367525911
ok updated the description so hopefully it's a little clearer whats going on
--
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
[GitHub] [airflow] dstandish commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1059057535
##########
airflow/cli/commands/task_command.py:
##########
@@ -280,38 +281,85 @@ def _extract_external_executor_id(args) -> str | None:
@contextmanager
-def _capture_task_logs(ti: TaskInstance) -> Generator[None, None, None]:
+def _move_task_handlers_to_root(ti: TaskInstance) -> Generator[None, None, None]:
"""
- Manage logging context for a task run.
+ Move handlers for task logging to root logger.
- - Replace the root logger configuration with the airflow.task configuration
- so we can capture logs from any custom loggers used in the task.
+ We want anything logged during task run to be propagated to task log handlers.
+ If running in a k8s executor pod, also keep the stream handler on root logger
+ so that logs are still emitted to stdout.
+ """
+ # if there are no task handlers, then we should not do anything
+ # because either the handlers were already moved by the LocalTaskJob
+ # invocation of task_run (which wraps the --raw invocation), or
+ # user is doing something custom / unexpected
+ if not ti.log.handlers or settings.DONOT_MODIFY_HANDLERS:
+ yield
+ return
+ is_k8s_executor_pod = os.environ.get("AIRFLOW_IS_K8S_EXECUTOR_POD")
+ root_logger = logging.getLogger()
+ root_level = root_logger.level
+ task_logger = ti.log
+ task_propagate = task_logger.propagate
+ task_handlers = task_logger.handlers.copy()
+
+ # these are already copied to root logger, so remove and propagate
+ # (this ensures they are not handled more than they need to be)
+ # we'll add them back later
+ for h in task_logger.handlers[:]:
+ task_logger.removeHandler(h)
+ task_logger.propagate = True
+ root_logger.setLevel(task_logger.level)
+ root_handlers = root_logger.handlers.copy()
+ root_logger.handlers[:] = task_handlers
+
+ # task log handler reads from k8s pod logs when pod still running
+ # so we need to keep the console handler and set it to not respect
+ # redirection because we read live pod logging from container stdout
+ if is_k8s_executor_pod:
+ for h in root_handlers:
+ if isinstance(h, RedirectStdHandler):
+ root_logger.addHandler(h)
+ # "redirect stdout" context managers work by changing what lives at
+ # sys.stdout. If respect_redirection is True, then we reference whatever
+ # "sys.stdout" is at runtime. If False, this handler will store the value
+ # of sys.stdout at handler *init* time, and stream to *that* instead of whatever
+ # the current value is.
+ h.respect_redirection = False
+ break
+ try:
+ yield
+ finally:
+ task_logger.propagate = task_propagate
+ root_logger.setLevel(root_level)
+ root_logger.handlers[:] = root_handlers
+ task_logger.handlers[:] = task_handlers
+ if is_k8s_executor_pod:
+ for h in root_handlers:
+ if isinstance(h, RedirectStdHandler):
+ h.respect_redirection = True
- - Redirect stdout and stderr to the task instance log, as INFO and WARNING
- level messages, respectively.
+@contextmanager
+def _redirect_stdout_to_ti_log(ti: TaskInstance) -> Generator[None, None, None]:
"""
- modify = not settings.DONOT_MODIFY_HANDLERS
- if modify:
- root_logger, task_logger = logging.getLogger(), logging.getLogger("airflow.task")
+ Redirect stdout to ti logger.
- orig_level = root_logger.level
- root_logger.setLevel(task_logger.level)
- orig_handlers = root_logger.handlers.copy()
- root_logger.handlers[:] = task_logger.handlers
Review Comment:
@malthe observe, we already copy (temporarily) to root logger the thandlers from airflow.task
This causes problems because we have to have complicated propagation rules at the airflow.task logger because we leave it there.
While my solution here is a few more lines (and a lot more comments) it's not very complicated. There are two parts.
1. Instead of _copying_ the handlers to root, I _move_ them to root. We don't need them at task if they are already at root. This could ultimately allow for simplification of our propagation logic.
2. Previously we _remove_ our console handler from root at run time. Now, if we're in a k8s executor pod, I keep the console handler there.
--
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
[GitHub] [airflow] dstandish commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1064944840
##########
airflow/cli/commands/task_command.py:
##########
@@ -390,11 +400,19 @@ def task_run(args, dag=None):
log.info("Running %s on host %s", ti, hostname)
+ # IMPORTANT, have to re-configure ORM with the NullPool, otherwise, each "run" command may leave
+ # behind multiple open sleeping connections while heartbeating, which could
+ # easily exceed the database connection limit when
+ # processing hundreds of simultaneous tasks.
+ # this should be last thing before running, to reduce likelihood of an open session
+ # which can cause trouble if running process in a fork.
+ settings.reconfigure_orm(disable_connection_pool=True)
Review Comment:
i don't *think* this will fail if we remove this change. you could call it a drive by and i can remove if you like.
task runner tests failed with this change. then i was like "oh why is it just throwing away whole log config, let's try and not do that". then after leaving the FTH intact for that test, that i discovered there was an extraneous session being created in created in _render_filename. as part of troubleshooting the failures and discovering that though, also saw that this reconfigure is happening earlier than it probably ought to. should be done as late as possible i think.
--
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
[GitHub] [airflow] dstandish commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1060442177
##########
airflow/utils/log/file_task_handler.py:
##########
@@ -62,19 +62,29 @@ def __init__(self, base_log_folder: str, filename_template: str | None = None):
# handler, not the one that calls super()__init__.
stacklevel=(2 if type(self) == FileTaskHandler else 3),
)
+ self.maintain_propagate: bool = False
def set_context(self, ti: TaskInstance) -> None | SetContextPropagate:
"""
Provide task_instance context to airflow task handler.
+ Generally speaking returns None. But if attr `maintain_propagate` has
Review Comment:
Meta private affects documentation and has been used to signal not public api
--
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
[GitHub] [airflow] malthe commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
malthe commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1060425716
##########
airflow/utils/log/file_task_handler.py:
##########
@@ -62,19 +62,29 @@ def __init__(self, base_log_folder: str, filename_template: str | None = None):
# handler, not the one that calls super()__init__.
stacklevel=(2 if type(self) == FileTaskHandler else 3),
)
+ self.maintain_propagate: bool = False
def set_context(self, ti: TaskInstance) -> None | SetContextPropagate:
"""
Provide task_instance context to airflow task handler.
+ Generally speaking returns None. But if attr `maintain_propagate` has
Review Comment:
I don't really get this one. The `self.maintain_propagate` attribute is available for anyone to check as it is. The point of the method is to set the handler to refer to the provided `ti`.
--
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
[GitHub] [airflow] ashb commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1064530937
##########
airflow/cli/commands/task_command.py:
##########
@@ -390,11 +400,19 @@ def task_run(args, dag=None):
log.info("Running %s on host %s", ti, hostname)
+ # IMPORTANT, have to re-configure ORM with the NullPool, otherwise, each "run" command may leave
+ # behind multiple open sleeping connections while heartbeating, which could
+ # easily exceed the database connection limit when
+ # processing hundreds of simultaneous tasks.
+ # this should be last thing before running, to reduce likelihood of an open session
+ # which can cause trouble if running process in a fork.
+ settings.reconfigure_orm(disable_connection_pool=True)
Review Comment:
Is this related to this logging PR in anyway?
##########
tests/cli/commands/test_task_command.py:
##########
@@ -69,6 +72,15 @@ def move_back(old_path, new_path):
shutil.move(new_path, old_path)
+@pytest.fixture()
+def freeze_time():
+ timestamp = "2022-06-10T12:02:44+00:00"
+ freezer = time_machine.travel(timestamp, tick=False)
+ freezer.start()
+ yield
+ freezer.stop()
Review Comment:
This fixture doesn't seem to be used anywhere?
##########
airflow/utils/log/file_task_handler.py:
##########
@@ -62,19 +62,29 @@ def __init__(self, base_log_folder: str, filename_template: str | None = None):
# handler, not the one that calls super()__init__.
stacklevel=(2 if type(self) == FileTaskHandler else 3),
)
+ self.maintain_propagate: bool = False
+ """If true, overrides default behavior of setting propagate=False
Review Comment:
```suggestion
"""
If true, overrides default behavior of setting propagate=False
```
##########
tests/cli/commands/test_task_command.py:
##########
@@ -600,6 +612,48 @@ def test_external_executor_id_present_for_process_run_task(self, mock_local_job)
external_executor_id="ABCD12345",
)
+ @pytest.mark.parametrize("is_k8s", ["true", ""])
+ def test_logging_with_run_task_stdout_k8s_executor_pod(self, is_k8s):
+ """
+ When running task --local as k8l executor pod, all logging should make it to stdout.
Review Comment:
```suggestion
When running task --local as k8s executor pod, all logging should make it to stdout.
```
##########
airflow/utils/log/file_task_handler.py:
##########
@@ -62,19 +62,29 @@ def __init__(self, base_log_folder: str, filename_template: str | None = None):
# handler, not the one that calls super()__init__.
stacklevel=(2 if type(self) == FileTaskHandler else 3),
)
+ self.maintain_propagate: bool = False
+ """If true, overrides default behavior of setting propagate=False
Review Comment:
(The blank line is the important bit here)
##########
tests/task/task_runner/test_standard_task_runner.py:
##########
@@ -131,38 +141,34 @@ def test_notifies_about_start_and_stop(self):
)
dag = dagbag.dags.get("test_example_bash_operator")
task = dag.get_task("runme_1")
+ dag.create_dagrun(
+ run_id="test",
+ data_interval=(DEFAULT_DATE, DEFAULT_DATE),
+ state=State.RUNNING,
+ start_date=DEFAULT_DATE,
+ )
+ ti = TaskInstance(task=task, run_id="test")
+ job1 = LocalTaskJob(task_instance=ti, ignore_ti_state=True)
+ runner = StandardTaskRunner(job1)
+ runner.start()
+
+ # Wait until process sets its pgid to be equal to pid
Review Comment:
```suggestion
# Wait until process makes itself the leader of it's own process group
```
##########
tests/task/task_runner/test_standard_task_runner.py:
##########
@@ -131,38 +141,34 @@ def test_notifies_about_start_and_stop(self):
)
dag = dagbag.dags.get("test_example_bash_operator")
task = dag.get_task("runme_1")
+ dag.create_dagrun(
+ run_id="test",
+ data_interval=(DEFAULT_DATE, DEFAULT_DATE),
+ state=State.RUNNING,
+ start_date=DEFAULT_DATE,
+ )
+ ti = TaskInstance(task=task, run_id="test")
+ job1 = LocalTaskJob(task_instance=ti, ignore_ti_state=True)
+ runner = StandardTaskRunner(job1)
+ runner.start()
+
+ # Wait until process sets its pgid to be equal to pid
Review Comment:
Though why do we have this loop, and then a 10s timeout directly below it?
--
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
[GitHub] [airflow] dstandish commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1064887458
##########
tests/task/task_runner/test_standard_task_runner.py:
##########
@@ -131,38 +141,34 @@ def test_notifies_about_start_and_stop(self):
)
dag = dagbag.dags.get("test_example_bash_operator")
task = dag.get_task("runme_1")
+ dag.create_dagrun(
+ run_id="test",
+ data_interval=(DEFAULT_DATE, DEFAULT_DATE),
+ state=State.RUNNING,
+ start_date=DEFAULT_DATE,
+ )
+ ti = TaskInstance(task=task, run_id="test")
+ job1 = LocalTaskJob(task_instance=ti, ignore_ti_state=True)
+ runner = StandardTaskRunner(job1)
+ runner.start()
+
+ # Wait until process sets its pgid to be equal to pid
Review Comment:
do not know, existing code :)
--
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
[GitHub] [airflow] dstandish commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1055117252
##########
airflow/cli/commands/task_command.py:
##########
@@ -280,38 +281,80 @@ def _extract_external_executor_id(args) -> str | None:
@contextmanager
-def _capture_task_logs(ti: TaskInstance) -> Generator[None, None, None]:
+def _move_task_handlers_to_root(ti: TaskInstance) -> Generator[None, None, None]:
"""
- Manage logging context for a task run.
-
- - Replace the root logger configuration with the airflow.task configuration
- so we can capture logs from any custom loggers used in the task.
-
- - Redirect stdout and stderr to the task instance log, as INFO and WARNING
- level messages, respectively.
+ Move handlers for task logging to root logger.
+ We want anything logged during task run to be propagated to task log handlers.
+ If running in a k8s executor pod, also keep the stream handler on root logger
+ so that logs are still emitted to stdout.
"""
modify = not settings.DONOT_MODIFY_HANDLERS
+ did_modify = False
+ is_k8s_executor_pod = os.environ.get("AIRFLOW_IS_K8S_EXECUTOR_POD")
if modify:
- root_logger, task_logger = logging.getLogger(), logging.getLogger("airflow.task")
+ root_logger = logging.getLogger()
+ root_level = root_logger.level
+ task_logger = ti.log
+ task_propagate = task_logger.propagate
+ task_handlers = task_logger.handlers.copy()
+
+ # if there are no task handlers, then we should not do anything
+ # because either the handlers were already moved by the LocalTaskJob
+ # invocation of task_run (which wraps the --raw invocation), or
+ # user is doing something custom / unexpected
+ if task_handlers:
Review Comment:
removed the modify flag too
--
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
[GitHub] [airflow] dstandish commented on pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on PR #28440:
URL: https://github.com/apache/airflow/pull/28440#issuecomment-1362454819
two changes that are sort of elective in this one are (1) overridable "maintain propagate" and (2) checking if ti is actually a TI
both of these were required because while fixing some tests that were failing i noticed that the test replaced the whole logging config with something completely different and i curious why that was necessary so i dug in. we needed to check TI because in some of the test it's a mock value (and checking doesn't hurt anything -- if it's not TI it will fail anyway. then overriding maintain propagate is necessary to ensure that pytest caplog works... again... it doesn't hurt anything being there and allows us to do less manipulation in the tests which, all else equal, is a good thing.
--
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
[GitHub] [airflow] malthe commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
malthe commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1060422798
##########
airflow/utils/log/file_task_handler.py:
##########
@@ -62,19 +62,29 @@ def __init__(self, base_log_folder: str, filename_template: str | None = None):
# handler, not the one that calls super()__init__.
stacklevel=(2 if type(self) == FileTaskHandler else 3),
)
+ self.maintain_propagate: bool = False
def set_context(self, ti: TaskInstance) -> None | SetContextPropagate:
"""
Provide task_instance context to airflow task handler.
+ Generally speaking returns None. But if attr `maintain_propagate` has
+ been set to propagate, then returns sentinel MAINTAIN_PROPAGATE. This
+ has the effect of overriding the default behavior to set `propagate`
+ to False whenever set_context is called. At time of writing, this
+ functionality is only used in unit testing.
+
:param ti: task instance object
"""
- local_loc = self._init_file(ti)
- self.handler = NonCachingFileHandler(local_loc, encoding="utf-8")
- if self.formatter:
- self.handler.setFormatter(self.formatter)
- self.handler.setLevel(self.level)
- return None
+ from airflow.models.taskinstance import TaskInstance
+
+ if isinstance(ti, TaskInstance):
+ local_loc = self._init_file(ti)
+ self.handler = NonCachingFileHandler(local_loc, encoding="utf-8")
+ if self.formatter:
+ self.handler.setFormatter(self.formatter)
+ self.handler.setLevel(self.level)
Review Comment:
It does hurt somebody if they end up scratching their head to make sense of it!
--
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
[GitHub] [airflow] dstandish commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1060441192
##########
airflow/utils/log/file_task_handler.py:
##########
@@ -62,19 +62,29 @@ def __init__(self, base_log_folder: str, filename_template: str | None = None):
# handler, not the one that calls super()__init__.
stacklevel=(2 if type(self) == FileTaskHandler else 3),
)
+ self.maintain_propagate: bool = False
def set_context(self, ti: TaskInstance) -> None | SetContextPropagate:
"""
Provide task_instance context to airflow task handler.
+ Generally speaking returns None. But if attr `maintain_propagate` has
Review Comment:
Look at how it is used in the standard task runner test
--
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
[GitHub] [airflow] dstandish commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1064888364
##########
tests/task/task_runner/test_standard_task_runner.py:
##########
@@ -131,38 +141,34 @@ def test_notifies_about_start_and_stop(self):
)
dag = dagbag.dags.get("test_example_bash_operator")
task = dag.get_task("runme_1")
+ dag.create_dagrun(
+ run_id="test",
+ data_interval=(DEFAULT_DATE, DEFAULT_DATE),
+ state=State.RUNNING,
+ start_date=DEFAULT_DATE,
+ )
+ ti = TaskInstance(task=task, run_id="test")
+ job1 = LocalTaskJob(task_instance=ti, ignore_ti_state=True)
+ runner = StandardTaskRunner(job1)
+ runner.start()
+
+ # Wait until process sets its pgid to be equal to pid
Review Comment:
looking, though, i think the timeout is related to "done" while the loop is about "started"
--
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
[GitHub] [airflow] dstandish commented on pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on PR #28440:
URL: https://github.com/apache/airflow/pull/28440#issuecomment-1367442823
> @dstandish I think an abstraction point in the executor framework would be better rather than the `if`-statements in this changeset. That is, we should make the executor pluggable in terms of supporting this logging capability.
>
> In terms of actual implementation, rather than adjusting log handlers at runtime, perhaps an internal pub/sub mechanism could be devised.
We already adjust handlers. This fixes a deficiency
--
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
[GitHub] [airflow] dstandish commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1060398353
##########
airflow/utils/log/file_task_handler.py:
##########
@@ -62,19 +62,29 @@ def __init__(self, base_log_folder: str, filename_template: str | None = None):
# handler, not the one that calls super()__init__.
stacklevel=(2 if type(self) == FileTaskHandler else 3),
)
+ self.maintain_propagate: bool = False
def set_context(self, ti: TaskInstance) -> None | SetContextPropagate:
"""
Provide task_instance context to airflow task handler.
+ Generally speaking returns None. But if attr `maintain_propagate` has
Review Comment:
in this case, to do so, we would have to obliterate code that i'd rather keep. if we mack set_context (or the free function of same name) then we don't execute them. doing this lets us keep that code live. it ends up removed in this followup PR (https://github.com/apache/airflow/pull/28571)
which i need to rebase here after some recent changes... and the diff will be horrendous for now don't bother looking :)
--
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
[GitHub] [airflow] uranusjr commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
uranusjr commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1060270338
##########
airflow/cli/commands/task_command.py:
##########
@@ -281,38 +283,62 @@ def _extract_external_executor_id(args) -> str | None:
@contextmanager
-def _capture_task_logs(ti: TaskInstance) -> Generator[None, None, None]:
+def _move_task_handlers_to_root(ti: TaskInstance) -> Generator[None, None, None]:
"""
- Manage logging context for a task run.
+ Move handlers for task logging to root logger.
- - Replace the root logger configuration with the airflow.task configuration
- so we can capture logs from any custom loggers used in the task.
+ We want anything logged during task run to be propagated to task log handlers.
+ If running in a k8s executor pod, also keep the stream handler on root logger
+ so that logs are still emitted to stdout.
+ """
+ # nothing to do
+ if not ti.log.handlers or settings.DONOT_MODIFY_HANDLERS:
+ yield
+ return
+
+ def get_console_handler(logger):
+ for h in logger.handlers:
+ if h.name == "console":
+ return h
+
+ def ensure_handler(logger, handler):
+ if not handler:
+ return
+ if handler not in logger.handlers:
+ logger.addHandler(handler)
+
+ # Move task handlers to root and reset task logger and restore original logger settings after exit.
+ # If k8s executor, we need to ensure that root logger has a console handler, so that
+ # task logs propagate to stdout (this is how webserver retrieves them while task is running).
+ root_logger = logging.getLogger()
+ console_handler = get_console_handler(root_logger)
Review Comment:
Can be simplified into
```python
console_handler = next((h for h in logger.handlers if h.name == "console"), None)
```
Also it seems logic in `ensure_handler` can also be inlined?
--
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
[GitHub] [airflow] dstandish commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1060326360
##########
airflow/utils/log/file_task_handler.py:
##########
@@ -92,16 +102,17 @@ def _render_filename(self, ti: TaskInstance, try_number: int) -> str:
with create_session() as session:
dag_run = ti.get_dagrun(session=session)
template = dag_run.get_log_template(session=session).filename
- str_tpl, jinja_tpl = parse_template_string(template)
+ str_tpl, jinja_tpl = parse_template_string(template)
Review Comment:
if you look a few lines below, we need to keep `context = ti.get_template_context(session=session)`
in the same session. otherwise it creates a new session (provided by provide_session), and this can cause problems post-os.fork. this is why the test for standard task runner previously blew away the entire log config because. with this, we don't have to do that anymore.
--
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
[GitHub] [airflow] dstandish commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1060411483
##########
airflow/utils/log/file_task_handler.py:
##########
@@ -62,19 +62,29 @@ def __init__(self, base_log_folder: str, filename_template: str | None = None):
# handler, not the one that calls super()__init__.
stacklevel=(2 if type(self) == FileTaskHandler else 3),
)
+ self.maintain_propagate: bool = False
def set_context(self, ti: TaskInstance) -> None | SetContextPropagate:
"""
Provide task_instance context to airflow task handler.
+ Generally speaking returns None. But if attr `maintain_propagate` has
+ been set to propagate, then returns sentinel MAINTAIN_PROPAGATE. This
+ has the effect of overriding the default behavior to set `propagate`
+ to False whenever set_context is called. At time of writing, this
+ functionality is only used in unit testing.
+
:param ti: task instance object
"""
- local_loc = self._init_file(ti)
- self.handler = NonCachingFileHandler(local_loc, encoding="utf-8")
- if self.formatter:
- self.handler.setFormatter(self.formatter)
- self.handler.setLevel(self.level)
- return None
+ from airflow.models.taskinstance import TaskInstance
+
+ if isinstance(ti, TaskInstance):
+ local_loc = self._init_file(ti)
+ self.handler = NonCachingFileHandler(local_loc, encoding="utf-8")
+ if self.formatter:
+ self.handler.setFormatter(self.formatter)
+ self.handler.setLevel(self.level)
Review Comment:
i think it came up when i was mocking a TI.... and it doesn't hurt anybody.... but i've removed it for now let's see what the tests say
--
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
[GitHub] [airflow] uranusjr commented on a diff in pull request #28440: Propagate logs to stdout when in k8s executor pod
Posted by GitBox <gi...@apache.org>.
uranusjr commented on code in PR #28440:
URL: https://github.com/apache/airflow/pull/28440#discussion_r1055103624
##########
airflow/cli/commands/task_command.py:
##########
@@ -280,38 +281,80 @@ def _extract_external_executor_id(args) -> str | None:
@contextmanager
-def _capture_task_logs(ti: TaskInstance) -> Generator[None, None, None]:
+def _move_task_handlers_to_root(ti: TaskInstance) -> Generator[None, None, None]:
"""
- Manage logging context for a task run.
-
- - Replace the root logger configuration with the airflow.task configuration
- so we can capture logs from any custom loggers used in the task.
-
- - Redirect stdout and stderr to the task instance log, as INFO and WARNING
- level messages, respectively.
+ Move handlers for task logging to root logger.
+ We want anything logged during task run to be propagated to task log handlers.
+ If running in a k8s executor pod, also keep the stream handler on root logger
+ so that logs are still emitted to stdout.
"""
modify = not settings.DONOT_MODIFY_HANDLERS
+ did_modify = False
+ is_k8s_executor_pod = os.environ.get("AIRFLOW_IS_K8S_EXECUTOR_POD")
if modify:
- root_logger, task_logger = logging.getLogger(), logging.getLogger("airflow.task")
+ root_logger = logging.getLogger()
+ root_level = root_logger.level
+ task_logger = ti.log
+ task_propagate = task_logger.propagate
+ task_handlers = task_logger.handlers.copy()
+
+ # if there are no task handlers, then we should not do anything
+ # because either the handlers were already moved by the LocalTaskJob
+ # invocation of task_run (which wraps the --raw invocation), or
+ # user is doing something custom / unexpected
+ if task_handlers:
Review Comment:
I think it’d be good enough if we simply do a
```python
if not task_handlers:
yield
return
```
and get rid of the `did_modify` flag? This also allows everything after this to be unindented one level.
--
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