fix task_instance_mutation_hook

.github/workflows/ci.yml

@@ -152,7 +152,7 @@ jobs:
       - name: Cache virtualenv for kubernetes testing
         uses: actions/cache@v2
         env:
-          cache-name: cache-kubernetes-tests-virtualenv-v3
+          cache-name: cache-kubernetes-tests-virtualenv-v4
         with:
           path: .build/.kubernetes_venv
           key: "${{ env.cache-name }}-${{ github.job }}-\
@@ -306,6 +306,17 @@ ${{ hashFiles('requirements/requirements-python${{matrix.python-version}}.txt')
       - name: "Tests"
         run: ./scripts/ci/ci_run_airflow_testing.sh
 
+  helm-tests:
+    timeout-minutes: 5
+    name: "Checks: Helm tests"
+    runs-on: ubuntu-latest
+    env:
+      CI_JOB_TYPE: "Tests"
+    steps:
+      - uses: actions/checkout@master
+      - name: "Helm Tests"
+        run: ./scripts/ci/ci_run_helm_testing.sh
+
   requirements:
     timeout-minutes: 80
     name: "Requirements"

CI.rst

@@ -33,7 +33,7 @@ environments we use. Most of our CI jobs are written as bash scripts which are e
 the CI jobs and we are mapping all the CI-specific environment variables to generic "CI" variables.
 The only two places where CI-specific code might be are:
 
