kaxil commented on code in PR #74789:
URL: https://github.com/apache/airflow/pull/74789#discussion_r4394356736
##########
providers/git/docs/changelog.rst:
##########
@@ -25,6 +25,7 @@ Changelog
Bug Fixes
~~~~~~~~~
+* ``Allow Dag processing to continue from unchanged Git bundle files after a
fetch failure on Airflow 3.4+ (#74789)``
Review Comment:
This sits under 1.0.1, which is already released, and bug-fix entries are
generated from the commit log at release time anyway (`providers/AGENTS.md`).
I'd drop the line.
##########
providers/git/src/airflow/providers/git/bundles/git.py:
##########
@@ -472,10 +473,20 @@ def refresh(self) -> None:
with self.lock():
cm = self.hook.configure_hook_env() if self.hook else nullcontext()
Review Comment:
With GitHub App auth, `configure_hook_env()` mints the token when it is
entered, which is outside the `try`. A failed token request therefore still
pauses parsing, although no git command has run yet. Could the token step get
the same `source_unchanged=True` wrapping?
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -926,33 +959,76 @@ def _refresh_dag_bundles(self, known_files: dict[str,
set[DagFileInfo]]):
current_version_matches_db = True
previously_seen = bundle.name in self._bundle_versions
- if self.should_skip_refresh(
- bundle=bundle,
- elapsed_time_since_refresh=elapsed_time_since_refresh,
- current_version_matches_db=current_version_matches_db,
- previously_seen=previously_seen,
+ if (
+ bundle.name not in self._bundles_waiting_for_refresh
+ and bundle.name not in self._bundles_needing_refresh
+ and bundle.name in self._bundle_parse_sources
+ and self.should_skip_refresh(
+ bundle=bundle,
+ elapsed_time_since_refresh=elapsed_time_since_refresh,
+ current_version_matches_db=current_version_matches_db,
+ previously_seen=previously_seen,
+ )
):
self.log.debug("Not time to refresh bundle %s", bundle.name)
continue
+ if bundle.refresh_changes_source and bundle.name in active_bundles:
+ # Forced refreshes wait too: refreshing under an import would
discard its result.
+ self._bundles_waiting_for_refresh.add(bundle.name)
+ continue
+
+ self._bundles_waiting_for_refresh.discard(bundle.name)
self.log.info("Refreshing bundle %s", bundle.name)
+ previous_source = self._bundle_parse_sources.get(bundle.name)
+ if bundle.refresh_changes_source:
+ self._invalidate_bundle_parse_source(bundle.name)
try:
bundle.refresh()
any_refreshed = True
- except Exception:
+ except Exception as error:
self.log.exception("Error refreshing bundle %s", bundle.name)
+ source_unchanged = not bundle.refresh_changes_source or (
Review Comment:
Nothing tests the `not bundle.refresh_changes_source` half: deleting it
leaves `tests/unit/dag_processing` green, since `_make_refresh_bundle` sets the
attribute to True and the `LocalDagBundle` test only covers a refresh that
succeeds. Without it, a forced refresh that raises on such a bundle leaves it
in `_force_refresh_bundles`, and `_get_dispatch_source` then holds every file
in it. Could one case raise from `refresh()` with `refresh_changes_source =
False` and `force=True`, and assert the file still dispatches with the previous
source?
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -926,33 +959,76 @@ def _refresh_dag_bundles(self, known_files: dict[str,
set[DagFileInfo]]):
current_version_matches_db = True
previously_seen = bundle.name in self._bundle_versions
- if self.should_skip_refresh(
- bundle=bundle,
- elapsed_time_since_refresh=elapsed_time_since_refresh,
- current_version_matches_db=current_version_matches_db,
- previously_seen=previously_seen,
+ if (
+ bundle.name not in self._bundles_waiting_for_refresh
+ and bundle.name not in self._bundles_needing_refresh
Review Comment:
During a Git fetch outage the bundle stays in `_bundles_needing_refresh`, so
this skips `should_skip_refresh` and the refresh retries on every
`bundle_refresh_check_interval` (5s by default). Each retry first waits at line
976 for the bundle's in-flight parses to finish, so with one slow file the
bundle stalls every 5s for as long as the outage lasts. Could a bundle that
kept its source wait `refresh_interval` since its last failed attempt before
retrying?
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1624,9 +1819,8 @@ def prepare_file_queue(self, known_files: dict[str,
set[DagFileInfo]]):
Note this method is only called when the file path queue is empty
"""
- # We only emit metrics after processing all files in the queue. If
`self._parsing_start_time` is None
- # when this method is called, no files have yet been added to the
queue so we shouldn't emit metrics.
- if self._parsing_start_time is not None:
+ # Deferred files are still part of this queue cycle even while other
bundles refill the queue.
+ if self._parsing_start_time is not None and not self._file_queue and
not self._files_awaiting_source:
Review Comment:
A bundle paused after a failed refresh (S3, GCS, a failed Git reset, or a
reparse request on a Git bundle during a fetch outage) keeps files in
`_files_awaiting_source` until a refresh succeeds. So this stops
`total_parse_time`, `dagbag_size` and `dag_processing.import_errors` for every
bundle in the deployment for the whole outage, and the first reading afterwards
spans all of it. Could this wait only on files held for a drain
(`_bundles_waiting_for_refresh` / `_force_refresh_bundles`) and leave paused
bundles out?
--
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]
