kaxil commented on code in PR #70586:
URL: https://github.com/apache/airflow/pull/70586#discussion_r3690147824
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -1046,6 +1046,72 @@ 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.
+ """
+ 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_path),
+ bundle_name="testing",
+ )
+ dagbag = DagBag(
+ test_zip_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._files = [test_dag_path]
Review Comment:
I don't think either `manager._files` or the `MagicMock` bundle above is
read on this code path. `_files` isn't an attribute the manager itself uses
(only tests assign it), and `deactivate_stale_dags` reads bundle state from the
DB, which the `testing_dag_bundle` fixture provides. Looks carried over from
the sibling test, these lines can be dropped.
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -1046,6 +1046,72 @@ 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.
+ """
+ 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_path),
Review Comment:
Since `bundle_path` isn't set here, `test_dag_path.bundle_path` is None,
`DagBag._get_relative_fileloc` falls through, and the DB row ends up with an
absolute `relative_fileloc`. Production always stores bundle-relative paths,
and the sibling `test_scan_stale_dags` above sets
`bundle_path=TEST_DAGS_FOLDER` with a true relative path. Using
`rel_path=Path("test_zip.zip"), bundle_path=Path(test_zip_path).parent` (and
building the `DagBag` from `test_dag_path.absolute_path`) would exercise the
same key shapes as production, and stops the test depending on pytest's tmp dir
name not containing `.zip`.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -445,6 +445,20 @@ def cleanup_stale_bundle_versions(self) -> None:
"""Clean up stale DAG bundle version usage records."""
BundleUsageTrackingManager().remove_stale_bundle_versions()
+ @staticmethod
+ def _file_name_from_fileloc(fileloc: str) -> str:
+ """
+ If a python file comes from a zip module, return just the path to the
zip file.
+
+ file parsing stats are keyed by the top-level filename
(path/to/archive.zip).
+ in order to correctly link DAGs to their relevant parsing stats, we
need to normalize
+ fileloc to the archive file location, if applicable.
+ """
+ if ".zip" in fileloc:
Review Comment:
This matches `.zip` anywhere in the string, so any fileloc that merely
contains those four characters gets truncated: `dags/report.zip.py` becomes
`dags/report.zip`, and a file under a directory named `extracted.zip/` loses
its filename too. The truncated key won't be in `last_parsed`, so the staleness
check is silently skipped for such files (the same failure mode this PR fixes),
and if the truncated path collides with a real archive's stats key, a live DAG
inherits the archive's `last_finish_time` and can be wrongly deactivated.
One way to avoid the string surgery entirely is to try the exact key first
and fall back to the parent on a miss. Zip members are only imported from the
archive root (`_load_modules_from_zip` skips entries with `len(zip_path.parts)
> 1`), so the parent of a zip-inner fileloc is always the archive itself:
```python
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.
file_info = DagFileInfo(rel_path=rel_path.parent,
bundle_name=dag.bundle_name)
```
This also covers archives not named `*.zip`, which do get parsed today since
discovery and routing are content-based (`zipfile.is_zipfile`) but would never
match the `.zip` check here. Reusing `correct_maybe_zipped` isn't an option
either, its `zipfile.is_zipfile` gate would resolve the bundle-relative path
against the processor's cwd and return the fileloc unchanged.
--
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]
