Setup Airflow on Kubernetes with remote logging to GCS

Published in

Mils’ Blog

5 min readNov 7, 2021

Kubernetes series:
1. Setup Spark job and history server on K8s with GCS log
2. Setup Airflow on Kubernetes with remote logging to GCS
3. Create external database with PVC for Airflow Kubernetes

วันนี้เรากลับมาด้วยซีรี่ย์ Kubernetes เหมือนเดิม แต่คราวนี้เราจะมา setup ตัว Airflow กัน และจะ remote log ไปไว้ที่ Google Cloud Storage กันค่ะ รวมถึงการทำ git sync สำหรับ DAG ด้วย

Prerequisite นะคะ เราจะใช้ tools ทั้ง helm และ kubectl นะคะ ลง 2 ตัวนี้ให้เรียบร้อย

ก่อนอื่นเลย เราจะสร้าง namespace ที่ชื่อ airflow-cluster ก่อนค่ะ

kubectl create namespace airflow-cluster

ในบทความนี้เราจะใช้ helm chart จากทาง Official Airflow นะคะ หรือใครสะดวกจะไปใช้ helm ของ Airflow-helm (User Community) ก็ได้ ตรงนั้นก็เขียนละเอียดดีค่ะ

helm repo add apache-airflow https://airflow.apache.org
helm repo update

แล้วเราก็ install กันเลย

helm install airflow apache-airflow/airflow — n airflow-cluster

แล้วรอแปบนึง เมื่อรันสำเร็จ เราจะเห็น instruction รายละเอียดต่างๆที่ทาง airflow เขาจัดให้ ตั้งแต่ port-forward, username, port อะไรต่างๆ

เราจะลองเช็คว่าตอนนี้ pods เรามีอะไรบ้าง ด้วยคำสั่ง

kubectl get pods -n airflow-cluster

ถ้ารันสำเร็จแล้ว ทุกอย่างก็จะขึ้น status ว่า RUNNING แบบนี้นะคะ

Step ถัดไป เราจะปรับ Airflow ให้ใช้ Kubernetes Executor กันค่ะ ><
ต่อจากนี้เราจะใช้ values.yaml เป็นหลักค่ะ

ขั้นตอนแรก download ไฟล์นี้มาก่อนเลย ซึ่งเราจะจัดให้ ไฟล์ values.yaml อยู่ใน folder ที่ชื่อ airflow_charts ไว้เพื่อความเรียบร้อย

mkdir airflow_charts
helm show values apache-airflow/airflow > airflow_charts/values.yaml

Directory ตอนนี้เราจะหน้าตาประมาณนี้ค่ะ

└── airflow_charts
 └── values.yaml

คราวนี้ เราจะแก้ไขไฟล์ values.yaml กัน โดยจะเปลี่ยน default ที่เขาให้มาคือ CeleryExecutor เป็น KubernetesExecutor ค่ะ

```values.yaml# Airflow executor
# Options: LocalExecutor, CeleryExecutor, KubernetesExecutor, CeleryKubernetesExecutor
executor: "KubernetesExecutor" <<---เปลี่ยนตัว executor เป็น k8s

คราวนี้เราก็จะ upgrade ที่เราแก้ไขไปด้วยคำสั่ง

cd airflow_chartshelm upgrade --install airflow apache-airflow/airflow -n airflow-cluster -f values.yaml

รันดู pods ซิ

เราจะเห็นว่า pods หายไปบางส่วนจนเหลือ 4 ตัวค่ะ รวมถึง worker ด้วย

ถึงตอนนี้เราลองเข้า Airflow Web UI ให้ชื่นใจกันดีกว่า

kubectl port-forward svc/airflow-webserver 8080:8080 --namespace airflow-cluster

โดย username เป็น admin และ password เป็น admin ตามที่เห็นในตอน upgrade chart ค่ะ

แต่เนื่องจากตอนนี้เรายังไม่มี DAGs เลย เลยยังไม่มี DAG ปรากฏใน web
ขั้นต่อไปเราจะไปใส่ DAG กันค่ะ

Step ต่อไป เราจะใช้ Git Sync กันค่ะ
Git Sync จะช่วยให้เราบอก Airflow ของเราว่าจะไปจิ้มอ่าน DAG ได้ที่ไหนผ่าน Git repository ค่ะ ในกรณีที่เราจะแยก DAG ออกไปอยู่ repo อื่นเช่นเคสนี้

``` values.yamldags:
 gitSync:
    enabled: true        #<---- ปรับเป็น true
    repo: ssh://git@gitlab.com/<user>/<repo_name>.git #<---- ชื่อ git repo
    branch: main        #<---- branch ที่เราจะใช้
    rev: HEAD
    depth: 1
    maxFailures: 0
    subPath: "dags/"         #<-- folder ที่เราจะ sync จากตรงนั้น
    sshKeySecret: airflow-ssh-git-secret    #<-- ชื่อ secret ที่เราจะใช้
    wait: 10                 #<---- ปรับ interval ให้เร็วขึ้นเป็น 10sec
    containerName: git-sync
    uid: 65533
    extraVolumeMounts: []
    env: []
    resources: {}

