kaxil commented on code in PR #69597:
URL: https://github.com/apache/airflow/pull/69597#discussion_r3680893652
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -240,7 +240,10 @@ def execute_workload(input: str) -> None:
log.info("[%s] Executing workload in Celery: %s", celery_task_id, workload)
try:
- BaseExecutor.run_workload(workload)
+ BaseExecutor.run_workload(
+ workload,
+ subprocess_logs_to_stdout=conf.getboolean("celery",
"task_logs_to_stdout", fallback=False),
Review Comment:
The edge3 worker has the same gap
(`providers/edge3/src/airflow/providers/edge3/cli/worker.py:438` calls
`run_workload` without this kwarg, and `supervise` at :451 on the pre-3.3
branch), so a `[celery]`-scoped key means every executor provider ends up
carrying its own copy of the same option. That is the reason #64481 put it in
`[logging]`, which is what multimeric's comment above is pointing at. Would it
work to have `run_workload` default `subprocess_logs_to_stdout` from a core
`[logging]` key when the caller does not pass one, and keep `[celery]
task_logs_to_stdout` as the worker-level override? That is the same two-level
lookup this provider already uses for `[celery] json_logs` falling back to
`[logging] json_logs`.
##########
providers/celery/tests/unit/celery/executors/test_celery_executor.py:
##########
@@ -1147,6 +1147,91 @@ def
test_execute_workload_runs_execute_task_before_airflow_3_3():
assert mock_supervise.call_args.kwargs["log_path"] == "test.log"
[email protected](not AIRFLOW_V_3_0_PLUS, reason="execute_workload is only
used for Airflow 3+")
[email protected](AIRFLOW_V_3_3_PLUS, reason="pre-3.3 compatibility path
only applies before Airflow 3.3")
[email protected](
+ ("config_value", "expected"),
+ [("True", True), ("False", False)],
+)
+def
test_execute_workload_forwards_task_logs_to_stdout_before_airflow_3_3(config_value,
expected):
Review Comment:
These two new tests are near-verbatim copies of
`test_execute_workload_runs_execute_task_before_airflow_3_3` and
`test_execute_workload_runs_base_executor_workload_on_airflow_3_3_plus` sitting
right beside them: same workload construction, same `mock_app` setup, differing
only in the final assert. Could the `conf_vars` parametrize be added to those
two existing tests instead, or the workload/mock setup pulled into a fixture?
85 lines for two assertions is a lot to keep in sync.
##########
providers/celery/provider.yaml:
##########
@@ -228,6 +228,17 @@ config:
type: boolean
example: ~
default: ~
+ task_logs_to_stdout:
+ description: |
+ Also forward task subprocess stdout/stderr to the Celery worker's
own stdout,
+ so task logs reach a container-level log collector (e.g.
Kubernetes/Loki) in
+ addition to the task-log handler and the UI. This gives the Celery
worker parity
+ with the LocalExecutor and the KubernetesExecutor per-task pod,
which always
+ forward task logs to stdout. Disabled by default to preserve
existing behaviour.
Review Comment:
`celery_executor.rst` has a "Worker logging" section that walks through
`[celery] json_logs` with an ini example and a version note, and that is where
someone hunting for "how do I get task logs into my container log collector"
will look, so this option is worth covering there rather than only in the
generated config ref. Separately, the provider still supports
`apache-airflow>=2.11.0`, where tasks go through `execute_command` and never
reach `supervise`, so setting this on Airflow 2 does nothing at all -- saying
"Airflow 3+ only" here would match how the json_logs note calls out 3.2.
--
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]
