ashb commented on code in PR #71704:
URL: https://github.com/apache/airflow/pull/71704#discussion_r3803995289
##########
airflow-core/tests/unit/jobs/test_scheduler_dagbag.py:
##########
Review Comment:
This test feels like overkill for the size of the until fn/class. I'd
personally not bother with this, but especially not
`test_from_config_does_not_load_op_links`
##########
airflow-core/docs/faq.rst:
##########
@@ -717,22 +717,29 @@ The API server caches serialized Dag objects in memory.
Over time, as Dag versio
There are two complementary approaches:
-**1. Bounded DAG caching (available since Airflow 3.3.0)**
+**1. Dag cache eviction (available since Airflow 3.2.2)**
Review Comment:
What version is the caching described below released in?
##########
dev/airflow_perf/dag_bag_cache_overhead.py:
##########
Review Comment:
An in memory cache lookup by dict key is always going to be many orders of
magnitude quicker than loading from DB over a network and deserializing the
JSON.
I appreciate the thought, but lets not commit this file as it doesn't
provide any long term value to the repo.
##########
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:
Isn't this testing cachetools, not our own code?
##########
airflow-core/src/airflow/jobs/scheduler_dagbag.py:
##########
Review Comment:
I don't think this really needs a whole new file and class just for one
utility method -- massive overkill.
Lets move it as a utility fn/helper class in scheduler job please
##########
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:
This change isn't quite true is it? It says 0 = no eviction, but if the
cache size has a size then TTL=0 will mean least-recently-used (LRU) is in
force?
##########
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:
What is the default value for this? A default config that results in
unlimited memory growth doesn't seem right.
##########
shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml:
##########
@@ -395,6 +413,12 @@ metrics:
legacy_name: "-"
name_variables: []
+ - name: "scheduler.dag_bag.cache_size"
Review Comment:
Nit: why is this one not with the other scheduler.dag_bag metrics.
--
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]
