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

Airflow 쿠버네티스 로컬 환경 구축과 Vault로 보안 강화하기

로컬 Airflow on K8S 구축에 Vault 곁들이기 - Airflow 개발 환경 고도화
Do not index
Do not index
안녕하세요 오토피디아 데이터 엔지니어 김건우입니다.
오토피디아에서는 여러 배치 작업에 Airflow를 사용하고 있습니다. 서비스가 많아지고 요구사항이 늘어나면서 Airflow의 DAG가 점차 늘어났는데요. 그러다보니 초반에 사용하던 방식 그대로 Airflow를 활용하면 불편한 부분이 생겨났고 이러한 변화에 따라 Airflow 사용 환경을 개선할 필요성이 굳어졌습니다.
이번 글에서는 기존에 오토피디아에서 Airflow를 사용하던 방식과 이를 사용하면서 겪은 문제들을 어떻게 해결하였는지 공유드리려고 합니다.

문제점

지금까지 Airflow를 운영하면서 겪은 여러 불편함이 있었는데요. 모든 문제를 한 번에 해결할 수는 없기 때문에 일단 가장 시급한 문제들부터 우선순위를 매겨 해결하려 하였습니다. 제가 정의한 문제는 총 4가지로 아래와 같습니다.
  1. DAG의 로컬 테스트가 안됨
  1. 수 많은 secret들이 github에 존재함
  1. Airflow 사용에 대한 공통 사용 가이드가 없음
  1. 다른 on-premise 리소스들은 대부분 ArgoCD로 관리되고 있는데 Airflow만 아닌 상황
이번 글에서는 1, 2, 4번 문제에 대해 자세히 알아보고 어떤 식으로 해결하였는지를 공유드리려고 합니다.

DAG 로컬 테스트

가장 해결이 필요하다고 생각한 것은 DAG 로컬 테스트입니다. 오토피디아에서는 Airflow를 dev, prod로 환경을 구분하여 사용하고 있습니다. 각 환경은 별도의 Airflow 깃허브 브랜치로 관리되고 있고, git-sync를 통해 해당 브랜치에 코드를 merge하면 10초 간격으로 Airflow와 싱크됩니다.
지금까지는 새로운 DAG를 만들고 테스트를 하기 위해서는 dev 브랜치에 머지를 하고 dev 환경에서 추가한 DAG를 실행하며 테스트해야 했습니다. 이렇게 되면 2가지 큰 불편함이 존재했습니다.
  • DAG 테스트를 위해서는 테스트를 할 때마다 dev 환경에 코드를 머지하고 싱크되기까지(10초) 기다려야 한다.
  • 만약 dev에 테스트를 하고 있는데 다른 누군가가 dev의 자신의 코드를 prod로 merge 해야한다면 병목이 될 수 있다.
따라서 이러한 문제들을 해결하기 위해서 로컬에서 prod, dev와 동일한 환경으로 DAG를 테스트할 수 있도록 하여 최대한 위와 같은 비효율을 줄이는 것이 필요했습니다.

하드코딩된 secret

이제까지는 별도로 secret을 관리하는 방법이 없다보니 만약 필요하다면 secret 파일을 직접 넣거나 하드코딩 하였습니다. 그러다보니 모든 시크릿이 github에 올라가있어 보안적으로도 좋지 않았고, 만약 .gitignore로 secret을 제외하면 다른 누군가가 해당 DAG를 수정할 때 불편함을 겪기도 하였습니다.
이전까지는 이러한 문제를 해결할 방안이 없었기 때문에 어려움을 겪었지만 사내에 Vault를 도입하면서 모든 중요 secret들을 중앙에서 관리할 수 있게되며 문제를 조금 더 쉽게 해결할 수 있게 되었습니다.

ArgoCD 적용

오토피디아에서는 대부분의 쿠버네티스 서비스들을 ArgoCD를 통해 배포하고 있습니다. ArgoCD를 이용하면 굉장히 편리하게 쿠버네티스에 어플리케이션을 배포할 수 있고, 무엇보다 모든 서비스의 모니터링과 유지보수가 더 쉬워집니다.
대부분의 서비스가 ArgoCD로 배포가 되었지만 Airflow만 별도의 레포지토리에서 helm values.yaml을 관리하고 있었습니다. 이를 파악하고 이번 기회에 Airflow까지 ArgoCD를 통해 배포함으로써 모든 쿠버네티스 어플리케이션을 ArgoCD 하나로 관리하고자 하였습니다.
ArgoCD를 통해 배포된 어플리케이션들
ArgoCD를 통해 배포된 어플리케이션들

해결과정

Airflow 로컬 개발 환경 구축

