diff --git a/airflow-core/docs/faq.rst b/airflow-core/docs/faq.rst index 46c783317db92..e6030d576c6cb 100644 --- a/airflow-core/docs/faq.rst +++ b/airflow-core/docs/faq.rst @@ -728,17 +728,18 @@ 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)** @@ -767,6 +768,46 @@ 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 0000000000000..18daa37a8f250 --- /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 49ae0c1f3090d..870b97ff48b3f 100644 --- a/airflow-core/src/airflow/config_templates/config.yml +++ b/airflow-core/src/airflow/config_templates/config.yml @@ -1714,10 +1714,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 @@ -2664,6 +2664,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 6b08dab3f0277..ea35e11b6d1df 100644 --- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py +++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py @@ -156,17 +156,6 @@ 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 @@ -301,6 +290,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", fallback=512) + cache_ttl = conf.getint("scheduler", "dag_cache_ttl", fallback=0) + 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 ( @@ -383,12 +396,7 @@ def __init__( 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 9f14dee0b786c..e43f0d71a4e1b 100644 --- a/airflow-core/tests/unit/jobs/test_scheduler_job.py +++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py @@ -64,7 +64,7 @@ from airflow.executors.executor_utils import ExecutorName from airflow.executors.local_executor import LocalExecutor from airflow.jobs.job import Job, run_job -from airflow.jobs.scheduler_job_runner import SCHEDULER_DAG_CACHE_SIZE, SchedulerJobRunner +from airflow.jobs.scheduler_job_runner import SchedulerJobRunner from airflow.models.asset import ( AssetActive, AssetAliasModel, @@ -423,7 +423,7 @@ def test_scheduler_dag_bag_is_bounded(self): 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 d4c5e66314b58..f00ee95e568e5 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.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 @@ def test_api_server_reports_under_its_own_namespace(self): 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"):