shahar1 commented on code in PR #74915:
URL: https://github.com/apache/airflow/pull/74915#discussion_r4367631126
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
@@ -691,6 +825,10 @@ def get_bundle(
:return: The DAG bundle.
"""
+ if name not in self._bundle_config:
+ if name not in _get_cached_bundle_config_snapshot().names:
+ _refresh_bundle_config_snapshot(force=True)
Review Comment:
**minor** — Every `get_bundle()` call for a name that is not configured
forces a fresh read of the command or secret, and ignores
`bundle_refresh_check_interval`. Nothing limits how often this happens. In the
Dag processor this runs in the main loop: `prepare_callback_bundle()` calls
`get_bundle()` once for each versioned callback. After a bundle is removed,
every callback for its in-flight runs starts a synchronous command or
secrets-backend call. A slow or remote source then stalls parsing. I confirmed
it at 42d2b4ef with a mocked `run_command`: five `get_bundle("removed-bundle")`
calls ran the command five more times, after the one read at startup.
**Suggestion:** Rate-limit the forced read as well, e.g. at most once per
`bundle_refresh_check_interval`, or remember the names a forced read did not
find until the next scheduled read. Add a test that two lookups of the same
unknown name read the source only once.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1003,15 +1036,65 @@ def _refresh_dag_bundles(self, known_files: dict[str,
set[DagFileInfo]]):
self._resort_file_queue()
self._add_new_files_to_queue(known_files=known_files)
+ def _reconcile_bundles(self, known_files: dict[str, set[DagFileInfo]]) ->
None:
+ if not is_bundle_config_reload_enabled():
+ return
+ try:
+ dag_bundle_manager = DagBundlesManager()
+ bundle_names = set(dag_bundle_manager.get_all_bundle_names())
+ except Exception:
+ self.log.exception("Could not load Dag bundles; keeping the
current bundles")
+ return
+
+ loaded_bundles = {bundle.name: bundle for bundle in self._dag_bundles}
+ if self.bundle_names_to_parse:
+ bundle_names.intersection_update(self.bundle_names_to_parse)
+ removed_names = loaded_bundles.keys() - bundle_names
+ bundle_instances: dict[str, BaseDagBundle] = {}
+ for name in sorted(bundle_names - loaded_bundles.keys()):
+ try:
+ bundle_instances[name] = dag_bundle_manager.get_bundle(name)
+ except Exception:
+ self.log.exception("Could not create Dag bundle %s; will
retry", name)
+
+ try:
+ # Other processors can overwrite active state even when our
configuration is unchanged.
+ dag_bundle_manager.sync_bundles_to_db(
+ deactivate_missing=not self.bundle_names_to_parse,
+ bundle_instances=bundle_instances,
+ )
+ except Exception:
+ self.log.exception("Could not reconcile Dag bundles; keeping the
current bundles")
Review Comment:
**minor** — `_refresh_bundle_config_snapshot()` accepts a new bundle without
checking that its `team_name` exists. `sync_bundles_to_db()` then raises `Team
'...' does not exist` on every check, and `_reconcile_bundles()` returns before
it applies anything. One typo in one new bundle's team therefore blocks all
unrelated additions and removals in this processor until the list changes
again. The author also cannot fix the typo in place, because changing an
accepted bundle's definition is rejected ("Use a new bundle name ..."). The
bundle has to be removed or renamed. In the meantime the API server and workers
have already accepted the list with that bundle in it.
**Suggestion:** Check that the teams exist when the list is accepted (the
Dag processor has DB access), so the bad list is rejected as a whole like an
invalid classpath. Or skip only the failing bundle and apply the rest of the
list. Either way, add a test with a new bundle whose team does not exist next
to a removal.
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
@@ -387,40 +498,49 @@ def _extract_and_sign_template(bundle_name: str) ->
tuple[str | None, dict]:
self.log.debug("Signed URL template for bundle %s",
bundle_name)
return new_template_, new_params_
- stored = {b.name: b for b in
session.scalars(select(DagBundleModel)).all()}
+ stored = {
+ b.name: b
+ for b in
session.scalars(select(DagBundleModel).options(selectinload(DagBundleModel.teams))).all()
+ }
bundle_to_team = {
bundle.name: bundle.teams[0].name if len(bundle.teams) == 1 else
None
for bundle in stored.values()
}
+ requested_teams = {config.team_name for config in
self._bundle_config.values() if config.team_name}
+ teams = {
+ team.name: team for team in
session.scalars(select(Team).where(Team.name.in_(requested_teams)))
+ }
+
for name, config in self._bundle_config.items():
team: Team | None = None
if config.team_name:
- team = session.scalars(select(Team).where(Team.name ==
config.team_name)).one_or_none()
+ team = teams.get(config.team_name)
if not team:
raise _bundle_item_exc(f"Team '{config.team_name}' does
not exist")
- try:
- new_template, new_params = _extract_and_sign_template(name)
- except Exception as e:
- self.log.exception("Error creating bundle '%s': %s", name, e)
- continue
-
- if bundle := stored.pop(name, None):
- bundle.active = True
+ bundle = stored.pop(name, None)
+ update_template = bundle_instances is None or name in
bundle_instances
+ if update_template:
+ try:
+ new_template, new_params = _extract_and_sign_template(name)
+ except Exception as e:
+ self.log.exception("Error creating bundle '%s': %s", name,
e)
+ if bundle is None and bundle_instances is None:
+ continue
+ update_template = False
Review Comment:
**nit** — A newly added bundle is signed only in the reconcile that adds it,
while it is still in `bundle_instances`. If `view_url_template()` or the
parameter extraction fails at that moment, the row is created active and has no
template. Later reconciles pass that bundle as already loaded, so
`update_template = name in bundle_instances` stays `False`. The URL is then
missing until the processor restarts. This differs from the "failed
construction is retried" behaviour in the docs.
**Suggestion:** Retry signing while the stored template is missing, or keep
the bundle in a "needs signing" set until signing succeeds.
##########
airflow-core/src/airflow/dag_processing/processor.py:
##########
@@ -116,6 +117,9 @@ class DagFileParseRequest(BaseModel):
bundle_name: str
"""Bundle name for team-specific executor validation."""
+ bundle_config: dict[str, Any] | None = None
Review Comment:
**minor** — The PR body says the change adds no "`DagFileParseRequest`
change". The diff does change it: it adds `bundle_config` to
`DagFileParseRequest` and to `StartLangSDKRuntime`, adds a `VersionChange`
(`AddBundleConfigToDagFileParseRequest`) to the in-progress supervisor schema
version, and regenerates the Go and TypeScript schemas. The schema steps follow
`task-sdk/src/airflow/sdk/execution_time/schema/AGENTS.md`, so the code is
fine. But reviewers of the lang-SDK contract need to know about this change.
The field is also an untyped `dict[str, Any]` in the cross-language schema,
although only the Python parser reads it.
**Suggestion:** Correct the body and say why the field is needed: parse
children must not read the bundle config source again. Consider a typed model
(`importers`, `team_name`) instead of `dict[str, Any]`, so the cross-language
schema says what the field holds.
--
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]
