kaxil commented on code in PR #74793:
URL: https://github.com/apache/airflow/pull/74793#discussion_r4469209439
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1442,41 +1795,317 @@ def handle_parsing_result(
team_name=team_name,
)
- if proc.parsing_result is not None:
+ try:
+ if self.api_client is not None and not is_callback_only:
+ if proc.parsing_result is not None and proc.api_request_error
is None:
+ self._pending_publications[file] =
self._build_pending_publication(
+ file, proc, run_duration, next_stat
+ )
+ else:
+ self._queue_failed_parse_result(
+ file,
+ proc,
+ run_duration,
+ next_stat,
+ reason="the parsing process exited without a result",
+ )
+ return
+ if proc.parsing_result is not None:
+ with create_session() if session is None else
nullcontext(session) as session:
+ self.persist_parsing_result(
+ bundle_name=file.bundle_name,
+ bundle_version=proc.parse_source.bundle_version,
+ version_data=(
+ dict(proc.parse_source.version_data)
+ if proc.parse_source.version_data is not None
+ else None
+ ),
+ parsing_result=proc.parsing_result,
+ run_duration=run_duration,
+ relative_fileloc=str(file.rel_path),
+ session=session,
+ )
+ except Exception:
+ self.log.exception(
+ "Failed to persist parsing result for %s in bundle %s; "
+ "keeping previous persisted stats while throttling retries. "
+ "Other files in this cycle are still processed.",
+ str(file.rel_path),
+ file.bundle_name,
+ )
+ self._record_failed_publication(file, next_stat)
+ return
+
+ self._file_stats[file] = next_stat
+
+ def _queue_failed_parse_result(
+ self,
+ file: DagFileInfo,
+ proc: BaseDagFileProcessorProcess,
+ run_duration: float,
+ stat: DagFileStat,
+ *,
+ reason: str,
+ ) -> None:
+ source = proc.parse_source
+ if source.refresh_generation !=
self._bundle_refresh_generations.get(file.bundle_name, 0):
+ self._add_files_to_queue([file], mode="front")
+ return
+ if proc.api_request_error is not None:
+ # The Task SDK answers a failed request as a missing value, so the
Dags may hold a fallback.
+ reason = f"an Execution API request failed while parsing:
{proc.api_request_error}"
+ # A restarted manager cannot see when the file's Dags were last
parsed, so it waits from its start.
+ last_published = self._file_stats[file].last_finish_time or
self._started_at
+ if timezone.utcnow() - last_published <=
timedelta(seconds=self.stale_dag_threshold):
Review Comment:
This waits on this processor's own last publication, but the import error it
then sends makes `_update_import_errors` mark every Dag at the file stale with
no freshness check. With two Dag processor replicas, one that keeps timing out
on a file deactivates the Dags the other replica published seconds earlier,
once per threshold window. Could the grace period move to the server, so a
failure report only marks Dags stale whose `last_parsed_time` is past the
cutoff, like the missing-Dag loop in `dag_parsing.py` already does?
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -6188,6 +6281,361 @@ def
test_a_sweep_pays_the_fixed_cost_once_per_call(self, session, testing_dag_bu
f"({SWEEP_CALLS} x {fixed} fixed + {SWEEP_FILES} x {per_dag} per
Dag)."
)
+ @pytest.mark.parametrize(("publication_fails", "expected_num_dags"),
[(False, 0), (True, 2)])
+ @mock.patch.object(DagFileProcessorManager, "persist_parsing_result",
autospec=True)
+ @mock.patch.object(DagFileProcessorManager, "_create_process",
autospec=True)
+ def test_api_publication_uses_dispatch_identity_without_local_persistence(
+ self, create, persist, session, publication_fails, expected_num_dags
+ ):
+ client = mock.create_autospec(DagProcessorAPIClient, instance=True)
+ manager = DagFileProcessorManager(max_runs=1, api_client=client)
+ manager._bundle_parse_sources["testing"] =
DagParseSource(bundle_version="v1")
+ file = _get_file_infos(["dag.py"])[0]
+ manager._file_stats[file] = DagFileStat(num_dags=2)
+ manager._file_queue[file] = None
+ proc, read_end = self.mock_processor()
+ proc.had_callbacks = False
+ proc.parsing_result = DagFileParsingResult(fileloc="dag.py",
serialized_dags=[])
+ proc.parse_source = manager._bundle_parse_sources["testing"]
+ create.return_value = proc
+ if publication_fails:
+ client.publish_parse_result.side_effect =
RuntimeError("publication failed")
+ try:
+ manager._start_new_processes()
+ create.assert_called_once_with(manager, file,
source=proc.parse_source)
+ manager.handle_parsing_result(file, proc, session=session)
+ client.publish_parse_result.assert_not_called()
+ manager._publish_pending_results()
+ body = client.publish_parse_result.call_args.args[0]
+ assert body.attempt_id == proc.id
+ assert body.dispatch_sequence == proc.dispatch_sequence == 1
+ assert body.bundle_version == "v1"
+ assert body.relative_fileloc == "dag.py"
+ assert body.serialized_dags == []
+ assert manager._file_stats[file].run_count == 1
+ assert manager._file_stats[file].num_dags == expected_num_dags
+ persist.assert_not_called()
+ finally:
+ read_end.close()
+ proc.close()
+
+ def test_api_publication_reports_a_result_the_api_would_refuse(self,
session):
+ client = mock.create_autospec(DagProcessorAPIClient, instance=True)
+ manager = DagFileProcessorManager(max_runs=1, api_client=client)
+ file = _get_file_infos(["dag.py"])[0]
+ dag = SdkDAG("published", schedule=None)
+ dag.fileloc = "/worker/dag.py"
+ dag.relative_fileloc = "dag.py"
+ # Mutating the set skips the SDK's constructor check, so only the API
model rejects the tag.
+ dag.tags.add("x" * 101)
+ revision = uuid7()
+ proc, read_end = self.mock_processor()
+ proc.had_callbacks = False
+ proc.dispatch_sequence = 1
+ proc.parse_source = DagParseSource(bundle_version="v1",
bundle_revision=revision)
+ proc.parsing_result = DagFileParsingResult(
+ fileloc=dag.fileloc,
serialized_dags=[LazyDeserializedDAG.from_dag(dag)]
+ )
+ try:
+ manager.handle_parsing_result(file, proc, session=session)
+ manager._publish_pending_results()
+ finally:
+ read_end.close()
+ proc.close()
+
+ report = client.publish_parse_result.call_args.args[0]
+ assert report.serialized_dags == []
+ assert (report.bundle_version, report.bundle_revision) == ("v1",
revision)
+ assert report.import_errors == {
+ "dag.py": "The parsing result was not published: it is not a valid
Dag publication "
+ "(Value error, Dag tags must be strings of at most 100
characters)."
+ }
+ assert manager._file_stats[file].import_errors == 1
+ assert not manager._pending_publications
+
+ @pytest.mark.parametrize("source_available", [True, False])
+ def
test_api_publication_carries_diagnostics_and_explicit_source_availability(self,
source_available):
+ client = mock.create_autospec(DagProcessorAPIClient, instance=True)
+ manager = DagFileProcessorManager(max_runs=1, api_client=client)
+ file = _get_file_infos(["dag.py"])[0]
+ dag = SdkDAG("published", schedule=None)
+ dag.fileloc = "/worker/dag.py"
+ dag.relative_fileloc = "dag.py"
+ proc, read_end = self.mock_processor()
+ proc.dispatch_sequence = 1
+ warning = {"dag_id": dag.dag_id, "warning_type": "python:deprecated",
"message": "warning"}
+ unserialized_warning = {"dag_id": "failed", "warning_type":
"python:deprecated", "message": "warning"}
+ proc.parsing_result = DagFileParsingResult(
+ fileloc=dag.fileloc,
+ serialized_dags=[LazyDeserializedDAG.from_dag(dag)],
+ warnings=[warning, unserialized_warning],
+ parsed_definitions=["dag.py"],
+ import_errors={"dag.py": "import error"},
+ dag_source_codes={dag.dag_id:
DagSourceCode(source_code="captured", language="python")}
+ if source_available
+ else {},
+ )
+ try:
+ body = manager._build_parse_result(file, proc, 0.5)
+ assert [published.model_dump() for published in body.warnings] ==
[warning]
+ assert body.import_errors == {"dag.py": "import error"}
+ assert body.parsed_definitions == ["dag.py"]
+ assert
body.source_codes[body.dag_source_indexes[dag.dag_id]].source_code == (
+ "captured" if source_available else None
+ )
+ finally:
+ read_end.close()
+ proc.close()
+
+ @pytest.mark.parametrize("with_valid_member", [True, False],
ids=["beside-valid-member", "only-member"])
+ def
test_api_publication_drops_container_members_with_unrepresentable_paths(
+ self, with_valid_member, caplog
+ ):
+ client = mock.create_autospec(DagProcessorAPIClient, instance=True)
+ manager = DagFileProcessorManager(max_runs=1, api_client=client)
+ file = _get_file_infos(["archive.zip"])[0]
+ invalid_location = r"archive.zip/sub\invalid.py"
+ invalid_error_location = r"archive.zip/sub\broken.py"
+ locations = ["archive.zip/valid.py", invalid_location] if
with_valid_member else [invalid_location]
+ dags = []
+ for index, location in enumerate(locations):
+ dag = SdkDAG(f"dag_{index}", schedule=None)
+ dag.fileloc = f"/worker/{location}"
+ dag.relative_fileloc = location
+ dags.append(dag)
+ import_errors = {invalid_error_location: "import error"}
+ if with_valid_member:
+ import_errors["archive.zip/broken.py"] = "import error"
+ proc, read_end = self.mock_processor()
+ proc.dispatch_sequence = 1
+ proc.parsing_result = DagFileParsingResult(
+ fileloc="/worker/archive.zip",
+ serialized_dags=[LazyDeserializedDAG.from_dag(dag) for dag in
dags],
+ warnings=[
+ {"dag_id": dag.dag_id, "warning_type": "python:deprecated",
"message": "warning"}
+ for dag in dags
+ ],
+ parsed_definitions=locations,
+ import_errors=import_errors,
+ dag_source_codes={
+ dag.dag_id: DagSourceCode(source_code=dag.dag_id,
language="python") for dag in dags
+ },
+ )
+ try:
+ body = manager._build_parse_result(file, proc, 0.5)
+ finally:
+ read_end.close()
+ proc.close()
+
+ published = ["dag_0"] if with_valid_member else []
+ assert [dag["dag"]["dag_id"] for dag in body.serialized_dags] ==
published
+ assert body.parsed_definitions == locations[:-1]
+ assert body.import_errors == ({"archive.zip/broken.py": "import
error"} if with_valid_member else {})
+ assert [warning.dag_id for warning in body.warnings] == published
+ assert [source.source_code for source in body.source_codes] ==
published
+ assert {
+ "event": "Not publishing definitions with an unrepresentable API
path",
+ "relative_fileloc": "archive.zip",
+ "locations": [invalid_error_location, invalid_location],
+ } in caplog
+
+ @pytest.mark.parametrize("different_language", [False, True])
+ def test_api_publication_deduplicates_factory_source(self,
different_language):
+ client = mock.create_autospec(DagProcessorAPIClient, instance=True)
+ manager = DagFileProcessorManager(max_runs=1, api_client=client)
+ file = _get_file_infos(["factory.py"])[0]
+ source_text = "#" * (42 * 1024)
+ dags = []
+ sources = {}
+ for index in range(400):
+ dag = SdkDAG(f"factory_{index}", schedule=None)
+ dag.fileloc = "/worker/factory.py"
+ dag.relative_fileloc = "factory.py"
+ dags.append(LazyDeserializedDAG.from_dag(dag))
+ sources[dag.dag_id] = DagSourceCode(
+ source_code=source_text, language="sql" if different_language
and index == 0 else "python"
+ )
+ proc, read_end = self.mock_processor()
+ proc.dispatch_sequence = 1
+ proc.parsing_result = DagFileParsingResult(
+ fileloc="/worker/factory.py", serialized_dags=dags,
dag_source_codes=sources
+ )
+ try:
+ body = manager._build_parse_result(file, proc, 0.5)
+ finally:
+ read_end.close()
+ proc.close()
+
+ assert len(source_text.encode()) * len(dags) > MAX_PUBLICATION_BYTES
+ assert len(body.model_dump_json().encode()) < MAX_PUBLICATION_BYTES
+ assert len(body.source_codes) == (2 if different_language else 1)
+ assert set(body.dag_source_indexes) == set(sources)
+ for dag_id, source in sources.items():
+ published = body.source_codes[body.dag_source_indexes[dag_id]]
+ assert (published.source_code, published.language) ==
(source.source_code, source.language)
+
+ @conf_vars({("workers", "execution_api_retries"): "1"})
+ @pytest.mark.parametrize("failure", [413, 422, 403, 500, "timeout"])
+ def test_rejected_publication_preserves_previous_success_stats(self,
time_machine, failure):
+ time_machine.move_to("2026-10-05T12:00:00Z", tick=False)
+ client = mock.create_autospec(DagProcessorAPIClient, instance=True)
+ request = httpx2.Request("POST", "http://api/jobs/1/parse-results";)
+ client.publish_parse_result.side_effect = (
+ httpx2.ReadTimeout("Lost response", request=request)
+ if failure == "timeout"
+ else httpx2.HTTPStatusError("Rejected", request=request,
response=httpx2.Response(failure))
+ )
+ manager = DagFileProcessorManager(max_runs=1, api_client=client)
+ file = _get_file_infos(["dag.py"])[0]
+ previous = timezone.utcnow() -
timedelta(seconds=manager.stale_dag_threshold + 1)
+ manager._file_stats[file] = DagFileStat(num_dags=1,
last_finish_time=previous)
+ dag = SdkDAG("published", schedule=None)
+ dag.fileloc = "/worker/dag.py"
+ dag.relative_fileloc = "dag.py"
+ proc, read_end = self.mock_processor()
+ proc.dispatch_sequence = 1
+ proc.parsing_result = DagFileParsingResult(
+ fileloc=dag.fileloc,
serialized_dags=[LazyDeserializedDAG.from_dag(dag)]
+ )
+ try:
+ manager.handle_parsing_result(file, proc)
+ manager._publish_pending_results()
+ assert manager._file_stats[file].last_finish_time == previous
+ assert manager._file_stats[file].num_dags == 1
+ assert manager.processed_recently(timezone.utcnow(), file)
+ assert not manager._pending_publications
+ finally:
+ read_end.close()
+ proc.close()
+
+ @pytest.mark.parametrize("status_code", [None, 408, 429])
+ def test_publication_retries_leave_the_loop_responsive(self, mocker,
status_code):
+ client = mock.create_autospec(DagProcessorAPIClient, instance=True)
+ error = (
+ httpx2.ReadTimeout("Lost acknowledgment")
+ if status_code is None
+ else _make_status_error(status_code)
+ )
+ client.publish_parse_result.side_effect = [error, None, None]
+ heartbeat_observations = []
+ with selectors.DefaultSelector() as selector:
+ manager = DagFileProcessorManager(
+ max_runs=1,
+ api_client=client,
+ selector=selector,
+ heartbeat=lambda:
heartbeat_observations.append(client.publish_parse_result.call_count),
+ )
+ for method in (
+ "_refresh_dag_bundles",
+ "_queue_requested_files_for_parsing",
+ "_scan_stale_dags",
+ "_cleanup_stale_bundle_versions",
+ "print_stats",
+ ):
+ mocker.patch.object(DagFileProcessorManager, method,
autospec=True)
+ mocker.patch.object(DagFileProcessorManager, "fetch_callbacks",
autospec=True, return_value=[])
+ for file in _get_file_infos(["a.py", "b.py"]):
+ proc, read_end = self.mock_processor()
+ proc.dispatch_sequence = 1
+ proc.parsing_result =
DagFileParsingResult(fileloc=str(file.rel_path), serialized_dags=[])
+ try:
+ manager.handle_parsing_result(file, proc)
+ finally:
+ read_end.close()
+ proc.close()
+ assert not manager.max_runs_reached()
+ manager._run_parsing_loop()
+ calls = client.publish_parse_result.call_args_list
+ assert [call.args[0].relative_fileloc for call in calls] == ["a.py",
"b.py", "a.py"]
+ assert calls[0].args[0] is calls[2].args[0]
+ assert {0, 1}.issubset(heartbeat_observations)
+ assert not manager._pending_publications
+ assert all(stat.run_count == 1 for stat in
manager._file_stats.values())
+
+ @pytest.mark.parametrize("change", ["refresh", "delete"])
+ def test_pending_publication_is_discarded_when_source_changes(self,
change):
+ client = mock.create_autospec(DagProcessorAPIClient, instance=True)
+ manager = DagFileProcessorManager(max_runs=1, api_client=client)
+ file = _get_file_infos(["dag.py"])[0]
+ proc, read_end = self.mock_processor()
+ proc.dispatch_sequence = 1
+ proc.parsing_result = DagFileParsingResult(fileloc="dag.py",
serialized_dags=[])
+ try:
+ manager.handle_parsing_result(file, proc)
+ if change == "refresh":
+ manager._invalidate_bundle_parse_source("testing")
+ else:
+ manager.remove_orphaned_file_stats(present=set())
+ manager._publish_pending_results()
+ client.publish_parse_result.assert_not_called()
+ assert not manager._pending_publications
+ assert list(manager._file_queue) == ([file] if change == "refresh"
else [])
+ finally:
+ read_end.close()
+ proc.close()
+
+ @mock.patch.object(DagFileProcessorManager, "_create_process",
autospec=True)
+ def test_pending_publications_bound_admission_and_prevent_reimports(self,
create):
+ client = mock.create_autospec(DagProcessorAPIClient, instance=True)
+ manager = DagFileProcessorManager(max_runs=1, api_client=client,
parallelism=1)
+ manager._file_parsing_sort_mode = "alphabetical"
+ file, other = _get_file_infos(["dag.py", "other.py"])
+ proc, read_end = self.mock_processor()
+ proc.dispatch_sequence = 1
+ proc.parsing_result = DagFileParsingResult(fileloc="dag.py",
serialized_dags=[])
+ manager._bundle_parse_sources["testing"] = DagParseSource()
+ try:
+ manager.handle_parsing_result(file, proc)
+ manager.prepare_file_queue(known_files={"testing": {file, other}})
+ assert list(manager._file_queue) == [other]
+ manager._start_new_processes()
+ create.assert_not_called()
+ assert list(manager._file_queue) == [other]
+ manager._parallelism = 2
+ manager._file_queue = OrderedDict.fromkeys([file])
+ manager._start_new_processes()
+ create.assert_not_called()
+ finally:
+ read_end.close()
+ proc.close()
+
+ @pytest.mark.parametrize("failure", ["invalid_path", "too_many_files"])
+ @mock.patch("airflow.dag_processing.manager.time", autospec=True)
+ @mock.patch.object(DagFileProcessorManager, "_get_observed_filelocs",
autospec=True)
+ def
test_unusable_inventory_preserves_version_and_recovers_after_refresh(self,
observed, clock, failure):
+ clock.monotonic.return_value = 100
+ observed.return_value = (
+ {r"bad\dag.py"} if failure == "invalid_path" else {f"{index}.py"
for index in range(100001)}
Review Comment:
`_filter_api_discovery_result` now drops paths like `bad\dag.py` before the
inventory is built, so the `invalid_path` case only fails because this mock
hands the path back to the inventory step. It tests a state production can't
reach anymore; maybe keep just `too_many_files` here?
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -906,167 +1157,249 @@ def _refresh_dag_bundles(self, known_files: dict[str,
set[DagFileInfo]]):
if periodic_check_due:
self._bundles_last_refreshed = now_seconds
+ catalog: dict[str, BundleState] | None = None
+ if self.api_client is not None and self._dag_bundles:
+ try:
+ bundles = self.api_client.get_bundles()
+ catalog = {
+ bundle.name: BundleState(bundle.last_refreshed,
bundle.version, bundle.revision)
+ for bundle in bundles
+ }
+ self._bundle_name_to_team_name = {bundle.name:
bundle.team_name for bundle in bundles}
+ missing_bundles = {bundle.name for bundle in
self._dag_bundles} - catalog.keys()
+ for bundle_name in sorted(missing_bundles -
self._bundles_missing_from_catalog):
+ self.log.warning(
+ "Bundle missing from API catalog; parsing paused",
bundle_name=bundle_name
+ )
+ self._bundles_missing_from_catalog = missing_bundles
+ self._catalog_retry_at = 0.0
+ except httpx2.HTTPError:
+ self.log.warning("Unable to read the bundle catalog; retrying
on a later refresh")
+ self._catalog_retry_at = now_seconds +
self.bundle_refresh_check_interval
+
any_refreshed = False
- active_bundles = {file.bundle_name for file in self._processors if
file.bundle_version is None}
+ active_bundles = {
+ file.bundle_name
+ for file in self._processors.keys() |
self._pending_publications.keys()
+ if file.bundle_version is None
+ }
for bundle in self._dag_bundles:
if not periodic_check_due and bundle.name not in urgent_bundles:
continue
- if bundle.name not in urgent_bundles and now_seconds <
self._bundle_refresh_retry_at.get(
- bundle.name, 0
- ):
- continue
- # TODO: AIP-66 handle errors in the case of incomplete cloning?
And test this.
- # What if the cloning/refreshing took too long(longer than the
dag processor timeout)
- if not bundle.is_initialized:
- self._invalidate_bundle_parse_source(bundle.name)
+ with self._use_bundle(bundle.name):
+ if catalog is not None and bundle.name not in catalog:
+ self._discard_unusable_bundle_source(bundle.name)
+ continue
+ if now_seconds <
self._bundle_inventory_retry_times.get(bundle.name, 0):
+ continue
+ if bundle.name in self._pending_inventories:
+ # A listing still waiting for the API must be published
before the bundle is listed again.
+ continue
+ if self.api_client is not None and catalog is None:
+ continue
+ if bundle.name not in urgent_bundles and now_seconds <
self._bundle_refresh_retry_at.get(
+ bundle.name, 0
+ ):
+ continue
+ # TODO: AIP-66 handle errors in the case of incomplete
cloning? And test this.
+ # What if the cloning/refreshing took too long(longer than
the dag processor timeout)
+ if not bundle.is_initialized:
+ self._invalidate_bundle_parse_source(bundle.name)
+ try:
+ bundle.initialize()
+ any_refreshed = True
+ except Exception as e:
+ self.log.exception("Error initializing bundle %s: %s",
bundle.name, e)
+ self._bundles_needing_refresh.add(bundle.name)
+ continue
try:
- bundle.initialize()
- any_refreshed = True
- except Exception as e:
- self.log.exception("Error initializing bundle %s: %s",
bundle.name, e)
- self._bundles_needing_refresh.add(bundle.name)
+ bundle_state = (
+ catalog.get(bundle.name)
+ if catalog is not None
+ else self.get_bundle_state(bundle.name)
+ )
+ except Exception:
+ self.log.exception("Error fetching state for bundle %s",
bundle.name)
+ continue
+ if bundle_state is None:
+ self.log.warning("Bundle model not found for %s",
bundle.name)
+ continue
+ elapsed_time_since_refresh = (
+ now - (bundle_state.last_refreshed or utc_epoch())
+ ).total_seconds()
+ if bundle.supports_versioning:
+ # we will also check the version of the bundle to see if
another DAG processor has seen
+ # a new version
+ pre_refresh_version =
self._bundle_versions.get(bundle.name)
+ # Use `is None` (not falsy) so an empty-string version is
treated as a valid cached value.
+ if pre_refresh_version is None:
+ pre_refresh_version, _ =
unpack_bundle_version(bundle.get_current_version(), bundle)
+ current_version_matches_db = pre_refresh_version ==
bundle_state.version
+ else:
+ # With no versioning, it always "matches"
+ current_version_matches_db = True
+
+ previously_seen = bundle.name in self._bundle_versions
+ 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
- try:
- bundle_state = self.get_bundle_state(bundle.name)
- except Exception:
- self.log.exception("Error fetching state for bundle %s",
bundle.name)
- continue
- if bundle_state is None:
- self.log.warning("Bundle model not found for %s", bundle.name)
- continue
- elapsed_time_since_refresh = (now - (bundle_state.last_refreshed
or utc_epoch())).total_seconds()
- if bundle.supports_versioning:
- # we will also check the version of the bundle to see if
another DAG processor has seen
- # a new version
- pre_refresh_version = self._bundle_versions.get(bundle.name)
- # Use `is None` (not falsy) so an empty-string version is
treated as a valid cached value.
- if pre_refresh_version is None:
- pre_refresh_version, _ =
unpack_bundle_version(bundle.get_current_version(), bundle)
- current_version_matches_db = pre_refresh_version ==
bundle_state.version
- else:
- # With no versioning, it always "matches"
- current_version_matches_db = True
-
- previously_seen = bundle.name in self._bundle_versions
- 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
+ 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._bundle_refresh_retry_at.pop(bundle.name, None)
- self.log.info("Refreshing bundle %s", bundle.name)
+ self._bundles_waiting_for_refresh.discard(bundle.name)
+ self._bundle_refresh_retry_at.pop(bundle.name, None)
+ 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 as error:
- self.log.exception("Error refreshing bundle %s", bundle.name)
- source_unchanged = not bundle.refresh_changes_source or (
- isinstance(error, DagBundleRefreshError) and
error.source_unchanged
- )
- if source_unchanged and previous_source is not None:
- self._bundle_parse_sources[bundle.name] = attrs.evolve(
- previous_source,
-
refresh_generation=self._bundle_refresh_generations[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 as error:
+ self.log.exception("Error refreshing bundle %s",
bundle.name)
+ source_unchanged = not bundle.refresh_changes_source or (
+ isinstance(error, DagBundleRefreshError) and
error.source_unchanged
)
- self._force_refresh_bundles.discard(bundle.name)
- self._bundle_refresh_retry_at[bundle.name] =
time.monotonic() + bundle.refresh_interval
- if bundle.name not in self._bundles_needing_refresh:
+ if source_unchanged and previous_source is not None:
+ self._bundle_parse_sources[bundle.name] = attrs.evolve(
+ previous_source,
+
refresh_generation=self._bundle_refresh_generations[bundle.name],
+ )
+ self._force_refresh_bundles.discard(bundle.name)
+ self._bundle_refresh_retry_at[bundle.name] = (
+ time.monotonic() + bundle.refresh_interval
+ )
+ if bundle.name not in self._bundles_needing_refresh:
+ self.log.warning(
+ "Refresh failed without changing the source;
continuing to parse bundle",
+ bundle_name=bundle.name,
+ )
+ elif previous_source is not None or bundle.name not in
self._bundles_needing_refresh:
self.log.warning(
- "Refresh failed without changing the source;
continuing to parse bundle",
- bundle_name=bundle.name,
+ "Parsing paused until bundle refresh succeeds",
bundle_name=bundle.name
)
- elif previous_source is not None or bundle.name not in
self._bundles_needing_refresh:
- self.log.warning("Parsing paused until bundle refresh
succeeds", bundle_name=bundle.name)
- self._bundles_needing_refresh.add(bundle.name)
- continue
+ self._bundles_needing_refresh.add(bundle.name)
+ continue
+
+ self._bundles_needing_refresh.discard(bundle.name)
+ self._force_refresh_bundles.discard(bundle.name)
- self._bundles_needing_refresh.discard(bundle.name)
- self._force_refresh_bundles.discard(bundle.name)
+ if bundle.supports_versioning:
+ version_after_refresh, version_data_after_refresh =
unpack_bundle_version(
+ bundle.get_current_version(), bundle
+ )
+ else:
+ version_after_refresh = None
+ version_data_after_refresh = None
- if bundle.supports_versioning:
- version_after_refresh, version_data_after_refresh =
unpack_bundle_version(
- bundle.get_current_version(), bundle
+ self._bundle_parse_sources[bundle.name] = DagParseSource(
+ bundle_version=version_after_refresh,
+ version_data=deepcopy(version_data_after_refresh),
+
refresh_generation=self._bundle_refresh_generations[bundle.name],
)
- else:
- version_after_refresh = None
- version_data_after_refresh = None
- self._bundle_parse_sources[bundle.name] = DagParseSource(
- bundle_version=version_after_refresh,
- version_data=deepcopy(version_data_after_refresh),
-
refresh_generation=self._bundle_refresh_generations[bundle.name],
- )
+ if bundle.supports_versioning:
+ # We can short-circuit the rest of this if (1) bundle was
seen before by
+ # this dag processor and (2) the version of the bundle did
not change
+ # after refreshing it
+ if (
+ self.api_client is None
+ and previously_seen
+ and pre_refresh_version == version_after_refresh
+ ):
+ self.log.debug(
+ "Bundle %s version not changed after refresh: %s",
+ bundle.name,
+ version_after_refresh,
+ )
+ try:
+ self.update_bundle_state(bundle.name,
last_refreshed=now, version=None)
+ except Exception:
+ self.log.exception("Error persisting state for
bundle %s", bundle.name)
+ self._delay_draining_refresh_retry(bundle,
known_files.get(bundle.name, ()))
+ continue
- if bundle.supports_versioning:
- # We can short-circuit the rest of this if (1) bundle was seen
before by
- # this dag processor and (2) the version of the bundle did not
change
- # after refreshing it
- if previously_seen and pre_refresh_version ==
version_after_refresh:
- self.log.debug(
- "Bundle %s version not changed after refresh: %s",
- bundle.name,
- version_after_refresh,
+ self.log.info(
+ "Version changed for %s, new version: %s",
bundle.name, version_after_refresh
)
- try:
- self.update_bundle_state(bundle.name,
last_refreshed=now, version=None)
- except Exception:
- self.log.exception("Error persisting state for bundle
%s", bundle.name)
+ try:
+ discovery = self._find_files_in_bundle(bundle)
+ except Exception:
+ # Keep the bundle's known files and Dags, and leave its
version unadvanced so the
+ # next refresh lists it again.
+ self.log.exception("Error listing Dag definitions in
bundle %s", bundle.name)
+ if self.api_client is not None:
+ self._bundle_parse_sources.pop(bundle.name, None)
Review Comment:
A listing that raises pops the parse source and sets no retry time, so the
skip check is bypassed and every `bundle_refresh_check_interval` (5s by
default) runs `bundle.refresh()` (a git fetch for Git bundles) and lists again.
The `ValidationError` branch below calls `_defer_unusable_bundle_inventory`;
should this one do the same?
--
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]