가장 먼저 착수한 작업은 Airflow 로컬 개발 환경 구축입니다. 구축을 위해서 2가지 사항을 가장 중요하게 고려하였습니다.
  • dev, prod과 동일한(거의 비슷한) 환경이어야 한다.
  • 모든 사용자들은 동일한 환경을 공유해야 한다.
하지만 첫번째 조건부터 큰 장벽을 만났는데요. 실제 프로덕션 환경이 쿠버네티스를 사용하였기 때문에 로컬에 완전히 동일한 환경을 구성하는 것이 어려운 상황이었습니다. 그래서 쏘카의 Airflow 고도화 글을 참고하여 Docker compose로 구축을 하기로 결정을 하려던 그때..!!
문득 로컬에서도 간단하게 미니 쿠버네티스 환경을 구축할 수 있다는 사실이 떠올랐습니다. 쿠버네티스 자체가 워낙 무겁다보니 이를 최대한 경량화해서 사용하려는 노력들이 있었는데요. 대표적으로 minikube, k3s 등 로컬 쿠버네티스를 구성할 수 있는 다양한 툴이 있습니다. 저는 그 중에서도 가장 구축이 간단하다고 판단한 kind를 이용하여 로컬 쿠버네티스 환경을 구성하기로 하였습니다.
kind는 kubernetes in docker의 줄임말로 기존에 사용하던 도커만 있으면 쿠버네티스를 사용할 수 있게 해줍니다. 기본적으로 오토피디아 내에 개발자들은 도커 데스크탑이 설치되어 있기 때문에 별도의 설치 과정없이 편리하게 사용이 가능하다고 판단하였습니다.
이후 실제 클러스터를 구축하는 작업을 진행하였습니다. 맥의 경우 brew install kind 로 kind 패키지 매니저를 설치하고 kind cli를 사용할 준비를 하면 이제 Airflow를 위한 kind 클러스터를 생성해야 합니다. 클러스터는 kind 명령어로 아주 쉽게 생성할 수 있지만 그전에 중요한 작업이 있습니다. 로컬에 있는 DAG를 클러스터에서 사용하기 위해서는 로컬과 클러스터 간에 폴더를 공유할 방법이 필요합니다. 이를 위해서 볼륨 마운트를 사용하여 클러스터가 로컬에서 필요한 폴더를 공유할 수 있도록 연결하였습니다. 클러스터 구축을 위한 config 코드는 아래와 같습니다.
# 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는 yaml 파일을 통해서 클러스터를 생성할 때 설정을 통해 다양한 설정을 할 수 있습니다. 위의 yaml은 컨트롤 플레인 역할을 하는 노드 하나로 구성된 클러스터입니다. 1개의 노드만 생성한 이유는 애초에 해당 클러스터는 Airflow의 KubernetesOperator의 로컬 테스트를 위한 역할이기 때문에 많은 노드가 필요가 없었기 때문입니다. (다시 말하면 KubernetesOperator가 돌아가기만 하면 됐기 때문입니다)
그리고 해당 노드는 extraMounts 설정을 통해서 로컬에서 마운트가 필요한 폴더와 볼륨 마운트되도록 하였습니다. 이를 통해 클러스터와 로컬 환경은 실시간으로 서로의 폴더의 변경사항이 반영되게 됩니다.
클러스터 설정이 끝났다면 이제 본격적으로 kind 클러스터를 생성해야 합니다. 이는 아래의 kind 명령어로 아주 손쉽게 구축할 수 있습니다.
kind create cluster --name airflow --config kind-cluster.yaml --image kindest/node:v1.23.5
이제 Airflow가 실행될 클러스터는 모두 세팅이 마무리 되었습니다. 이제 본격적으로 Airflow를 클러스터에 설치하여야 합니다. Airflow는 기존처럼 helm 차트를 이용하여 설치하였습니다.
여기서 중요한 점은 한번 더 볼륨 마운트를 해주어야 한다는 것입니다. 처음에 클러스터와 로컬 폴더를 마운트하였었는데요. Airflow는 클러스터 안에서 다시 또 컨테이너(pod)로 실행되기 때문에 이번에는 클러스터와 컨테이너 사이의 볼륨 마운트가 한 번 더 필요합니다. 이는 아래와 같은 Airflow helm의 value.yaml 설정으로 구축할 수 있었습니다.
---
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 value를 통해서 본격적으로 Airflow 배포를 진행합니다.
helm upgrade --install airflow-local airflow-stable/airflow  --values ./local.values.yaml
배포가 모두 끝났다면 Airflow web pod와 포트 포워딩을 해주고 localhost:8080 를 통해 airflow web에 접속을 해보면 접속이 잘 되는 것을 볼 수 있습니다.
notion image
이제 기본적인 PythonOperator부터 KubernetesOperator까지 dev 환경에 머지할 필요없이 로컬 환경에서 빠르게 테스트할 수 있게 되었습니다.

