Usage

  webservers:
    envOverrides:
      AIRFLOW__WEBSERVER__AUTO_REFRESH_INTERVAL: "8"
    roleGroups:
      default:
        replicas: 1

  webservers:
    roleGroups:
      default:
        envOverrides:
          AIRFLOW__WEBSERVER__AUTO_REFRESH_INTERVAL: "8"
        replicas: 1

---
apiVersion: v1
kind: ConfigMap
metadata:
  name: cm-dag (1)
data:
  test_airflow_dag.py: | (2)
    from datetime import datetime, timedelta
    from airflow import DAG
    from airflow.operators.bash import BashOperator
    from airflow.operators.dummy import DummyOperator

    with DAG(
        dag_id='test_airflow_dag',
        schedule_interval='0 0 * * *',
        start_date=datetime(2021, 1, 1),
        catchup=False,
        dagrun_timeout=timedelta(minutes=60),
        tags=['example', 'example2'],
        params={"example_key": "example_value"},
    ) as dag:
        run_this_last = DummyOperator(
            task_id='run_this_last',
        )

        # [START howto_operator_bash]
        run_this = BashOperator(
            task_id='run_after_loop',
            bash_command='echo 1',
        )
        # [END howto_operator_bash]

        run_this >> run_this_last

        for i in range(3):
            task = BashOperator(
                task_id='runme_' + str(i),
                bash_command='echo "{{ task_instance_key_str }}" && sleep 1',
            )
            task >> run_this

        # [START howto_operator_bash_template]
        also_run_this = BashOperator(
            task_id='also_run_this',
            bash_command='echo "run_id={{ run_id }} | dag_run={{ dag_run }}"',
        )
        # [END howto_operator_bash_template]
        also_run_this >> run_this_last

    # [START howto_operator_bash_skip]
    this_will_skip = BashOperator(
        task_id='this_will_skip',
        bash_command='echo "hello world"; exit 99;',
        dag=dag,
    )
    # [END howto_operator_bash_skip]
    this_will_skip >> run_this_last

    if __name__ == "__main__":
        dag.cli()

---
apiVersion: airflow.stackable.tech/v1alpha1
kind: AirflowCluster
metadata:
  name: airflow
spec:
  version: 2.2.5-python39-stackable0.3.0
  statsdExporterVersion: v0.22.4
  executor: CeleryExecutor
  loadExamples: false
  exposeConfig: false
  credentialsSecret: simple-airflow-credentials
  volumes:
    - name: cm-dag (3)
      configMap:
        name: cm-dag (4)
  volumeMounts:
    - name: cm-dag (5)
      mountPath: /dags/test_airflow_dag.py (6)
      subPath: test_airflow_dag.py (7)
  webservers:
    roleGroups:
      default:
        envOverrides:
          AIRFLOW__CORE__DAGS_FOLDER: "/dags" (8)
        replicas: 1
  workers:
    roleGroups:
      default:
        envOverrides:
          AIRFLOW__CORE__DAGS_FOLDER: "/dags" (8)
        replicas: 2
  schedulers:
    roleGroups:
      default:
        envOverrides:
          AIRFLOW__CORE__DAGS_FOLDER: "/dags" (8)
        replicas: 1

---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: pvc-airflow (1)
spec:
  accessModes:
    - ReadWriteOnce
  resources:
    requests:
      storage: 1Gi
---
apiVersion: batch/v1
kind: Job (2)
metadata:
  name: airflow-dags
spec:
  template:
    spec:
      restartPolicy: Never
      volumes:
        - name: external-dags (3)
          persistentVolumeClaim:
            claimName: pvc-airflow (4)
      initContainers:
        - name: dest-dir
          image: docker.stackable.tech/stackable/tools:0.2.0-stackable0
          env:
            - name: DEST_DIR
              value: "/stackable/externals"
          command:
            [
              "bash",
              "-x",
              "-c",
              "mkdir -p $DEST_DIR && chown stackable:stackable ${DEST_DIR} && chmod -R a=,u=rwX,g=rwX ${DEST_DIR}",
            ]
          securityContext:
            runAsUser: 0
          volumeMounts:
            - name: external-dags (5)
              mountPath: /stackable/externals (6)
      containers:
        - name: airflow-dags
          image: docker.stackable.tech/stackable/tools:0.2.0-stackable0
          env:
            - name: DEST_DIR
              value: "/stackable/externals"
          command: (7)
            [
              "bash",
              "-x",
              "-c",
              "curl -L  https://raw.githubusercontent.com/apache/airflow/2.2.5/airflow/example_dags/example_bash_operator.py \
              -o ${DEST_DIR}/example_bash_operator.py && \
              curl -L  https://raw.githubusercontent.com/apache/airflow/2.2.5/airflow/example_dags/example_complex.py \
              -o ${DEST_DIR}/example_complex.py && \
              curl -L  https://raw.githubusercontent.com/apache/airflow/2.2.5/airflow/example_dags/example_branch_datetime_operator.py \
              -o ${DEST_DIR}/example_branch_datetime_operator.py",
            ]
          volumeMounts:
            - name: external-dags (5)
              mountPath: /stackable/externals (6)

---
apiVersion: airflow.stackable.tech/v1alpha1
kind: AirflowCluster
metadata:
  name: airflow
spec:
  version: 2.2.4-python39-stackable0.3.0
  statsdExporterVersion: v0.22.4
  executor: CeleryExecutor
  loadExamples: false
  exposeConfig: false
  credentialsSecret: simple-airflow-credentials
  volumes:
    - name: external-dags (8)
      persistentVolumeClaim:
        claimName: pvc-airflow (9)
  volumeMounts:
    - name: external-dags (10)
      mountPath: /stackable/external-dags (11)
  webservers:
    roleGroups:
      default:
        envOverrides:
          AIRFLOW__CORE__DAGS_FOLDER: "/stackable/external-dags" (12)
        replicas: 1
  workers:
    roleGroups:
      default:
        envOverrides:
          AIRFLOW__CORE__DAGS_FOLDER: "/stackable/external-dags" (12)
        replicas: 2
  schedulers:
    roleGroups:
      default:
        envOverrides:
          AIRFLOW__CORE__DAGS_FOLDER: "/stackable/external-dags" (12)
        replicas: 1