potiuk commented on code in PR #74915:
URL: https://github.com/apache/airflow/pull/74915#discussion_r4437639477
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
@@ -108,28 +115,116 @@ def _bundle_item_exc(msg):
)
-class _BundleConfigSnapshot(NamedTuple):
- """The configured Dag bundles, as names only and as full configs."""
+@dataclass
+class _BundleConfigSnapshot:
+ """The last accepted bundle configuration in this process."""
- configs: tuple[_ExternalBundleConfig, ...]
+ configs: dict[str, _ExternalBundleConfig]
names: frozenset[str]
+ team_names: dict[str, str | None]
+ last_checked: float = field(default_factory=lambda: time.monotonic())
+
+
+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
[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:
+ 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 is_bundle_config_reload_enabled() -> bool:
+ """Return whether bundle configuration comes from a command or secrets
backend."""
+ return _get_bundle_config_reload_option() is not None
+
+
[email protected]
+def _get_cached_bundle_config_snapshot(*, metadata_only: bool = False) ->
_BundleConfigSnapshot:
+ """Return the cached bundle snapshot, reading configuration on first
use."""
+ if metadata_only:
+ return replace(_get_cached_bundle_config_snapshot())
+ return _create_bundle_config_snapshot(_read_bundle_config_list())
+
+
+def _refresh_bundle_config_snapshot(
+ *,
+ force: bool = False,
+ metadata_only: bool = False,
+ snapshot: _BundleConfigSnapshot | None = None,
+) -> _BundleConfigSnapshot:
+ cached = (
+ _get_cached_bundle_config_snapshot(metadata_only=True)
+ if metadata_only
+ else _get_cached_bundle_config_snapshot()
+ )
+ if snapshot is None:
+ snapshot = cached
+ if not is_bundle_config_reload_enabled():
+ return snapshot
+ now = time.monotonic()
+ try:
+ if not force and now - cached.last_checked < conf.getint(
+ "dag_processor", "bundle_refresh_check_interval"
+ ):
+ if snapshot.configs is cached.configs:
+ return snapshot
+ config_list = [cached.configs[name] for name in cached.team_names]
+ else:
+ cached.last_checked = now
+ config_list = _read_bundle_config_list()
+ if {config.name: config for config in config_list} == {
+ name: snapshot.configs[name] for name in snapshot.team_names
+ }:
+ current = snapshot
+ else:
+ for config in config_list:
+ existing = snapshot.configs.get(config.name)
+ if existing is not None:
+ if existing != config:
+ raise _bundle_item_exc(
+ f"Use a new bundle name when changing
configuration for '{config.name}'"
Review Comment:
**minor** — A changed entry for an existing bundle name rejects the whole
read, not just that bundle: the `raise` here falls through to
`log.exception(...)` at line 220, so any additions and removals published in
the same list are not applied either, and the same error is logged on every
`bundle_refresh_check_interval` until the list is corrected or the processor
restarts. `dag-bundles.rst:67` says to use a new bundle name when changing
configuration, but not that the rest of the list is held back. Could the docs
(and ideally the log message) say so, e.g. "the whole list is rejected until
the changed entry is reverted or renamed"? Applying the other
additions/removals while skipping only the changed bundle would also be
reasonable, if you prefer that behaviour.
--
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]
