Use LoggingMixin for DagProcessorManager (apache#46408)

Since we always have a standalone DAG processor now, we no longer need
the special logging config to have a separate log file for the
DagProcessorManager. Using LoggingMixin instead, all of the logs end up
on stdout, just like other Airflow components.

Author: jedcunningham

airflow/cli/commands/remote_commands/config_command.py

@@ -84,27 +84,30 @@ class ConfigChange:
     :param config: The configuration parameter being changed.
     :param suggestion: A suggestion for replacing or handling the removed configuration.
     :param renamed_to: The new section and option if the configuration is renamed.
+    :param was_deprecated: If the config is removed, whether the old config was deprecated.
     """
 
     config: ConfigParameter
     suggestion: str = ""
     renamed_to: ConfigParameter | None = None
+    was_deprecated: bool = True
 
     @property
     def message(self) -> str:
         """Generate a message for this configuration change."""
         if self.renamed_to:
             if self.config.section != self.renamed_to.section:
                 return (
-                    f"`{self.config.option}` configuration parameter moved from `{self.config.section}` section to `"
-                    f"{self.renamed_to.section}` section as `{self.renamed_to.option}`."
+                    f"`{self.config.option}` configuration parameter moved from `{self.config.section}` section to "
+                    f"`{self.renamed_to.section}` section as `{self.renamed_to.option}`."
                 )
             return (
                 f"`{self.config.option}` configuration parameter renamed to `{self.renamed_to.option}` "
                 f"in the `{self.config.section}` section."
             )
         return (
-            f"Removed deprecated `{self.config.option}` configuration parameter from `{self.config.section}` section. "
+            f"Removed{' deprecated' if self.was_deprecated else ''} `{self.config.option}` configuration parameter "
+            f"from `{self.config.section}` section. "
             f"{self.suggestion}"
         )
 
@@ -203,6 +206,12 @@ def message(self) -> str:
         config=ConfigParameter("core", "dag_file_processor_timeout"),
         renamed_to=ConfigParameter("dag_processor", "dag_file_processor_timeout"),
     ),
+    ConfigChange(
+        config=ConfigParameter("core", "dag_processor_manager_log_location"),
+    ),
+    ConfigChange(
+        config=ConfigParameter("core", "log_processor_filename_template"),
+    ),
     # api
     ConfigChange(
         config=ConfigParameter("api", "access_control_allow_origin"),
@@ -218,6 +227,18 @@ def message(self) -> str:
         suggestion="Remove TaskContextLogger: Replaced by the Log table for better handling of task log "
         "messages outside the execution context.",
     ),
+    ConfigChange(
+        config=ConfigParameter("logging", "dag_processor_manager_log_location"),
+        was_deprecated=False,
+    ),
+    ConfigChange(
+        config=ConfigParameter("logging", "dag_processor_manager_log_stdout"),
+        was_deprecated=False,
+    ),
+    ConfigChange(
+        config=ConfigParameter("logging", "log_processor_filename_template"),
+        was_deprecated=False,
+    ),
     # metrics
     ConfigChange(
         config=ConfigParameter("metrics", "metrics_use_pattern_match"),

airflow/config_templates/airflow_local_settings.py

@@ -20,7 +20,6 @@
 from __future__ import annotations
 
 import os
-from pathlib import Path
 from typing import Any
 from urllib.parse import urlsplit
 
@@ -53,17 +52,6 @@
 
 PROCESSOR_LOG_FOLDER: str = conf.get_mandatory_value("scheduler", "CHILD_PROCESS_LOG_DIRECTORY")
 
-DAG_PROCESSOR_MANAGER_LOG_LOCATION: str = conf.get_mandatory_value(
-    "logging", "DAG_PROCESSOR_MANAGER_LOG_LOCATION"
-)
-
-DAG_PROCESSOR_MANAGER_LOG_STDOUT: str = conf.get_mandatory_value(
-    "logging", "DAG_PROCESSOR_MANAGER_LOG_STDOUT"
-)
-
-
-PROCESSOR_FILENAME_TEMPLATE: str = conf.get_mandatory_value("logging", "LOG_PROCESSOR_FILENAME_TEMPLATE")
-
 DEFAULT_LOGGING_CONFIG: dict[str, Any] = {
     "version": 1,
     "disable_existing_loggers": False,
@@ -99,27 +87,8 @@
             "base_log_folder": os.path.expanduser(BASE_LOG_FOLDER),
             "filters": ["mask_secrets"],
         },
-        "processor": {
-            "class": "airflow.utils.log.file_processor_handler.FileProcessorHandler",
-            "formatter": "airflow",
-            "base_log_folder": os.path.expanduser(PROCESSOR_LOG_FOLDER),
-            "filename_template": PROCESSOR_FILENAME_TEMPLATE,
-            "filters": ["mask_secrets"],
-        },
-        "processor_to_stdout": {
-            "class": "airflow.utils.log.logging_mixin.RedirectStdHandler",
-            "formatter": "source_processor",
-            "stream": "sys.stdout",
-            "filters": ["mask_secrets"],
-        },
     },
     "loggers": {
-        "airflow.processor": {
-            "handlers": ["processor_to_stdout" if DAG_PROCESSOR_LOG_TARGET == "stdout" else "processor"],
-            "level": LOG_LEVEL,
-            # Set to true here (and reset via set_context) so that if no file is configured we still get logs!
-            "propagate": True,
-        },
         "airflow.task": {
             "handlers": ["task"],
             "level": LOG_LEVEL,
@@ -152,54 +121,6 @@
     }
     DEFAULT_LOGGING_CONFIG["loggers"].update(new_loggers)
 
-DEFAULT_DAG_PARSING_LOGGING_CONFIG: dict[str, dict[str, dict[str, Any]]] = {
-    "handlers": {
-        "processor_manager": {
-            "class": "airflow.utils.log.non_caching_file_handler.NonCachingRotatingFileHandler",
-            "formatter": "airflow",
-            "filename": DAG_PROCESSOR_MANAGER_LOG_LOCATION,
-            "mode": "a",
-            "maxBytes": 104857600,  # 100MB
-            "backupCount": 5,
-        }
-    },
-    "loggers": {
-        "airflow.processor_manager": {
-            "handlers": ["processor_manager"],
-            "level": LOG_LEVEL,
-            "propagate": False,
-        }
-    },
-}
-
-if DAG_PROCESSOR_MANAGER_LOG_STDOUT == "True":
-    DEFAULT_DAG_PARSING_LOGGING_CONFIG["handlers"].update(
-        {
-            "console": {
-                "class": "airflow.utils.log.logging_mixin.RedirectStdHandler",
-                "formatter": "airflow",
-                "stream": "sys.stdout",
-                "filters": ["mask_secrets"],
-            }
-        }
-    )
-    DEFAULT_DAG_PARSING_LOGGING_CONFIG["loggers"]["airflow.processor_manager"]["handlers"].append("console")
-
-# Only update the handlers and loggers when CONFIG_PROCESSOR_MANAGER_LOGGER is set.
-# This is to avoid exceptions when initializing RotatingFileHandler multiple times
-# in multiple processes.
-if os.environ.get("CONFIG_PROCESSOR_MANAGER_LOGGER") == "True":
-    DEFAULT_LOGGING_CONFIG["handlers"].update(DEFAULT_DAG_PARSING_LOGGING_CONFIG["handlers"])
-    DEFAULT_LOGGING_CONFIG["loggers"].update(DEFAULT_DAG_PARSING_LOGGING_CONFIG["loggers"])
-
-    # Manually create log directory for processor_manager handler as RotatingFileHandler
-    # will only create file but not the directory.
-    processor_manager_handler_config: dict[str, Any] = DEFAULT_DAG_PARSING_LOGGING_CONFIG["handlers"][
-        "processor_manager"
-    ]
-    directory: str = os.path.dirname(processor_manager_handler_config["filename"])
-    Path(directory).mkdir(parents=True, exist_ok=True, mode=0o755)
-
 ##################
 # Remote logging #
 ##################

airflow/config_templates/config.yml

@@ -912,28 +912,6 @@ logging:
       default: "dag_id={{ ti.dag_id }}/run_id={{ ti.run_id }}/task_id={{ ti.task_id }}/\
                {%% if ti.map_index >= 0 %%}map_index={{ ti.map_index }}/{%% endif %%}\
                attempt={{ try_number|default(ti.try_number) }}.log"
-    log_processor_filename_template:
-      description: |
-        Formatting for how airflow generates file names for log
-      version_added: 2.0.0
-      type: string
-      example: ~
-      is_template: true
-      default: "{{ filename }}.log"
-    dag_processor_manager_log_location:
-      description: |
-        Full path of dag_processor_manager logfile.
-      version_added: 2.0.0
-      type: string
-      example: ~
-      default: "{AIRFLOW_HOME}/logs/dag_processor_manager/dag_processor_manager.log"
-    dag_processor_manager_log_stdout:
-      description: |
-        Whether DAG processor manager will write logs to stdout
-      version_added: 2.9.0
-      type: boolean
-      example: ~
-      default: "False"
     task_log_reader:
       description: |
         Name of handler to read task instance logs.

airflow/dag_processing/manager.py

@@ -63,6 +63,7 @@
 from airflow.traces.tracer import Trace
 from airflow.utils import timezone
 from airflow.utils.file import list_py_file_paths, might_contain_dag
+from airflow.utils.log.logging_mixin import LoggingMixin
 from airflow.utils.net import get_hostname
 from airflow.utils.process_utils import (
     kill_child_processes_by_pids,
@@ -97,9 +98,6 @@ class DagFileStat:
     last_num_of_db_queries: int = 0
 
 
-log = logging.getLogger("airflow.processor_manager")
-
-
 @dataclass(frozen=True)
 class DagFileInfo:
     """Information about a DAG file."""
@@ -135,7 +133,7 @@ def _resolve_path(instance: Any, attribute: attrs.Attribute, val: str | os.PathL
 
 
 @attrs.define
-class DagFileProcessorManager:
+class DagFileProcessorManager(LoggingMixin):
     """
     Manage processes responsible for parsing DAGs.
 
@@ -167,8 +165,6 @@ class DagFileProcessorManager:
         factory=_config_int_factory("dag_processor", "stale_dag_threshold")
     )
 
