Streaming Inference 잡 관리

Epic #1084 M2 가 추가한 선언적 스트리밍 잡 관리 운영 가이드. M1 (PR #1106) 에서 인프라 / Bytewax 워커 / Feast online / KServe HTTP 클라이언트 위에서 동작.

본 문서는 (1) 데이터 모델 (2) 잡 라이프사이클 (3) 운영자 시나리오 (4) M3 위임 범위 를 다룬다. API 호출법은 API Reference 참조.

1. 데이터 모델

streaming_job (선언적 정의) + streaming_job_run (per-execution 이력) 두 테이블만.

streaming_job

컬럼	타입	비고
id	UUID PK
workspace_id	UUID FK → workspaces	NULLABLE (M2 backfill 윈도우)
name	VARCHAR(128)	UNIQUE(workspace_id, name)
description	TEXT
engine	VARCHAR(16)	CHECK ('bytewax', 'flink'), default 'bytewax'
source_topic	VARCHAR(255)	Kafka 입력 토픽
sink_topic	VARCHAR(255)	성공 예측 출력 토픽
dlq_topic	VARCHAR(255)	DLQ 토픽 (사전 생성 필수)
model_name	VARCHAR(255)	MLflow registered model 명
model_alias	VARCHAR(64)	default 'Production'
consumer_group	VARCHAR(255)	Kafka offset 커서
bytewax_image	VARCHAR(512)	default genosprodacr.azurecr.io/gend-streaming:latest
status	VARCHAR(32)	CHECK ('active','paused','deleted'), default 'active'
created_by	UUID	Keycloak sub
created_at / updated_at	TIMESTAMPTZ

streaming_job_run

컬럼	타입	비고
id	UUID PK
job_id	UUID FK → streaming_job CASCADE
k8s_deployment_name	VARCHAR(255)	NULL (M3 reconciler 가 채움)
replicas	INT	트리거 시 요청 값
status	VARCHAR(32)	CHECK ('queued','running','succeeded','failed','cancelled')
started_at / finished_at	TIMESTAMPTZ	NULL until M3 reconciler
processed_count / dlq_count	BIGINT	default 0
consumer_lag	BIGINT	NULL until M3 Prometheus poll
error_message	TEXT
created_at	TIMESTAMPTZ

인덱스: idx_streaming_job_run_job_started (job_id, started_at) — "이 잡의 최근 실행" 핫패스.

2. 잡 라이프사이클

• active → paused → active: 일시정지 + 재시작 (PUT status). M3 reconciler 가 Deployment replicas=0 로 스케일 다운.
• active / paused → deleted: 소프트 삭제 (DELETE). 새 트리거는 409. 실행 이력은 보존 (FK CASCADE 는 M3 의 진짜 hard delete 에서 적용).
• trigger → queued: row insert 만. M3 가 K8s Deployment scale + KEDA ScaledObject 생성.

3. 운영자 시나리오

시나리오 A — 새 잡 등록

# 1. 사전: Kafka 토픽 3종 (source / sink / DLQ) 생성. feedback_otel_kafka_prereq 참조.
kafka-topics --create --topic events.fraud --bootstrap-server kafka:9092 ...
kafka-topics --create --topic predictions.fraud ...
kafka-topics --create --topic dlq.fraud ...

# 2. POST 로 잡 정의 (admin JWT 필요).
curl -X POST -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "anomaly-fraud",
    "source_topic": "events.fraud",
    "sink_topic": "predictions.fraud",
    "dlq_topic": "dlq.fraud",
    "model_name": "fraud_detection",
    "consumer_group": "gend-streaming-fraud"
  }' \
  https://gend.genon.ai/api/v1/streaming/jobs

# 3. 트리거 — M2 는 queued row 만 추가.
curl -X POST -H "Authorization: Bearer $JWT" \
  -d '{"replicas": 2}' \
  https://gend.genon.ai/api/v1/streaming/jobs/<uuid>/trigger

# 4. M2 운영자 브릿지 — queued row 확인 후 수동 K8s scale.
kubectl --context aks-genos-prod -n gend scale deployment/gend-streaming-anomaly --replicas=2

시나리오 B — 잡 일시정지

curl -X PUT -H "Authorization: Bearer $JWT" \
  -d '{"status": "paused"}' \
  https://gend.genon.ai/api/v1/streaming/jobs/<uuid>

# M2 운영자 브릿지: Deployment scale to 0
kubectl --context aks-genos-prod -n gend scale deployment/gend-streaming-anomaly --replicas=0

