MLOps 2차 아키텍처 — Airflow를 걷어내고 Kafka 이벤트 + DevOps ArgoCD로 배포하기

앞선 글에서 MLOps 플랫폼 1차 아키텍처를 정리했다. 1차는 Airflow가 학습부터 배포까지 직접 오케스트레이션하는 구조였다. 2차에서는 이걸 두 갈래로 갈아엎었다. ① 무거운 Airflow를 걷어내고 실행을 Kafka 이벤트로 위임했고, ② 빌드·배포는 직접 하지 않고 DevOps 팀의 Jenkins·ArgoCD 인프라를 그대로 재사용했다. 이 글은 그 전환을 인프라 구성과 실제 코드로 정리한다.

핵심 요약

• 1차: MLOps 백엔드 코어(FastAPI)가 Airflow DAG를 트리거하고, KubernetesPodOperator로 학습·평가·배포를 한 그래프 안에서 돌렸다.
• 2차: MLOps 백엔드는 "커맨드를 Kafka로 발행하고, 결과를 Kafka로 받는" 역할만 한다. 무거운 일은 각 에이전트가 나눠 맡는다.
• 학습 = Kafka RUN_JOB 커맨드 발행 → MLOps 잡 에이전트가 K8s Job 실행.
• 빌드 = DevOps에 위임 → Jenkins가 이미지 빌드.
• 배포 = DevOps 배포 에이전트가 GitOps 레포의 image tag를 바꾸고 → ArgoCD가 클러스터를 그 상태로 동기화.
• 모든 결과는 다시 Kafka로 돌아와, commandId로 매칭해 상태를 갱신한다(멱등).

1. 배경 — 1차 Airflow의 한계

1차에서 Airflow는 강력했다. 학습→평가→조건→배포를 DAG 한 장으로 표현할 수 있었으니까. 하지만 운영하면서 부담이 쌓였다.

• Airflow 자체가 무겁다. 스케줄러·웹서버·메타DB를 상시 띄워야 한다. 단일 노드 개발 클러스터에서는 이 상주 비용이 컸다.
• 동적 DAG가 복잡했다. 사용자가 짠 파이프라인을 런타임 DAG로 변환하고, 메트릭을 XCom으로 넘기고, Pod 스크립트 안에서 모델 등록까지 시키는 흐름이 손이 많이 갔다.
• MSA인데 인프라를 MLOps가 들고 있었다. 빌드(Jenkins 트리거)와 배포(서빙)를 Airflow DAG 안에 욱여넣었는데, 사실 옆 팀(DevOps)이 이미 빌드·배포 인프라를 잘 갖추고 있었다.

그래서 2차의 방향은 분명해졌다. "직접 다 하지 말고, 이벤트로 위임하자."

2. 핵심 아이디어 — 이벤트로 위임하기

비유: 1차는 주문받고, 요리하고, 배달까지 혼자 다 하는 1인 식당이었다. 2차는 주문 전표(이벤트)만 띄우면, 조리(빌드)와 배달(배포)은 각 전문 파트가 가져가 처리하는 구조다. 홀(MLOps 백엔드)은 전표를 넣고 "나왔습니다(결과 이벤트)"를 받기만 한다. 각 파트는 서로를 직접 호출하지 않고 전표(Kafka 토픽)만 본다 — 한쪽이 바빠도 전표는 큐에 쌓여 기다리므로, 느슨하게 결합된다.

구체적으로 두 가지를 바꿨다.

• 오케스트레이션: Airflow DAG → Kafka 이벤트(커맨드). MLOps 백엔드가 커맨드를 발행하면, 잡 에이전트가 소비해 K8s Job을 띄운다.
• 빌드·배포: MLOps 자체 처리 → DevOps 인프라 재사용. 빌드는 DevOps Jenkins, 배포는 DevOps ArgoCD(GitOps).

3. 인프라 구성 — 누가 무엇을 하나

중심에 Kafka(이벤트 버스)가 있고, 양옆으로 MLOps 백엔드와 여러 에이전트가 토픽을 주고받는다.

