diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py index df0f8f8748135..a4d4917958c11 100644 --- a/airflow-core/src/airflow/dag_processing/manager.py +++ b/airflow-core/src/airflow/dag_processing/manager.py @@ -506,7 +506,11 @@ def deactivate_stale_dags( # When the Dag's last_parsed_time is more than the stale_dag_threshold older than the # Dag file's last_finish_time, the Dag is considered stale as has apparently been removed from the file, # This is especially relevant for Dag files that generate Dags in a dynamic manner. - file_info = DagFileInfo(rel_path=Path(dag.relative_fileloc), bundle_name=dag.bundle_name) + rel_path = Path(dag.relative_fileloc) + file_info = DagFileInfo(rel_path=rel_path, bundle_name=dag.bundle_name) + if file_info not in last_parsed: + # Zip-packaged dags are keyed by the archive path, not the inner file, so try the parent as well + file_info = DagFileInfo(rel_path=rel_path.parent, bundle_name=dag.bundle_name) if last_finish_time := last_parsed.get(file_info, None): if dag.last_parsed_time + timedelta(seconds=self.stale_dag_threshold) < last_finish_time: self.log.info( diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py b/airflow-core/tests/unit/dag_processing/test_manager.py index f1fb4b66c81fd..b136a4d9e8fe3 100644 --- a/airflow-core/tests/unit/dag_processing/test_manager.py +++ b/airflow-core/tests/unit/dag_processing/test_manager.py @@ -576,7 +576,6 @@ def test_files_sorted_random_seeded_by_host(self): assert manager._file_queue == expected # Verify running it again produces same order - manager._files = [] manager.prepare_file_queue(known_files=known_files) assert manager._file_queue == expected @@ -1016,7 +1015,6 @@ def test_scan_stale_dags(self, session): run_count=1, last_num_of_db_queries=1, ) - manager._files = [test_dag_path] manager._file_stats[test_dag_path] = stat active_dag_count = session.scalar( @@ -1046,6 +1044,73 @@ def test_scan_stale_dags(self, session): # SerializedDagModel gives history about Dags assert serialized_dag_count == 1 + @pytest.mark.usefixtures("testing_dag_bundle") + def test_scan_stale_dags_deactivates_zip_packaged_dags(self, session, test_zip_path): + """ + Ensure that zip-packaged DAGs are marked inactive when the file is parsed but the + DagModel.last_parsed_time is not updated, testing fallback to the parent path when + comparing DAG.relative_fileloc to last_parsed entries. + """ + manager = DagFileProcessorManager( + max_runs=1, + processor_timeout=10 * 60, + ) + bundle = MagicMock() + bundle.name = "testing" + manager._dag_bundles = [bundle] + + test_dag_path = DagFileInfo( + rel_path=Path("test_zip.zip"), + bundle_path=Path(test_zip_path).parent, + bundle_name="testing", + ) + dagbag = DagBag( + test_dag_path.absolute_path, + bundle_path=test_dag_path.bundle_path, + ) + + # Add stale DAG to the DB + dag = dagbag.get_dag("test_zip_dag") + sync_dag_to_db(dag, session=session) + + # Add DAG to the file_parsing_stats + stat = DagFileStat( + num_dags=1, + import_errors=0, + last_finish_time=timezone.utcnow() + timedelta(hours=1), + last_duration=1, + run_count=1, + last_num_of_db_queries=1, + ) + manager._file_stats[test_dag_path] = stat + + active_dag_count = session.scalar( + select(func.count(DagModel.dag_id)).where( + DagModel.dag_id == "test_zip_dag", + ~DagModel.is_stale, + DagModel.relative_fileloc == str(test_dag_path.rel_path / "test_zip.py"), + ) + ) + assert active_dag_count == 1 + + manager._scan_stale_dags() + + active_dag_count = session.scalar( + select(func.count(DagModel.dag_id)).where( + DagModel.dag_id == "test_zip_dag", + ~DagModel.is_stale, + DagModel.relative_fileloc == str(test_dag_path.rel_path / "test_zip.py"), + ) + ) + assert active_dag_count == 0 + + serialized_dag_count = session.scalar( + select(func.count(SerializedDagModel.dag_id)).where(SerializedDagModel.dag_id == dag.dag_id) + ) + # Deactivating the DagModel should not delete the SerializedDagModel + # SerializedDagModel gives history about Dags + assert serialized_dag_count == 1 + @pytest.mark.usefixtures("testing_dag_bundle") def test_deactivate_stale_dags_marks_dags_in_inactive_bundles(self, session): """Dags whose bundle is no longer active should be marked stale even without a parse signal."""