시나리오 C — 모델 업그레이드 (alias 전환)

# 새 모델 버전을 Production alias 로 승격 후
curl -X PUT -H "Authorization: Bearer $JWT" \
  -d '{"model_alias": "Production"}' \
  https://gend.genon.ai/api/v1/streaming/jobs/<uuid>

# 동작 중인 Pod 는 부팅 시 resolve 한 version 유지 — 재시작 필요.
kubectl --context aks-genos-prod -n gend rollout restart deployment/gend-streaming-anomaly

model_name 변경은 새 잡 등록이 권장 (PUT 에서 422 거부) — audit trail 분리 + DLQ 격리 보존.

시나리오 D — 워커 이미지 업그레이드

# 새 이미지 push (ACR)
docker tag gend/gend-streaming:0.2.0 genosprodacr.azurecr.io/gend-streaming:0.2.0
docker push genosprodacr.azurecr.io/gend-streaming:0.2.0

# 잡 정의 갱신
curl -X PUT -H "Authorization: Bearer $JWT" \
  -d '{"bytewax_image": "genosprodacr.azurecr.io/gend-streaming:0.2.0"}' \
  https://gend.genon.ai/api/v1/streaming/jobs/<uuid>

# M2 운영자 브릿지: Deployment image 동기 + rollout
kubectl --context aks-genos-prod -n gend set image \
  deployment/gend-streaming-anomaly streaming-worker=genosprodacr.azurecr.io/gend-streaming:0.2.0

이미지는 ACR (genosprodacr.azurecr.io/) 또는 Kind 개발용 gend/ prefix 만 허용 — Docker Hub 등 외부 registry 는 422 로 거부 (supply-chain guard, tests/test_no_bytewax_image_hardcoded.py).

4. Audit 추적

모든 mutation 은 gend.audit 로거에 구조화 한 줄 emit. AKS prod 에서는 fluent-bit 가 stdout 캡처 후 OpenSearch 인덱싱.

# Pod 로그에서 streaming.job.* 만 추출
kubectl --context aks-genos-prod -n gend logs deployment/gend-api --tail=1000 \
  | grep "streaming\\.job\\."

# OpenSearch (M3)
GET /gend-audit-*/_search?q=action:streaming.job.*

audit_chain HMAC 통합 + 자동 검증 CronJob 은 M3 — 현재는 stdout / OpenSearch 단방향만.

5. 워크스페이스 격리 (Epic #1018)

• workspace_id 는 L0 tenant boundary (FK → workspaces.id).
• 읽기 (GET): 자기 워크스페이스 + legacy NULL 만 노출.
• 쓰기 (POST): admin-with-tenant_slug 는 그 워크스페이스 stamp; admin-without-tenant_slug 는 NULL stamp (M2 backfill posture).
• tenant_slug 가 JWT 에 있으나 매칭되는 Workspace.slug 가 없으면 fail-closed 403 — 묵시적 fall-through 금지 (PR #1095 review pattern).

UNIQUE 제약은 (workspace_id, name) 복합 — 같은 이름이라도 워크스페이스가 다르면 충돌 없음. PG / SQLite 모두 NULL 을 distinct 로 취급하므로 워크스페이스 stamp 가 없는 admin 끼리는 같은 이름으로 여러 row 생성 가능 (M3 backfill NOT NULL 전환 시 해소).

6. M3 위임 범위

본 PR (M2) 은 API + DB + audit 까지. 다음 항목은 M3 로 위임:

• 실 K8s apps/v1 Deployment scale 자동화 (M2 운영자가 수동으로 kubectl scale).
• KEDA ScaledObject 자동 생성 (Kafka consumer_group lag 기반 autoscale).
• Redis Sentinel 3-replica (M1 은 단일 replica).
• lineage 라우터 확장 — kafka_topic / kserve_endpoint / streaming_job 노드 추가.
• UI /admin/streaming 페이지 (현재 API 전용).
• deleted_at / deleted_by 컬럼 + 진짜 tombstone.
• Flink engine 옵션 (engine='flink' 가 wire layer 에서 거부됨 — Literal "bytewax" 만).
• gRPC KServe protocol.
• exactly-once (Flink 2PC + Kafka EOS v2).
• GPU node pinning.

7. 참고

• M1 PoC: PR #1106 — Bytewax + Feast + KServe HTTP.
• M1 운영 설치: Streaming Inference 운영 설치.
• M2 API: Streaming Inference API Reference.
• 아키텍처 개요: Streaming Overview.