구성요소	역할
MLOps 백엔드(Spring)	커맨드 발행 + 결과 수신. 메타·상태를 DB에 기록.
Kafka	이벤트 버스. 커맨드/결과 토픽을 중계.
MLOps 잡 에이전트(Python)	RUN_JOB 소비 → K8s 학습 Job 실행 → 결과 회신.
DevOps 빌드 에이전트	빌드 요청 소비 → Jenkins로 이미지 빌드 → Docker Registry push → 결과 회신.
DevOps 배포 에이전트(Python/FastAPI)	배포 요청 소비 → GitOps image tag 갱신 → ArgoCD sync → health check → 결과 회신.
ArgoCD	GitOps 레포를 감시하다 변경되면 클러스터를 그 상태로 동기화.

주고받는 주요 Kafka 토픽은 이렇다.

service.mlops.command   # MLOps 백엔드 → 잡 에이전트 (RUN_JOB 등)
service.mlops.result    # 잡 에이전트 → MLOps 백엔드 (학습 결과)
service.build.result    # DevOps 빌드 에이전트 → MLOps 백엔드 (빌드 결과)
service.deploy.result   # DevOps 배포 에이전트 → MLOps 백엔드 (배포 결과)

4. 흐름 ① 학습 실행 — Kafka RUN_JOB

스테이지(학습 등)를 실행하면, 백엔드는 Airflow를 부르지 않고 RUN_JOB 커맨드 하나를 Kafka로 던진다.

// PipelineExecSvcImpl.runStage() — RUN_JOB 커맨드 발행 (요지)
Map<String,Object> payload = new LinkedHashMap<>();
payload.put("mlopsRunId", stageRunInternalId);
payload.put("jobType", stage.getType());        // TRAIN / EVALUATE ...
payload.put("templateName", "python-ml");
payload.put("artifactRef", image);              // 실행할 잡 이미지
payload.put("command", cmdArgs[0]);
payload.put("args", cmdArgs[1]);
payload.put("namespace", "mlops-job");
payload.put("resource", resource);              // CPU/GPU/메모리
payload.put("envVars", envVars);                // MinIO/MLflow 접근정보 등

Map<String,Object> command = new LinkedHashMap<>();
command.put("commandId", commandId);
command.put("serviceId", serviceId);
command.put("type", "RUN_JOB");                 // 커맨드 타입
command.put("payload", payload);
commandPublisher.publish(String.valueOf(serviceId), command);   // serviceId를 key로 발행

발행 자체는 얇은 퍼블리셔가 담당한다. JSON으로 직렬화해 토픽에 보내는 게 전부다.

// MlopsCommandPublisher.publish()
public void publish(String key, Map<String,Object> message) {
    try {
        String json = objectMapper.writeValueAsString(message);
        kafkaTemplate.send(topic, key, json);    // topic = service.mlops.command
        log.info("[MlopsCommand] 발행 — type={}", message.get("type"));
    } catch (Exception e) {
        throw new RuntimeException("Kafka 커맨드 발행 실패", e);
    }
}

그러면 잡 에이전트가 이 커맨드를 소비해 python-ml 템플릿으로 K8s Job을 만들어 실행하고, 끝나면 service.mlops.result로 결과를 돌려보낸다. 백엔드는 그 결과를 받아 상태를 갱신한다.

// MlopsResultListener — 학습 결과 수신
@KafkaListener(topics = "${kafka.topic.mlops-result}", ...)
public void onMlopsResult(String rawJson) {
    JsonNode root = objectMapper.readTree(rawJson);
    String type = root.path("type").asText(null);
    if ("JOB_RUN_RESULT".equals(type)) {   // 스테이지 실행 결과
        handleJobRunResult(root);          // commandId로 매칭해 DB 상태 갱신
    }
    ...
}

Airflow의 스케줄러·DAG·XCom이 통째로 사라지고, "발행 → 소비 → 결과 회신"이라는 단순한 파이프만 남았다.

