uranusjr commented on code in PR #74915:
URL: https://github.com/apache/airflow/pull/74915#discussion_r4522646366
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
@@ -113,25 +120,175 @@ def _bundle_item_exc(msg):
class _BundleConfigSnapshot(NamedTuple):
"""The configured Dag bundles, as names only and as full configs."""
- configs: tuple[_ExternalBundleConfig, ...]
+ configs: dict[str, _ExternalBundleConfig]
names: frozenset[str]
+ team_names: dict[str, str | None]
[email protected]
-def _load_bundle_config_snapshot() -> _BundleConfigSnapshot:
- """Read and validate the configured Dag bundles, without importing their
classes."""
- bundle_config_list = _read_bundle_config_list()
- if not bundle_config_list:
- return _BundleConfigSnapshot(configs=(), names=frozenset())
- if conf.getboolean("core", "LOAD_EXAMPLES"):
+def _create_bundle_config_snapshot(bundle_config_list:
list[_ExternalBundleConfig]) -> _BundleConfigSnapshot:
+ """Validate the configured Dag bundles, without importing their classes."""
+ team_names = {cfg.name: cfg.team_name for cfg in bundle_config_list}
+ if bundle_config_list and conf.getboolean("core", "LOAD_EXAMPLES"):
_add_example_dag_bundle(bundle_config_list)
_add_provider_example_dags_to_bundle(bundle_config_list)
return _BundleConfigSnapshot(
- configs=tuple(bundle_config_list),
+ configs={cfg.name: cfg for cfg in bundle_config_list},
names=frozenset(cfg.name for cfg in bundle_config_list),
+ team_names=team_names,
+ )
+
+
+def _get_bundle_config_reload_option() -> tuple[str, str] | None:
+ section, key = "dag_processor", "dag_bundle_config_list"
+ env_key = "AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST"
+ if env_key in os.environ:
+ return None
+ for suffix in ("cmd", "secret"):
+ env_option = f"{env_key}_{suffix.upper()}"
+ if env_option in os.environ:
+ return suffix, os.environ[env_option]
+ if ConfigParser.has_option(conf, section, key):
+ return None
+ for suffix in ("cmd", "secret"):
+ option = f"{key}_{suffix}"
+ if ConfigParser.has_option(conf, section, option):
+ return suffix, ConfigParser.get(conf, section, option)
+ return None
+
+
+def _build_internal_bundle_config(config: _ExternalBundleConfig) ->
_InternalBundleConfig:
+ if config.team_name and not conf.getboolean("core", "multi_team"):
+ raise AirflowConfigException(
+ "Section `dag_processor` key `dag_bundle_config_list` "
+ "cannot have a team name when multi-team mode is disabled."
+ "To enable multi-team, you need to update section `core` key
`multi_team` in your config."
+ )
+ return _InternalBundleConfig(
+ bundle_class=import_string(config.classpath),
+ kwargs=config.kwargs,
+ team_name=config.team_name,
+ importers=config.importers,
)
+class _StaticBundleConfigSource:
+ """The bundle configuration of this process, read from
``dag_bundle_config_list`` on first use."""
+
+ @functools.cached_property
+ def _accepted(self) -> _BundleConfigSnapshot:
+ return _create_bundle_config_snapshot(_read_bundle_config_list())
+
+ def current(self) -> _BundleConfigSnapshot:
+ """Return the accepted list, which is checked without importing bundle
classes."""
+ return self._accepted
+
+ def loadable(self) -> _BundleConfigSnapshot:
+ """Return the newest accepted list whose bundle classes load in this
process."""
+ return self._accepted
+
+ def refresh(self, adopted: _BundleConfigSnapshot | None = None) -> None:
+ """
+ Read the list again if the check interval has passed.
+
+ The new list may not change the definition of a bundle in *adopted*,
which defaults to the
+ accepted list.
+ """
+
+ def refresh_for_missing(self, name: str) -> None:
+ """Read the list again for a bundle *name* that is not loadable,
unless it missed recently."""
+
+
+class _ReloadingBundleConfigSource(_StaticBundleConfigSource):
+ """
+ The output of ``dag_bundle_config_list_cmd`` or
``dag_bundle_config_list_secret``.
+
+ It is read again at most once per ``bundle_refresh_check_interval``, plus
once per interval for
+ each missing bundle name. A list that changes the definition of an adopted
bundle is rejected
+ as a whole.
+ """
+
+ _loadable: _BundleConfigSnapshot
+
+ def __init__(self, reload_option: tuple[str, str]) -> None:
+ self._reload_option = reload_option
+ self._last_read = 0.0
+ self._unloadable: _BundleConfigSnapshot | None = None
+ self._missed_at: dict[str, float] = {}
+
+ @functools.cached_property
+ def _accepted(self) -> _BundleConfigSnapshot:
+ self._last_read = time.monotonic()
+ # The first list is not checked, so a bundle class that cannot be
imported fails the
+ # managers as it does with a static list.
+ self._loadable =
_create_bundle_config_snapshot(_read_bundle_config_list(self._reload_option))
+ return self._loadable
+
+ def loadable(self) -> _BundleConfigSnapshot:
+ accepted = self._accepted
+ loadable = self._loadable
+ if accepted is loadable or accepted is self._unloadable:
+ return loadable
+ try:
+ for name, config in accepted.configs.items():
+ if loadable.configs.get(name) != config:
+ _build_internal_bundle_config(config)
+ except Exception:
+ log.exception("Could not load the new Dag bundles; keeping the
last loadable list")
+ self._unloadable = accepted
+ return loadable
+ self._loadable = accepted
+ return accepted
+
+ def refresh(self, adopted: _BundleConfigSnapshot | None = None) -> None:
+ accepted = self._accepted
+ if time.monotonic() - self._last_read >= conf.getint(
+ "dag_processor", "bundle_refresh_check_interval"
+ ):
+ self._read(accepted if adopted is None else adopted)
+
+ def refresh_for_missing(self, name: str) -> None:
+ now = time.monotonic()
+ interval = conf.getint("dag_processor",
"bundle_refresh_check_interval")
+ self._missed_at = {missed: at for missed, at in
self._missed_at.items() if now - at < interval}
+ if name not in self._missed_at:
+ # A bundle whose class failed to load was never used, so a
corrected definition is allowed.
+ self._read(self.loadable())
+ if name not in self.loadable().configs:
+ self._missed_at[name] = now
+
+ def _read(self, adopted: _BundleConfigSnapshot) -> None:
+ accepted = self._accepted
+ self._last_read = time.monotonic()
+ self._unloadable = None
+ try:
+ snapshot =
_create_bundle_config_snapshot(_read_bundle_config_list(self._reload_option))
+ if snapshot.configs == accepted.configs:
+ return
+ for name, config in snapshot.configs.items():
+ existing = adopted.configs.get(name)
+ if existing is not None and existing != config:
+ raise _bundle_item_exc(
+ f"Configuration for bundle '{name}' has changed.
Rejecting the entire "
+ "configuration list and keeping the last accepted
list. "
+ "Revert this entry or use a new bundle name."
+ )
+ self._accepted = snapshot
+ except Exception:
+ log.exception("Could not reload Dag bundle configuration; keeping
the last accepted list")
+
+
[email protected]
+def _get_bundle_config_source() -> _StaticBundleConfigSource:
+ if (reload_option := _get_bundle_config_reload_option()) is None:
+ return _StaticBundleConfigSource()
+ return _ReloadingBundleConfigSource(reload_option)
+
+
+def is_bundle_config_reload_enabled() -> bool:
+ """Return whether this process reads the bundle configuration again from a
command or secrets backend."""
+ return isinstance(_get_bundle_config_source(),
_ReloadingBundleConfigSource)
Review Comment:
I don’t think we need to send a context in static mode (and change the
parsing path). We can add `parse_context(name) -> BundleParseContext | None` to
the source. The static source returns `None`, so the static path stays exactly
as it is today. The reloading source returns the accepted context. The same
goes for the HTTP-mode guard and `_reconcile_bundles` (`source.reconcilable`).
The call sites then ask the source what to do, instead of what it is, and
`is_bundle_config_reload_enabled` can go. I'd rather this land in this PR than
in a follow-up since the follow-up would touch the same call sites again.
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
@@ -113,25 +120,175 @@ def _bundle_item_exc(msg):
class _BundleConfigSnapshot(NamedTuple):
"""The configured Dag bundles, as names only and as full configs."""
- configs: tuple[_ExternalBundleConfig, ...]
+ configs: dict[str, _ExternalBundleConfig]
names: frozenset[str]
+ team_names: dict[str, str | None]
[email protected]
-def _load_bundle_config_snapshot() -> _BundleConfigSnapshot:
- """Read and validate the configured Dag bundles, without importing their
classes."""
- bundle_config_list = _read_bundle_config_list()
- if not bundle_config_list:
- return _BundleConfigSnapshot(configs=(), names=frozenset())
- if conf.getboolean("core", "LOAD_EXAMPLES"):
+def _create_bundle_config_snapshot(bundle_config_list:
list[_ExternalBundleConfig]) -> _BundleConfigSnapshot:
+ """Validate the configured Dag bundles, without importing their classes."""
+ team_names = {cfg.name: cfg.team_name for cfg in bundle_config_list}
+ if bundle_config_list and conf.getboolean("core", "LOAD_EXAMPLES"):
_add_example_dag_bundle(bundle_config_list)
_add_provider_example_dags_to_bundle(bundle_config_list)
return _BundleConfigSnapshot(
- configs=tuple(bundle_config_list),
+ configs={cfg.name: cfg for cfg in bundle_config_list},
names=frozenset(cfg.name for cfg in bundle_config_list),
+ team_names=team_names,
+ )
+
+
+def _get_bundle_config_reload_option() -> tuple[str, str] | None:
+ section, key = "dag_processor", "dag_bundle_config_list"
+ env_key = "AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST"
+ if env_key in os.environ:
+ return None
+ for suffix in ("cmd", "secret"):
+ env_option = f"{env_key}_{suffix.upper()}"
+ if env_option in os.environ:
+ return suffix, os.environ[env_option]
+ if ConfigParser.has_option(conf, section, key):
+ return None
+ for suffix in ("cmd", "secret"):
+ option = f"{key}_{suffix}"
+ if ConfigParser.has_option(conf, section, option):
+ return suffix, ConfigParser.get(conf, section, option)
+ return None
Review Comment:
This is still a copy from AirflowConfigParser. Expose a supported helper in
the shared configuration module that returns the `_cmd`/`_secret` source for an
option, and build the reloading source from that.
##########
airflow-core/src/airflow/dag_processing/bundles/base.py:
##########
@@ -223,7 +224,9 @@ def remove_stale_bundle_versions(self):
log.info("checking for stale bundle versions locally")
- bundles = list(DagBundlesManager().get_all_dag_bundles())
+ manager = DagBundlesManager()
+ manager.get_all_bundle_names()
Review Comment:
This is too unobvious. Give the manager an explicit `refresh()` instead.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -2261,7 +2352,13 @@ def _create_process(
logger, logger_filehandle = self._get_logger_for_dag_file(dag_file)
subprocess_logs_to_stdout = conf.get("logging",
"dag_processor_log_target") == "stdout"
- if get_claiming_importer(dag_file.absolute_path, dag_file.bundle_name)
is not None:
+ bundle_config = None
+ if is_bundle_config_reload_enabled():
+ bundle_config =
self._dag_bundle_manager.get_bundle_parse_context(dag_file.bundle_name)
Review Comment:
`get_bundle_parse_context()` still raises `ValueError` for a name that isn't
configured. If a callback file for a bundle removed in that check is still
queued, the main loop crashes here. Purge `_callback_to_execute` for removed
bundles, or handle the error, and add a test either way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
