diff --git a/airflow-core/newsfragments/69845.bugfix.rst b/airflow-core/newsfragments/69845.bugfix.rst new file mode 100644 index 0000000000000..c9cbc2bdb2bb4 --- /dev/null +++ b/airflow-core/newsfragments/69845.bugfix.rst @@ -0,0 +1 @@ +Fix DAG import errors being silently dropped (never shown in the *Import Errors* view). While recording an import error, ``_update_import_errors`` called ``ParseImportError.full_file_path()`` to build the import-error listener argument. ``full_file_path()`` re-instantiates the DAG bundle and, for some bundle types (e.g. the Git bundle), resolves an Airflow Connection; doing so inside the DAG-parsing transaction could roll back the not-yet-flushed import error row, leaving the DAG marked as broken with nothing shown in the UI. The already-known bundle path is now threaded down to ``_update_import_errors`` (from both the DAG processor and ``sync_bag_to_db``) and used to build the listener filename directly, so the bundle is no longer re-instantiated inside the transaction. Additionally, when no listener plugin is registered, the listener arguments are no longer computed at all. diff --git a/airflow-core/src/airflow/dag_processing/collection.py b/airflow-core/src/airflow/dag_processing/collection.py index 8fa5209ff15fb..864e448930596 100644 --- a/airflow-core/src/airflow/dag_processing/collection.py +++ b/airflow-core/src/airflow/dag_processing/collection.py @@ -71,6 +71,7 @@ if TYPE_CHECKING: from collections.abc import Collection, Iterable, Iterator + from pathlib import Path from sqlalchemy.orm import Session from sqlalchemy.sql import Select @@ -397,9 +398,36 @@ def _update_import_errors( bundle_name: str, import_errors: dict[tuple[str, str], str], session: Session, + *, + bundle_path: Path | None = None, ): from airflow.listeners.listener import get_listener_manager + def _listener_filename( + relative_fileloc: str, bundle_name_: str, import_error: ParseImportError | None = None + ) -> str | None: + """ + Build the file path handed to import-error listeners. + + When the bundle path is known (the DAG processor threads it down from the + already-initialised bundle) we build the path directly. This avoids calling + ``ParseImportError.full_file_path()``, which re-instantiates the DAG bundle -- + and, for some bundle types (e.g. the Git bundle), resolves a Connection -- + inside this parsing transaction, where it can roll back the pending import + error row. Falls back to ``full_file_path()`` only when the bundle path was + not provided. + """ + if bundle_path is not None: + return str(bundle_path / relative_fileloc) + if import_error is None: + import_error = session.scalar( + select(ParseImportError).where( + ParseImportError.bundle_name == bundle_name_, + ParseImportError.filename == relative_fileloc, + ) + ) + return import_error.full_file_path() if import_error is not None else None + # Check existing import errors BEFORE deleting, so we can determine if we should update or create existing_import_error_files = set( session.execute(select(ParseImportError.bundle_name, ParseImportError.filename)) @@ -436,16 +464,19 @@ def _update_import_errors( # sending notification when an existing dag import error occurs try: # todo: make listener accept bundle_name and relative_filename - import_error = session.scalar( - select(ParseImportError).where( - ParseImportError.bundle_name == bundle_name_, - ParseImportError.filename == relative_fileloc, - ) - ) - if import_error is not None: - get_listener_manager().hook.on_existing_dag_import_error( - filename=import_error.full_file_path(), stacktrace=stacktrace - ) + # When no listener plugin is registered, skip building the listener + # arguments entirely: for callers that do not thread ``bundle_path`` + # down, the fallback would otherwise query ``ParseImportError`` and + # call the bundle-instantiating ``full_file_path()`` inside this + # transaction just to feed a hook nobody listens to. The check stays + # inside the try block so a listener-manager failure is logged rather + # than aborting the recording of import errors. + if get_listener_manager().has_listeners: + filename = _listener_filename(relative_fileloc, bundle_name_) + if filename is not None: + get_listener_manager().hook.on_existing_dag_import_error( + filename=filename, stacktrace=stacktrace + ) except Exception: log.exception("error calling listener") else: @@ -458,9 +489,12 @@ def _update_import_errors( session.add(import_error) # sending notification when a new dag import error occurs try: - get_listener_manager().hook.on_new_dag_import_error( - filename=import_error.full_file_path(), stacktrace=stacktrace - ) + if get_listener_manager().has_listeners: + filename = _listener_filename(relative_fileloc, bundle_name_, import_error) + if filename is not None: + get_listener_manager().hook.on_new_dag_import_error( + filename=filename, stacktrace=stacktrace + ) except Exception: log.exception("error calling listener") session.execute( @@ -494,6 +528,7 @@ def update_dag_parsing_results_in_db( DagWarningType.RUNTIME_VARYING_VALUE, ), files_parsed: set[tuple[str, str]] | None = None, + bundle_path: Path | None = None, ): """ Update everything to do with DAG parsing in the DB. @@ -515,6 +550,10 @@ def update_dag_parsing_results_in_db( :param files_parsed: Set of (bundle_name, relative_fileloc) tuples for all files that were parsed. If None, will be inferred from dags and import_errors. Passing this explicitly ensures that import errors are cleared for files that were parsed but no longer contain DAGs. + :param bundle_path: Filesystem path of the already-initialised DAG bundle. When provided, it is + used to build the file path passed to import-error listeners without re-instantiating the + bundle inside this transaction. If None, ``ParseImportError.full_file_path()`` is used as a + fallback. """ # Retry 'DAG.bulk_write_to_db' & 'SerializedDagModel.bulk_sync_to_db' in case # of any Operational Errors @@ -571,6 +610,7 @@ def update_dag_parsing_results_in_db( bundle_name=bundle_name, import_errors=import_errors, session=session, + bundle_path=bundle_path, ) except Exception: log.exception("Error logging import errors!") diff --git a/airflow-core/src/airflow/dag_processing/dagbag.py b/airflow-core/src/airflow/dag_processing/dagbag.py index 3bf82804bffe1..8d4879864765e 100644 --- a/airflow-core/src/airflow/dag_processing/dagbag.py +++ b/airflow-core/src/airflow/dag_processing/dagbag.py @@ -579,4 +579,5 @@ def sync_bag_to_db( session=session, version_data=version_data, files_parsed=files_parsed, + bundle_path=dagbag.bundle_path, ) diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py index 3f60cb3f4c4de..7c6565038bab5 100644 --- a/airflow-core/src/airflow/dag_processing/manager.py +++ b/airflow-core/src/airflow/dag_processing/manager.py @@ -1243,6 +1243,7 @@ def handle_parsing_result( parsing_result=proc.parsing_result, run_duration=run_duration, relative_fileloc=str(file.rel_path), + bundle_path=file.bundle_path, session=session, ) except Exception: @@ -1276,6 +1277,7 @@ def persist_parsing_result( run_duration: float, relative_fileloc: str | None, session: Session, + bundle_path: Path | None = None, ) -> None: """Persist parsed DAG data to the metadata database.""" import_errors: dict[tuple[str, str], str] = {} @@ -1305,6 +1307,7 @@ def persist_parsing_result( warnings=set(warnings), session=session, files_parsed=files_parsed, + bundle_path=bundle_path, ) def _collect_results(self): diff --git a/airflow-core/tests/unit/dag_processing/test_collection.py b/airflow-core/tests/unit/dag_processing/test_collection.py index ad21350be6bcb..8f15220a48af7 100644 --- a/airflow-core/tests/unit/dag_processing/test_collection.py +++ b/airflow-core/tests/unit/dag_processing/test_collection.py @@ -22,6 +22,7 @@ import warnings from collections.abc import Generator from datetime import timedelta +from pathlib import Path from typing import TYPE_CHECKING from unittest import mock from unittest.mock import patch @@ -39,6 +40,7 @@ _get_latest_runs_stmt, _get_latest_runs_stmt_partitioned, _update_dag_tags, + _update_import_errors, update_dag_parsing_results_in_db, ) from airflow.exceptions import SerializationError @@ -905,6 +907,156 @@ def test_serialized_dag_errors_are_import_errors( assert len(dag_import_error_listener.existing) == 0 assert dag_import_error_listener.new["abc.py"] == import_error.stacktrace + @patch.object(ParseImportError, "full_file_path") + @pytest.mark.usefixtures("clean_db") + def test_import_error_listener_uses_bundle_path_without_full_file_path( + self, mock_full_path, session, dag_import_error_listener, testing_dag_bundle + ): + """ + Import errors are persisted and the listener filename is built from ``bundle_path``. + + ``ParseImportError.full_file_path()`` re-instantiates the DAG bundle (and, for some + bundle types such as the Git bundle, resolves an Airflow Connection); calling it inside + the parsing transaction could roll back the not-yet-flushed import error row, leaving the + DAG marked as broken while nothing showed up in the UI. When the caller threads the + already-known ``bundle_path`` down, the filename is built directly and + ``full_file_path()`` must not be called at all. + """ + bundle_path = Path("/opt/airflow/bundles/testing") + import_errors = {("testing", "abc.py"): "AnImportError"} + _update_import_errors( + files_parsed={("testing", "abc.py")}, + bundle_name="testing", + import_errors=import_errors, + session=session, + bundle_path=bundle_path, + ) + session.flush() + + # The bundle-instantiating full_file_path() must not be used when bundle_path is given. + mock_full_path.assert_not_called() + + # The import error is still persisted. + persisted = session.scalars(select(ParseImportError)).all() + assert len(persisted) == 1 + assert persisted[0].bundle_name == "testing" + assert persisted[0].filename == "abc.py" + assert persisted[0].stacktrace == "AnImportError" + + # The listener still received the correct filename, built from bundle_path. + assert dag_import_error_listener.new[str(bundle_path / "abc.py")] == "AnImportError" + + @patch.object(ParseImportError, "full_file_path") + @pytest.mark.usefixtures("clean_db") + def test_existing_import_error_listener_uses_bundle_path_without_full_file_path( + self, mock_full_path, session, dag_import_error_listener, testing_dag_bundle + ): + """ + The existing-error branch also builds the listener filename from ``bundle_path``. + + When an import error row already exists for the file, ``on_existing_dag_import_error`` + is notified instead of ``on_new_dag_import_error``; that branch previously ran an extra + ``select(ParseImportError)`` and called ``full_file_path()`` just to build the listener + argument. With ``bundle_path`` threaded down, neither must happen. + """ + mock_full_path.side_effect = AssertionError( + "full_file_path() must not be called when bundle_path is provided" + ) + bundle_path = Path("/opt/airflow/bundles/testing") + session.add( + ParseImportError( + filename="abc.py", bundle_name="testing", timestamp=tz.utcnow(), stacktrace="OldError" + ) + ) + session.flush() + + _update_import_errors( + files_parsed={("testing", "abc.py")}, + bundle_name="testing", + import_errors={("testing", "abc.py"): "NewImportError"}, + session=session, + bundle_path=bundle_path, + ) + session.flush() + + # The existing row is updated in place. + persisted = session.scalars(select(ParseImportError)).all() + assert len(persisted) == 1 + assert persisted[0].stacktrace == "NewImportError" + + mock_full_path.assert_not_called() + + # The existing-error listener received the bundle_path-based filename. + assert dag_import_error_listener.new == {} + assert dag_import_error_listener.existing[str(bundle_path / "abc.py")] == "NewImportError" + + @patch.object(ParseImportError, "full_file_path") + @pytest.mark.usefixtures("clean_db") + def test_sync_bag_to_db_does_not_call_full_file_path( + self, mock_full_path, tmp_path, session, dag_import_error_listener, testing_dag_bundle + ): + """ + ``sync_bag_to_db`` threads ``dagbag.bundle_path`` down to ``_update_import_errors``. + + This exercises a production caller end-to-end (instead of the private helper), so it + fails if any caller in the chain stops forwarding ``bundle_path`` and the code falls + back to the bundle-instantiating ``full_file_path()`` inside the parsing transaction. + """ + from airflow.dag_processing.dagbag import DagBag, sync_bag_to_db + + mock_full_path.side_effect = AssertionError( + "full_file_path() must not be called when bundle_path is provided" + ) + bundle_path = tmp_path / "bundle" + bundle_path.mkdir() + dag_file = bundle_path / "broken.py" + # Must mention both "airflow" and "dag" or DAG_DISCOVERY_SAFE_MODE skips the file. + dag_file.write_text("from airflow.sdk import DAG\nraise ImportError('broken dag')") + + dagbag = DagBag(dag_folder=str(dag_file), bundle_path=bundle_path, bundle_name="testing") + assert dagbag.import_errors + + sync_bag_to_db(dagbag, "testing", None, session=session) + session.flush() + + persisted = session.scalars(select(ParseImportError)).all() + assert len(persisted) == 1 + assert persisted[0].bundle_name == "testing" + assert persisted[0].filename == "broken.py" + + mock_full_path.assert_not_called() + assert dag_import_error_listener.new[str(bundle_path / "broken.py")] == persisted[0].stacktrace + + @patch.object(ParseImportError, "full_file_path") + @pytest.mark.usefixtures("clean_db") + def test_import_error_persisted_without_listener_or_bundle_path( + self, mock_full_path, session, testing_dag_bundle + ): + """ + With no listener registered, the ``full_file_path()`` fallback must not run at all. + + Callers that do not thread ``bundle_path`` down would otherwise still hit the + bundle-instantiating fallback inside the parsing transaction just to build an argument + for a hook nobody listens to -- the original silent-drop bug. + """ + mock_full_path.side_effect = AssertionError( + "full_file_path() must not be called when no listener is registered" + ) + _update_import_errors( + files_parsed={("testing", "abc.py")}, + bundle_name="testing", + import_errors={("testing", "abc.py"): "AnImportError"}, + session=session, + ) + session.flush() + + persisted = session.scalars(select(ParseImportError)).all() + assert len(persisted) == 1 + assert persisted[0].filename == "abc.py" + assert persisted[0].stacktrace == "AnImportError" + + mock_full_path.assert_not_called() + @patch.object(ParseImportError, "full_file_path") @mark_fab_auth_manager_test @conf_vars({("core", "min_serialized_dag_update_interval"): "5"}) diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py b/airflow-core/tests/unit/dag_processing/test_manager.py index 74da895bca30a..83c2c59614ba2 100644 --- a/airflow-core/tests/unit/dag_processing/test_manager.py +++ b/airflow-core/tests/unit/dag_processing/test_manager.py @@ -1338,6 +1338,7 @@ def test_handle_parsing_result_updates_stats_after_successful_persist(self, sess parsing_result=processor.parsing_result, run_duration=mock.ANY, relative_fileloc="abc.txt", + bundle_path=TEST_DAGS_FOLDER, session=session, ) assert manager._file_stats[file] is not original_stat