5. 흐름 ② 빌드 — DevOps Jenkins에 위임

이미지 빌드는 MLOps가 직접 하지 않는다. DevOps 빌드 API를 호출해 위임하고, 돌아올 결과를 매칭할 commandId만 기록해 둔다.

// PipelineBuildSvcImpl.triggerStageBuild() (요지)
String subPath = slug + "/" + stageDir;       // 예: mlsvc06023/train

DevopsPipelineTriggerReqDto req = DevopsPipelineTriggerReqDto.builder()
        .branch(branch).triggerType("MANUAL")
        .subPath(subPath).jobType(jobType).build();

Long devopsRunId = devopsApiService.triggerBuild(serviceId, req);  // DevOps에 위임
String commandId = "build." + devopsRunId;     // 결과 매칭용 키

// 빌드 스테이지를 DB에 BUILDING 으로 기록 (devopsRunId를 soft-ref)
params.put("commandId", commandId);
params.put("devopsPipelineRunId", devopsRunId);
pipelineBuildDao.regBuildStage(params);

DevOps Jenkins가 빌드를 끝내면 service.build.result로 결과가 오고, 백엔드는 commandId="build.{runId}"로 어느 스테이지인지 찾아 상태와 이미지 경로를 채운다. DevOps가 결과를 두 번 보내도 안전하게 멱등 처리한다.

// DevopsBuildResultListener (요지)
@KafkaListener(topics = "${kafka.topic.devops-build-result}", ...)
public void handleBuildResult(String rawJson) {
    String commandId = root.path("commandId").asText(null);
    long devopsRunId = Long.parseLong(commandId.substring("build.".length()));
    Map<String,Object> stageRow = pipelineBuildDao.getBuildStageByDevopsRunId(devopsRunId);
    if ("SUCCEEDED".equals(stageRow.get("status"))) return;   // 멱등 — 이미 끝난 건 skip
    String image = root.path("result").path("imageFullPath").asText(null);
    pipelineBuildDao.modBuildStageResult(devopsRunId, "SUCCEEDED", image, ...);
}

6. 흐름 ③ 배포 — DevOps GitOps + ArgoCD (이 글의 핵심)

2차에서 가장 많이 바뀐 게 배포다. 1차는 Airflow DAG가 배포까지 직접 했지만, 2차는 DevOps의 GitOps + ArgoCD 인프라 위에 얹었다. MLOps는 "배포해줘"라고 이벤트만 던지고, 실제 클러스터에 파드를 올리는 건 ArgoCD가 한다. 이 장은 그 배포가 한 단계씩 어떻게 진행되는지를, 실제 ArgoCD 설정·실제 GitOps 레포·실제 에이전트 코드로 풀어본다.

비유: GitOps는 집 안의 자동 온도조절기(보일러 thermostat)다. 우리는 벽에 붙은 패널에 "24도"라는 목표만 적어둔다(= Git 레포에 "이 서비스는 이 이미지로 떠 있어야 한다"고 적기). 그러면 조절기가 실제 방 온도를 계속 재면서, 목표보다 추우면 보일러를 켜고 더우면 끈다(= ArgoCD가 클러스터 실제 상태를 목표에 맞춰 수렴). 사람이 직접 보일러 밸브를 돌리지(=kubectl apply 치지) 않는다. 그래서 배포 = "패널의 숫자 한 줄(이미지 태그)을 바꾸는 일"이 된다.

6-1. 등장인물 셋 — deploy 레포 · 배포 에이전트 · ArgoCD

배포에는 세 주연이 있다. 이름만 먼저 익히자.

