diff --git a/demos/airflow-scheduled-job/03-enable-and-run-spark-dag.yaml b/demos/airflow-scheduled-job/03-enable-and-run-spark-dag.yaml index 124f4c66..f7b52582 100644 --- a/demos/airflow-scheduled-job/03-enable-and-run-spark-dag.yaml +++ b/demos/airflow-scheduled-job/03-enable-and-run-spark-dag.yaml @@ -23,16 +23,16 @@ spec: - | kubectl rollout status --watch statefulset/airflow-webserver-default kubectl rollout status --watch statefulset/airflow-scheduler-default - AIRFLOW_ADMIN_PASSWORD=$(cat /airflow-credentials/adminUser.password) + AIRFLOW_ADMIN_PASSWORD=$(cat /airflow-admin-credentials/adminUser.password) ACCESS_TOKEN=$(curl -XPOST http://airflow-webserver-default-headless:8080/auth/token -H 'Content-Type: application/json' -d '{"username": "admin", "password": "'$AIRFLOW_ADMIN_PASSWORD'"}' | jq -r .access_token) curl -H "Authorization: Bearer $ACCESS_TOKEN" -H 'Content-Type: application/json' -XPATCH http://airflow-webserver-default-headless:8080/api/v2/dags/sparkapp_dag -d '{"is_paused": false}' | jq curl -H "Authorization: Bearer $ACCESS_TOKEN" -H 'Content-Type: application/json' -XPOST http://airflow-webserver-default-headless:8080/api/v2/dags/sparkapp_dag/dagRuns -d '{"logical_date": null}' | jq volumeMounts: - - name: airflow-credentials - mountPath: /airflow-credentials + - name: airflow-admin-credentials + mountPath: /airflow-admin-credentials volumes: - - name: airflow-credentials + - name: airflow-admin-credentials secret: - secretName: airflow-credentials + secretName: airflow-admin-credentials restartPolicy: OnFailure backoffLimit: 20 # give some time for the Airflow cluster to be available diff --git a/demos/airflow-scheduled-job/04-enable-and-run-date-dag.yaml b/demos/airflow-scheduled-job/04-enable-and-run-date-dag.yaml index 78ec737d..94165cb6 100644 --- a/demos/airflow-scheduled-job/04-enable-and-run-date-dag.yaml +++ b/demos/airflow-scheduled-job/04-enable-and-run-date-dag.yaml @@ -23,16 +23,16 @@ spec: - | kubectl rollout status --watch statefulset/airflow-webserver-default kubectl rollout status --watch statefulset/airflow-scheduler-default - AIRFLOW_ADMIN_PASSWORD=$(cat /airflow-credentials/adminUser.password) + AIRFLOW_ADMIN_PASSWORD=$(cat /airflow-admin-credentials/adminUser.password) ACCESS_TOKEN=$(curl -XPOST http://airflow-webserver-default-headless:8080/auth/token -H 'Content-Type: application/json' -d '{"username": "admin", "password": "'$AIRFLOW_ADMIN_PASSWORD'"}' | jq -r .access_token) curl -H "Authorization: Bearer $ACCESS_TOKEN" -H 'Content-Type: application/json' -XPATCH http://airflow-webserver-default-headless:8080/api/v2/dags/date_demo -d '{"is_paused": false}' | jq curl -H "Authorization: Bearer $ACCESS_TOKEN" -H 'Content-Type: application/json' -XPOST http://airflow-webserver-default-headless:8080/api/v2/dags/date_demo/dagRuns -d '{"logical_date": null}' | jq volumeMounts: - - name: airflow-credentials - mountPath: /airflow-credentials + - name: airflow-admin-credentials + mountPath: /airflow-admin-credentials volumes: - - name: airflow-credentials + - name: airflow-admin-credentials secret: - secretName: airflow-credentials + secretName: airflow-admin-credentials restartPolicy: OnFailure backoffLimit: 20 # give some time for the Airflow cluster to be available diff --git a/demos/airflow-scheduled-job/05-enable-and-run-kafka-dag.yaml b/demos/airflow-scheduled-job/05-enable-and-run-kafka-dag.yaml index 2cbcb2b4..84a26c6d 100644 --- a/demos/airflow-scheduled-job/05-enable-and-run-kafka-dag.yaml +++ b/demos/airflow-scheduled-job/05-enable-and-run-kafka-dag.yaml @@ -47,7 +47,7 @@ spec: kubectl rollout status --watch statefulset/airflow-scheduler-default # Airflow: activate DAG - AIRFLOW_ADMIN_PASSWORD=$(cat /airflow-credentials/adminUser.password) + AIRFLOW_ADMIN_PASSWORD=$(cat /airflow-admin-credentials/adminUser.password) ACCESS_TOKEN=$(curl -XPOST http://airflow-webserver-default-headless:8080/auth/token -H 'Content-Type: application/json' -d '{"username": "admin", "password": "'$AIRFLOW_ADMIN_PASSWORD'"}' | jq -r .access_token) curl -H "Authorization: Bearer $ACCESS_TOKEN" -H 'Content-Type: application/json' -XPATCH http://airflow-webserver-default-headless:8080/api/v2/dags/kafka_watcher -d '{"is_paused": false}' | jq @@ -58,11 +58,11 @@ spec: --producer.config /stackable/config/client.properties \ --topic test-topic' volumeMounts: - - name: airflow-credentials - mountPath: /airflow-credentials + - name: airflow-admin-credentials + mountPath: /airflow-admin-credentials volumes: - - name: airflow-credentials + - name: airflow-admin-credentials secret: - secretName: airflow-credentials + secretName: airflow-admin-credentials restartPolicy: OnFailure backoffLimit: 20 # give some time for the Airflow cluster to be available diff --git a/demos/airflow-scheduled-job/06-create-opa-users.yaml b/demos/airflow-scheduled-job/06-create-opa-users.yaml index f9834cc9..3a77e437 100644 --- a/demos/airflow-scheduled-job/06-create-opa-users.yaml +++ b/demos/airflow-scheduled-job/06-create-opa-users.yaml @@ -42,11 +42,11 @@ spec: --password "richard.roe" \ --role "User" volumeMounts: - - name: airflow-credentials - mountPath: /airflow-credentials + - name: airflow-admin-credentials + mountPath: /airflow-admin-credentials volumes: - - name: airflow-credentials + - name: airflow-admin-credentials secret: - secretName: airflow-credentials + secretName: airflow-admin-credentials restartPolicy: OnFailure backoffLimit: 20 # give some time for the Airflow cluster to be available diff --git a/docs/modules/demos/pages/airflow-scheduled-job.adoc b/docs/modules/demos/pages/airflow-scheduled-job.adoc index f7438863..f7b807ee 100644 --- a/docs/modules/demos/pages/airflow-scheduled-job.adoc +++ b/docs/modules/demos/pages/airflow-scheduled-job.adoc @@ -57,7 +57,7 @@ To list the installed Stackable services run the following command: [source,console] ---- -$ stackablectl stacklet list +$ stackablectl stacklet list -n airflow-demo ┌─────────┬───────────────┬───────────┬───────────────────────────────────────────────────────────────────────────────────────────────────────────┬─────────────────────────────────┐ │ PRODUCT ┆ NAME ┆ NAMESPACE ┆ ENDPOINTS ┆ CONDITIONS │ diff --git a/stacks/airflow/airflow.yaml b/stacks/airflow/airflow.yaml index eb42ccd4..b42a03ec 100644 --- a/stacks/airflow/airflow.yaml +++ b/stacks/airflow/airflow.yaml @@ -9,6 +9,11 @@ spec: productVersion: 3.2.2 pullPolicy: IfNotPresent clusterConfig: + metadataDatabase: + postgresql: + host: postgresql-airflow + database: airflow + credentialsSecretName: airflow-postgresql-credentials authorization: opa: configMapName: opa @@ -17,7 +22,7 @@ spec: entryTimeToLive: 5s maxEntries: 10 loadExamples: false - credentialsSecret: airflow-credentials + credentialsSecretName: airflow-admin-credentials volumes: - name: airflow-dags configMap: @@ -105,6 +110,11 @@ spec: image: oci.stackable.tech/sdp/airflow:3.1.6-stackable0.0.0-dev imagePullPolicy: IfNotPresent env: + # Needed by the DAGs run by KubernetesPodOperator + - name: NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace - name: AWS_SECRET_ACCESS_KEY valueFrom: secretKeyRef: @@ -140,7 +150,14 @@ data: from airflow import DAG from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator from kubernetes.client import models as k8s - from kubernetes.client import V1EnvVar, V1EnvVarSource, V1SecretKeySelector + from kubernetes.client import V1EnvVar, V1EnvVarSource, V1ObjectFieldSelector, V1SecretKeySelector + import os + + # NOTE: This code is executed on the Pod launched by the KubernetesPodOperator + # Therefore, the NAMESPACE env must be set on that. + # See AirflowCluster.spec.kubernetesExecutor.podOverrides. + namespace = os.environ.get("NAMESPACE", "default") + trino_coordinator_hostname = f"trino-coordinator-default-headless.{namespace}.svc.cluster.local" tls_volume = k8s.V1Volume( name="server-tls-mount", @@ -188,6 +205,12 @@ data: volumes=[tls_volume], volume_mounts=[tls_volume_mount], env_vars=[ + V1EnvVar( + name="NAMESPACE", + value_from=V1EnvVarSource( + field_ref=V1ObjectFieldSelector(field_path="metadata.namespace") + ) + ), V1EnvVar( name="TRINO_PASSWORD", value_from=V1EnvVarSource( @@ -198,7 +221,7 @@ data: ) ), V1EnvVar(name="TRINO_USER", value="admin"), - V1EnvVar(name="TRINO_HOST", value="trino-coordinator-default-headless.{{ NAMESPACE }}.svc.cluster.local"), + V1EnvVar(name="TRINO_HOST", value=trino_coordinator_hostname), V1EnvVar(name="TRINO_PORT", value="8443"), V1EnvVar(name="CERT_PATH", value="/dbt/trusted/ca.crt"), ], @@ -571,7 +594,7 @@ data: apiVersion: v1 kind: Secret metadata: - name: airflow-credentials + name: airflow-admin-credentials type: Opaque stringData: adminUser.username: admin @@ -579,5 +602,12 @@ stringData: adminUser.lastname: Admin adminUser.email: airflow@airflow.com adminUser.password: "{{ airflowAdminPassword }}" - connections.secretKey: "{{ airflowSecretKey }}" - connections.sqlalchemyDatabaseUri: postgresql+psycopg2://airflow:airflow@postgresql-airflow/airflow +--- +apiVersion: v1 +kind: Secret +metadata: + name: airflow-postgresql-credentials +type: Opaque +stringData: + username: airflow + password: airflow