Hi everyone! I'm a new contributor to Airflow and I would like to discuss adding support for dynamically discovering new DAG bundle configurations without restarting Airflow.
*The Problem:* Today, DAG bundles are configured through [dag_processor] dag_bundle_config_list. DagBundlesManager loads this configuration during process startup, and the DAG processor retains the resulting bundle set for its lifetime. This becomes operationally difficult when we intend to manage multiple bundles, where bundles are added over time. Right now, it requires deployment of Airflow services. There's also similar discussion going over: - Runtime CRUD proposal: https://github.com/apache/airflow/issues/61645 - Automatic bundle configuration loading: https://github.com/apache/airflow/discussions/59799 - File-based hot-reload PR: https://github.com/apache/airflow/pull/63928 *Proposal* The core behavior change would be to let the DAG processor periodically reload the available bundle configurations from its existing processing loop. To support different deployment environments, I propose introducing a BaseDagBundleConfigProvider abstraction used by DagBundlesManager. Reference: https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/dag_processing/bundles/manager.py Conceptually, the provider interface could look like: class BaseDagBundleConfigProvider(ABC): @abstractmethod def get_all_bundle_configs(self) -> Collection[DagBundleConfig]: """Return all bundle configurations currently available.""" def get_bundle_config(self, name: str) -> DagBundleConfig | None: """Return configuration for a specific bundle.""" The provider would return the same configuration shape currently accepted by dag_bundle_config_list: name, classpath, kwargs, and optional team name. The idea is to make the source of DAG bundle configurations pluggable, while keeping the bundle implementations and their existing runtime behavior unchanged. A provider could read configurations from a file, API, etc. When no provider is configured, dag_bundle_config_list behavior would remain unchanged. *Scheduler and worker behavior:* The scheduler would continue passing the existing bundle name, bundle version, and relative DAG path. No provider lookup would be introduced in the scheduler enqueue path. Workers already resolve bundles through DagBundlesManager.get_bundle(). For a provider-only bundle, DagBundlesManager could use provider.get_bundle_config(name) to obtain the classpath and kwargs before instantiating the requested bundle version. I would like to get the community feedback on this approach. If there is alignment on this approach, I can also add a detailed proposal. Thanks, Sameer Raj