• deploy 레포(GitOps 레포) = "목표 상태가 붙는 게시판". 서비스마다 따로 하나씩 있는 Git 레포다. 여기 deployment.yaml(어떤 이미지를 몇 개 띄울지)이 들어있다. 이 게시판이 곧 "정답"이다.
• 배포 에이전트(Python) = "게시판을 고쳐 붙이고 조절기를 깨우는 사람". 배포 커맨드를 받으면 게시판의 이미지 태그를 새 버전으로 바꿔 커밋하고, ArgoCD에게 "이거 봐" 하고 알린다.
• ArgoCD = "게시판과 실제 주방을 대조하는 자동 주방장". 게시판(Git)과 실제 클러스터를 계속 비교하다, 다르면 클러스터를 게시판대로 맞춘다(sync).

핵심 사실 하나: 서비스 하나 = ArgoCD 앱 1개 + deploy 레포 1개로 1:1 대응된다. 실제 우리 클러스터에서 ArgoCD 앱을 조회하면 이렇게 묶여 있다.

# ArgoCD 앱 ↔ deploy 레포 매핑 (실제 조회 결과)
앱 이름            : ml06083-mlsvc06083-dev
└ 소스 레포(deploy): .../platoops-gitops-deploy/ml06083-mlsvc06083-deploy.git
└ 경로            : dev          # 레포 안 dev/ 폴더를 본다
└ 대상 네임스페이스 : mlsvc06083
└ 떠 있는 이미지   : 192.168.2.56:30500/ml06083-mlsvc06083/mlsvc06083:1

6-2. deploy 레포는 소스 코드와 "따로" 있다

처음 헷갈렸던 게 이거다. 서비스 코드가 있는 소스 레포와, 배포 상태가 적힌 deploy 레포는 별개다. 소스 레포에는 main.py·Dockerfile이 있고, deploy 레포에는 K8s 매니페스트만 있다. 왜 굳이 나눌까?

• "코드"와 "지금 떠 있어야 할 상태"는 다른 관심사다. 코드 커밋이 배포를 건드리지 않고, 반대로 배포(이미지 태그 변경) 이력이 소스 히스토리를 더럽히지 않는다.
• 배포 이력이 Git 커밋으로 남는다. "언제 어떤 이미지로 바꿨나"가 deploy 레포 커밋 로그에 그대로 찍힌다 — 되돌리기(rollback)도 그 커밋을 되돌리면 된다.

deploy 레포의 실제 내용은 이렇다. provisioning(서비스 생성) 단계에서 자동 생성되며, 손으로 안 만든다.

# deploy 레포: plato-admin/platoops-gitops-deploy/{프로젝트}-{서비스}-deploy
#   └ dev/deployment.yaml   ← 아래 (서빙 파드 정의)
#   └ dev/service.yaml      ← 그 아래 (클러스터 내부 접근용 Service)

# ===== dev/deployment.yaml (실제) =====
apiVersion: apps/v1
kind: Deployment
metadata:
  name: mlsvc06092
  namespace: ml06092
  labels: { app: mlsvc06092, project: ml06092, env: dev, runtime: python }
spec:
  replicas: 1                         # 파드 몇 개 띄울지
  selector:
    matchLabels: { app: mlsvc06092, env: dev }
  template:
    metadata:
      labels: { app: mlsvc06092, project: ml06092, env: dev, runtime: python }
    spec:
      containers:
      - name: mlsvc06092
        image: {레지스트리}/ml06092-mlsvc06092/mlsvc06092:latest   # ★ 이 한 줄이 배포 때 바뀐다
        imagePullPolicy: Always
        ports:
        - { name: http, containerPort: 8080 }
        livenessProbe:                # 살아있나 (죽으면 K8s가 재시작)
          httpGet: { path: /health, port: 8080 }
          initialDelaySeconds: 10
          periodSeconds: 30
        readinessProbe:               # 트래픽 받을 준비됐나 (안 되면 Service에서 제외)
          httpGet: { path: /health, port: 8080 }
          initialDelaySeconds: 15
          periodSeconds: 10
        resources:
          requests: { cpu: 256m, memory: 1Gi }   # 최소 보장
          limits:   { cpu: 1,    memory: 2Gi }   # 상한
      imagePullSecrets:
      - name: registry-pull-secret    # 사설 레지스트리 인증