ตรง repo เอาแบบแม่นสุดก็ไปเอาตรง clone ใน git มาแปะนี่แหล่ะค่ะ
ท่านี้เราจะฝังแบบ ssh นะคะ ดังนั้นเราจะใช้ attribute ที่ชื่อว่า sshKeySecret เพื่อบอกว่า secret ของเราจะเป็น private key บนเครื่อง ในการยืนยันว่าเราสามารถเข้าถึง git repo ของเราได้
ดังนั้นเราจะต้องสร้าง secret เพิ่มด้วย โดยจะเข้าไปที่ .ssh

cd ~/.ssh/kubectl create secret generic airflow-ssh-git-secret --from-file=gitSshKey=<private_key_file_name> -n airflow-cluster

โดยเราจะต้องฝัง public key ไว้ที่ repo ของเราด้วยนะคะ ถ้าเป็น Gitlab ก็ต้องเข้าไปที่ repository นั้น
Settings > repository > Deploy keys

เสร็จแล้วก็ upgrade ได้เลย

helm upgrade --install airflow apache-airflow/airflow -n airflow-cluster -f values.yaml

ถึงขั้นตอนนี้ ถ้าเรากลับมาดูที่ Airflow Web UI จะเห็นว่ามี DAG ขึ้นแล้ว หรือไม่ก็จะ import errors แบบของเรา5555555
เหตุผลเพราะตอนนี้ DAG เราเพราะต้อง install package ที่ต้องใช้เพิ่ม

แต่ถือว่า sync ไป git ได้แล้วนะทุกคนนน เย้ๆๆๆ

Step ต่อมา เราจะ build image เพิ่มเพื่อให้ values.yaml ไปใช้ image ที่มีการลง package เสริมตามใจเรา

step นี้ทำตามแบบระมัดระวังด้วยเน้อ

ใน Dockerfile ของเราก็จะมี package หลักๆคือ

apache-airflow[google] (เอาไว้ต่อให้ airflow ต่อไป GCS ในอนาคต)
และ package ต่างๆที่เราต้องการเพิ่มเติมหลัง pip install ได้เลย

