stable/airflow: added horizontal pod autoscaler for Celery workers (#21870)

* Added horizontal pod autoscaler for Celery workers.
Also added possibility to set "resources" for git-sync container, to be able to use hpa with pods it is attached to.

Signed-off-by: Maksym Shalak <mshalak@n-ix.com>

* Described how to use an autoscaler with memory metric

Signed-off-by: Maksym Shalak <mshalak@n-ix.com>

stable/airflow/Chart.yaml

@@ -1,7 +1,7 @@
 apiVersion: v1
 description: Airflow is a platform to programmatically author, schedule and monitor workflows
 name: airflow
-version: 6.6.1
+version: 6.7.0
 appVersion: 1.10.4
 icon: https://airflow.apache.org/_images/pin_large.png
 home: https://airflow.apache.org/

stable/airflow/README.md

@@ -159,6 +159,46 @@ requests the logs from each workers individually.
 This requires to expose a port (8793) and ensure the pod DNS is accessible to the web server pod,
 which is why StatefulSet is for.
 
+#### Worker autoscaler
+Celery workers can be scaled using the Horizontal Pod Autoscaler.
+To enable it you need to set `workers.autoscaling.enabled=true` and provide `workers.autoscaling.maxReplicas`.
+`workers.replicas` value will be used as a minimum amount. 
+Provide custom metrics for autoscaler in `workers.autoscaling.metrics`, for more information see the
+[Kubernetes documentation](https://kubernetes.io/docs/tasks/run-application/horizontal-pod-autoscale/#support-for-metrics-apis).
+Also make sure you set requested resources for git-sync side car container in `dags.git.gitSync.resources` (if you use it for pulling DAGs),
+otherwise worker pods will not scale. 
+
+For example: you have a worker pod which is allowed to execute 10 tasks at once.
+```
+airflow:
+  config:
+    AIRFLOW__CELERY__WORKER_CONCURRENCY: 10
+```
+Every task a worker executes consumes approximately 200MB of worker's memory, so that makes memory a good metric for monitoring.
+Also do not forget to setup resources requested for the metric you monitor, both for a worker and git-sync (if used).
+For a git-sync 50MB should be enough. For a worker pod you can calculate it: WORKER_CONCURRENCY * 200MB. So if running 10 tasks 
+at the same time, a worker will consume ~2GB of memory.
+```
+workers:
+  replicas: 1
+  resources:
+    requests:
+      memory: "2Gi"
+  autoscaling:
+    enabled: true
+    maxReplicas: 30
+    metrics:
+    - type: Resource
+      resource:
+        name: memory
+        target:
+          type: Utilization
+          averageUtilization: 80
+```
+With this setup if a worker consumes 80% of 2GB (which will happen if it runs 9-10 tasks at the same time),
+an autoscaling will be triggered and a new worker will be added. If you have many tasks
+in a queue, Kubernetes will keep adding workers until maxReplicas reached.
+
 #### Worker secrets
 
 You can add kubernetes secrets which will be mounted as volumes on the worker nodes
@@ -457,6 +497,9 @@ The following table lists the configurable parameters of the Airflow chart and t
 | `workers.secretsDir`                     | directory in which to mount secrets on worker nodes     | /var/airflow/secrets      |
 | `workers.secrets`                        | secrets to mount as volumes on worker nodes             | []                        |
 | `workers.securityContext`                 | (optional) security context for the worker statefulSet  | `{}`                      |
+| `workers.autoscaling.enabled`            | enable workers autoscaling                              | `false`                   |
+| `workers.autoscaling.maxReplicas`        | max worker pods allowed after autoscaling               | `2`              |
+| `workers.autoscaling.metrics`            | Kubernetes metrics used for autoscaling rules           | `{}`                      |
 | `nodeSelector`                           | Node labels for pod assignment                          | `{}`                      |
 | `affinity`                               | Affinity labels for pod assignment                      | `{}`                      |
 | `tolerations`                            | Toleration labels for pod assignment                    | `[]`                      |
@@ -504,6 +547,7 @@ The following table lists the configurable parameters of the Airflow chart and t
 | `dags.git.gitSync.image.repository`      | Image of the sidecar container that syncs dags          | `alpine/git`                  |
 | `dags.git.gitSync.image.tag`             | Image tag of sidecar container that syncs dags          | `1.0.4`                  |
 | `dags.git.gitSync.refreshTime`           | How often the sidecar container syncs dags up with repository(in seconds)         | nil                  |
+| `dags.git.gitSync.resources`             | custom resource configuration for the git-sync side car container    | `{}`                      |
 | `logs.path`                              | mount path for logs persistent volume                   | `/usr/local/airflow/logs` |
 | `rbac.create`                            | create RBAC resources                                   | `true`                    |
 | `serviceAccount.create`                  | create a service account                                | `true`                    |

stable/airflow/templates/autoscaler-workers.yaml

@@ -0,0 +1,23 @@
+{{- if .Values.workers.enabled -}}
+{{- if .Values.workers.autoscaling.enabled -}}
+apiVersion: autoscaling/v2beta2
+kind: HorizontalPodAutoscaler
+metadata:
+  name: {{ template "airflow.fullname" . }}-worker
+  labels:
+    app: {{ template "airflow.name" . }}
+    component: worker
+    chart: {{ template "airflow.chart" . }}
+    release: {{ .Release.Name }}
+    heritage: {{ .Release.Service }}
+spec:
+  scaleTargetRef:
+    apiVersion: apps/v1
+    kind: StatefulSet
+    name: {{ template "airflow.fullname" . }}-worker
+  minReplicas: {{ .Values.workers.replicas }}
+  maxReplicas: {{ .Values.workers.autoscaling.maxReplicas }}
+  metrics:
+{{ toYaml .Values.workers.autoscaling.metrics | indent 2 }}
+{{- end }}
+{{- end }}

stable/airflow/templates/deployments-scheduler.yaml

@@ -159,6 +159,8 @@ spec:
             - name: git-clone-secret
               mountPath: /keys
             {{- end }}
+          resources:
+{{ toYaml .Values.dags.git.gitSync.resources | indent 12 }}
         {{- end }}
         - name: {{ .Chart.Name }}-scheduler
           image: {{ .Values.airflow.image.repository }}:{{ .Values.airflow.image.tag }}

stable/airflow/templates/deployments-web.yaml

@@ -130,6 +130,8 @@ spec:
             - name: git-clone-secret
               mountPath: /keys
             {{- end }}
+          resources:
+{{ toYaml .Values.dags.git.gitSync.resources | indent 12 }}
         {{- end }}
         - name: {{ .Chart.Name }}-web
           image: {{ .Values.airflow.image.repository }}:{{ .Values.airflow.image.tag }}

stable/airflow/templates/statefulsets-workers.yaml

@@ -132,6 +132,8 @@ spec:
             - name: git-clone-secret
               mountPath: /keys
             {{- end }}
+          resources:
+{{ toYaml .Values.dags.git.gitSync.resources | indent 12 }}
         {{- end }}
         - name: {{ .Chart.Name }}-worker
           imagePullPolicy: {{ .Values.airflow.image.pullPolicy }}

stable/airflow/values.yaml

@@ -312,6 +312,18 @@ workers:
   ##
   ## Number of workers pod to launch
   replicas: 1
+  ##
+  ## Autoscaling settings for workers
+  autoscaling:
+    enabled: false
+    maxReplicas: 2
+    metrics: {}
+#    - type: Resource
+#      resource:
+#        name: memory
+#        target:
+#          type: Utilization
+#          averageUtilization: 80
   ##
   ## Gracefull termination period
   terminationPeriod: 30
@@ -575,6 +587,8 @@ dags:
         pullPolicy: IfNotPresent
       ## The amount of time in seconds to git pull dags
       refreshTime: 0
+      ## If using workers.autoscaling=true with git-sync, git-sync needs to have resources requested
+      resources: {}
   initContainer:
     ## Fetch the source code when the pods starts
     enabled: false