# ===== dev/service.yaml (실제) =====
apiVersion: v1
kind: Service
metadata: { name: mlsvc06092, namespace: ml06092 }
spec:
  type: ClusterIP                     # 클러스터 내부에서만 접근(8080)
  ports:
  - { name: http, port: 8080, targetPort: 8080, protocol: TCP }
  selector: { app: mlsvc06092, env: dev }   # 이 라벨 가진 파드로 트래픽 전달

이미지 한 줄을 잘 보자. {레지스트리}/{프로젝트}-{서비스}/{서비스}:{태그} 형식이다. provisioning이 처음 만들 땐 태그가 :latest인데, 배포가 일어날 때마다 이 :latest(또는 이전 번호)가 새 빌드 번호(:1, :2 …)로 교체된다. 그 "교체+커밋"이 배포 에이전트의 일이고, 그걸 ArgoCD가 감지한다.

6-3. ArgoCD는 실제로 어떻게 움직이나 (선언형 동기화)

ArgoCD의 심장은 "비교 후 수렴(reconcile)" 루프다. 쉬지 않고 이 세 가지를 반복한다.

• ① 목표(desired) 읽기: deploy 레포 dev/의 매니페스트 = "이렇게 떠 있어야 한다".
• ② 실제(live) 읽기: 클러스터에 지금 실제로 뭐가 떠 있나.
• ③ 비교 → 수렴: 둘이 같으면 Synced. 다르면 OutOfSync → (자동 동기화면) 클러스터를 목표대로 고친다. 사람이 kubectl apply를 치는 그 동작을 ArgoCD가 대신한다.

그래서 배포 에이전트가 deploy 레포의 이미지 태그를 :1로 바꿔 커밋하면 → ArgoCD가 "어, 목표는 :1인데 실제는 :latest네(OutOfSync)" → 새 이미지로 파드를 갈아끼운다 → Synced. 배포가 'Git 커밋 한 번'으로 시작되는 이유가 이거다.

상태는 두 축으로 본다. 헷갈리기 쉬워서 꼭 구분하자.

• sync = 목표(Git)와 실제(클러스터)가 일치하나? (Synced / OutOfSync) — "게시판대로 됐나"
• health = 그 리소스가 정상 동작하나? (Healthy / Progressing / Degraded) — 파드가 Ready인가, 즉 "음식이 실제로 잘 나오나"

배포 성공 = Synced + Healthy 둘 다. 우리 실제 앱 한 개를 ArgoCD에서 그대로 떠 보면 이렇게 생겼다.

# ArgoCD Application: ml06083-mlsvc06083-dev  (실제 구성)
spec:
  source:                                  # ← 어디(Git)를 보나
    repoURL: .../ml06083-mlsvc06083-deploy.git
    path: dev                              #   레포 안 dev/ 폴더
    targetRevision: main                   #   main 브랜치
  destination:                             # ← 어디(클러스터)에 배포하나
    server: https://kubernetes.default.svc
    namespace: mlsvc06083
  syncPolicy:
    automated: { prune: true, selfHeal: true }   # 자동 동기화
    syncOptions: [ CreateNamespace=true ]
status:
  sync:   Synced                           # 목표(Git) == 실제(클러스터)
  health: Healthy                          # 파드 Ready, 정상 동작
  resources:                               # ArgoCD가 관리하는 실제 리소스
    - Deployment/mlsvc06083
    - Service/mlsvc06083
  history:                                 # 배포 이력 (각 리비전 = 되돌릴 수 있는 지점)
    - id=2  revision=9a70763f  deployedAt=2026-06-07T22:08:10Z

syncPolicy 세 옵션이 운영 편의의 핵심이다.

• automated: Git이 바뀌면 사람이 sync 버튼을 안 눌러도 ArgoCD가 알아서 반영.
• prune: true: Git에서 지운 리소스는 클러스터에서도 삭제(쓰레기 안 남김).
• selfHeal: true: 누가 클러스터를 손으로 바꿔놔도(kubectl edit 등) Git 상태로 되돌린다. "게시판이 항상 이긴다."
• CreateNamespace=true: 대상 네임스페이스가 없으면 만들어 준다.

