Helm: Scope KEDA_DB_CONN to worker and triggerer pods

chart/templates/_helpers.yaml

@@ -84,16 +84,6 @@ If release name contains chart name it will be used as a full name.
         name: {{ template "airflow_metadata_secret" . }}
         key: connection
   {{- end }}
-  {{- $triggererKedaEnabled := and .Values.triggerer.enabled .Values.triggerer.keda.enabled }}
-  {{- $workersKedaNeedsDbConn := and .Values.workers.celery.keda.enabled (or (eq .Values.data.metadataConnection.protocol "mysql") (and .Values.pgbouncer.enabled (not .Values.workers.celery.keda.usePgbouncer))) }}
-  {{- $triggererKedaNeedsDbConn := and $triggererKedaEnabled (or (eq .Values.data.metadataConnection.protocol "mysql") (and .Values.pgbouncer.enabled (not .Values.triggerer.keda.usePgbouncer))) }}
-  {{- if or $workersKedaNeedsDbConn $triggererKedaNeedsDbConn }}
-  - name: KEDA_DB_CONN
-    valueFrom:
-      secretKeyRef:
-        name: {{ template "airflow_metadata_secret" . }}
-        key: kedaConnection
-  {{- end }}
   {{- if .Values.enableBuiltInSecretEnvVars.AIRFLOW__API__SECRET_KEY }}
   - name: AIRFLOW__API__SECRET_KEY
     valueFrom:
@@ -158,6 +148,17 @@ If release name contains chart name it will be used as a full name.
   {{- end }}
 {{- end }}
 
+{{/* KEDA scaler database connection for the worker and triggerer autoscalers */}}
+{{- define "keda_airflow_environment" }}
+  {{- if or (eq .Values.data.metadataConnection.protocol "mysql") (and .Values.pgbouncer.enabled (not .UsePgbouncer)) }}
+  - name: KEDA_DB_CONN
+    valueFrom:
+      secretKeyRef:
+        name: {{ template "airflow_metadata_secret" . }}
+        key: kedaConnection
+  {{- end }}
+{{- end }}
+
 {{/* User defined Airflow environment variables */}}
 {{- define "custom_airflow_environment" }}
   # Dynamically created environment variables

chart/templates/triggerer/triggerer-deployment.yaml

@@ -201,6 +201,9 @@ spec:
           env:
             {{- include "custom_airflow_environment" . | indent 10 }}
             {{- include "standard_airflow_environment" (merge (dict "IncludeJwtSecret" false) .) | indent 10 }}
+            {{- if $keda }}
+            {{- include "keda_airflow_environment" (merge (dict "UsePgbouncer" .Values.triggerer.keda.usePgbouncer) .) | indent 10 }}
+            {{- end }}
             {{- include "container_extra_envs" (list . .Values.triggerer.env) | nindent 10 }}
           livenessProbe:
             initialDelaySeconds: {{ .Values.triggerer.livenessProbe.initialDelaySeconds }}

chart/templates/workers/worker-deployment.yaml

@@ -326,6 +326,9 @@ spec:
               value: "0"
             {{- include "custom_airflow_environment" . | indent 10 }}
             {{- include "standard_airflow_environment" (merge (dict "IncludeJwtSecret" false) .) | indent 10 }}
+            {{- if $keda }}
+            {{- include "keda_airflow_environment" (merge (dict "UsePgbouncer" .Values.workers.celery.keda.usePgbouncer) .) | indent 10 }}
+            {{- end }}
             {{- include "container_extra_envs" (list . .Values.workers.celery.env) | indent 10 }}
             {{- if .Values.workers.celery.kerberosSidecar.enabled }}
             - name: KRB5_CONFIG

chart/tests/helm_tests/other/test_keda.py

@@ -406,3 +406,47 @@ def test_overwrite_keda_max_replica_count(self):
         )
 
         assert jmespath.search("spec.maxReplicaCount", docs[0]) == 5
+
+    @staticmethod
+    def _env_names(doc):
+        return jmespath.search("spec.template.spec.containers[].env[].name", doc) or []
+
+    def test_worker_keda_db_conn_not_leaked_to_other_components(self):
+        docs = render_chart(
+            values={
+                "workers": {"celery": {"keda": {"enabled": True, "usePgbouncer": False}}},
+                "executor": "CeleryExecutor",
+                "pgbouncer": {"enabled": True},
+            },
+            show_only=[
+                "templates/workers/worker-deployment.yaml",
+                "templates/scheduler/scheduler-deployment.yaml",
+                "templates/api-server/api-server-deployment.yaml",
+                "templates/dag-processor/dag-processor-deployment.yaml",
+                "templates/triggerer/triggerer-deployment.yaml",
+            ],
+        )
+        worker, scheduler, api_server, dag_processor, triggerer = docs
+        assert "KEDA_DB_CONN" in self._env_names(worker)
+        assert "KEDA_DB_CONN" not in self._env_names(scheduler)
+        assert "KEDA_DB_CONN" not in self._env_names(api_server)
+        assert "KEDA_DB_CONN" not in self._env_names(dag_processor)
+        assert "KEDA_DB_CONN" not in self._env_names(triggerer)
+
+    def test_triggerer_keda_db_conn_not_leaked_to_other_components(self):
+        docs = render_chart(
+            values={
+                "triggerer": {"keda": {"enabled": True, "usePgbouncer": False}},
+                "executor": "CeleryExecutor",
+                "pgbouncer": {"enabled": True},
+            },
+            show_only=[
+                "templates/triggerer/triggerer-deployment.yaml",
+                "templates/workers/worker-deployment.yaml",
+                "templates/scheduler/scheduler-deployment.yaml",
+            ],
+        )
+        triggerer, worker, scheduler = docs
+        assert "KEDA_DB_CONN" in self._env_names(triggerer)
+        assert "KEDA_DB_CONN" not in self._env_names(worker)
+        assert "KEDA_DB_CONN" not in self._env_names(scheduler)