Changing triggerer config option default_capacity to capacity (apache#48032)

Author: jroachgolf84

airflow-core/src/airflow/cli/commands/remote_commands/config_command.py

@@ -499,6 +499,11 @@ def message(self) -> str | None:
     ConfigChange(
         config=ConfigParameter("database", "load_default_connections"),
     ),
+    # triggerer
+    ConfigChange(
+        config=ConfigParameter("triggerer", "default_capacity"),
+        renamed_to=ConfigParameter("triggerer", "capacity"),
+    ),
 ]
 
 

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

airflow-core/src/airflow/configuration.py

@@ -332,6 +332,7 @@ def sensitive_config_values(self) -> set[tuple[str, str]]:
         ("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).

airflow-core/src/airflow/jobs/triggerer_job_runner.py

@@ -95,7 +95,7 @@ def __init__(
     ):
         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:

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") .) -}}

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.",

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

helm_tests/airflow_core/test_triggerer.py

@@ -812,7 +812,7 @@ def test_include_event_source_container_name_in_scaled_object_for_triggerer(self
             ),
             # 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",
             ),