비유 정리: deploy 레포 = 벽에 붙은 목표 패널, ArgoCD = 감시 카메라 달린 자동 주방장. 주방장은 패널(Git)과 주방(클러스터)을 끊임없이 대조해서, 손님(개발자)이 주방에 몰래 들어가 냄비를 바꿔놔도(selfHeal) 패널대로 되돌려 놓는다. 그래서 "지금 무엇이 떠 있어야 하는가"의 유일한 정답은 항상 Git이다.

6-4. 배포 에이전트의 6단계 (실제 코드)

그럼 MLOps가 "배포해줘" 이벤트를 던진 뒤, 배포 에이전트(DeployOrchestrator.execute())가 정확히 무슨 일을 하나. 딱 6단계다.

# deploy_orchestrator.py — 배포 6단계 (실제, 주석 보강)
async def execute(self, command_id, service_id, payload):
    gitops_repo_url = payload["gitopsRepoUrl"]   # 이 서비스의 deploy 레포 주소
    gitops_path     = payload.get("gitopsPath", ".")   # 보통 "dev"
    image_tag       = payload["imageTag"]        # 새로 띄울 빌드 번호 (예: "1")
    namespace       = payload["namespace"]       # 예: "mlsvc06083"
    app_name        = payload["appName"]         # 예: "ml06083-mlsvc06083-dev"

    # [1] ArgoCD에 deploy 레포 등록 (이미 있으면 통과) — "이 게시판 주소 기억해둬"
    await self._argocd.add_repository(gitops_repo_url, username="oauth2", password=self._gitlab_token)

    # [2] ★ deploy 레포의 이미지 태그를 새 버전으로 바꿔 커밋 (= 목표 상태 변경)
    project_id = await self._gitops.get_project_id_by_url(gitops_repo_url)
    await self._gitops.update_image_tag(project_id, "main", f"{gitops_path}/deployment.yaml", image_tag)

    # [3] ArgoCD 앱이 없으면 생성 (있으면 그대로 둠) — "이 게시판을 감시하라"
    existing = await self._argocd.get_application_status(app_name)
    if existing["syncStatus"] == "NotFound":
        await self._argocd.create_application(app_name, gitops_repo_url, gitops_path, namespace)

    # [4] Sync 트리거 — "지금 당장 게시판과 주방을 맞춰" (새 앱은 네임스페이스 생성 대기 후 재시도)
    await self._argocd.sync_application(app_name)

    # [5] Healthy + Synced 될 때까지 폴링 (최대 300초)
    monitor = await self._monitor.monitor_deployment(app_name, service_id, command_id)
    status = "FAILED" if monitor.get("error") else "SUCCESS"

    # [6] 접속 엔드포인트 추출 + 결과를 Kafka로 회신 + CP에 콜백
    endpoint = extract_endpoint(monitor["resources"])     # Ingress/NodePort에서 URL 뽑기
    await self._publisher.send_result(topic="service.deploy.result",
        command_id=command_id, service_id=service_id, status=status,
        result={ "appName": app_name, "endpoint": endpoint, "imageTag": image_tag })

이 중 [2]·[3]·[4]가 진짜 핵심이다. 하나씩 코드로 보자.

[2] 이미지 태그 교체. deploy 레포의 deployment.yaml을 GitLab API로 읽어, image: …:태그의 마지막 태그만 정규식으로 새 값으로 바꾸고 커밋한다. 이 커밋 한 번이 곧 "배포 명령"이다.

