Add celery_logging_level

airflow/cli/commands/celery_command.py

@@ -134,6 +134,10 @@ def worker(args):
             # it, we raced to create the tables and lost.
             pass
 
+    # backwards-compatible: https://github.com/apache/airflow/pull/21506#pullrequestreview-879893763
+    celery_log_level = conf.get('logging', 'CELERY_LOGGING_LEVEL')
+    if not celery_log_level:
+        celery_log_level = conf.get('logging', 'LOGGING_LEVEL')
     # Setup Celery worker
     options = [
         'worker',
@@ -146,7 +150,7 @@ def worker(args):
         '--hostname',
         args.celery_hostname,
         '--loglevel',
-        conf.get('logging', 'LOGGING_LEVEL'),
+        celery_log_level,
         '--pidfile',
         pid_file_path,
     ]

airflow/config_templates/config.yml

@@ -531,6 +531,15 @@
       type: string
       example: ~
       default: "INFO"
+    - name: celery_logging_level
+      description: |
+        Logging level for celery. If not set, it uses the value of logging_level
+
+        Supported values: ``CRITICAL``, ``ERROR``, ``WARNING``, ``INFO``, ``DEBUG``.
+      version_added: 2.3.0
+      type: string
+      example: ~
+      default: ""
     - name: fab_logging_level
       description: |
         Logging level for Flask-appbuilder UI.

airflow/config_templates/default_airflow.cfg

@@ -302,6 +302,11 @@ encrypt_s3_logs = False
 # Supported values: ``CRITICAL``, ``ERROR``, ``WARNING``, ``INFO``, ``DEBUG``.
 logging_level = INFO
 
+# Logging level for celery. If not set, it uses the value of logging_level
+#
+# Supported values: ``CRITICAL``, ``ERROR``, ``WARNING``, ``INFO``, ``DEBUG``.
+celery_logging_level =
+
 # Logging level for Flask-appbuilder UI.
 #
 # Supported values: ``CRITICAL``, ``ERROR``, ``WARNING``, ``INFO``, ``DEBUG``.

airflow/config_templates/default_test.cfg

@@ -50,6 +50,7 @@ store_serialized_dags = False
 [logging]
 base_log_folder = {AIRFLOW_HOME}/logs
 logging_level = INFO
+celery_logging_level = WARN
 fab_logging_level = WARN
 log_filename_template = {{{{ ti.dag_id }}}}/{{{{ ti.task_id }}}}/{{{{ ts }}}}/{{{{ try_number }}}}.log
 log_processor_filename_template = {{{{ filename }}}}.log

airflow/configuration.py

@@ -211,6 +211,8 @@ class AirflowConfigParser(ConfigParser):
         ("scheduler", "file_parsing_sort_mode"): ["modified_time", "random_seeded_by_host", "alphabetical"],
         ("logging", "logging_level"): _available_logging_levels,
         ("logging", "fab_logging_level"): _available_logging_levels,
+        # celery_logging_level can be empty, which uses logging_level as fallback
+        ("logging", "celery_logging_level"): _available_logging_levels + [''],
     }
 
     # This method transforms option names on every read, get, or set operation.

tests/cli/commands/test_celery_command.py

@@ -221,7 +221,7 @@ def test_worker_started_with_required_arguments(self, mock_celery_app, mock_pope
                 '--hostname',
                 celery_hostname,
                 '--loglevel',
-                conf.get('logging', 'LOGGING_LEVEL'),
+                conf.get('logging', 'CELERY_LOGGING_LEVEL'),
                 '--pidfile',
                 pid_file,
                 '--autoscale',