This is an automated email from the ASF dual-hosted git repository.
ephraimbuddy pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 786dd79c284 Keep ZIP-archived Dags active when dag_discovery_safe_mode
is False (#68518)
786dd79c284 is described below
commit 786dd79c284af3930c6012743cc0fccb4280f717
Author: deepinsight coder <[email protected]>
AuthorDate: Mon Aug 17 05:55:17 2026 -0700
Keep ZIP-archived Dags active when dag_discovery_safe_mode is False (#68518)
* 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
---------
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 91796b1bbca..5d2b4925345 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 c065b277716..6d2438fd213 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 e8a8a2c5036..395ec25b371 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"""
@@ -955,8 +960,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
@@ -974,7 +987,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 4599c8ea583..6d83d9c69ab 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"