Vault를 이용한 secret 관리

바로 이어서 착수한 작업은 Airflow의 secret 관리입니다. 앞서 말씀드린 것처럼 지금까지는 Airflow에서 별도로 secret을 관리하는 방법이 없다보니 폴더에 필요한 secret을 그대로 넣어주었습니다. 이는 보안에 굉장히 치명적이기도 하고 시크릿 파일이 이리저리 산재하다보니 관리 측면에서도 비효율적이었습니다.
하지만 최근 시크릿 매니저 오픈소스인 Vault를 사내에 도입하면서 한층 더 강화된 secret 관리가 가능해졌고 airflow에도 이를 도입해보고자 하였습니다. 어떤 식으로 도입할 지에는 다양한 방법이 있었는데, 저는 쉽게 vault를 이용할 수 있도록 하는 파이썬 라이브러리인 hvac을 이용해보기로 하였습니다.
vault를 도입할 때 가장 신경써야 했던 부분은 크게 두가지입니다.
  1. 모든 DAG에서 vault에 쉽게 접근할 수 있도록 한다.
  1. vault에 접근하는 token에 대한 보안
먼저 첫번째 사항을 위해 저는 common.py를 이용했습니다. 오토피디아 airflow에서는 모든 DAG가 기본적으로 사용해야 하는 DAG 설정과 라이브러리를 가지고 있는 common.py를 만들어 각 DAG에서 필수적으로 임포트하여 사용하고 있습니다. 이러한 점을 이용하여 “vault의 secret을 가져오는 getter 함수를 common.py에 만들어서 각 DAG에서 임포트할 수 있도록 한다면 편리하게 이용할 수 있지 않을까” 생각하게 되었습니다. 이를 위해 아래와 같이 common.py에 vault secret getter 함수를 만들고 각 DAG에서 사용할 수 있도록 하였습니다.
# 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 getter 함수를 보시면 hvac 클라이언트 안에 token 파라미터에 환경변수를 이용하여 vault token을 넣어주고 있습니다. 아무리 vault의 토큰값이 쿠버네티스 secret을 통해 관리되고 있다고 하더라도 만약 해당 토큰이 유출되어 버린다면 모든 값이 속수무책으로 유출되어 버린다는 크나큰 문제가 있었습니다.
그리고 이를 해결하기 위해 vault의 Approle 기능을 사용하기로 하였습니다. Approle 인증 방식은 어플리케이션별로 로그인 기능을 분리할 수 있습니다. Airflow에서 접근하는 것이라면 airflow_role을 만들어 해당 앱만 사용할 수 있도록 만들어주는 것이죠. 아래는 Approle의 작동방식을 도식화한 그림입니다.
https://irezyigit.medium.com/approle-authentication-with-vault-7dab4ce0b75e
Approle에서는 인증을 위해 roleID와 secretID를 사용하는데, 이는 쉽게 말해 우리가 흔히 사용하는 아이디와 비밀번호와 같다고 생각하면 됩니다.
Approle을 만드는 과정은 아래와 같습니다. 가장 먼저 approle이 사용할 policy를 만들어줍니다. 이 policy를 통해 approle이 접근할 수 있는 폴더와 권한을 설정할 수가 있습니다. 이것이 Approle을 사용했을 때 얻을 수 있는 이점 중 하나이죠. 만약 approle이 airflow secret 폴더에서 읽기 권한만 갖기를 원한다면 아래와 같이 설정합니다.
vault policy write airflow_policy - <<EOF
path "secret/airflow/*" {
  capabilities = ["read"]
}
EOF
이제 본격적으로 airflow가 사용할 approle을 만들어줍니다. approle에는 다양한 설정이 있는데요. 토큰의 수명과 토큰을 가져오는 secret_id의 최대 사용 수를 지정하는 secret_id_num_uses, 허용 ip 등 다양한 설정값을 지정할 수 있습니다. 여기서 가장 중요한 값은 앞서 만들어두었던 policy를 사용하기 위한 token_policies 파라미터입니다. 여기에 아까 만들어놓은 policy를 지정합니다.
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
이후 vault read auth/approle/role/airflow_role/role-id 을 통해 role이 잘 생성된 것을 확인할 수 있습니다. 이제 role을 사용하기 위한 secret_id를 발급받아야 하는데요. vault write -f auth/approle/role/airflow_role/secret-id 명령어를 통해 secret_id를 발급받을 수 있습니다. 해당 값은 발급 이후로는 확인하지 못하기 때문에 잘 보관하여야 합니다. 저는 이를 vault에 잘 저장하였습니다.
그리고 새롭게 생성한 approle을 이용하여 airflow에서 vault에 로그인을 해서 secret 값을 받아오기 위한 작업을 진행했습니다. 아까 만들어놓은 secret getter 함수를 아래와 같이 살짝 수정해주기만 하면 되는데요.
# 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
기존에는 token값을 바로 넣어주었던 것과 다르게 secret_id를 통해 vault에 접근할 수 있게 되었습니다. 이렇게 Approle을 사용하면 만약 secret_id가 유출된다고 하더라도 어플리케이션별로 접근 권한이 분리되어 있어 모든 secret 값이 유출되지 않고, 빠르게 새로운 secret_id를 다시 생성하여 대응할 수 있다는 장점이 있습니다.

