jason810496 commented on code in PR #71704:
URL: https://github.com/apache/airflow/pull/71704#discussion_r3804837787
##########
airflow-core/docs/administration-and-deployment/web-stack.rst:
##########
@@ -142,8 +142,8 @@ The following configuration options are available in the
``[api]`` section:
- ``server_type``: ``uvicorn`` (default) or ``gunicorn``
- ``worker_refresh_interval``: Seconds between worker refresh cycles (0 =
disabled, default)
- ``worker_refresh_batch_size``: Number of workers to refresh per cycle
(default: 1)
-- ``dag_cache_size``: Max cached SerializedDAG versions in the API server
(default: 64, 0 = unbounded)
-- ``dag_cache_ttl``: TTL in seconds for cached DAGs (default: 3600, 0 = LRU
only)
+- ``dag_cache_size``: Max cached SerializedDAG versions in the API server
(default: 64, 0 = no size limit)
+- ``dag_cache_ttl``: Idle timeout in seconds for cached Dags (default: 3600, 0
= no TTL; both 0 = no eviction)
Review Comment:
The line now lists `0 = no TTL` and `both 0 = no eviction` as separate
cases, so TTL=0 with a size set is LRU.
##########
airflow-core/docs/faq.rst:
##########
@@ -758,9 +765,47 @@ See :ref:`config:api__server_type`,
:ref:`config:api__worker_refresh_interval`,
.. note::
Worker recycling handles memory growth from *any* source, not just the Dag
cache.
- For production deployments, using both bounded caching and gunicorn worker
recycling
+ 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 serialized Dag objects for the life of the process, so as
Dag versions
+accumulate (see :ref:`faq:dag-version-inflation`) its memory grows until the
process is
+restarted or OOM killed. Configure eviction in the ``[scheduler]`` section:
Review Comment:
Addressed in ef09e82d74. As discussed offline that it would be better to set
`dag_cache_size = 1024` with `dag_cache_ttl = 0` for scheduler, so a size limit
is the default rather than an opt-in.
##########
airflow-core/src/airflow/jobs/scheduler_dagbag.py:
##########
Review Comment:
Addressed in dfca4564db.
##########
airflow-core/tests/unit/jobs/test_scheduler_dagbag.py:
##########
Review Comment:
Addressed in dfca4564db.
##########
airflow-core/tests/unit/models/test_dagbag.py:
##########
@@ -246,33 +268,39 @@ def make_lazy(task_ids):
class TestDBDagBagCache:
"""Tests for DBDagBag optional caching behavior."""
- def test_no_caching_by_default(self):
- """Test that DBDagBag uses a simple dict without caching by default."""
- dag_bag = DBDagBag()
- assert dag_bag._use_cache is False
- assert isinstance(dag_bag._dags, dict)
-
- def test_lru_cache_enabled_with_cache_size(self):
- """Test that LRU cache is enabled when cache_size is provided."""
- dag_bag = DBDagBag(cache_size=10)
- assert dag_bag._use_cache is True
- assert isinstance(dag_bag._dags, LRUCache)
-
- def test_ttl_cache_enabled_with_cache_size_and_ttl(self):
- """Test that TTL cache is enabled when both cache_size and cache_ttl
are provided."""
- dag_bag = DBDagBag(cache_size=10, cache_ttl=60)
- assert dag_bag._use_cache is True
- assert isinstance(dag_bag._dags, TTLCache)
-
- def test_zero_cache_size_uses_unbounded_dict(self):
- """Test that cache_size=0 uses unbounded dict (same as no caching)."""
- dag_bag = DBDagBag(cache_size=0, cache_ttl=60)
- assert dag_bag._use_cache is False
- assert isinstance(dag_bag._dags, dict)
+ @pytest.mark.parametrize(
+ ("cache_size", "cache_ttl", "expected_type", "expected_maxsize"),
+ [
+ pytest.param(None, None, dict, None, id="no_args_plain_dict"),
+ pytest.param(0, 0, dict, None, id="both_zero_plain_dict"),
+ pytest.param(0, None, dict, None,
id="size_zero_no_ttl_plain_dict"),
+ pytest.param(10, None, LRUCache, 10, id="size_only_lru"),
+ pytest.param(10, 0, LRUCache, 10, id="ttl_zero_lru"),
+ pytest.param(10, 60, TTLCache, 10, id="size_and_ttl_bounded_ttl"),
+ pytest.param(0, 60, TTLCache, math.inf, id="ttl_only_uncapped"),
+ pytest.param(None, 60, TTLCache, math.inf,
id="ttl_only_size_none_uncapped"),
+ pytest.param(-1, 60, TTLCache, math.inf,
id="negative_size_clamped_to_uncapped"),
+ pytest.param(10, -1, LRUCache, 10,
id="negative_ttl_clamped_to_lru"),
+ pytest.param(-1, -1, dict, None, id="both_negative_plain_dict"),
+ ],
+ )
+ def test_cache_selection(self, cache_size, cache_ttl, expected_type,
expected_maxsize):
+ dag_bag = _stub_dag_bag(cache_size=cache_size, cache_ttl=cache_ttl)
+ assert isinstance(dag_bag._dags, expected_type)
+ assert dag_bag._use_cache is (expected_type is not dict)
+ if expected_maxsize is not None:
+ assert dag_bag._dags.maxsize == expected_maxsize
+
+ def test_uncapped_ttl_cache_accepts_entries(self):
+ """A size of 0 must mean "no limit", not cachetools' zero-capacity
cache."""
+ dag_bag = _stub_dag_bag(cache_size=0, cache_ttl=60)
+ for i in range(200):
+ dag_bag._dags[f"version_{i}"] = MagicMock()
+ assert len(dag_bag._dags) == 200
Review Comment:
Addressed in dfca4564db.
##########
dev/airflow_perf/dag_bag_cache_overhead.py:
##########
Review Comment:
Addressed in dfca4564db.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