# gitops_service.py — image tag 치환 (실제, 발췌)
async def update_image_tag(self, project_id, branch, file_path, new_tag):
    enc = file_path.replace("/", "%2F")
    # 1) 현재 manifest 원문 읽기 (GitLab API)
    content = (await client.get(f"{base}/api/v4/projects/{project_id}/repository/files/{enc}/raw",
                                params={"ref": branch}, headers=self._headers)).text
    # 2) 'image: <레지스트리>/<이미지>:<태그>' 에서 ':태그' 부분만 new_tag 로 치환
    updated = re.sub(r'(image:\s*\S+:)\S+', rf'\g<1>{new_tag}', content)
    # 3) 바뀐 내용을 커밋 (← 이 커밋이 ArgoCD를 깨운다)
    await client.put(f"{base}/api/v4/projects/{project_id}/repository/files/{enc}",
        json={ "branch": branch, "content": updated,
               "commit_message": f"chore: update image tag to {new_tag}" },
        headers=self._headers)

[3] ArgoCD 앱 생성. 앱이 아직 없으면, "이 deploy 레포의 dev/ 경로를 보고 이 네임스페이스에 배포하라"는 정의를 만든다. 여기서 보내는 payload가 곧 6-3에서 본 앱 구성이다(코드 → 실제 앱이 1:1로 맞아떨어진다).

# argocd_service.py — 앱 생성 payload (실제, 발췌)
async def create_application(self, app_name, repo_url, path, namespace,
                             target_revision="main", auto_sync=True, self_heal=True):
    sync_policy = { "syncOptions": ["CreateNamespace=true"] }
    if auto_sync:
        sync_policy["automated"] = { "prune": True, "selfHeal": self_heal }  # 자동 동기화 + self-heal
    payload = {
        "metadata": { "name": app_name },
        "spec": {
            "project": "default",
            "source":      { "repoURL": repo_url, "path": path, "targetRevision": target_revision },
            "destination": { "server": "https://kubernetes.default.svc", "namespace": namespace },
            "syncPolicy":  sync_policy,
        },
    }
    await self._request("post", "/api/v1/applications", json=payload)   # ArgoCD REST API

# [4] 수동 Sync 트리거 (자동 동기화 외에 "지금 바로" 한 번 더 민다)
async def sync_application(self, app_name):
    await self._request("post", f"/api/v1/applications/{app_name}/sync", json={})

[5] 모니터링. 이제 ArgoCD에게 상태를 5초마다 물어보며, Healthy + Synced가 되면 성공, 망가지면(Degraded/Missing) 실패로 끝낸다(최대 300초).

# deploy_monitor.py — 배포 상태 폴링 (실제, 발췌)
while elapsed < timeout:                      # timeout=300초
    st = await self._argocd.get_application_status(app_name)
    health, sync = st["healthStatus"], st["syncStatus"]
    if health == "Healthy" and sync == "Synced":
        return { "healthStatus": health, "syncStatus": sync,
                 "pods": n, "readyPods": ready, "resources": resources }   # ✅ 성공
    if health in ("Degraded", "Missing", "NotFound"):
        return { "error": f"Deploy unhealthy: {health}" }                  # ❌ 실패
    if st["operationState"].get("phase") == "Failed":
        return { "error": "Sync failed: ..." }                             # ❌ Sync 자체 실패
    await asyncio.sleep(poll_interval)         # poll_interval=5초
    elapsed += poll_interval

6-5. 결과 회신 — commandId로 매칭, 멱등

에이전트는 끝나면 service.deploy.result로 결과를 회신한다(엔드포인트·상태 포함). MLOps 백엔드는 그걸 받아, 빌드와 똑같이 commandId="deploy.{runId}"로 어느 배포인지 찾아 상태를 마감한다. 결과가 두 번 와도 안전하게 멱등 처리한다.

