This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 3426221b7f9 [v3-3-test] Keep ZIP-archived Dags active when
dag_discovery_safe_mode is False (#68518) (#71714)
3426221b7f9 is described below
commit 3426221b7f9de498b3698c2f1c08920ace7ddfe5
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Aug 17 21:57:58 2026 +0200
[v3-3-test] Keep ZIP-archived Dags active when dag_discovery_safe_mode is
False (#68518) (#71714)
* Keep ZIP-archived Dags active when dag_discovery_safe_mode is False
The Dag file processor parsed and activated keyword-less Dags inside ZIP
archives but then immediately deactivated them, because the scan that
decides which files still exist ignored dag_discovery_safe_mode and always
applied the airflow/dag keyword heuristic. Resolve the setting once and use
it consistently for discovery and the deactivation scan, and log the
effective value so a misconfigured processor is diagnosable.
closes: #66104
* Cover safe-mode filtering in Dag bundle discovery
The file-discovery and observed-file paths must both honor safe mode so
wrapped Dag files are not silently skipped by a future regression.
* Shorten review-flagged comments in Dag discovery safe mode change
Review asked for one-line comments and no issue numbers in code, since
the PR link already carries that context.
* Validate configured Dag discovery safe mode
---------
(cherry picked from commit 786dd79c284af3930c6012743cc0fccb4280f717)
Co-authored-by: deepinsight coder
<[email protected]>
Co-authored-by: Cursor <[email protected]>
---
airflow-core/docs/core-concepts/dags.rst | 2 +-
airflow-core/newsfragments/68518.bugfix.rst | 1 +
.../src/airflow/config_templates/config.yml | 13 +++-
airflow-core/src/airflow/dag_processing/manager.py | 20 +++++-
.../tests/unit/dag_processing/test_manager.py | 84 ++++++++++++++++++++++
5 files changed, 115 insertions(+), 5 deletions(-)
diff --git a/airflow-core/docs/core-concepts/dags.rst
b/airflow-core/docs/core-concepts/dags.rst
index cf7d6ee4716..9b90be76710 100644
--- a/airflow-core/docs/core-concepts/dags.rst
+++ b/airflow-core/docs/core-concepts/dags.rst
@@ -166,7 +166,7 @@ While both Dag constructors get called when the file is
accessed, only ``dag_1``
When searching for Dags inside the Dag bundle, Airflow only considers
Python files that contain the strings ``airflow`` and ``dag``
(case-insensitively) as an optimization.
- To consider all Python files instead, disable the
``DAG_DISCOVERY_SAFE_MODE`` configuration flag.
+ To consider all Python files instead, set the ``[core]
dag_discovery_safe_mode`` configuration flag to ``False``. This is the setting
to use when your Dags are defined through a wrapper or abstraction whose source
does not contain those strings. The flag is read by the Dag file processor, so
set it on that component (and on anything that runs ``airflow dags
reserialize``) and restart it for the change to take effect -- otherwise Dags
may appear after a manual reserialize and then dis [...]
You can also provide an ``.airflowignore`` file inside your Dag bundle, or any
of its subfolders, which describes patterns of files for the loader to ignore.
It covers the directory it's in plus all subfolders underneath it. See
:ref:`.airflowignore <concepts:airflowignore>` below for details of the file
syntax.
diff --git a/airflow-core/newsfragments/68518.bugfix.rst
b/airflow-core/newsfragments/68518.bugfix.rst
new file mode 100644
index 00000000000..db2792de216
--- /dev/null
+++ b/airflow-core/newsfragments/68518.bugfix.rst
@@ -0,0 +1 @@
+Keep Dags from ZIP archives active when ``[core] dag_discovery_safe_mode`` is
``False``. Previously, Dags packaged in ZIP files whose source did not contain
the ``airflow``/``dag`` keywords were parsed and activated by the Dag file
processor but then immediately deactivated, because the scan that determines
which files still exist always applied the keyword heuristic regardless of the
configured ``dag_discovery_safe_mode``.
diff --git a/airflow-core/src/airflow/config_templates/config.yml
b/airflow-core/src/airflow/config_templates/config.yml
index 27d674b5802..42cefc13ba7 100644
--- a/airflow-core/src/airflow/config_templates/config.yml
+++ b/airflow-core/src/airflow/config_templates/config.yml
@@ -320,7 +320,18 @@ core:
default: "True"
dag_discovery_safe_mode:
description: |
- If enabled, Airflow will only scan files containing both ``DAG`` and
``airflow`` (case-insensitive).
+ If enabled, Airflow only scans files containing both ``DAG`` and
``airflow`` (case-insensitive)
+ when looking for Dags. Set this to ``False`` to scan every Python file
instead -- required when
+ your Dags are defined through a wrapper or abstraction whose source
does not contain those strings.
+
+ This setting is read by the Dag file processor, so it must be
configured for that component. In a
+ separate-process deployment (for example the Helm chart) set it on the
``dag-processor`` as well as
+ on anything that runs ``airflow dags reserialize``, and restart the
component for a change to take
+ effect. If only some components see ``False``, Dags can appear after a
manual reserialize and then
+ be deactivated again on the next processor scan.
+
+ As an alternative to scanning every file, set ``[core]
might_contain_dag_callable`` to a custom
+ heuristic that recognizes your wrapper.
version_added: 1.10.3
type: string
example: ~
diff --git a/airflow-core/src/airflow/dag_processing/manager.py
b/airflow-core/src/airflow/dag_processing/manager.py
index f0992e4494f..b214c0029e7 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -298,6 +298,11 @@ class DagFileProcessorManager(LoggingMixin):
factory=_config_get_factory("dag_processor", "file_parsing_sort_mode")
)
+ dag_discovery_safe_mode: bool = attrs.field(
+ factory=_config_bool_factory("core", "dag_discovery_safe_mode")
+ )
+ """Resolved once per process so file discovery and the deactivation scan
use the same value."""
+
_api_server: InProcessExecutionAPI = attrs.field(init=False,
factory=_make_execution_api)
"""API server to interact with Metadata DB"""
@@ -951,8 +956,16 @@ class DagFileProcessorManager(LoggingMixin):
"""Get relative paths for dag files from bundle dir."""
# Build up a list of Python files that could contain DAGs
self.log.info("Searching for files in %s at %s", bundle.name,
bundle.path)
- rel_paths = [Path(x).relative_to(bundle.path) for x in
list_py_file_paths(bundle.path)]
- self.log.info("Found %s files for bundle %s", len(rel_paths),
bundle.name)
+ rel_paths = [
+ Path(x).relative_to(bundle.path)
+ for x in list_py_file_paths(bundle.path,
safe_mode=self.dag_discovery_safe_mode)
+ ]
+ self.log.info(
+ "Found %s files for bundle %s (dag_discovery_safe_mode=%s)",
+ len(rel_paths),
+ bundle.name,
+ self.dag_discovery_safe_mode,
+ )
return rel_paths
@@ -970,7 +983,8 @@ class DagFileProcessorManager(LoggingMixin):
try:
with zipfile.ZipFile(abs_path) as z:
for info in z.infolist():
- if might_contain_dag(info.filename, True, z):
+ # Use the configured discovery safe mode
+ if might_contain_dag(info.filename,
self.dag_discovery_safe_mode, z):
yield os.path.join(abs_path, info.filename)
except zipfile.BadZipFile:
self.log.exception("There was an error accessing ZIP file %s",
abs_path)
diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py
b/airflow-core/tests/unit/dag_processing/test_manager.py
index 0bce3308c04..15dc8975e85 100644
--- a/airflow-core/tests/unit/dag_processing/test_manager.py
+++ b/airflow-core/tests/unit/dag_processing/test_manager.py
@@ -157,6 +157,40 @@ def
_create_zip_bundle_with_valid_and_broken_dags(zip_path: Path) -> None:
)
+def _create_zip_bundle_with_keywordless_dag(zip_path: Path) -> None:
+ """Build a zip with one keyword-bearing member and one keyword-less
("wrapped") member.
+
+ ``with_keywords.py`` contains the ``airflow``/``dag`` strings the
safe-mode heuristic looks
+ for. ``no_keywords.py`` mimics a custom wrapper whose source contains
neither ``airflow`` nor
+ ``dag``/``asset`` -- exactly the case ``dag_discovery_safe_mode=False``
exists to support.
+ """
+ with zipfile.ZipFile(zip_path, "w") as zf:
+ zf.writestr(
+ "with_keywords.py",
+ textwrap.dedent(
+ """
+ from airflow.sdk import DAG
+
+ with DAG(dag_id="zip_with_keywords"):
+ pass
+ """
+ ),
+ )
+ zf.writestr(
+ "no_keywords.py",
+ textwrap.dedent(
+ """
+ from mycompany.pipelines import flow
+
+
+ @flow(name="nightly")
+ def nightly():
+ run_step("extract")
+ """
+ ),
+ )
+
+
class TestDagFileProcessorManager:
@pytest.fixture(autouse=True)
def _disable_examples(self):
@@ -309,6 +343,56 @@ class TestDagFileProcessorManager:
manager.sync_bundles()
mock_bundles_manager.return_value.sync_bundles_to_db.assert_called_once_with(deactivate_missing=False)
+ @pytest.mark.parametrize(
+ "safe_mode",
+ [
+ pytest.param(False, id="safe-mode-off-includes-keywordless"),
+ pytest.param(True, id="safe-mode-on-filters-keywordless"),
+ ],
+ )
+ def test_find_files_in_bundle_respects_dag_discovery_safe_mode(self,
tmp_path, safe_mode):
+ (tmp_path / "with_keywords.py").write_text("from airflow.sdk import
DAG\n")
+ (tmp_path / "no_keywords.py").write_text("from mycompany.pipelines
import flow\n")
+ bundle = MagicMock(spec=BaseDagBundle)
+ bundle.name = "testing"
+ bundle.path = tmp_path
+
+ with conf_vars({("core", "dag_discovery_safe_mode"): str(safe_mode)}):
+ manager = DagFileProcessorManager(max_runs=1)
+
+ expected = {Path("with_keywords.py")}
+ if not safe_mode:
+ expected.add(Path("no_keywords.py"))
+ assert set(manager._find_files_in_bundle(bundle)) == expected
+
+ @pytest.mark.parametrize(
+ "safe_mode",
+ [
+ pytest.param(False, id="safe-mode-off-includes-keywordless"),
+ pytest.param(True, id="safe-mode-on-filters-keywordless"),
+ ],
+ )
+ def test_get_observed_filelocs_respects_dag_discovery_safe_mode(self,
tmp_path, safe_mode):
+ """ZIP-member discovery used for deactivation must honor the
configured safe_mode.
+
+ With ``dag_discovery_safe_mode=False`` a keyword-less (wrapped) zip
member is parsed and
+ activated, so it must also be reported as observed -- otherwise it is
deactivated right
+ after being parsed. This path previously hardcoded safe_mode=True.
+ """
+ zip_path = tmp_path / "test_zip.zip"
+ _create_zip_bundle_with_keywordless_dag(zip_path)
+
+ with conf_vars({("core", "dag_discovery_safe_mode"): str(safe_mode)}):
+ manager = DagFileProcessorManager(max_runs=1)
+ observed_filelocs = manager._get_observed_filelocs(
+ {DagFileInfo(bundle_name="testing", rel_path=Path("test_zip.zip"),
bundle_path=tmp_path)}
+ )
+
+ expected = {"test_zip.zip/with_keywords.py"}
+ if not safe_mode:
+ expected.add("test_zip.zip/no_keywords.py")
+ assert observed_filelocs == expected
+
@pytest.mark.usefixtures("clear_parse_import_errors")
def test_refresh_dag_bundles_keeps_zip_inner_file_errors(self, session,
tmp_path, configure_dag_bundles):
bundle_path = tmp_path / "bundleone"
