로컬 Airflow on K8S 구축에 Vault 곁들이기 - Airflow 개발 환경 고도화

문제점

1. DAG의 로컬 테스트가 안됨

2. 수 많은 secret들이 github에 존재함

3. Airflow 사용에 대한 공통 사용 가이드가 없음

4. 다른 on-premise 리소스들은 대부분 ArgoCD로 관리되고 있는데 Airflow만 아닌 상황

DAG 로컬 테스트

• DAG 테스트를 위해서는 테스트를 할 때마다 dev 환경에 코드를 머지하고 싱크되기까지(10초) 기다려야 한다.

• 만약 dev에 테스트를 하고 있는데 다른 누군가가 dev의 자신의 코드를 prod로 merge 해야한다면 병목이 될 수 있다.

하드코딩된 secret

ArgoCD 적용

해결과정

Airflow 로컬 개발 환경 구축

• dev, prod과 동일한(거의 비슷한) 환경이어야 한다.

• 모든 사용자들은 동일한 환경을 공유해야 한다.

# kind-cluster.yaml

apiVersion: kind.x-k8s.io/v1alpha4
kind: Cluster
nodes:
  - role: control-plane
    extraMounts:
      - hostPath: ./dags
        containerPath: /opt/airflow/dags
      - hostPath: ./dbt
        containerPath: /opt/airflow/dbt
      - hostPath: ./python_tasks
        containerPath: /opt/airflow/python_tasks
      - hostPath: ./logs
        containerPath: /opt/airflow/logs

kind create cluster --name airflow --config kind-cluster.yaml --image kindest/node:v1.23.5

---
airflow:
  image:
    repository: airflow-local
    pullPolicy: IfNotPresent
    tag: local

  executor: KubernetesExecutor

  extraVolumeMounts:
    - name: dags
      mountPath: /opt/airflow/dags
    - name: logs
      mountPath: /opt/airflow/logs
    - name: python-tasks
      mountPath: /opt/airflow/python_tasks
    - name: dbt
      mountPath: /opt/airflow/dbt

  extraVolumes:
    - name: dags
      hostPath:
        path: /opt/airflow/dags
    - name: logs
      hostPath:
        path: /opt/airflow/logs
    - name: python-tasks
      hostPath:
        path: /opt/airflow/python_tasks
    - name: dbt
      hostPath:
        path: /opt/airflow/dbt
...

helm upgrade --install airflow-local airflow-stable/airflow  --values ./local.values.yaml

Vault를 이용한 secret 관리

1. 모든 DAG에서 vault에 쉽게 접근할 수 있도록 한다.

2. vault에 접근하는 token에 대한 보안

# common.py
AIRFLOW_STAGE = os.getenv("AIRFLOW_STAGE")
SECRET_ID = os.getenv("VAULT_TOKEN")

def get_secret_from_vault():
    client = hvac.Client(url="...", token=os.environ['VAULT_TOKEN'])
    secret = client.secrets.kv.v2.read_secret_version(path=f"airflow/{AIRFLOW_STAGE}")
    secret_json = secret['data']['data']
    return secret_json

# test1_dag.py
from common import *

secret = get_secret_from_vault()
TEST1_SECRET = secret['test1_secret']

with BaseDag(
    dag_id="test1_dag", start_date=datetime(2021, 10, 21), schedule_interval="0 * * * *", catchup=False
) as dag:
    latest_only = LatestOnlyOperator(task_id="latest_only")

    test1 = BashOperator(
        task_id="test1",
        bash_command="echo $secret",
        env={"secret": TEST1_SECRET},
    )

		latest_only >> test1

vault policy write airflow_policy - <<EOF
path "secret/airflow/*" {
  capabilities = ["read"]
}
EOF

vault auth enable approle

vault write auth/approle/role/airflow_role \
    secret_id_ttl=10m \
    token_num_uses=10 \
    token_ttl=20m \
    token_max_ttl=30m \
    secret_id_num_uses=40 \
		secret_id_bound_cidrs="0.0.0.0/0" \
		token_policies=airflow_policy

# common.py
SECRET_ID = os.getenv("SECRET_ID")

def get_secret_from_vault():
    client = hvac.Client(url="...")
    client.auth.approle.login(role_id="airflow_role", secret_id=SECRET_ID)
    secret = client.secrets.kv.v2.read_secret_version(path=f"airflow/{AIRFLOW_STAGE}")
    secret_json = secret["data"]["data"]
    return secret_json

ArgoCD 적용

...
secretGenerator:
    - name: vault
      literals:
          - value=<path:secret/data/airflow/dev#secret_id>
...

...
helmCharts:
    - name: airflow
      repo: https://airflow-helm.github.io/charts
      version: 8.5.2
      releaseName: airflow
      namespace: airflow
      valuesFile: values.yaml
      includeCRDs: true
...

---
apiVersion: argoproj.io/v1alpha1
kind: ApplicationSet
metadata:
    name: airflow
spec:
    generators:
        - list:
              elements:
                  - env: prod
                  - env: dev
    template:
        metadata:
            name: airflow-{{env}}
        spec:
            project: default
            source:
                repoURL: <airflow repo url>
                targetRevision: HEAD
                path: .../airflow/overlays/{{env}}
            destination:
                server: <k8s destination address>
                namespace: airflow-{{env}}
...

마무리

개선된 점

1. 프로덕션과 유사한 스펙으로 Airflow 로컬 테스트 환경을 구축하여 테스트 속도를 1분 → 10초 정도로 줄여 빠른 개발이 가능하도록 하였음.

2. Vault 도입을 통해 보안 체계를 갖추고 Approle 인증 방식으로 어플리케이션별 권한 분리를 통해 보안을 더욱 강화하였음.

더 개선할 점

1. kind에서는 현재 사용하고 있는 Airflow 2.1.2 버전이 동작하지 않아 로컬 버전만 최신 버전을 사용(2.5.3)
1. 프로덕션 Airflow 환경도 버전업하여 동일한 버전으로 만들 예정

2. Airflow 사용에 대한 공통 사용 가이드 작성
1. DAG 생성 규칙이나 기타 컨벤션이 정해진 부분이 없어 여러 방식이 혼재되어 있는 상황

3. 아직까지 많은 DAG에 secret이 하드코딩 되어 있는 상황
1. 새로 생성하는 DAG부터 secret을 적용하고 이후 기존 DAG에 소급 적용할 예정

참고자료

전사 구성원들이 사용하는 배치 데이터 플랫폼 만들기 - Airflow Advanced

Airflow 운영 고도화를 위한 고군분투기

https://tech.socarcorp.kr/data/2022/11/09/advanced-airflow-for-databiz.html

Documentation | Vault | HashiCorp Developer

Vault reference documentation covering the main Vault concepts, feature FAQs, and CLI usage examples to start managing your secrets.

https://developer.hashicorp.com/vault/docs

Vault Part 5 - AppRole Authentication with Vault

AppRole authentication can be used to separate app based login capabilities for applications. For example, access to app1 secrets can be…

https://irezyigit.medium.com/approle-authentication-with-vault-7dab4ce0b75e

Argo CD Vault Plugin

https://argocd-vault-plugin.readthedocs.io/en/stable/