이 글에서 다루는 것

Airflow Sensor를 reschedule 모드로 리팩토링하고 성공 체인을 보장하는 DAG 구조를 만들고, FastAPI에 cert-manager 기반 TLS + IP 화이트리스트를 적용하는 과정을 다룹니다.

선수지식

이 단계에서 해결하려는 문제

모델을 등록하는 것보다 더 중요한 건 언제 Reload가 실행되느냐이다. 등록 실패나 READY 미도달 상태에서 /reload가 실행되면 잘못된 모델이 서빙될 수 있다. Airflow 흐름을 한 번 더 단단히 조이고, FastAPI 쪽은 HTTPS·화이트리스트까지 적용해 안전하게 자동화된다는 확신을 주는 구조를 만든다.

🎯 핵심 요약

  • Airflow: PythonSensor(mode="reschedule")로 READY 감시, TriggerRule.ALL_SUCCESS로 성공 체인에서만 /reload
  • 데이터/상태 꼬임 방지: Variable 검증 유틸 + XCom 키 표준
  • FastAPI: Ingress TLS(HTTPS) + IP 화이트리스트 강화
  • cert-manager + ClusterIssuer(self-signed) 자동 발급/갱신: 템플릿에 어노테이션 자동 주입 반영 완료

5-1️⃣ Airflow 안정화: Sensor 리팩토링 & XCom/Variable 표준

1) DAG 구조

mermaid-05-01.png

2) READY 감시 전용 센서

# dags/ml_code/sensor_model_ready.py
import mlflow
from ml_code.config import get_mlflow_client
from airflow.utils.log.logging_mixin import LoggingMixin

logger = LoggingMixin().log

def check_model_ready(model_name: str, version: str) -> bool:
    client = get_mlflow_client()
    mv = client.get_model_version(name=model_name, version=version)
    logger.info(f"[Sensor] 상태 체크: {model_name} v{version} → {mv.status}")
    if mv.status == "READY":
        return True
    elif mv.status == "FAILED_REGISTRATION":
        raise RuntimeError("모델 등록 실패: 상태 FAILED_REGISTRATION")
    return False

3) DAG 본문(핵심 발췌) — 성공 체인 보장 & 표준화된 변수/XCom

# dags/dag_ml_train_register_reload.py (핵심부)
from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator
from airflow.sensors.python import PythonSensor
from airflow.utils.trigger_rule import TriggerRule

# ... (import 및 헬퍼 함수 생략)

def train_and_evaluate(ti, **_):
    C = get_param("logreg_C", 1.0, float, lambda x: 0.001 <= x <= 10.0)
    max_iter = get_param("logreg_max_iter", 200, int, lambda x: x > 50)
    threshold = get_param("accuracy_threshold", 0.9, float, lambda x: 0.5 <= x <= 0.99)
    model_name = Variable.get("model_name")
    alias = Variable.get("mlflow_alias")
    acc, run_id = train_model(C=C, max_iter=max_iter)
    for k, v in {"run_id": run_id, "model_name": model_name,
                  "alias": alias, "acc": acc, "threshold": threshold}.items():
        ti.xcom_push(key=k, value=v)

with DAG("ml_train_register_and_reload", schedule=None, catchup=False,
         tags=["mlops", "train", "sensor", "reload"], on_failure_callback=alert_slack) as dag:
    train = PythonOperator(task_id="train_and_evaluate", python_callable=train_and_evaluate)
    branch = BranchPythonOperator(task_id="check_result", python_callable=check_result)
    register = PythonOperator(task_id="register_model", python_callable=register_model_task)
    sensor = PythonSensor(task_id="check_model_ready", python_callable=sensor_ready_func,
                          poke_interval=10, timeout=180, mode="reschedule")
    reload = PythonOperator(task_id="trigger_reload", python_callable=trigger_reload_task,
                            trigger_rule=TriggerRule.ALL_SUCCESS)  # 성공 체인에서만 실행

    train >> branch >> [register, failure]
    register >> sensor >> reload

운영 포인트

  • 센서는 reschedule로 워커 슬롯 잠식 방지
  • /reload는 등록 + READY 센서 모두 성공해야만 실행
  • Variable 캐스팅/검증 실패는 Slack 통지 + 안전 기본값으로 진행
  • XCom 키명: run_id / model_name / alias / acc / threshold / version 고정

