reload dag bundle configuration without restarts

airflow-core/docs/administration-and-deployment/dag-bundles.rst

@@ -61,6 +61,33 @@ Configuring Dag bundles
 
 Dag bundles are configured in :ref:`config:dag_processor__dag_bundle_config_list`. You can add one or more Dag bundles here.
 
+Dynamic bundle configuration
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Set :ref:`config:dag_processor__dag_bundle_provider` to the import path for the class Airflow uses to discover
+and construct Dag bundles. The configured implementation must return metadata for the complete list of currently
+configured bundles.
+The Dag processor requests the list every
+:ref:`config:dag_processor__bundle_refresh_check_interval` seconds and applies additions and removals without
+a restart. A bundle that is absent from the list becomes inactive.
+
+The configured implementation must raise an exception when it cannot read its source. It must return an empty
+list only when no bundles are configured. On an error, the Dag processor keeps the last valid list and tries again
+on the next refresh cycle.
+
+Airflow also reconciles the active list with the metadata database on later refresh cycles. This makes the
+result converge when Dag processors observe a change at different times. For example, an older Dag processor
+can temporarily reactivate a removed bundle. A later cycle from a processor with the current list makes it
+inactive again. This process does not use a shared configuration revision or a separate database table.
+
+A bundle name must keep the same construction settings. Use a new name when settings such as its bundle
+class, repository, branch, connection, or refresh interval change. The configured implementation must continue
+to resolve the previous name while retained Dag runs can still need it.
+
+The default implementation, ``ConfigDagBundleProvider``, reads ``dag_bundle_config_list``. Its values stay static
+while the Dag processor runs. When filtered Dag processors use partial static configuration, they do not
+deactivate bundles that are absent from their local configuration.
+
 .. warning:: Reference credentials through a Connection — do not inline them
 
     Bundle ``kwargs`` are stored in the ``[dag_processor] dag_bundle_config_list``

airflow-core/newsfragments/73217.feature.rst

@@ -0,0 +1 @@
+Reload Dag bundle configuration without restarting Dag processors.

airflow-core/src/airflow/api/common/airflow_health.py

@@ -23,7 +23,7 @@
 from sqlalchemy import select
 
 from airflow.configuration import conf
-from airflow.dag_processing.bundles.manager import _get_configured_bundle_team_names
+from airflow.dag_processing.bundles.manager import DagBundlesManager
 from airflow.jobs.dag_processor_job_runner import DagProcessorJobRunner
 from airflow.jobs.job import Job
 from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
@@ -112,7 +112,7 @@ def _dag_processor_instance_health(job: Job) -> dict[str, Any]:
 # behind forever and a restarted replica adds a second one. The denominator is therefore taken from
 # the declared work partition instead, which is unaffected by how replicas come and go:
 #
-# * Dag processor -- the bundles in ``[dag_processor] dag_bundle_config_list``.
+# * Dag processor -- the bundles returned by the configured Dag bundle provider.
 # * Triggerer -- the team scopes those bundles declare, since a triggerer only picks up triggers for
 #   its own team (see ``Trigger.ids_for_triggerer``).
 # * Scheduler -- schedulers are symmetric, so there is no partition and no partial state to report.
@@ -121,7 +121,7 @@ def _dag_processor_instance_health(job: Job) -> dict[str, Any]:
 def _configured_bundle_teams() -> dict[str, str | None]:
     """Map every configured Dag bundle to the team owning it, empty when the config is unreadable."""
     try:
-        return _get_configured_bundle_team_names()
+        return DagBundlesManager().get_configured_bundle_team_names()
     except Exception:
         # A health probe must not fail on malformed bundle config; callers fall back to liveness only.
         log.warning("Could not read the Dag bundle configuration", exc_info=True)

airflow-core/src/airflow/cli/commands/dag_processor_command.py

@@ -23,7 +23,7 @@
 
 from airflow.cli.commands.daemon_utils import run_command_with_daemon_option
 from airflow.configuration import conf
-from airflow.dag_processing.bundles.manager import _get_configured_bundle_team_names
+from airflow.dag_processing.bundles.manager import DagBundlesManager
 from airflow.dag_processing.manager import DagFileProcessorManager
 from airflow.jobs.dag_processor_job_runner import DagProcessorJobRunner
 from airflow.jobs.job import Job, run_job
@@ -52,7 +52,7 @@ def _get_team_names(bundle_names: list[str] | None) -> list[str]:
     if not conf.getboolean("core", "multi_team"):
         return []
 
-    configured = _get_configured_bundle_team_names()
+    configured = DagBundlesManager().get_configured_bundle_team_names()
     names = bundle_names or list(configured)
     return sorted({team for name in names if (team := configured.get(name)) is not None})
 

airflow-core/src/airflow/cli/commands/team_command.py

@@ -218,7 +218,7 @@ def team_sync(args, *, session=NEW_SESSION):
 
     dag_bundle_teams = {
         bundle.team_name
-        for bundle in DagBundlesManager()._bundle_config.values()
+        for bundle in DagBundlesManager().get_active_bundle_metadata()
         if bundle.team_name is not None
     }
     existing_teams = Team.get_all_team_names(session=session)
@@ -311,9 +311,9 @@ def team_verify(args, *, session=NEW_SESSION):
 
     existing_teams = {team.name for team in teams}
 
-    for bundle_name, bundle in DagBundlesManager()._bundle_config.items():
+    for bundle in DagBundlesManager().get_active_bundle_metadata():
         if bundle.team_name and bundle.team_name not in existing_teams:
-            issues.append(f"DAG bundle '{bundle_name}' references unknown team '{bundle.team_name}'.")
+            issues.append(f"DAG bundle '{bundle.name}' references unknown team '{bundle.team_name}'.")
 
     if issues:
         print("Verification failed.\n")

airflow-core/src/airflow/config_templates/config.yml

@@ -3244,6 +3244,22 @@ dag_processor:
       example: "/tmp/some-place"
       default: ~
 
+    dag_bundle_provider:
+      description: |
+        Import path for the class that provides configured Dag bundle metadata and constructs Dag bundles.
+        The class must extend ``airflow.dag_processing.bundles.provider.DagBundleProvider`` and have a
+        no-argument constructor.
+
+        The default implementation reads ``dag_bundle_config_list``. A custom implementation can read
+        bundle configuration from another source and must return the complete set of configured bundles.
+        Dag processors request this list every ``bundle_refresh_check_interval`` seconds. A missing
+        bundle is inactive. A bundle name must keep the same construction settings. Use a new name
+        when settings such as its class, repository, branch, connection, or refresh interval change.
+      version_added: 3.4.0
+      type: string
+      example: ~
+      default: "airflow.dag_processing.bundles.provider.ConfigDagBundleProvider"
+
     dag_bundle_config_list:
       description: |
         List of backend configs.  Must supply name, classpath, and kwargs for each backend.
@@ -3416,9 +3432,10 @@ dag_processor:
       default: "False"
     bundle_refresh_check_interval:
       description: |
-        How often the DAG processor should check if any DAG bundles are ready for a refresh, either by hitting
-        the bundles refresh_interval or because another DAG processor has seen a newer version of the bundle.
-        A low value means we check more frequently, and have a smaller window of time where DAG processors are
+        How often the Dag processor should request the active bundle configuration and check if any
+        Dag bundles are ready for a refresh, either by hitting the bundles refresh_interval or because
+        another Dag processor has seen a newer version of the bundle.
+        A low value means we check more frequently, and have a smaller window of time where Dag processors are
         out of sync with each other, parsing different versions of the same bundle.
       version_added: ~
       type: integer