pierrejeambrun commented on code in PR #69007:
URL: https://github.com/apache/airflow/pull/69007#discussion_r3481321163
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -2743,6 +2743,24 @@ scheduler:
type: integer
default: "20"
see_also: ":ref:`scheduler:ha:tunables`"
+ dag_cache_size:
+ description: |
+ Size of the LRU cache for SerializedDAG objects in the scheduler.
+ Set to 0 to use an unbounded dict with no eviction.
+ The cache is keyed by Dag version ID.
+ version_added: 3.3.0
+ type: integer
+ example: ~
+ default: "1024"
Review Comment:
This default might be low. That's the number of versions accross dags.
Having more than 1024 dags can be common. Trading the memory issue for repeated
db read.
It's configurable anyway, but maybe a bigger default can be better suited.
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -347,8 +347,25 @@ def __init__(
if log:
self._log = log
+ dag_cache_size = conf.getint("scheduler", "dag_cache_size",
fallback=1024)
+ dag_cache_ttl_config = conf.getint("scheduler", "dag_cache_ttl",
fallback=10800)
Review Comment:
Therer is already a default in the conf. I wouldn't put a fallback too.
```suggestion
dag_cache_size = conf.getint("scheduler", "dag_cache_size")
dag_cache_ttl_config = conf.getint("scheduler", "dag_cache_ttl")
```
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -347,8 +347,25 @@ def __init__(
if log:
self._log = log
+ dag_cache_size = conf.getint("scheduler", "dag_cache_size",
fallback=1024)
+ dag_cache_ttl_config = conf.getint("scheduler", "dag_cache_ttl",
fallback=10800)
- self.scheduler_dag_bag = DBDagBag(load_op_links=False)
+ if dag_cache_size < 0:
Review Comment:
Minor inconsistency: API server uses cache_size <= 0 → unbounded; scheduler
uses < 0 → warn+0, == 0 → unbounded. Both end at "0 = unbounded," so behavior
matches, just expressed differently.
##########
airflow-core/tests/unit/jobs/test_scheduler_job.py:
##########
@@ -408,6 +408,24 @@ def test_heartrate(self, heartrate):
_ = SchedulerJobRunner(job=scheduler_job,
executors=[self.null_exec])
assert scheduler_job.heartrate == heartrate
+ @patch("airflow.jobs.scheduler_job_runner.DBDagBag")
+ def test_scheduler_dag_bag_uses_scheduler_cache_config(self,
mock_db_dag_bag):
Review Comment:
Missing test for `< 0` for cache size -> raise warning + unbounded
--
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]