// DevopsDeployResultListener (요지)
@KafkaListener(topics = "${kafka.topic.devops-deploy-result}", ...)
public void handleDeployResult(String rawJson) {
    String commandId = root.path("commandId").asText(null);
    if (commandId == null || !commandId.startsWith("deploy.")) return;   // MLOps 배포 아님 → 무시
    long devopsRunId = Long.parseLong(commandId.substring("deploy.".length()));

    Map<String,Object> deployRow = pipelineDeployDao.getDeployByDevopsRunId(devopsRunId);
    if (deployRow == null) return;                              // 매칭 없음 → skip
    if ("SUCCEEDED".equals(deployRow.get("status"))) return;    // 이미 끝남 → 멱등 skip

    String status = root.path("status").asText("UNKNOWN");
    String deployStatus = "SUCCESS".equals(status) ? "SUCCEEDED" : "FAILED";
    String endpoint = root.path("result").path("endpoint").asText(null);   // 서비스 접속 주소
    pipelineDeployDao.modDeployResult(devopsRunId, deployStatus, endpoint, ...);
}

정리하면 배포 한 번은 이렇게 흐른다: MLOps "배포해줘"(Kafka) → 배포 에이전트가 deploy 레포 이미지 태그 커밋(목표 변경) → ArgoCD가 감지해 클러스터를 그 상태로 수렴(Synced+Healthy) → 결과를 Kafka로 회신 → MLOps가 commandId로 매칭해 마감. MLOps는 클러스터를 직접 만지지 않고, "목표를 Git에 적게" 시킬 뿐이다.

7. 무엇을 얻었나

• 가벼워졌다. Airflow 스케줄러·웹서버·메타DB가 사라졌다. MLOps 백엔드는 이벤트를 발행·수신하는 얇은 계층이 됐다.
• 관심사가 분리됐다. 빌드는 DevOps Jenkins, 배포는 DevOps ArgoCD가 맡고, MLOps는 ML 파이프라인 로직에 집중한다. 옆 팀이 잘하는 걸 다시 만들지 않았다.
• 느슨하게 결합됐다. 서로 REST로 직접 부르는 대신 Kafka 토픽으로 비동기 통신한다. 한쪽이 느리거나 잠깐 죽어도 메시지는 큐에 남아 이어 처리된다.
• 배포가 선언형이 됐다. "지금 무엇이 떠 있어야 하는지"가 Git에 남는다. ArgoCD가 자동 동기화·self-heal 하므로, 사람이 직접 클러스터를 만질 일이 줄었다.

8. 실전에서 배운 것

• 멱등은 선택이 아니라 필수. 결과 이벤트는 재발행될 수 있어서, 모든 결과 리스너가 commandId 매칭 + "이미 끝난 건 skip"으로 멱등하게 설계했다.
• 컨슈머 오프셋은 latest로. 한때 에이전트 재시작 시 과거 커맨드를 earliest로 일괄 재소비해, 학습 이미지가 한꺼번에 당겨지며 디스크를 압박한 사고가 있었다. 이후 auto-offset-reset=latest로 바꿔 "재시작했다고 과거를 다시 실행"하지 않게 했다.
• 결과 이벤트엔 DLT보다 재시도. 결과는 멱등·재발행되므로, 깨진 메시지는 즉시 버리되 일시적 실패는 일정 간격으로 여러 번 재시도하도록 에러 핸들러를 뒀다(고정 간격 × 수 회).

9. 마무리 — 1차 vs 2차

구분	1차	2차
오케스트레이션	Airflow DAG	Kafka 이벤트(커맨드/결과)
실행 위치	Airflow KubernetesPodOperator	MLOps 잡 에이전트(K8s Job)
빌드	DAG 안에서 Jenkins 트리거	DevOps 빌드 API 위임(Jenkins)
배포	DAG 안에서 처리	DevOps 배포 에이전트 → GitOps + ArgoCD
MLOps 백엔드 역할	오케스트레이터(무거움)	이벤트 발행·수신(가벼움)

1차가 "내가 다 한다"였다면, 2차는 "이벤트로 알리고, 잘하는 쪽에 맡긴다"였다. 무거운 오케스트레이터를 걷어내고 메시지 버스와 선언형 배포(GitOps) 위에 올리니, 시스템이 한결 가볍고 손이 덜 가는 구조가 됐다. MSA에서 팀 경계를 존중하면서도 끊김 없이 잇는 방법으로, 이벤트와 GitOps는 잘 맞는 짝이었다.