ArgoCD 적용

마지막으로 Airflow에만 적용되어 있지 않던 ArgoCD를 적용하였습니다. 일전에 ArgoCD Vault Plugin을 이용하여 ArgoCD에서 vault를 사용할 수 있게끔 설정해놓았았는데요. (자세한 내용은 이후에 공유될 예정입니다😀) 이를 통해 helm에 직접적으로 secret을 넣을 필요없이 vault의 secret을 주입할 수 있게됩니다.
Airflow 배포에는 ArgoCD와 함께 kustomize를 사용하였습니다. kustomize를 사용한 가장 큰 이유는 커스터마이즈의 기능 중 하나인 secretGenerator 와 vault를 함께 쓰기 위해서입니다. 기본적으로 오토피디아에서는 Airflow on K8S를 사용하고 있고, 시크릿값들을 K8S의 secret을 이용하여 관리하고 있습니다. 따라서 각 파드에서 이러한 secret들을 사용하려면 쿠버네티스에 secret을 일일이 만들어주어야 합니다.
하지만 secretGenerator 를 이용하면 이러한 secret 값들을 일일이 만들 필요없이 자동으로 만들 수 있고, 더욱이 플러그인을 함께 사용하면 아래와 같이 secret을 하드코딩할 필요없이 vault의 시크릿값을 주입할 수 있습니다.
...
secretGenerator:
    - name: vault
      literals:
          - value=<path:secret/data/airflow/dev#secret_id>
...
또한 kustomize에는 helm 차트에 대한 설정도 할 수 있어 사용할 Airflow chart와 세부 설정을 위한 values.yaml을 명시하였습니다.
...
helmCharts:
    - name: airflow
      repo: https://airflow-helm.github.io/charts
      version: 8.5.2
      releaseName: airflow
      namespace: airflow
      valuesFile: values.yaml
      includeCRDs: true
...
마지막으로 prod와 dev으로 구분된 환경을 한번에 구성하기 위해 ArgoCD의 ApplicationSet을 이용하여 준비를 모두 마친 뒤 실제 배포를 진행하였고 정상적으로 모두 배포가 완료된 것을 확인할 수 있었습니다.
---
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}}
...
notion image

마무리

개선된 점

  1. 프로덕션과 유사한 스펙으로 Airflow 로컬 테스트 환경을 구축하여 테스트 속도를 1분 → 10초 정도로 줄여 빠른 개발이 가능하도록 하였음.
  1. Vault 도입을 통해 보안 체계를 갖추고 Approle 인증 방식으로 어플리케이션별 권한 분리를 통해 보안을 더욱 강화하였음.

더 개선할 점

  1. kind에서는 현재 사용하고 있는 Airflow 2.1.2 버전이 동작하지 않아 로컬 버전만 최신 버전을 사용(2.5.3)
    1. 프로덕션 Airflow 환경도 버전업하여 동일한 버전으로 만들 예정
  1. Airflow 사용에 대한 공통 사용 가이드 작성
    1. DAG 생성 규칙이나 기타 컨벤션이 정해진 부분이 없어 여러 방식이 혼재되어 있는 상황
  1. 아직까지 많은 DAG에 secret이 하드코딩 되어 있는 상황
    1. 새로 생성하는 DAG부터 secret을 적용하고 이후 기존 DAG에 소급 적용할 예정
 
Airflow는 사내의 많은 배치 처리를 담당하고 있는 중요 서비스입니다. 위의 시도들을 통해 조금은 Airflow 사용성 향상에 도움을 줄 수 있었지만 아직까지 개선해나갈 부분이 많다고 느끼고 있습니다. 앞으로는 Airflow 뿐만 아니라 사내에서 사용하고 있는 다른 데이터 플랫폼들 또한 조금씩 보완하여 더욱 안정적이고 편리한 데이터 플랫폼을 만들어 나갈 예정입니다. 긴 글 읽어주셔서 감사하고 다음에 더 좋은 글로 찾아뵙겠습니다!

참고자료

오토피디아 채용에 관한 모든 것을 준비했어요

첨단기술을 통한 모빌리티 혁신, 함께 하고 싶다면?

채용 둘러보기

글쓴이

김건우
김건우

Data Engineer

    0 comments