kaxil commented on code in PR #74789:
URL: https://github.com/apache/airflow/pull/74789#discussion_r4336431581
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -926,17 +949,28 @@ 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 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.name in active_bundles:
Review Comment:
This wait also applies to `LocalDagBundle`, the default dags-folder bundle.
Its `refresh()` is a no-op and its version is always `None`, so neither failure
this PR guards against can happen there. Even so, every periodic refresh (300s
by default) and every UI reparse request (which force-refreshes the bundle) now
stops new dispatches for the whole bundle until its slowest in-flight parse
finishes, which can take up to `dag_file_processor_timeout`. The requested file
waits too. Under `airflow standalone` the dags folder refreshes on every 5s
check, so one slow file can leave the other slots idle for most of each cycle.
S3 and GCS sync files in place, so they do need the wait. Could bundles
declare whether `refresh()` changes files on disk (say a class attribute
defaulting to True, set to False on `LocalDagBundle`), and skip both this wait
and `_invalidate_bundle_parse_source` when it doesn't?
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -926,17 +949,28 @@ 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 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.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)
+ self._invalidate_bundle_parse_source(bundle.name)
Review Comment:
Because this runs before `refresh()`, a refresh that raises now pauses all
parsing of the bundle until a later refresh succeeds, UI reparse requests
included (they wait in `_files_awaiting_source`). The only thing logged is the
existing `Error refreshing bundle` line, which used to mean parsing carried on
against the current checkout.
For `GitDagBundle`, a remote outage or an expired token fails in `fetch`,
before `head.reset` touches the working tree. Could a failed refresh of a
versioned bundle restore the source (with the bumped generation) when
`get_current_version()` still equals the pre-refresh version? And if the pause
stays, I'd log a warning that parsing of bundle X is paused until a refresh
succeeds, and tag `files_awaiting_source` with `bundle_name`, so on-call can
tell why that bundle's Dags stopped updating.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1783,11 +1921,16 @@ def _add_files_to_queue(
stats.gauge("dag_processing.file_path_queue_size",
len(self._file_queue))
def max_runs_reached(self):
- """:return: whether all file paths have been processed max_runs
times."""
+ """Return whether all files reached max_runs; failed refreshes are not
parsing attempts."""
if self.max_runs == -1: # Unlimited runs.
return False
if self._num_run < self.max_runs:
return False
+ if any(
+ file not in self._callback_to_execute
+ for file in [*self._file_queue, *self._files_awaiting_source]
+ ) or any(not proc.had_callbacks for proc in self._processors.values()):
Review Comment:
If I delete this `had_callbacks` clause, or the `_force_refresh_bundles`
check in `_get_dispatch_source` (line 1609), the whole
`tests/unit/dag_processing` directory still passes. Could each one get a test?
For this one: an empty queue, a stat with `run_count >= max_runs`, and one
in-flight processor with `had_callbacks=False`, asserting that
`max_runs_reached()` is False. For the other: a file queued for a bundle in
`_force_refresh_bundles` that still has a parse source, asserting that
`_start_new_processes` holds the file.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -249,13 +253,18 @@ class DagFileProcessorManager(LoggingMixin):
"""An overridable heartbeat called once every time around the loop"""
_file_queue: OrderedDict[DagFileInfo, None] =
attrs.field(factory=OrderedDict, init=False)
+ _priority_files: set[DagFileInfo] = attrs.field(factory=set, init=False)
+ _files_awaiting_source: dict[DagFileInfo, None] =
attrs.field(factory=dict, init=False)
+ """Files held out of the queue until their bundle has a usable source, so
they never stop it emptying."""
Review Comment:
Letting the queue empty affects `dag_processing.total_parse_time`. Once a
waiting bundle's queued files move here, `prepare_file_queue` sees an empty
queue and emits the metric for a cycle that hasn't finished. Then
`_add_files_to_queue` starts a new timer when the files are readmitted. In a
single-bundle deployment, each deferral splits one cycle into two short
readings. Should the emit wait until both `_file_queue` and
`_files_awaiting_source` are empty?
##########
shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml:
##########
@@ -122,6 +122,14 @@ metrics:
legacy_name: "-"
name_variables: []
+ - name: "dag_processing.results_discarded_on_refresh"
+ description: "Parse attempts discarded because bundle initialization or
refresh overlapped the import.
+ Metric with bundle_name tagging, and team_name tagging in multi-team mode.
+ Repeated discards can indicate that refreshes are delaying parsing."
Review Comment:
I couldn't find a way for this counter to go above zero with in-tree
bundles. The generation only changes in `_invalidate_bundle_parse_source`. Its
refresh call site sits behind the `active_bundles` wait, and its init call site
needs `is_initialized` to flip back to False, which nothing in-tree does.
Callback-only runs are exempt from the discard. An operator who follows this
description would watch a counter that stays at zero, while refresh delays
actually show up in `dag_processing.files_awaiting_source`. If the discard is a
backstop, could the description say it should stay at 0 and point at
`files_awaiting_source` instead?
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -2066,6 +2073,478 @@ def test_terminate_normalizes_file_path_stats_tag(self):
)
processor.kill.assert_called_once_with(signal.SIGTERM,
escalation_delay=5.0)
+ @mock.patch.object(DagFileProcessorManager, "persist_parsing_result",
autospec=True)
+ @mock.patch.object(DagFileProcessorManager, "_create_process",
autospec=True)
+ def test_persistence_uses_the_source_captured_at_dispatch(self, create,
persist, session):
+ manager = DagFileProcessorManager(max_runs=1)
+ source = DagParseSource(bundle_version="v1", version_data={"files":
{"dag.py": "original"}})
+ manager._bundle_parse_sources["testing"] = source
+ file = DagFileInfo(bundle_name="testing", rel_path=Path("dag.py"),
bundle_path=TEST_DAGS_FOLDER)
+ manager._file_queue[file] = None
+ processor, _ = self.mock_processor()
+ processor.had_callbacks = False
+ processor.parsing_result = DagFileParsingResult(fileloc="dag.py",
serialized_dags=[])
+ create.return_value = processor
+
+ manager._start_new_processes()
+ dispatched = next(iter(manager._processors))
+ assert processor.parse_source is source
+ manager._bundle_parse_sources["testing"] = DagParseSource(
+ bundle_version="v2", version_data={"files": {"dag.py": "changed"}}
+ )
+ manager._bundle_versions["testing"] = "v2"
+ manager.handle_parsing_result(dispatched, processor, session=session)
+
+ assert dispatched == file
+ persist.assert_called_once_with(
+ manager,
+ bundle_name="testing",
+ bundle_version="v1",
+ version_data={"files": {"dag.py": "original"}},
+ parsing_result=processor.parsing_result,
+ run_duration=mock.ANY,
+ relative_fileloc="dag.py",
+ session=session,
+ )
+
+ @pytest.mark.parametrize("current_version", [None, "v1", "v2"])
+ @mock.patch.object(DagFileProcessorManager, "persist_parsing_result",
autospec=True)
+ @mock.patch.object(DagFileProcessorManager, "_create_process",
autospec=True)
+ def test_forced_refresh_waits_for_in_flight_imports(self, create, persist,
current_version):
+ manager = DagFileProcessorManager(max_runs=1)
+ bundle = self._make_refresh_bundle(
+ supports_versioning=current_version is not None,
current_version=current_version
+ )
+ manager._bundle_versions[bundle.name] = "v1"
+ file = DagFileInfo(bundle_name=bundle.name, rel_path=Path("dag.py"),
bundle_path=bundle.path)
+ manager._file_stats[file] = DagFileStat()
+ manager._file_queue = OrderedDict.fromkeys([file])
+ manager._bundle_parse_sources = {bundle.name: DagParseSource()}
+ processor = self.mock_processor()[0]
+ processor.had_callbacks = False
+ processor.parsing_result = DagFileParsingResult(fileloc="dag.py",
serialized_dags=[])
+ create.return_value = processor
+ manager._start_new_processes()
+
+ self._refresh_with_mocked_state(
+ manager, bundle, BundleState(last_refreshed=None, version="v1"),
force=True
+ )
+ bundle.refresh.assert_not_called()
+ assert manager._bundles_waiting_for_refresh == {bundle.name}
+
+ manager._collect_results()
+ persist.assert_called_once()
+ assert not manager._file_queue
+
+ self._refresh_with_mocked_state(
+ manager, bundle, BundleState(last_refreshed=None, version="v1"),
force=True
+ )
+ bundle.refresh.assert_called_once_with()
+
+ @pytest.mark.parametrize("has_result", [False, True], ids=["crashed",
"result"])
Review Comment:
`current_version` never reaches the code under test here. With `initialize`
raising, `_refresh_dag_bundles` hits `continue` before anything reads
`supports_versioning` or the version, and `handle_parsing_result` only reads
`proc.parse_source`. So three of the six cases repeat the other three. Could
this drop the parametrize and use `_make_refresh_bundle()`?
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1052,11 +1097,27 @@ def _get_observed_filelocs(present: set[DagFileInfo])
-> set[str]:
return {loc for file in present for loc in (str(file.rel_path),
*file.definition_locs)}
def deactivate_deleted_dags(self, bundle_name: str, present:
set[DagFileInfo]) -> None:
- """Deactivate DAGs that come from files that are no longer present in
bundle."""
+ """
+ Deactivate Dags whose files are no longer present in the bundle.
+
+ Failed discovery does not prove deletion; later import failures and
stale-Dag checks still apply.
+ """
observed_filelocs = self._get_observed_filelocs(present)
+ failed_sources = {loc for file in present for loc in
file.failed_sources}
with create_session() as session:
try:
with with_db_lock_timeout(session=session, lock_timeout=30):
+ if failed_sources:
+ observed_filelocs.update(
+ loc
+ for loc in session.scalars(
+ select(DagModel.relative_fileloc).where(
+ DagModel.bundle_name == bundle_name,
+ DagModel.relative_fileloc.is_not(None),
+ )
+ )
+ if loc is not None and
is_source_discovery_failed(loc, failed_sources)
Review Comment:
`loc is not None` repeats the `.is_not(None)` filter in the query above, and
`is_source_discovery_failed` already returns False for `None`. This could keep
just one of the three checks.
--
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]