```DockerfileFROM apache/airflow:2.2.0-python3.8RUN pip install apache-airflow[google]==2.2.0

แล้วเราก็ build image ตามปกติ

docker build -t <dockerhub>/<image_name>:v3 .
docker login
docker push <dockerhub>/<image_name>:v3

คราวนี้มาในไฟล์ values.yaml เราก็ไปแก้ชื่อ image ซะหน่อยย

# Default airflow repository -- overrides all the specific images below
defaultAirflowRepository: mesodiar/airflow-datahub ##<---- ชื่อ image ที่เราใช้# Default airflow tag to deploy
defaultAirflowTag: "v3" ##<----  version ของ image นั้น

แล้ว upgrade chart ใหม่จ้า

helm upgrade --install airflow apache-airflow/airflow -n airflow-cluster -f values.yaml

ลืมบอกไปว่าถ้า refresh ที่ airflow web ui แล้วพัง ก็ให้รันคำสั่ง port-forward ใหม่นะคะ

ถ้ารันแล้วไม่ติดปัญหาอะไรก็จะขึ้น dag สวยๆแบบนี้ค่ะ

หากเราได้เข้าไปกดรัน dag แล้วลองเข้าไปดู log ข้างใน จะพบว่า ไม่มี log อะไรให้ดู
เลยย มันจะขึ้นว่า unable to fetch logs from worker

เหตุผลเพราะว่าตัว pod worker เราจะถูกสร้างขึ้นมาก็ต่อเมื่อรัน DAG เท่านั้น พอทำงานเสร็จมันก็จะดับไป การจะดู log นั้นจึงยากมาก

ดังนั้น step ถัดไป เราจะทำการเก็บ log ไปที่ Cloud Storage bucket และให้ airflow จิ้มไปที่ bucket นั้น เพื่อที่เราจะสามารถดู log ผ่าน Airflow ได้

กลับมาที่ไฟล์ values.yaml อีกครั้ง

เราจะสร้าง connections ไปยัง GCP ให้ได้ก่อน

# Images
images:
  airflow:
    repository: ~
    tag: ~
    pullPolicy: IfNotPresent
    connections:
    - id: google_cloud_default   
      type: google_cloud_platform
      description: my GCP connection
      extra: |-
        { "extra__google_cloud_platform__scope": "https://www.googleapis.com/auth/cloud-platform"}

จริงๆแล้ว connections ที่ชื่อ google_cloud_default นี้จะมีให้อยู่แล้ว เพียงแต่เราจะ override ตัว scope เพิ่มนิดหน่อย
type จำเป็นจะต้องเป็น google_cloud_platform นะๆ
อีกจุดที่สำคัญมาก ก็คือ env variable ตัว GOOGLE_APPLICATION_CREDENTIALS ที่ต้องมี value เป็น path ไปหา JSON file
(แนะนำว่า สังเกต comment ดีๆ จะให้ทุก container สามารถเอาตัวนี้ไปใช้ได้ด้วยนะ)
หากใครงงกับ step นี้แนะนำให้อ่านวิธี authentication ของ GCP ค่ะ

# Environment variables for all airflow containers
env:
  - name: GOOGLE_APPLICATION_CREDENTIALS
    value: "/opt/airflow/cred/<secret_json_file>.json" <-- path ไปหาไฟล์ใน pod

และที่เหลือ เราจะทำการ mount volume เจ้าตัวไฟล์ json ในทุกๆ pod เลย
โดยที่เราจะให้ volumes ทำการอ่านจากค่า secret ที่ชื่อ google-sa-key
แล้ว volumeMount จะเป็นคน mount ไปยัง path ข้างใน pod อีกทีที่ /opt/airflow/cred
(นั่นคือเหตุผลว่าทำไม GOOGLE_APPLICATION_CREDENTIALS ถึงต้องมีค่านี้เหมือนกัน)

# Airflow scheduler settings
scheduler:...# Mount additional volumes into webserver.
  extraVolumes: 
    - name: google-cloud-key
      secret:
        secretName: google-sa-key
  extraVolumeMounts:
    - name: google-cloud-key
      mountPath: /opt/airflow/cred...# Airflow webserver settings
webserver:...
 # Mount additional volumes into webserver.
  extraVolumes: 
    - name: google-cloud-key
      secret:
        secretName: google-sa-key
  extraVolumeMounts:
    - name: google-cloud-key
      mountPath: /opt/airflow/cred...# Airflow Triggerer Config
triggerer:
..# Mount additional volumes into triggerer.
  extraVolumes: 
    - name: google-cloud-key
      secret:
        secretName: google-sa-key
  extraVolumeMounts:
    - name: google-cloud-key
      mountPath: /opt/airflow/cred

ซึ่ง step นี้ เราจะสร้าง k8s secret ด้วยคำสั่งด้านล่าง โดยเราจะไปที่ path ที่วาง .json ไว้แล้วรัน

kubectl create -n airflow-cluster secret generic google-sa-key --from-file <secret_json_file>.json

ยังไม่จบค่ะ เราต้องไปบอก config ของ airflow ว่าเราจะทำการลิ้งค์ไปที่ bucket ไหนด้วย

config:
  logging:
    remote_logging: 'True'        ##<--- ปรับเป็น true
    colored_console_log: 'True'       ##<--- ปรับเป็น true
    remote_base_log_folder: "gs://<bucket_name>" 
    remote_log_conn_id: "google_cloud_default"  
    google_key_path: "/opt/airflow/cred/<secret_json_file>.json"

เสร็จแล้วค่ะทุกคน เราจะไป upgrade helm อีกรอบ

helm upgrade --install airflow apache-airflow/airflow -n airflow-cluster -f values.yaml

คราวนี้เราก็กลับไป airflow web ui เพื่อรัน dag อีกครั้ง
ถ้าสำเร็จ เราก็จะเจอว่ามันขึ้นแบบรูปด้านล่าง

และใน bucket ของเราก็จะมีไฟล์ log ขึ้นมา

เป็นอันเสร็จสิ้นการทำ remote logging ค่าา ><

ขอบคุณที่อ่านมาถึงตรงนี้ค่ะ

ref:

Setup Airflow on Kubernetes with remote logging to GCS

Written by Burasakorn Sabyeying