-    log: logging.Logger = attrs.field(default=log, init=False)
-
     _last_deactivate_stale_dags_time: float = attrs.field(default=0, init=False)
     print_stats_interval: float = attrs.field(
         factory=_config_int_factory("dag_processor", "print_stats_interval")

newsfragments/46408.significant.rst

@@ -0,0 +1,30 @@
+DAG processor related config options removed
+
+The follow configuration options have been removed:
+
+- ``[logging] dag_processor_manager_log_location``
+- ``[logging] dag_processor_manager_log_stdout``
+- ``[logging] log_processor_filename_template``
+
+If these config options are still present, they will have no effect any longer.
+
+* Types of change
+
+  * [ ] Dag changes
+  * [x] Config changes
+  * [ ] API changes
+  * [ ] CLI changes
+  * [ ] Behaviour changes
+  * [ ] Plugin changes
+  * [ ] Dependency changes
+  * [ ] Code interface changes
+
+.. List the migration rules needed for this change (see https://github.com/apache/airflow/issues/41641)
+
+* Migration rules needed
+
+  * ``airflow config lint``
+
+    * Remove ``[logging] dag_processor_manager_log_location``
+    * Remove ``[logging] dag_processor_manager_log_stdout``
+    * Remove ``[logging] log_processor_filename_template``