This is an automated email from the ASF dual-hosted git repository.
jedcunningham pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 14e15722683 Changing triggerer config option `default_capacity` to
`capacity` (#48032)
14e15722683 is described below
commit 14e15722683ce392d9d0fdabf4aeaef2901f208c
Author: Jake Roach <[email protected]>
AuthorDate: Sat Mar 22 00:12:41 2025 -0400
Changing triggerer config option `default_capacity` to `capacity` (#48032)
---
.../src/airflow/cli/commands/remote_commands/config_command.py | 5 +++++
airflow-core/src/airflow/config_templates/config.yml | 2 +-
airflow-core/src/airflow/configuration.py | 1 +
airflow-core/src/airflow/jobs/triggerer_job_runner.py | 2 +-
chart/templates/_helpers.yaml | 6 ++++++
chart/values.schema.json | 2 +-
chart/values.yaml | 5 ++---
helm_tests/airflow_core/test_triggerer.py | 2 +-
8 files changed, 18 insertions(+), 7 deletions(-)
diff --git
a/airflow-core/src/airflow/cli/commands/remote_commands/config_command.py
b/airflow-core/src/airflow/cli/commands/remote_commands/config_command.py
index 0ab1650db69..63ccf561a51 100644
--- a/airflow-core/src/airflow/cli/commands/remote_commands/config_command.py
+++ b/airflow-core/src/airflow/cli/commands/remote_commands/config_command.py
@@ -499,6 +499,11 @@ CONFIGS_CHANGES = [
ConfigChange(
config=ConfigParameter("database", "load_default_connections"),
),
+ # triggerer
+ ConfigChange(
+ config=ConfigParameter("triggerer", "default_capacity"),
+ renamed_to=ConfigParameter("triggerer", "capacity"),
+ ),
]
diff --git a/airflow-core/src/airflow/config_templates/config.yml
b/airflow-core/src/airflow/config_templates/config.yml
index aa5abe71150..4bc56d83e01 100644
--- a/airflow-core/src/airflow/config_templates/config.yml
+++ b/airflow-core/src/airflow/config_templates/config.yml
@@ -2587,7 +2587,7 @@ scheduler:
triggerer:
description: ~
options:
- default_capacity:
+ capacity:
description: |
How many triggers a single Triggerer will run at once, by default.
version_added: 2.2.0
diff --git a/airflow-core/src/airflow/configuration.py
b/airflow-core/src/airflow/configuration.py
index 7f8cc758f39..7b57ef93f01 100644
--- a/airflow-core/src/airflow/configuration.py
+++ b/airflow-core/src/airflow/configuration.py
@@ -332,6 +332,7 @@ class AirflowConfigParser(ConfigParser):
("api", "ssl_cert"): ("webserver", "web_server_ssl_cert", "3.0"),
("api", "ssl_key"): ("webserver", "web_server_ssl_key", "3.0"),
("api", "access_logfile"): ("webserver", "access_logfile", "3.0"),
+ ("triggerer", "capacity"): ("triggerer", "default_capacity", "3.0"),
}
# A mapping of new section -> (old section, since_version).
diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
index b630f66ee13..bcf0bb9bb96 100644
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
@@ -95,7 +95,7 @@ class TriggererJobRunner(BaseJobRunner, LoggingMixin):
):
super().__init__(job)
if capacity is None:
- self.capacity = conf.getint("triggerer", "default_capacity",
fallback=1000)
+ self.capacity = conf.getint("triggerer", "capacity")
elif isinstance(capacity, int) and capacity > 0:
self.capacity = capacity
else:
diff --git a/chart/templates/_helpers.yaml b/chart/templates/_helpers.yaml
index c76cefa843e..4e9b9c787b5 100644
--- a/chart/templates/_helpers.yaml
+++ b/chart/templates/_helpers.yaml
@@ -664,6 +664,12 @@ server_tls_key_file = /etc/pgbouncer/server.key
{{- include "_serviceAccountName" (merge (dict "key" "triggerer") .) -}}
{{- end }}
+{{/* Determine trigger capacity, taking Airflow 2 and 3 config option
differences into account */}}
+{{- define "triggerer.capacity" -}}
+ {{- $triggerer_section := .Values.config.triggerer | default dict }}
+ {{- $triggerer_section.capacity | default
$triggerer_section.default_capacity | default 1000 | int -}}
+{{- end -}}
+
{{/* Create the name of the dag processor service account to use */}}
{{- define "dagProcessor.serviceAccountName" -}}
{{- include "_serviceAccountName" (merge (dict "key" "dagProcessor"
"nameSuffix" "dag-processor") .) -}}
diff --git a/chart/values.schema.json b/chart/values.schema.json
index d8c8be92ede..8c168e7d7aa 100644
--- a/chart/values.schema.json
+++ b/chart/values.schema.json
@@ -3580,7 +3580,7 @@
"query": {
"description": "Query to use for KEDA autoscaling.
Must return a single integer.",
"type": "string",
- "default": "SELECT ceil(COUNT(*)::decimal / {{
.Values.config.triggerer.default_capacity }}) FROM trigger"
+ "default": "SELECT ceil(COUNT(*)::decimal / {{
include \"triggerer.capacity\" . }}) FROM trigger"
},
"usePgbouncer": {
"description": "Whether to use PGBouncer to
connect to the database or not when it is enabled. This configuration will be
ignored if PGBouncer is not enabled.",
diff --git a/chart/values.yaml b/chart/values.yaml
index 80c3b57b2d2..a029360f0ac 100644
--- a/chart/values.yaml
+++ b/chart/values.yaml
@@ -1806,7 +1806,7 @@ triggerer:
# Query to use for KEDA autoscaling. Must return a single integer.
query: >-
- SELECT ceil(COUNT(*)::decimal / {{
.Values.config.triggerer.default_capacity }})
+ SELECT ceil(COUNT(*)::decimal / {{ include "triggerer.capacity" . }})
FROM trigger
# Whether to use PGBouncer to connect to the database or not when it is
enabled
@@ -2710,8 +2710,7 @@ config:
worker_container_repository: '{{ .Values.images.airflow.repository |
default .Values.defaultAirflowRepository }}'
worker_container_tag: '{{ .Values.images.airflow.tag | default
.Values.defaultAirflowTag }}'
multi_namespace_mode: '{{ ternary "True" "False"
.Values.multiNamespaceMode }}'
- triggerer:
- default_capacity: 1000
+
# yamllint enable rule:line-length
# Whether Airflow can launch workers and/or pods in multiple namespaces
diff --git a/helm_tests/airflow_core/test_triggerer.py
b/helm_tests/airflow_core/test_triggerer.py
index 711778379b0..001cb92e4e7 100644
--- a/helm_tests/airflow_core/test_triggerer.py
+++ b/helm_tests/airflow_core/test_triggerer.py
@@ -812,7 +812,7 @@ class TestTriggererKedaAutoScaler:
),
# test custom template query
(
- "SELECT ceil(COUNT(*)::decimal / {{ mul
.Values.config.triggerer.default_capacity 2 }})"
+ 'SELECT ceil(COUNT(*)::decimal / {{ mul (include
"triggerer.capacity" . | int) 2 }})'
" FROM trigger",
"SELECT ceil(COUNT(*)::decimal / 2000) FROM trigger",
),
