[v3-3-test] Guard 2 to 3+ migration against custom Dag bundle configs using startup-based initialization (#63185) - #70994
Merged
vatsrahul1001 merged 1 commit intoAug 3, 2026
Conversation
… using startup-based initialization (apache#63185) * Reassign Dags from unconfigured bundles at DFP startup The 0082_3_1_0_make_bundle_name_not_nullable migration writes a single hard-coded ``bundle_name='dags-folder'`` on every legacy DagModel row, so deployments whose runtime config uses any other bundle name (or multiple custom bundles) cannot resolve the row at trigger time and fail with ``Requested bundle 'dags-folder' is not configured.`` (see apache#63323). Fixing this inside the Alembic migration is wrong: migrations must not import application code, and the migration runs before bundles are constructible, so the user's real bundle config is not available there. Instead, at DagFileProcessorManager startup -- after sync_bundles_to_db flushes the latest bundle state -- scan DagModel for legacy-candidate rows (NULL relative_fileloc and no DagVersion for that dag_id) and route each row to the most-specific configured bundle whose absolute path contains the Dag's fileloc, writing relative_fileloc at the same time so fileloc-based stale-detection works later. Rows whose fileloc is not under any configured bundle's path are left untouched: writing bundle_name without a verified relative_fileloc would produce an active row task workers cannot execute. Skipped rows then self-heal via the staleness lifecycle -- no manual ``airflow dags reserialize`` required. Concurrency and edge-case hardening on the repair path: * Fast-skip via ``EXISTS(DagVersion)`` -- DagVersion is written only by the parse path, which overwrites both bundle_name and relative_fileloc on every parse (DagModelOperation.update_dags), so once any 3.x parse has run the parse path is the source of truth and reassign has no work it would not do itself. PK-index probe vs. a sequential scan of ``dag`` (no index on relative_fileloc). * Chunked UPDATEs (_REASSIGN_BATCH_SIZE=1000) ordered by dag_id, one internally-owned transaction per chunk via create_session(), so the row-lock window stays bounded and the repair never commits a caller-provided session. Per-row compare-and-swap WHERE clause re-asserts the legacy-candidate predicate on the UPDATE so a concurrent parser write wins the race. * SELECT and UPDATE chunks run in separate sessions; per-row fileloc matching runs without a DB connection held. * Parent-traversal guard in _best_bundle_for_fileloc lexically normalises both sides with os.path.normpath and rejects any relative result that is still absolute or contains ``..``, so a stored fileloc like ``/dags/foo/../../outside.py`` cannot escape a bundle root. Lexical only -- no symlink resolution. * multi_team-safe because a bundle path belongs to at most one team. * Stale-Dag scan skips rows with NULL relative_fileloc and emits one INFO line per cycle with the skip count, so operator-visible legacy rows that the repair could not route stay observable. Tests cover custom bundle names, multiple bundles, overlapping paths (deepest wins), unmatched fileloc (row skipped), missing fileloc, FK-safety when a configured bundle is missing from dag_bundle, the legacy relative_fileloc backfill path, concurrent-DFP startup, chunk-boundary batching, and the full sync -> repair -> stale-scan -> re-parse lifecycle. closes: apache#63323 * Add newsfragment and mention the workaround * Clarify bundle path matching and NULL relative_fileloc handling Normalize bundle paths once when building the active-bundle map so the fileloc match uses plain Path.relative_to instead of mixing os.path with pathlib per iteration, and explain why the lexical normpath is required. Expand the cryptic stale-check comment to describe the legacy 2.x NULL relative_fileloc case and link the tracking issue. * Use plain Path.relative_to for bundle fileloc routing Match BaseDagImporter.get_relative_path instead of normalizing with os.path.normpath. Filelocs come from the Dag processor parsing admin-controlled bundle files, so they are trusted and need no path-traversal defense, and using the same relative_to check means the startup repair writes the same relative_fileloc the next parse computes. * Add critial regression guard test case * Order reassignment scan by bundle name instead of re-sorting logs Scanning in bundle_name order groups the movements log by source bundle naturally, so the Python-side sorted() can go; the keyset cursor becomes compound (bundle_name, dag_id) to keep pagination correct under the new ordering. Addresses review feedback in apache#63185 (comment) * Combine the top create_session for the fast prob path * Rename _best_bundle_for_fileloc to _guess_best_bundle_for_fileloc and parameterize its tests The old name implied a definitive match; the function is a best-effort guess based on configured bundle paths. Consolidating its four near-identical test cases into one parametrized test also makes it easier to see the input/expected shape at a glance, and the DagVersion import moves to module scope since nothing about the test requires a lazy import. * Shorten significant.rst * Guard Dag bundle reassignment so a failure can't crash DFP startup Reassigning legacy Dags to configured bundles is a best-effort repair run during Dag processor startup. A failure there must not take down the processor, since affected Dags already recover on the next successful parse. The reassignment log line is also worded for accuracy: the source bundle may be a configured-but-wrong bundle, not necessarily an unconfigured one. * Experiment: drop redundant session.flush() from sync_bundles_to_db Every caller commits before reading bundle state (dag_reserialize and DAG.test commit immediately; the DFP and utils/cli go through provide_session, which commits on exit), and reassign_dags_with_unconfigured_bundles reads in its own session. The explicit flush no longer guards any read. Pushed to observe whether CI still depends on it. (cherry picked from commit 40ced09) Co-authored-by: Jason(Zhe-You) Liu <68415893+jason810496@users.noreply.github.com>
jason810496
marked this pull request as ready for review
August 3, 2026 14:14
jason810496
requested review from
ephraimbuddy and
jedcunningham
as code owners
August 3, 2026 14:14
vatsrahul1001
approved these changes
Aug 3, 2026
vatsrahul1001
pushed a commit
that referenced
this pull request
Aug 5, 2026
… using startup-based initialization (#63185) (#70994) * Reassign Dags from unconfigured bundles at DFP startup The 0082_3_1_0_make_bundle_name_not_nullable migration writes a single hard-coded ``bundle_name='dags-folder'`` on every legacy DagModel row, so deployments whose runtime config uses any other bundle name (or multiple custom bundles) cannot resolve the row at trigger time and fail with ``Requested bundle 'dags-folder' is not configured.`` (see #63323). Fixing this inside the Alembic migration is wrong: migrations must not import application code, and the migration runs before bundles are constructible, so the user's real bundle config is not available there. Instead, at DagFileProcessorManager startup -- after sync_bundles_to_db flushes the latest bundle state -- scan DagModel for legacy-candidate rows (NULL relative_fileloc and no DagVersion for that dag_id) and route each row to the most-specific configured bundle whose absolute path contains the Dag's fileloc, writing relative_fileloc at the same time so fileloc-based stale-detection works later. Rows whose fileloc is not under any configured bundle's path are left untouched: writing bundle_name without a verified relative_fileloc would produce an active row task workers cannot execute. Skipped rows then self-heal via the staleness lifecycle -- no manual ``airflow dags reserialize`` required. Concurrency and edge-case hardening on the repair path: * Fast-skip via ``EXISTS(DagVersion)`` -- DagVersion is written only by the parse path, which overwrites both bundle_name and relative_fileloc on every parse (DagModelOperation.update_dags), so once any 3.x parse has run the parse path is the source of truth and reassign has no work it would not do itself. PK-index probe vs. a sequential scan of ``dag`` (no index on relative_fileloc). * Chunked UPDATEs (_REASSIGN_BATCH_SIZE=1000) ordered by dag_id, one internally-owned transaction per chunk via create_session(), so the row-lock window stays bounded and the repair never commits a caller-provided session. Per-row compare-and-swap WHERE clause re-asserts the legacy-candidate predicate on the UPDATE so a concurrent parser write wins the race. * SELECT and UPDATE chunks run in separate sessions; per-row fileloc matching runs without a DB connection held. * Parent-traversal guard in _best_bundle_for_fileloc lexically normalises both sides with os.path.normpath and rejects any relative result that is still absolute or contains ``..``, so a stored fileloc like ``/dags/foo/../../outside.py`` cannot escape a bundle root. Lexical only -- no symlink resolution. * multi_team-safe because a bundle path belongs to at most one team. * Stale-Dag scan skips rows with NULL relative_fileloc and emits one INFO line per cycle with the skip count, so operator-visible legacy rows that the repair could not route stay observable. Tests cover custom bundle names, multiple bundles, overlapping paths (deepest wins), unmatched fileloc (row skipped), missing fileloc, FK-safety when a configured bundle is missing from dag_bundle, the legacy relative_fileloc backfill path, concurrent-DFP startup, chunk-boundary batching, and the full sync -> repair -> stale-scan -> re-parse lifecycle. closes: #63323 * Add newsfragment and mention the workaround * Clarify bundle path matching and NULL relative_fileloc handling Normalize bundle paths once when building the active-bundle map so the fileloc match uses plain Path.relative_to instead of mixing os.path with pathlib per iteration, and explain why the lexical normpath is required. Expand the cryptic stale-check comment to describe the legacy 2.x NULL relative_fileloc case and link the tracking issue. * Use plain Path.relative_to for bundle fileloc routing Match BaseDagImporter.get_relative_path instead of normalizing with os.path.normpath. Filelocs come from the Dag processor parsing admin-controlled bundle files, so they are trusted and need no path-traversal defense, and using the same relative_to check means the startup repair writes the same relative_fileloc the next parse computes. * Add critial regression guard test case * Order reassignment scan by bundle name instead of re-sorting logs Scanning in bundle_name order groups the movements log by source bundle naturally, so the Python-side sorted() can go; the keyset cursor becomes compound (bundle_name, dag_id) to keep pagination correct under the new ordering. Addresses review feedback in #63185 (comment) * Combine the top create_session for the fast prob path * Rename _best_bundle_for_fileloc to _guess_best_bundle_for_fileloc and parameterize its tests The old name implied a definitive match; the function is a best-effort guess based on configured bundle paths. Consolidating its four near-identical test cases into one parametrized test also makes it easier to see the input/expected shape at a glance, and the DagVersion import moves to module scope since nothing about the test requires a lazy import. * Shorten significant.rst * Guard Dag bundle reassignment so a failure can't crash DFP startup Reassigning legacy Dags to configured bundles is a best-effort repair run during Dag processor startup. A failure there must not take down the processor, since affected Dags already recover on the next successful parse. The reassignment log line is also worded for accuracy: the source bundle may be a configured-but-wrong bundle, not necessarily an unconfigured one. * Experiment: drop redundant session.flush() from sync_bundles_to_db Every caller commits before reading bundle state (dag_reserialize and DAG.test commit immediately; the DFP and utils/cli go through provide_session, which commits on exit), and reassign_dags_with_unconfigured_bundles reads in its own session. The explicit flush no longer guards any read. Pushed to observe whether CI still depends on it. (cherry picked from commit 40ced09)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
(cherry picked from commit 40ced09)