This is an automated email from the ASF dual-hosted git repository.
hussein-awala pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new da202b08f42 Honor json_logs config in Celery worker startup (#68916)
da202b08f42 is described below
commit da202b08f42391e4d3c802008f9f04b33df58515
Author: Xu Han <[email protected]>
AuthorDate: Mon Jun 29 08:53:30 2026 -0400
Honor json_logs config in Celery worker startup (#68916)
* feat(celery): honor json_logs config in Celery worker startup
Add two-level config lookup for JSON logging in the Celery worker:
- Check [celery] json_logs first (celery-specific override)
- Fall back to [logging] json_logs (global setting, added in 3.2.0)
- Fall back to False if neither is set
This mirrors the existing CELERY_LOGGING_LEVEL / LOGGING_LEVEL fallback
pattern in the same file.
Also adds [celery] json_logs to provider.yaml config schema.
Closes apache/airflow#68912
* fix(celery): rename newsfragment to PR number, set version_added to
undecided
* fix(celery): drop unneeded backend marker, remove provider newsfragment
* fix(celery): handle null schema default and avoid DB access in non-DB
test mode
* fix(celery): mock cli_action_loggers and restore executor_loader reload
in TestWorkerJsonLogs
---
providers/celery/docs/celery_executor.rst | 36 ++++++++++++++++
providers/celery/provider.yaml | 11 +++++
.../airflow/providers/celery/cli/celery_command.py | 7 ++-
.../airflow/providers/celery/get_provider_info.py | 7 +++
.../tests/unit/celery/cli/test_celery_command.py | 50 ++++++++++++++++++++++
5 files changed, 110 insertions(+), 1 deletion(-)
diff --git a/providers/celery/docs/celery_executor.rst
b/providers/celery/docs/celery_executor.rst
index 6017c68c83c..c75d147c728 100644
--- a/providers/celery/docs/celery_executor.rst
+++ b/providers/celery/docs/celery_executor.rst
@@ -203,6 +203,42 @@ During this process, two processes are created:
| [11] **WorkerProcess** saves status information in **ResultBackend**.
| [13] When **SchedulerProcess** asks **ResultBackend** again about the
status, it will get information about the status of the task.
+.. _celery_executor:logging:
+
+Worker logging
+--------------
+
+By default the Celery worker writes plain-text logs to stdout. To emit
structured
+JSON instead, enable it via the ``[logging]`` section (applies to all Airflow
+components) or override it for the worker alone with the ``[celery]`` section:
+
+.. code-block:: ini
+
+ # Global — affects the API server, scheduler, and workers:
+ [logging]
+ json_logs = True
+
+ # Or override for the Celery worker only, leaving other components
unchanged:
+ [celery]
+ json_logs = True
+
+The lookup order is:
+
+1. ``[celery] json_logs`` — if set, takes precedence.
+2. ``[logging] json_logs`` — used when the celery-specific key is absent.
+3. ``False`` — the default when neither key is configured.
+
+This mirrors the existing ``[logging] CELERY_LOGGING_LEVEL`` /
+``[logging] LOGGING_LEVEL`` fallback already present in the worker startup
+code.
+
+.. note::
+
+ ``[logging] json_logs`` was added in Airflow 3.2.0. On older 3.x versions
the
+ global key is silently ignored (``fallback=False``), so setting only
+ ``[celery] json_logs = True`` is the safe way to enable JSON logs
regardless
+ of the core version.
+
.. _celery_executor:queue:
Queues
diff --git a/providers/celery/provider.yaml b/providers/celery/provider.yaml
index 29d92b74642..a1068ca7cec 100644
--- a/providers/celery/provider.yaml
+++ b/providers/celery/provider.yaml
@@ -215,6 +215,17 @@ config:
type: boolean
example: ~
default: "true"
+ json_logs:
+ description: |
+ Emit Celery worker stdout in JSON format. When set, this takes
precedence over
+ the global ``[logging] json_logs`` setting, allowing the worker to
use a
+ different format than the rest of the deployment. When unset (the
default),
+ the value falls back to ``[logging] json_logs``, which itself
defaults to
+ ``False``.
+ version_added: ~
+ type: boolean
+ example: ~
+ default: ~
broker_url:
description: |
The Celery broker URL. Celery supports multiple broker types. See:
diff --git
a/providers/celery/src/airflow/providers/celery/cli/celery_command.py
b/providers/celery/src/airflow/providers/celery/cli/celery_command.py
index 3aebe0c2088..ed5b83c4478 100644
--- a/providers/celery/src/airflow/providers/celery/cli/celery_command.py
+++ b/providers/celery/src/airflow/providers/celery/cli/celery_command.py
@@ -263,7 +263,12 @@ def worker(args):
if AIRFLOW_V_3_0_PLUS:
from airflow.sdk.log import configure_logging
- configure_logging(output=sys.stdout.buffer)
+ _celery_json = conf.get("celery", "json_logs", fallback="")
+ if _celery_json and _celery_json.lower() != "none":
+ json_output = conf.getboolean("celery", "json_logs")
+ else:
+ json_output = conf.getboolean("logging", "json_logs",
fallback=False)
+ configure_logging(output=sys.stdout.buffer, json_output=json_output)
else:
# Disable connection pool so that celery worker does not hold an
unnecessary db connection
settings.reconfigure_orm(disable_connection_pool=True)
diff --git a/providers/celery/src/airflow/providers/celery/get_provider_info.py
b/providers/celery/src/airflow/providers/celery/get_provider_info.py
index b1a663e3495..5358cda7784 100644
--- a/providers/celery/src/airflow/providers/celery/get_provider_info.py
+++ b/providers/celery/src/airflow/providers/celery/get_provider_info.py
@@ -110,6 +110,13 @@ def get_provider_info():
"example": None,
"default": "true",
},
+ "json_logs": {
+ "description": "Emit Celery worker stdout in JSON
format. When set, this takes precedence over\nthe global ``[logging]
json_logs`` setting, allowing the worker to use a\ndifferent format than the
rest of the deployment. When unset (the default),\nthe value falls back to
``[logging] json_logs``, which itself defaults to\n``False``.\n",
+ "version_added": None,
+ "type": "boolean",
+ "example": None,
+ "default": None,
+ },
"broker_url": {
"description": "The Celery broker URL. Celery supports
multiple broker types.
See:\nhttps://docs.celeryq.dev/en/stable/getting-started/backends-and-brokers/index.html#broker-overview\n";,
"version_added": None,
diff --git a/providers/celery/tests/unit/celery/cli/test_celery_command.py
b/providers/celery/tests/unit/celery/cli/test_celery_command.py
index 49495e818a5..41332d545cf 100644
--- a/providers/celery/tests/unit/celery/cli/test_celery_command.py
+++ b/providers/celery/tests/unit/celery/cli/test_celery_command.py
@@ -411,6 +411,56 @@ class TestWorkerDuplicateHostnameCheck:
mock_pools_reset.assert_called_once()
[email protected](not AIRFLOW_V_3_0_PLUS, reason="configure_logging only
called in Airflow 3+")
[email protected]("airflow.utils.cli_action_loggers.on_pre_execution")
[email protected]("conf_stale_bundle_cleanup_disabled")
+class TestWorkerJsonLogs:
+ @classmethod
+ def setup_class(cls):
+ with conf_vars({("core", "executor"): "CeleryExecutor"}):
+ importlib.reload(executor_loader)
+ importlib.reload(cli_parser)
+ cls.parser = cli_parser.get_parser()
+
+ @mock.patch("airflow.providers.celery.cli.celery_command.Process")
+ @mock.patch("airflow.providers.celery.executors.celery_executor.app")
+ @mock.patch("airflow.sdk.log.configure_logging")
+ def test_json_logs_defaults_to_false(
+ self, mock_configure_logging, mock_celery_app, mock_popen,
mock_pre_exec
+ ):
+ args = self.parser.parse_args(["celery", "worker"])
+ celery_command.worker(args)
+ mock_configure_logging.assert_called_once()
+ _, kwargs = mock_configure_logging.call_args
+ assert kwargs.get("json_output") is False
+
+ @mock.patch("airflow.providers.celery.cli.celery_command.Process")
+ @mock.patch("airflow.providers.celery.executors.celery_executor.app")
+ @mock.patch("airflow.sdk.log.configure_logging")
+ def test_json_logs_uses_global_logging_config(
+ self, mock_configure_logging, mock_celery_app, mock_popen,
mock_pre_exec
+ ):
+ args = self.parser.parse_args(["celery", "worker"])
+ with conf_vars({("logging", "json_logs"): "True"}):
+ celery_command.worker(args)
+ mock_configure_logging.assert_called_once()
+ _, kwargs = mock_configure_logging.call_args
+ assert kwargs.get("json_output") is True
+
+ @mock.patch("airflow.providers.celery.cli.celery_command.Process")
+ @mock.patch("airflow.providers.celery.executors.celery_executor.app")
+ @mock.patch("airflow.sdk.log.configure_logging")
+ def test_celery_json_logs_overrides_global(
+ self, mock_configure_logging, mock_celery_app, mock_popen,
mock_pre_exec
+ ):
+ args = self.parser.parse_args(["celery", "worker"])
+ with conf_vars({("logging", "json_logs"): "True", ("celery",
"json_logs"): "False"}):
+ celery_command.worker(args)
+ mock_configure_logging.assert_called_once()
+ _, kwargs = mock_configure_logging.call_args
+ assert kwargs.get("json_output") is False
+
+
@pytest.mark.backend("mysql", "postgres")
@pytest.mark.usefixtures("conf_stale_bundle_cleanup_disabled")
class TestFlowerCommand:
