scheduler-deployment.yaml

################################
## Airflow Scheduler Deployment/StatefulSet
#################################

# Are we using a local/sequential executor?
{{- $local := or (eq .Values.executor "LocalExecutor") (eq .Values.executor "SequentialExecutor") }}
# Are we using the kubernetes executor?
{{- $kube := eq .Values.executor "KubernetesExecutor" }}
# Is persistence enabled on the _workers_?
# This is important because in $local mode, the scheduler assumes the role of the worker
{{- $persistence := .Values.workers.persistence.enabled }}
# If we're using a StatefulSet
{{- $stateful := and $local $persistence }}
# If we're using elasticsearch logging
{{- $elasticsearch := .Values.elasticsearch.enabled }}

kind: {{ if $stateful }}StatefulSet{{ else }}Deployment{{ end }}
apiVersion: apps/v1
metadata:
  name: {{ .Release.Name }}-scheduler
  labels:
    tier: airflow
    component: scheduler
    release: {{ .Release.Name }}
    chart: "{{ .Chart.Name }}-{{ .Chart.Version }}"
    heritage: {{ .Release.Service }}
{{- with .Values.labels }}
{{ toYaml . | indent 4 }}
{{- end }}
spec:
{{- if $stateful }}
  serviceName: {{ .Release.Name }}-scheduler
{{- end }}
  replicas: {{ .Values.scheduler.replicas }}
{{- if not $stateful }}
  strategy:
    type: Recreate
{{- end }}
  selector:
    matchLabels:
      tier: airflow
      component: scheduler
      release: {{ .Release.Name }}
  template:
    metadata:
      labels:
        tier: airflow
        component: scheduler
        release: {{ .Release.Name }}
{{- with .Values.labels }}
{{ toYaml . | indent 8 }}
{{- end }}
      annotations:
        checksum/metadata-secret: {{ include (print $.Template.BasePath "/secrets/metadata-connection-secret.yaml") . | sha256sum }}
        checksum/result-backend-secret: {{ include (print $.Template.BasePath "/secrets/result-backend-connection-secret.yaml") . | sha256sum }}
        checksum/pgbouncer-config-secret: {{ include (print $.Template.BasePath "/secrets/pgbouncer-config-secret.yaml") . | sha256sum }}
        checksum/airflow-config: {{ include (print $.Template.BasePath "/configmap.yaml") . | sha256sum }}
        {{- if .Values.scheduler.safeToEvict }}
        cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
        {{- end }}
        {{- if .Values.airflowPodAnnotations }}
        {{- toYaml .Values.airflowPodAnnotations | nindent 8 }}
        {{- end }}
    spec:
      nodeSelector:
{{ toYaml .Values.nodeSelector | indent 8 }}
      affinity:
{{ toYaml .Values.scheduler.affinity | indent 8 }}
      tolerations:
{{ toYaml .Values.tolerations | indent 8 }}
      restartPolicy: Always
      terminationGracePeriodSeconds: 10
      serviceAccountName: {{ .Release.Name }}-scheduler-serviceaccount
      securityContext:
        runAsUser: {{ .Values.uid }}
        fsGroup: {{ .Values.gid }}
      {{- if or .Values.registry.secretName .Values.registry.connection }}
      imagePullSecrets:
        - name: {{ template "registry_secret" . }}
      {{- end }}
      initContainers:
        - name: wait-for-airflow-migrations
          image: {{ template "default_airflow_image" . }}
          imagePullPolicy: {{ .Values.images.airflow.pullPolicy }}
          args: ["airflow-migration-spinner", "--timeout=60"]
          env:
      {{- include "standard_airflow_environment" . | indent 10 }}
{{- if .Values.scheduler.extraInitContainers }}
{{- toYaml .Values.scheduler.extraInitContainers | nindent 8 }}
{{- end }}
      containers:
        # Always run the main scheduler container.
        - name: scheduler
          image: {{ template "airflow_image" . }}
          imagePullPolicy: {{ .Values.images.airflow.pullPolicy }}
          args: ["airflow", "scheduler"]
          {{- include "airflow_environment_variables" . | indent 10 }}
          # If the scheduler stops heartbeating for 5 minutes (10*30s) kill the
          # scheduler and let Kubernetes restart it.
          # Use the same for readiness and liveness.
          {{- range tuple "readinessProbe" "livenessProbe" }}
          {{ . }}:
            initialDelaySeconds: {{ $.Values.scheduler.livenessProbe.initialDelaySeconds | default 0 }}
            timeoutSeconds: {{ $.Values.scheduler.livenessProbe.timeoutSeconds | default 5 }}
            failureThreshold: {{ $.Values.scheduler.livenessProbe.failureThreshold | default 10 }}
            periodSeconds: {{ $.Values.scheduler.livenessProbe.periodSeconds | default 30 }}
            exec:
              command:
              - /usr/bin/env
              - AIRFLOW__CORE__LOGGING_LEVEL=ERROR
              - python3
              - -Wignore
              - -c
              - |
                from airflow.jobs.scheduler_job import SchedulerJob
                from airflow.utils.db import create_session
                from airflow.utils.net import get_hostname
                import sys

                with create_session() as session:
                  job = session.query(SchedulerJob).filter_by(hostname=get_hostname()).order_by(SchedulerJob.latest_heartbeat.desc()).limit(1).first()
                sys.exit(0 if job.is_alive() else 1)
          {{- end }}
          resources:
{{ toYaml .Values.scheduler.resources | indent 12 }}
          volumeMounts:
            - name: logs
              mountPath: {{ template "airflow_logs" . }}
            - name: config
              mountPath: {{ template "airflow_config_path" . }}
              subPath: airflow.cfg
              readOnly: true
            - name: config
              mountPath: {{ template "airflow_local_setting_path" . }}
              subPath: airflow_local_settings.py
              readOnly: true
            - name: config
              subPath: pod_template_file.yaml
              mountPath: {{ include "airflow_pod_template_file" . }}/pod_template_file.yaml
              readOnly: true
{{- if .Values.scheduler.extraVolumeMounts }}
{{ toYaml .Values.scheduler.extraVolumeMounts | indent 12 }}
{{- end }}
        # Always start the garbage collector sidecar.
        - name: scheduler-gc
          image: {{ template "airflow_image" . }}
          imagePullPolicy: {{ .Values.images.airflow.pullPolicy }}
          args: ["/usr/local/bin/clean-airflow-logs"]
          volumeMounts:
            - name: logs
              mountPath: {{ template "airflow_logs" . }}
{{- if and $local (not $elasticsearch) }}
        # Start the sidecar log server if we're in local mode and
        # we don't have elasticsearch enabled.
        - name: scheduler-logs
          image: {{ template "airflow_image" . }}
          args: ["airflow", "serve_logs"]
          ports:
            - name: worker-logs
              containerPort: {{ .Values.ports.workerLogs }}
          volumeMounts:
            - name: logs
              mountPath: {{ template "airflow_logs" . }}
            - name: config
              mountPath: {{ template "airflow_config_path" . }}
              subPath: airflow.cfg
              readOnly: true
            - name: config
              mountPath: {{ include "airflow_pod_template_file" . }}/pod_template_file.yaml
              subPath: pod_template_file.yaml
              readOnly: true
          {{- include "airflow_environment_variables" . | indent 10 }}
{{- end }}
{{- if .Values.scheduler.extraContainers }}
{{- toYaml .Values.scheduler.extraContainers | nindent 8 }}
{{- end }}
      volumes:
        - name: config
          configMap:
            name: {{ template "airflow_config" . }}
{{- if .Values.scheduler.extraVolumes }}
{{ toYaml .Values.scheduler.extraVolumes | indent 8 }}
{{- end }}
{{- if not $stateful }}
        - name: logs
          emptyDir: {}
{{- else }}
  volumeClaimTemplates:
    - metadata:
        name: logs
      spec:
      {{- if .Values.workers.persistence.storageClassName }}
        storageClassName: {{ .Values.workers.persistence.storageClassName }}
      {{- end }}
        accessModes: ["ReadWriteOnce"]
        resources:
          requests:
            storage: {{ .Values.workers.persistence.size }}
{{- end }}