# Copyright (c) 2018 AT&T Intellectual Property. All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. {{- if .Values.manifests.statefulset_airflow_worker }} {{- $envAll := . }} {{- $serviceAccountName := "airflow-worker" }} {{ tuple $envAll "airflow_server" $serviceAccountName | include "helm-toolkit.snippets.kubernetes_pod_rbac_serviceaccount" }} --- apiVersion: rbac.authorization.k8s.io/v1beta1 kind: ClusterRole metadata: name: airflow-worker-runner rules: - apiGroups: - "" resources: - endpoints - nodes - nodes/proxy - pods - pods/log - services verbs: - delete - get - list - watch - apiGroups: - "" resources: - configmaps verbs: - create - patch --- apiVersion: rbac.authorization.k8s.io/v1beta1 kind: ClusterRoleBinding metadata: name: run-airflow-worker subjects: - kind: ServiceAccount name: {{ $serviceAccountName }} namespace: {{ .Release.Namespace }} roleRef: kind: ClusterRole name: airflow-worker-runner apiGroup: rbac.authorization.k8s.io --- apiVersion: apps/v1beta1 kind: StatefulSet metadata: name: airflow-worker annotations: {{ tuple $envAll | include "helm-toolkit.snippets.release_uuid" }} spec: serviceName: {{ tuple "airflow_worker" "discovery" $envAll | include "helm-toolkit.endpoints.hostname_short_endpoint_lookup" }} podManagementPolicy: "Parallel" # NOTE: We are using 'OnDelete' strategy instead of 'RollingUpdate' # so that the upgrade of airflow worker will only start after the # completion of the 'update_site' workflow (the worker pods will get # deleted by the workflow at the very end, after everything is completed). # This will ensure availability of airflow worker during update/upgrade # and prevent any disruption to the workflow. updateStrategy: type: OnDelete replicas: {{ .Values.pod.replicas.airflow.worker }} template: metadata: labels: {{ tuple $envAll "airflow" "worker" | include "helm-toolkit.snippets.kubernetes_metadata_labels" | indent 8 }} spec: serviceAccountName: {{ $serviceAccountName }} affinity: {{ tuple $envAll "airflow" "worker" | include "helm-toolkit.snippets.kubernetes_pod_anti_affinity" | indent 8 }} nodeSelector: {{ .Values.labels.airflow.node_selector_key }}: {{ .Values.labels.airflow.node_selector_value }} initContainers: {{ tuple $envAll "airflow_server" list | include "helm-toolkit.snippets.kubernetes_entrypoint_init_container" | indent 8 }} - name: worker-perms image: {{ .Values.images.tags.airflow }} imagePullPolicy: {{ .Values.images.pull_policy }} securityContext: runAsUser: 0 {{ tuple $envAll $envAll.Values.pod.resources.airflow.worker | include "helm-toolkit.snippets.kubernetes_resources" | indent 10 }} command: - chown - -R - "airflow:airflow" - {{ .Values.conf.airflow.core.base_log_folder }} volumeMounts: - name: airflow-logs mountPath: {{ .Values.conf.airflow.core.base_log_folder }} containers: - name: airflow-scheduler image: {{ .Values.images.tags.airflow }} imagePullPolicy: {{ .Values.images.pull_policy }} {{ tuple $envAll $envAll.Values.pod.resources.airflow.scheduler | include "helm-toolkit.snippets.kubernetes_resources" | indent 10 }} env: - name: AIRFLOW_CONN_AIRFLOWS_OWN_DB valueFrom: secretKeyRef: name: {{ .Values.secrets.postgresql_airflow_db.user }} key: AIRFLOW_DATABASE_URI # Set to -1 to stop scheduler from going into crash loops args: ["scheduler", "-n", "-1" ] volumeMounts: - name: airflow-etc mountPath: {{ .Values.conf.airflow_config_file.path }} subPath: airflow.cfg readOnly: true - name: shipyard-etc mountPath: /usr/local/airflow/plugins/shipyard.conf subPath: shipyard.conf readOnly: true - name: airflow-logs mountPath: {{ .Values.conf.airflow.core.base_log_folder }} - name: airflow-worker image: {{ .Values.images.tags.airflow }} imagePullPolicy: {{ .Values.images.pull_policy }} {{ tuple $envAll $envAll.Values.pod.resources.airflow.worker | include "helm-toolkit.snippets.kubernetes_resources" | indent 10 }} env: - name: AIRFLOW_CONN_AIRFLOWS_OWN_DB valueFrom: secretKeyRef: name: {{ .Values.secrets.postgresql_airflow_db.user }} key: AIRFLOW_DATABASE_URI ports: - containerPort: {{ tuple "airflow_worker" "internal" "airflow_worker" $envAll | include "helm-toolkit.endpoints.endpoint_port_lookup" }} args: ["worker"] readinessProbe: tcpSocket: port: {{ .Values.network.airflow.worker.port }} volumeMounts: - name: airflow-etc mountPath: {{ .Values.conf.airflow_config_file.path }} subPath: airflow.cfg readOnly: true - name: shipyard-etc mountPath: /usr/local/airflow/plugins/shipyard.conf subPath: shipyard.conf readOnly: true - name: airflow-logs mountPath: {{ .Values.conf.airflow.core.base_log_folder }} - name: airflow-logrotate image: {{ .Values.images.tags.airflow }} imagePullPolicy: {{ .Values.images.pull_policy }} {{ tuple $envAll $envAll.Values.pod.resources.airflow.logrotate | include "helm-toolkit.snippets.kubernetes_resources" | indent 10 }} env: - name: PERCENT_MAX_LOG_FS_USAGE value: {{ .Values.logrotate.percent_max_log_fs_usage | quote }} - name: DAYS_BEFORE_LOG_DELETION value: {{ .Values.logrotate.days_before_deletion | quote }} - name: LOGROTATE_PATH value: {{ .Values.conf.airflow.core.base_log_folder }} - name: LOGROTATE_ENABLED value: "enabled" readinessProbe: exec: command: - sh - -c - |- set -ex ps -ef | grep 'sleep 3600' | grep -v 'grep' initialDelaySeconds: 15 periodSeconds: 15 volumeMounts: - name: airflow-logs mountPath: {{ .Values.conf.airflow.core.base_log_folder }} volumes: - name: airflow-etc configMap: name: airflow-etc defaultMode: 0444 - name: shipyard-etc configMap: name: shipyard-etc defaultMode: 0444 - name: airflow-bin configMap: name: airflow-bin defaultMode: 0555 volumeClaimTemplates: - metadata: name: airflow-logs spec: accessModes: [ "ReadWriteOnce" ] resources: requests: storage: {{ .Values.volume_worker.size }} storageClassName: {{ .Values.volume_worker.class_name }} {{- end }}