-- CI-specific declaration file (for example it is `<.github/workflow/ci.yml>`_ for GitHub Actions
+- CI-specific declaration file (for example it is `<.github/workflows/ci.yml>`_ for GitHub Actions
 - The ``get_environment_for_builds_on_ci`` function in `<scripts/ci/libraries/_build_images.sh>`_ where mapping is
   performed from the CI-environment specific to generic values. Example for that is CI_EVENT_TYPE variable
   which determines whether we are running a ``push``. ``schedule`` or ``pull_request`` kind of CI job. For

airflow/__init__.py

@@ -40,12 +40,14 @@
 from airflow import utils
 from airflow import settings
 from airflow.configuration import conf
-from airflow.models import DAG
 from flask_admin import BaseView
 from importlib import import_module
 from airflow.exceptions import AirflowException
 
 settings.initialize()
+# Delay the import of airflow.models to be after the settings initialization to make sure that
+# any reference to a settings' functions (e.g task_instance_mutation_hook) holds the expected implementation
+from airflow.models import DAG  # noqa: E402
 
 login = None  # type: Any
 log = logging.getLogger(__name__)

airflow/kubernetes/worker_configuration.py

@@ -134,6 +134,10 @@ def _get_init_containers(self):
                     name='GIT_SSH_KEY_FILE',
                     value='/etc/git-secret/ssh'
                 ),
+                k8s.V1EnvVar(
+                    name='GIT_SYNC_ADD_USER',
+                    value='true'
+                ),
                 k8s.V1EnvVar(
                     name='GIT_SYNC_SSH',
                     value='true'

breeze

@@ -509,7 +509,7 @@ function prepare_command_files() {
     export COMPOSE_PROD_FILE
 
     # Base python image for the build
-    export PYTHON_BASE_IMAGE=python:${PYTHON_MAJOR_MINOR_VERSION}-slim-buster
+    export PYTHON_BASE_IMAGE=python:${PYTHON_BASE_IMAGE_VERSION}-slim-buster
     export AIRFLOW_CI_IMAGE="${DOCKERHUB_USER}/${DOCKERHUB_REPO}:${BRANCH_NAME}-python${PYTHON_MAJOR_MINOR_VERSION}-ci"
     export AIRFLOW_PROD_IMAGE="${DOCKERHUB_USER}/${DOCKERHUB_REPO}:${BRANCH_NAME}-python${PYTHON_MAJOR_MINOR_VERSION}"
     export BUILT_IMAGE_FLAG_FILE="${BUILD_CACHE_DIR}/${BRANCH_NAME}/.built_${PYTHON_MAJOR_MINOR_VERSION}"

chart/README.md

@@ -66,7 +66,7 @@ The command removes all the Kubernetes components associated with the chart and
 
 ## Updating DAGs
 
-The recommended way to update your DAGs with this chart is to build a new docker image with the latest code (`docker build -t my-company/airflow:8a0da78 .`), push it to an accessible registry (`docker push my-company/airflow:8a0da78`), then update the Airflow pods with that image:
+The recommended way to update your DAGs with this chart is to build a new docker image with the latest DAG code (`docker build -t my-company/airflow:8a0da78 .`), push it to an accessible registry (`docker push my-company/airflow:8a0da78`), then update the Airflow pods with that image:
 
 ```bash
 helm upgrade airflow . \
@@ -77,6 +77,42 @@ helm upgrade airflow . \
 For local development purppose you can also u
 You can also build the image locally and use it via deployment method described by Breeze.
 
+## Mounting DAGS using Git-Sync side car with Persistence enabled
+
+This option will use a Persistent Volume Claim with an accessMode of `ReadWriteMany`. The scheduler pod will sync DAGs from a git repository onto the PVC every configured number of seconds. The other pods will read the synced DAGs. Not all volume  plugins have support for `ReadWriteMany` accessMode. Refer [Persistent Volume Access Modes](https://kubernetes.io/docs/concepts/storage/persistent-volumes/#access-modes) for details
+
+```bash
+helm upgrade airflow . \
+  --set dags.persistence.enabled=true \
+  --set dags.gitSync.enabled=true
+  # you can also override the other persistence or gitSync values
+  # by setting the  dags.persistence.* and dags.gitSync.* values
+  # Please refer to values.yaml for details
+```
+
+## Mounting DAGS using Git-Sync side car without Persistence
+This option will use an always running Git-Sync side car on every scheduler,webserver and worker pods. The Git-Sync side car containers will sync DAGs from a git repository every configured number of seconds. If you are using the KubernetesExecutor, Git-sync will run as an initContainer on your worker pods.
+
+```bash
+helm upgrade airflow . \
+  --set dags.persistence.enabled=false \
+  --set dags.gitSync.enabled=true
+  # you can also override the other gitSync values
+  # by setting the  dags.gitSync.* values
+  # Refer values.yaml for details
+```
+
+## Mounting DAGS from an externally populated PVC
+In this approach, Airflow will read the DAGs from a PVC which has `ReadOnlyMany` or `ReadWriteMany` accessMode. You will have to ensure that the PVC is populated/updated with the required DAGs(this won't be handled by the chart). You can pass in the name of the  volume claim to the chart
+
+```bash
+helm upgrade airflow . \
+  --set dags.persistence.enabled=true \
+  --set dags.persistence.existingClaim=my-volume-claim
+  --set dags.gitSync.enabled=false
+```
+
+
 ## Parameters
 
 The following tables lists the configurable parameters of the Airflow chart and their default values.
@@ -160,6 +196,8 @@ The following tables lists the configurable parameters of the Airflow chart and
 | `webserver.resources.requests.cpu`                    | CPU Request of webserver                                                                                     | `~`                                               |
 | `webserver.resources.requests.memory`                 | Memory Request of webserver                                                                                  | `~`                                               |
 | `webserver.defaultUser`                               | Optional default airflow user information                                                                    | `{}`                                              |
+| `dags.persistence.*`                               | Dag persistence configutation                                                                    | Please refer to `values.yaml`                                    |
+| `dags.gitSync.*`                               | Git sync configuration                                                                   | Please refer to `values.yaml`                                    |
 
 
 Specify each parameter using the `--set key=value[,key=value]` argument to `helm install`. For example,

chart/templates/_helpers.yaml

@@ -93,6 +93,80 @@
   {{ end }}
 {{- end }}
 
+{{/*  Git ssh key volume */}}
+{{- define "git_sync_ssh_key_volume"}}
+- name: git-sync-ssh-key
+  secret:
+    secretName: {{ .Values.dags.gitSync.sshKeySecret }}
+    defaultMode: 288
+{{- end }}
+
+{{/*  Git sync container */}}
+{{- define "git_sync_container"}}
+- name: {{ .Values.dags.gitSync.containerName }}
+  image: "{{ .Values.dags.gitSync.containerRepository }}:{{ .Values.dags.gitSync.containerTag }}"
+  env:
+    {{- if .Values.dags.gitSync.sshKeySecret }}
+    - name: GIT_SSH_KEY_FILE
+      value: "/etc/git-secret/ssh"
+    - name: GIT_SYNC_SSH
+      value: "true"
+    {{- if .Values.dags.gitSync.knownHosts }}
+    - name: GIT_KNOWN_HOSTS
+      value: "true"
+    - name: GIT_SSH_KNOWN_HOSTS_FILE
+      value: "/etc/git-secret/known_hosts"
+    {{- else }}
+    - name: GIT_KNOWN_HOSTS
+      value: "false"
+    {{- end }}
+    {{ else if .Values.dags.gitSync.credentialsSecret }}
+    - name: GIT_SYNC_USERNAME
+      valueFrom:
+        secretKeyRef:
+          name: {{ .Values.dags.gitSync.credentialsSecret | quote }}
+          key: GIT_SYNC_USERNAME
+    - name: GIT_SYNC_PASSWORD
+      valueFrom:
+        secretKeyRef:
+          name: {{ .Values.dags.gitSync.credentialsSecret | quote }}
+          key: GIT_SYNC_PASSWORD
+    {{- end }}
+    - name: GIT_SYNC_REV
+      value: {{ .Values.dags.gitSync.rev | quote }}
+    - name: GIT_SYNC_BRANCH
+      value: {{ .Values.dags.gitSync.branch | quote }}
+    - name: GIT_SYNC_REPO
+      value: {{ .Values.dags.gitSync.repo | quote }}
+    - name: GIT_SYNC_DEPTH
+      value: {{ .Values.dags.gitSync.depth | quote }}
+    - name: GIT_SYNC_ROOT
+      value: {{ .Values.dags.gitSync.root | quote }}
+    - name: GIT_SYNC_DEST
+      value: {{ .Values.dags.gitSync.dest | quote }}
+    - name: GIT_SYNC_ADD_USER
+      value: "true"
+    - name: GIT_SYNC_WAIT
+      value: {{ .Values.dags.gitSync.wait | quote }}
+    - name: GIT_SYNC_MAX_SYNC_FAILURES
+      value: {{ .Values.dags.gitSync.maxFailures | quote }}
+  volumeMounts:
+  - name: dags
+    mountPath: {{ .Values.dags.gitSync.root }}
+  {{- if and  .Values.dags.gitSync.enabled  .Values.dags.gitSync.sshKeySecret }}
+  - name: git-sync-ssh-key
+    mountPath: /etc/git-secret/ssh
+    readOnly: true
+    subPath: gitSshKey
+  {{- if .Values.dags.gitSync.knownHosts }}
+  - name: config
+    mountPath: /etc/git-secret/known_hosts
+    readOnly: true
+    subPath: known_hosts
+  {{- end }}
+  {{- end }}
+{{- end }}
+
 # This helper will change when customers deploy a new image.
 {{ define "airflow_image" -}}
 {{ printf "%s:%s" (.Values.images.airflow.repository | default .Values.defaultAirflowRepository) (.Values.images.airflow.tag | default .Values.defaultAirflowTag) }}
@@ -185,9 +259,30 @@ log_connections = {{ .Values.pgbouncer.logConnections }}
 {{ (printf "%s/logs" .Values.airflowHome) | quote }}
 {{- end }}
 
+{{ define "airflow_dags" -}}
+{{- if .Values.dags.gitSync.enabled -}}
+{{ (printf "%s/dags/%s/%s" .Values.airflowHome .Values.dags.gitSync.dest .Values.dags.gitSync.subPath ) }}
+{{- else -}}
+{{ (printf "%s/dags" .Values.airflowHome) }}
+{{- end -}}
+{{- end -}}
+
+{{ define "airflow_dags_volume_claim" -}}
+{{- if and .Values.dags.persistence.enabled .Values.dags.persistence.existingClaim -}}
+{{ .Values.dags.persistence.existingClaim }}
+{{- else -}}
+{{ .Release.Name }}-dags
+{{- end -}}
+{{- end -}}
+
+{{ define "airflow_dags_mount_path" -}}
+{{ (printf "%s/dags" .Values.airflowHome) }}
+{{- end }}
+
 {{ define "airflow_config_path" -}}
 {{ (printf "%s/airflow.cfg" .Values.airflowHome) | quote }}
 {{- end }}
+
 {{ define "airflow_webserver_config_path" -}}
 {{ (printf "%s/webserver_config.py" .Values.airflowHome) | quote }}
 {{- end }}

chart/templates/configmap.yaml

@@ -30,83 +30,17 @@ metadata:
     heritage: {{ .Release.Service }}
 {{- with .Values.labels }}
 {{ toYaml . | indent 4 }}
-{{- end }}
+{{- end -}}
+{{- $Global := . }}
 data:
   # These are system-specified config overrides.
-  airflow.cfg: |
-    [core]
-    load_examples = False
-    colored_console_log = False
-    executor = {{ .Values.executor }}
-{{- if .Values.elasticsearch.enabled }}
-    remote_logging = True
-{{- end }}
-
-    [logging]
-    logging_level = DEBUG
-    [webserver]
-    enable_proxy_fix = True
-    expose_config = True
-    rbac = True
-
-{{- if eq .Values.executor "CeleryExecutor" }}
-    [celery]
-    default_queue = celery
-{{- end }}
-
-    [scheduler]
-    scheduler_heartbeat_sec = 5
-{{- if .Values.statsd.enabled }}
-    statsd_on = True
-    statsd_port = 9125
-    statsd_prefix = airflow
-    statsd_host = {{ printf "%s-statsd" .Release.Name }}
-{{- end }}
-    # Restart Scheduler every 41460 seconds (11 hours 31 minutes)
-    # The odd time is chosen so it is not always restarting on the same "hour" boundary
-    run_duration = 41460
-
-{{- if .Values.elasticsearch.enabled }}
-    [elasticsearch]
-    # The elasticsearch variables were updated to the shorter names in v1.10.4
-    write_stdout = True
-    json_format = True
-    log_id_template = {dag_id}_{task_id}_{execution_date}_{try_number}
-
-    [elasticsearch_configs]
-    max_retries = 3
-    timeout = 30
-    retry_timeout = True
-{{- end }}
-
-{{- if eq .Values.executor "KubernetesExecutor" }}
-    [kubernetes]
-    namespace = {{ .Release.Namespace }}
-    airflow_configmap = {{ include "airflow_config" . }}
-    airflow_local_settings_configmap = {{ include "airflow_config" . }}
-    worker_container_repository = {{ .Values.images.airflow.repository }}
-    worker_container_tag = {{ .Values.images.airflow.tag }}
-    worker_container_image_pull_policy = {{ .Values.images.airflow.pullPolicy }}
-    worker_service_account_name = {{ .Release.Name }}-worker-serviceaccount
-    image_pull_secrets = {{ template "registry_secret" . }}
-    dags_in_image = True
-    delete_worker_pods = True
-
-    [kubernetes_secrets]
-    AIRFLOW__CORE__SQL_ALCHEMY_CONN = {{ printf "%s=connection" (include "airflow_metadata_secret" .) }}
-    AIRFLOW__CORE__FERNET_KEY = {{ printf "%s=fernet-key" (include "fernet_key_secret" .) }}
-{{- if .Values.elasticsearch.enabled }}
-    AIRFLOW__ELASTICSEARCH__ELASTICSEARCH_HOST = {{ printf "%s=connection" (include "elasticsearch_secret" .) }}
-{{- end }}
-
-    [kubernetes_labels]
-    tier = airflow
-    component = worker
-    release = {{ .Release.Name }}
-    {{- range $key, $value := .Values.labels }}
-    {{ printf "%s = %s" $key $value }}
+  airflow.cfg: |-
+    {{- range $section, $settings := .Values.config }}
+    [{{ $section }}]
+    {{- range $key, $val := $settings }}
+    {{ $key }} = {{ tpl ($val | toString) $Global }}
     {{- end }}
-{{- end }}
+    {{ end }}
 
 {{- if .Values.webserver.webserverConfig }}
   webserver_config.py: |
@@ -117,3 +51,7 @@ data:
   airflow_local_settings.py: |
     {{ .Values.scheduler.airflowLocalSettings | nindent 4 }}
 {{- end }}
+{{- if and .Values.dags.gitSync.enabled  .Values.dags.gitSync.knownHosts }}
+  known_hosts: |
+    {{ .Values.dags.gitSync.knownHosts | nindent 4 }}
+{{- end }}

chart/templates/dags-persistent-volume-claim.yaml

@@ -0,0 +1,41 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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 and (not .Values.dags.persistence.existingClaim ) .Values.dags.persistence.enabled }}
+kind: PersistentVolumeClaim
+apiVersion: v1
+metadata:
+  name: {{ .Release.Name }}-dags
+  labels:
+    tier: airflow
+    component: dags-pvc
+    release: {{ .Release.Name }}
+    chart: "{{ .Chart.Name }}-{{ .Chart.Version }}"
+    heritage: {{ .Release.Service }}
+spec:
+  accessModes: [{{ .Values.dags.persistence.accessMode | quote }}]
+  resources:
+    requests:
+      storage: {{ .Values.dags.persistence.size | quote }}
+  {{- if .Values.dags.persistence.storageClass }}
+  {{- if (eq "-" .Values.dags.persistence.storageClass) }}
+  storageClassName: ""
+  {{- else }}
+  storageClassName: "{{ .Values.dags.persistence.storageClass }}"
+  {{- end }}
+  {{- end }}
+{{- end }}