5-2️⃣ FastAPI HTTPS & cert-manager 자동 발급

1) 구조

mermaid-05-02.png

TLS는 Ingress에서 종단(복호화)되고, 백엔드는 HTTP로 유지합니다. 인증서 발급·갱신은 cert-manager가 ClusterIssuer(self-signed)를 통해 자동 처리합니다.

2) Values(dev/prod) — 실환경 값과 일치

# charts/fastapi/values/dev.yaml (발췌)
ingresses:
  - name: fastapi-local
    host: fastapi.local
    className: nginx
    whitelist: 192.168.18.0/24
    paths:
      - path: /
        pathType: Prefix
    tls:
      enabled: true
      secretName: fastapi-dev-tls
      issuerName: selfsigned-issuer

# charts/fastapi/values/prod.yaml (발췌)
ingresses:
  - name: fastapi-prod
    host: fastapi.prod
    className: nginx
    whitelist: 192.168.18.0/24
    paths:
      - path: /
        pathType: Prefix
    tls:
      enabled: true
      secretName: fastapi-prod-tls
      issuerName: selfsigned-issuer

3) Ingress 템플릿 — 어노테이션 자동 주입

tls.issuerName이 설정되면 cert-manager 어노테이션이 자동으로 주입됩니다.

# charts/fastapi/templates/fastapi-ingress.yaml (핵심부)
{{- range .Values.ingresses }}
---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: {{ $.Release.Name }}-{{ .name }}
  annotations:
    nginx.ingress.kubernetes.io/whitelist-source-range: "{{ .whitelist }}"
    nginx.ingress.kubernetes.io/ssl-redirect: "true"
{{- if and .tls .tls.enabled .tls.issuerName }}
    cert-manager.io/cluster-issuer: "{{ .tls.issuerName }}"
{{- end }}
spec:
  ingressClassName: {{ $.className | default "nginx"}}
  rules:
    - host: {{ .host }}
      # ... (paths 생략)
{{- if and .tls .tls.enabled }}
  tls:
    - hosts:
        - {{ .host }}
      secretName: {{ .tls.secretName }}
{{- end }}
{{- end }}

4) ClusterIssuer (공통 1회)

apiVersion: cert-manager.io/v1
kind: ClusterIssuer
metadata:
  name: selfsigned-issuer
spec:
  selfSigned: {}

5) 검증

# 리소스 확인
kubectl get clusterissuer
kubectl -n fastapi-dev get certificate
kubectl -n fastapi-dev get secret | grep fastapi-dev-tls

# Ingress에 어노테이션 반영 여부
kubectl -n fastapi-dev get ing -o yaml | grep -A3 annotations:

# HTTPS 응답(로컬/테스트)
curl -vkI https://fastapi.local --resolve fastapi.local:443:<LB_IP>

체크리스트

  • Airflow PythonSensor(mode="reschedule") 적용
  • /trigger_reload는 TriggerRule.ALL_SUCCESS로 성공 체인 전용
  • get_param() 유틸로 Variable 검증/캐스팅/Slack 통지
  • XCom 키 표준: run_id / model_name / alias / acc / threshold / version
  • Ingress 템플릿에 cert-manager 어노테이션 자동 주입 반영 완료
  • dev/prod 별 host / secretName / whitelist / issuerName 값 일치 확인

🧩 팁

증상원인해결
Sensor 타임아웃READY 미도달모델 등록 실패 로그 확인, 파라미터/데이터 재점검
Reload가 먼저 실행TriggerRule 누락/trigger_reload를 ALL_SUCCESS로 고정
인증서 Pendingcert-manager 이벤트 오류kubectl describe certificate로 원인 확인
SAN 불일치 경고host 변경 잔존해당 Certificate 삭제 → cert-manager 자동 재발급
403 접근 거부화이트리스트 불일치whitelist-source-range 대역 확인

설계 판단 (Why This Way?)

PythonSensor mode=reschedule로 대기 시간이 긴 센서의 워커 슬롯 낭비를 방지하고, TriggerRule.ALL_SUCCESS로 선행 Task 실패 시 잘못된 모델이 서빙되는 것을 차단했습니다. 내부망 환경에서는 cert-manager self-signed 인증서가 현실적입니다.

다음에 읽을 글

→ MLOps 운영 고도화 6단계: GitOps 고도화 — ArgoCD 동기화 전략 정교화