This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new dc34ea43799 [v3-4-test] Make the scheduler Dag cache size and TTL
configurable (#71816) (#75087)
dc34ea43799 is described below
commit dc34ea437995d3378aedd8009dfd4d3619ff6d38
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Oct 19 18:43:55 2026 +0530
[v3-4-test] Make the scheduler Dag cache size and TTL configurable (#71816)
(#75087)
* Make the scheduler Dag cache size and TTL configurable
A fixed 512-entry cache cannot fit every deployment: larger active working
sets churn while memory-constrained schedulers may need a lower cap. Exposing
size and TTL controls lets operators trade memory for refetch and
deserialization cost while preserving the current defaults.
* Remove redundant scheduler Dag cache config fallbacks
(cherry picked from commit d01d3510589e2c91f025600aae31461b7fb2acc4)
Co-authored-by: Jason(Zhe-You) Liu
<[email protected]>
---
airflow-core/docs/faq.rst | 52 +++++++++++++++++---
airflow-core/newsfragments/71816.feature.rst | 1 +
.../src/airflow/config_templates/config.yml | 42 ++++++++++++++--
.../src/airflow/jobs/scheduler_job_runner.py | 42 +++++++++-------
airflow-core/tests/unit/jobs/test_scheduler_job.py | 3 +-
airflow-core/tests/unit/models/test_dagbag.py | 57 ++++++++++++++++++++++
6 files changed, 167 insertions(+), 30 deletions(-)
diff --git a/airflow-core/docs/faq.rst b/airflow-core/docs/faq.rst
index e9104117042..8cdfe95475a 100644
--- a/airflow-core/docs/faq.rst
+++ b/airflow-core/docs/faq.rst
@@ -725,17 +725,17 @@ this in the ``[api]`` section:
dag_cache_size = 64 ; max cached versions (0 = no size limit)
dag_cache_ttl = 3600 ; seconds before a cached entry expires (0 = no TTL)
-``dag_cache_size`` is the only hard ceiling on memory. An entry's TTL is
refreshed only when the
-entry is checked against the database after ``[core]
min_serialized_dag_update_interval``, not on
-every request. With a shorter TTL, even frequently requested entries can
expire and reload
-between checks. Setting both options to 0 uses an unbounded dict with no
eviction, matching the
-behavior before 3.2.2.
+``dag_cache_size`` is the only hard ceiling on memory. An entry's TTL is
refreshed only when the entry is
+checked against the database after ``[core]
min_serialized_dag_update_interval``, not on every request.
+With a shorter TTL, even frequently requested entries can expire and reload
between checks. Setting both
+options to 0 disables eviction, so the API server retains every deserialized
Dag it loads until restart,
+matching the behavior before 3.2.2.
The cache is keyed by Dag version ID. After a Dag is updated, the API server
may serve the
previous version until the cached entry expires (controlled by
``dag_cache_ttl``).
-See :ref:`config:api__dag_cache_size` and :ref:`config:api__dag_cache_ttl` for
the full
-configuration reference.
+See :ref:`config:api__dag_cache_size` and :ref:`config:api__dag_cache_ttl` for
the full configuration
+reference. The scheduler has the same pair of options under ``[scheduler]``
(see :ref:`faq:scheduler-memory-growth`).
**2. Gunicorn with rolling worker restarts (available since Airflow 3.2.0)**
@@ -764,6 +764,44 @@ See :ref:`config:api__server_type`,
:ref:`config:api__worker_refresh_interval`,
For production deployments, using both cache eviction and gunicorn worker
recycling
provides the best results.
+.. _faq:scheduler-memory-growth:
+
+How to prevent scheduler memory growth?
+----------------------------------------
+
+The scheduler caches deserialized Dag objects. Airflow 3.3.2 capped this cache
at 512 Dag versions to
+prevent memory from growing with every version the scheduler had ever seen
(see :ref:`faq:dag-version-inflation`).
+Starting in Airflow 3.4.0, you can adjust that size limit and optionally add
time-based eviction in the
+``[scheduler]`` section:
+
+.. code-block:: ini
+
+ [scheduler]
+ dag_cache_size = 512 ; max cached versions, evicting least recently used
+ dag_cache_ttl = 0 ; seconds before an idle cached entry expires (0 =
no TTL)
+
+The defaults cap the cache at 512 Dag versions and evict the least recently
used one beyond that, so memory
+cannot grow with the number of versions the scheduler has ever seen. The
scheduler reaches the cache through
+the Dag version of each active Dag run, so its working set is the versions
with runs in flight; 512 is meant
+to sit above that for a typical deployment.
+
+If the scheduler is still OOM killed, lower ``dag_cache_size``. Raise it if
you have more Dag versions with
+runs in flight than the limit, since a limit below the working set evicts
versions that are still being
+scheduled and costs a database fetch and a deserialization on the next loop —
watch
+``scheduler.dag_bag.cache_miss`` to tell the two apart. Setting
``dag_cache_size = 0`` switches to no size
+limit, leaving eviction to ``dag_cache_ttl``, which bounds memory by the
concurrently active set rather than
+outright: each re-check resets an entry's expiry, so a TTL reclaims a version
only once its runs finish and
+it stops being requested. Setting both to 0 disables eviction, so the
scheduler retains every deserialized
+Dag it loads until restart, matching the behavior before 3.3.2.
+
+Neither option affects how quickly the scheduler picks up a Dag change. A Dag
update that creates a new
+version is seen immediately, because the new version is a different cache key.
A version rewritten in place
+is re-checked against its current hash once
:ref:`config:core__min_serialized_dag_update_interval` has elapsed since the
entry was last validated,
+so that option, not ``dag_cache_ttl``, bounds how long a rewritten version can
be served stale.
+The same applies to the API server.
+
+See :ref:`config:scheduler__dag_cache_size` and
:ref:`config:scheduler__dag_cache_ttl` for the full configuration reference.
+
MySQL and MySQL variant Databases
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
diff --git a/airflow-core/newsfragments/71816.feature.rst
b/airflow-core/newsfragments/71816.feature.rst
new file mode 100644
index 00000000000..18daa37a8f2
--- /dev/null
+++ b/airflow-core/newsfragments/71816.feature.rst
@@ -0,0 +1 @@
+Added ``[scheduler] dag_cache_size`` and ``[scheduler] dag_cache_ttl`` options
for tuning deserialized Dag cache eviction. The cache defaults to an LRU limit
of 512 versions with no TTL. A value of 0 disables the corresponding eviction
limit; setting both options to 0 disables eviction.
diff --git a/airflow-core/src/airflow/config_templates/config.yml
b/airflow-core/src/airflow/config_templates/config.yml
index 9bae77fb9bf..b87f27c5093 100644
--- a/airflow-core/src/airflow/config_templates/config.yml
+++ b/airflow-core/src/airflow/config_templates/config.yml
@@ -1763,10 +1763,10 @@ api:
default: "False"
dag_cache_size:
description: |
- Max number of deserialized SerializedDAG objects the API server keeps
in memory.
- Set to 0 for no size limit, leaving eviction to ``dag_cache_ttl``. Set
both this and
- ``dag_cache_ttl`` to 0 to use an unbounded dict with no eviction,
matching the behavior
- before 3.2.2.
+ Max number of deserialized SerializedDAG objects the API server keeps
in memory. Set to 0 for no
+ size limit, leaving eviction to ``dag_cache_ttl``. Setting both this
and ``dag_cache_ttl`` to 0
+ disables eviction, so the API server retains every deserialized Dag it
loads until restart, matching
+ the behavior before 3.2.2.
The cache is keyed by Dag version ID, so lookups by Dag ID
(e.g., viewing a Dag's details) always query the database for the
latest
@@ -2780,6 +2780,40 @@ scheduler:
type: integer
example: ~
default: "60"
+ dag_cache_size:
+ description: |
+ Max number of deserialized SerializedDAG objects the scheduler keeps
in memory, keyed by Dag version
+ ID. Once full, the least recently used version is evicted, so this is
a hard ceiling on how much the
+ cache can grow no matter how many Dag versions accumulate.
+
+ Raise it if the scheduler has more Dag versions with runs in flight
than this; a limit below the
+ working set evicts versions that are still being scheduled, costing a
database fetch and a
+ deserialization on the next loop. Set to 0 for no size limit, leaving
eviction to ``dag_cache_ttl``.
+ Setting both this and ``dag_cache_ttl`` to 0 disables eviction, so the
scheduler retains every
+ deserialized Dag it loads until restart, matching the behavior before
3.3.2.
+ version_added: 3.4.0
+ type: integer
+ example: ~
+ default: "512"
+ dag_cache_ttl:
+ description: |
+ Seconds a deserialized SerializedDAG stays in the scheduler's cache.
Defaults to 0, which disables
+ TTL and leaves eviction to the ``dag_cache_size`` LRU policy.
+
+ Note that each re-check resets an entry's expiry, so a TTL only
reclaims Dag versions that stop being
+ requested for the whole interval. Versions still referenced by active
Dag runs are kept, so
+ ``dag_cache_size`` remains the only hard ceiling and a TTL alone
(``dag_cache_size = 0``) bounds
+ memory by the concurrently active set rather than outright.
+
+ Note: this does not govern staleness. A Dag update that creates a new
version is picked up
+ immediately, because the new version is a different cache key. A
version rewritten in place is
+ re-checked against its current ``dag_hash`` once ``[core]
min_serialized_dag_update_interval`` has
+ elapsed since the entry was last validated, so that option bounds how
long a rewritten version can be
+ served stale.
+ version_added: 3.4.0
+ type: integer
+ example: ~
+ default: "0"
pool_metrics_interval:
description: |
How often (in seconds) should pool usage stats be sent to StatsD (if
statsd_on is enabled)
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index c5e7732f6d8..899632c277c 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -166,17 +166,6 @@ DM = DagModel
TASK_STUCK_IN_QUEUED_RESCHEDULE_EVENT = "stuck in queued reschedule"
""":meta private:"""
-SCHEDULER_DAG_CACHE_SIZE = 512
-"""
-Max deserialized Dag versions the scheduler keeps in memory.
-
-The scheduler reaches its DagBag through the Dag version of each active Dag
run, so an
-unbounded cache retains every version the process has ever seen and grows for
the life of
-the process. Sized to sit above the versions-with-runs-in-flight working set
of a typical
-deployment, so eviction costs a re-fetch only where that working set is
genuinely larger.
-
-:meta private:
-"""
# Per-tick cap on pending AssetPartitionDagRun rows the scheduler evaluates.
# Bounds the per-tick transaction so executor heartbeats and regular scheduling
@@ -311,6 +300,30 @@ def _is_parent_process() -> bool:
return multiprocessing.current_process().name == "MainProcess"
+def _create_scheduler_dag_bag() -> CachedDBDagBag:
+ """
+ Build the scheduler's DagBag from the ``[scheduler]`` cache options.
+
+ Defaults to an LRU cache of 512 versions with no TTL. A size limit is the
only hard ceiling:
+ each re-check resets an entry's expiry, so a TTL reclaims a version only
once its runs finish
+ and it stops being requested, bounding memory by the concurrently active
set rather than
+ outright. With both limits disabled, retain loaded Dags without eviction
for the process
+ lifetime.
+ """
+ cache_size = conf.getint("scheduler", "dag_cache_size")
+ cache_ttl = conf.getint("scheduler", "dag_cache_ttl")
+ if cache_size < 0:
+ raise ValueError("[scheduler] dag_cache_size must be greater than or
equal to 0")
+ if cache_ttl < 0:
+ raise ValueError("[scheduler] dag_cache_ttl must be greater than or
equal to 0")
+ return CachedDBDagBag(
+ load_op_links=False,
+ cache_size=cache_size,
+ cache_ttl=cache_ttl,
+ stats_prefix="scheduler.dag_bag",
+ )
+
+
def _get_current_dr_task_concurrency(states: Iterable[TaskInstanceState]) ->
Subquery:
"""Get the dag_run IDs and how many tasks are in the provided states for
each one."""
return (
@@ -393,12 +406,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
if log:
self._log = log
- self.scheduler_dag_bag = CachedDBDagBag(
- load_op_links=False,
- cache_size=SCHEDULER_DAG_CACHE_SIZE,
- cache_ttl=0,
- stats_prefix="scheduler.dag_bag",
- )
+ self.scheduler_dag_bag = _create_scheduler_dag_bag()
# Set of (dag_id, asset_name, asset_uri) tuples for trigger policies
that
# are permanently unreachable for the rollup window's cardinality — the
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index 576cb59a898..8b575a8369a 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -69,7 +69,6 @@ from airflow.executors.workloads import WorkloadType
from airflow.executors.workloads.types import TaskInstanceUuid
from airflow.jobs.job import Job, run_job
from airflow.jobs.scheduler_job_runner import (
- SCHEDULER_DAG_CACHE_SIZE,
TASK_STUCK_IN_QUEUED_RESCHEDULE_EVENT,
SchedulerJobRunner,
)
@@ -527,7 +526,7 @@ class TestSchedulerJob:
assert isinstance(job_runner.scheduler_dag_bag, CachedDBDagBag)
assert isinstance(job_runner.scheduler_dag_bag._dags, LRUCache)
- assert job_runner.scheduler_dag_bag._dags.maxsize ==
SCHEDULER_DAG_CACHE_SIZE
+ assert job_runner.scheduler_dag_bag._dags.maxsize == 512
# Reported separately from the API server's cache, not folded into it.
assert job_runner.scheduler_dag_bag._stats_prefix ==
"scheduler.dag_bag"
diff --git a/airflow-core/tests/unit/models/test_dagbag.py
b/airflow-core/tests/unit/models/test_dagbag.py
index d4c5e66314b..f00ee95e568 100644
--- a/airflow-core/tests/unit/models/test_dagbag.py
+++ b/airflow-core/tests/unit/models/test_dagbag.py
@@ -17,6 +17,7 @@
from __future__ import annotations
import math
+import re
import time
from concurrent.futures import ThreadPoolExecutor
from unittest.mock import MagicMock, patch
@@ -36,6 +37,7 @@ from airflow.serialization.serialized_objects import
LazyDeserializedDAG, Serial
from airflow.utils.session import create_session
from tests_common.test_utils import db
+from tests_common.test_utils.config import conf_vars
pytestmark = pytest.mark.db_test
@@ -316,6 +318,61 @@ class TestDBDagBagCache:
assert create_dag_bag()._stats_prefix == "api_server.dag_bag"
+ @pytest.mark.parametrize(
+ ("cache_size", "cache_ttl", "expected_bag_type",
"expected_cache_type", "expected_maxsize"),
+ [
+ pytest.param(None, None, CachedDBDagBag, LRUCache, 512,
id="defaults_bounded_lru"),
+ pytest.param("1024", "0", CachedDBDagBag, LRUCache, 1024,
id="size_only_lru"),
+ pytest.param("1024", "3600", CachedDBDagBag, TTLCache, 1024,
id="size_and_ttl"),
+ pytest.param("0", "3600", CachedDBDagBag, TTLCache, math.inf,
id="ttl_only_uncapped"),
+ pytest.param("0", "0", CachedDBDagBag, dict, None,
id="both_zero_no_eviction"),
+ ],
+ )
+ def test_scheduler_cache_follows_its_config(
+ self, cache_size, cache_ttl, expected_bag_type, expected_cache_type,
expected_maxsize
+ ):
+ from airflow.jobs.scheduler_job_runner import _create_scheduler_dag_bag
+
+ overrides = {}
+ if cache_size is not None:
+ overrides[("scheduler", "dag_cache_size")] = cache_size
+ if cache_ttl is not None:
+ overrides[("scheduler", "dag_cache_ttl")] = cache_ttl
+
+ with conf_vars(overrides):
+ dag_bag = _create_scheduler_dag_bag()
+
+ assert type(dag_bag) is expected_bag_type
+ assert isinstance(dag_bag._dags, expected_cache_type)
+ if expected_maxsize is not None:
+ assert dag_bag._dags.maxsize == expected_maxsize
+
+ @pytest.mark.parametrize(
+ ("cache_size", "cache_ttl", "expected_message"),
+ [
+ pytest.param(
+ "-1",
+ "0",
+ "[scheduler] dag_cache_size must be greater than or equal to
0",
+ id="negative_size",
+ ),
+ pytest.param(
+ "512",
+ "-1",
+ "[scheduler] dag_cache_ttl must be greater than or equal to 0",
+ id="negative_ttl",
+ ),
+ ],
+ )
+ def test_scheduler_cache_rejects_negative_config(self, cache_size,
cache_ttl, expected_message):
+ from airflow.jobs.scheduler_job_runner import _create_scheduler_dag_bag
+
+ with conf_vars(
+ {("scheduler", "dag_cache_size"): cache_size, ("scheduler",
"dag_cache_ttl"): cache_ttl}
+ ):
+ with pytest.raises(ValueError, match=re.escape(expected_message)):
+ _create_scheduler_dag_bag()
+
def test_cached_bag_requires_non_empty_stats_prefix(self):
"""A cache with no namespace to report under must fail at wiring time,
not mid-request."""
with pytest.raises(ValueError, match="requires a stats_prefix"):
