Streaming Inference 운영 설치
Epic #1084 (실시간 ML 추론 Kafka × Feast × KServe) M1 PR #1106 가 추가한 (a) infra/redis-feast/ (Feast online) (b) apps/streaming/ (Bytewax workspace) (c) infra/streaming/bytewax-base/ (Deployment) 를 실 배포하는 운영 절차. Issue #1113.
본 가이드는 (1) Redis 실배포 (2) gend-streaming 이미지 ACR push (3) Bytewax Deployment 적용 (4) e2e smoke 검증.
1. Redis (Feast online) 실배포
PR #1106 가 추가한 infra/redis-feast/ (StatefulSet redis:7.2-alpine, PVC 2Gi RWO, USER 999, readOnlyRootFS).
# AKS prod
kustomize build infra/redis-feast/ | kubectl --context aks-genos-prod apply -f -
kubectl --context aks-genos-prod rollout status statefulset/redis-feast -n gend --timeout=180s
# Kind dev
kustomize build infra/redis-feast/ | kubectl --context kind-datax-local apply -f -
kubectl --context kind-datax-local rollout status statefulset/redis-feast -n gend --timeout=120s
검증:
kubectl --context aks-genos-prod exec -n gend statefulset/redis-feast -- redis-cli PING
# expected: PONG
NetworkPolicy 가 ingress 를 같은 namespace 로만 제한 — gend-streaming worker 와 Feast materialize job 만 접근 가능.
2. gend-streaming 이미지 ACR push
Issue #1113 가 추가한 .github/workflows/streaming-image.yml 가 main push 시 자동 실행.
수동 트리거:
gh workflow run streaming-image.yml --ref main -f tag_suffix=manual-$(date +%H%M)
gh run watch $(gh run list --workflow=streaming-image.yml --limit=1 --json databaseId --jq '.[0].databaseId')
검증:
az acr repository show --name genosprodacr --image gend-streaming:latest
az acr repository show-tags --name genosprodacr --repository gend-streaming --orderby time_desc | head -3
3. Bytewax Deployment 적용
PR #1106 + Issue #1113 가 함께 만든 infra/streaming/bytewax-base/ (image ACR FQDN + ConfigMap env + Prometheus annotations port 9090).
# ConfigMap env 검증 (KServe / Feast / Kafka endpoint 일치)
kubectl --context aks-genos-prod -n gend get configmap gend-streaming-config -o yaml | head -30
# 적용
kustomize build infra/streaming/bytewax-base/ | kubectl --context aks-genos-prod apply -f -
kubectl --context aks-genos-prod rollout status deploy/gend-streaming-anomaly -n gend --timeout=180s
확인:
kubectl --context aks-genos-prod -n gend get pod -l app.kubernetes.io/name=gend-streaming
kubectl --context aks-genos-prod -n gend logs deploy/gend-streaming-anomaly --tail=50
기대 로그:
[gend_streaming.bytewax_jobs.anomaly_detection] flow built; consumer_group=...
[gend_streaming.metrics] Prometheus server on :9090
4. e2e smoke
Kafka topic 에 합성 메시지 publish → Bytewax 처리 → KServe inference → sink topic 결과 확인.
전제: KServe iris-classifier 가 배포되어 있어야 함 (Issue #1111 의 e2e_kserve_iris.py 가 만든 endpoint).
# 토픽 사전 생성 (feedback_otel_kafka_prereq)
kubectl --context aks-genos-prod -n gend exec -it kafka-0 -- kafka-topics.sh \
--bootstrap-server localhost:9092 --create \
--topic source-db.public.operating_event --partitions 3 --replication-factor 3 \
--if-not-exists
kubectl --context aks-genos-prod -n gend exec -it kafka-0 -- kafka-topics.sh \
--bootstrap-server localhost:9092 --create \
--topic streaming.anomaly_result --partitions 3 --replication-factor 3 \
--if-not-exists
kubectl --context aks-genos-prod -n gend exec -it kafka-0 -- kafka-topics.sh \
--bootstrap-server localhost:9092 --create \
--topic source-db.public.operating_event.dlq --partitions 3 --replication-factor 3 \
--if-not-exists
# 합성 입력 publish (100건)
for i in $(seq 1 100); do
echo "{\"equipment_id\": \"eq-${i}\", \"vibration_rms\": $(awk -v r=$RANDOM 'BEGIN{print r/32767}'), \"temperature\": 70, \"event_timestamp\": \"$(date -u +%FT%TZ)\"}"
done | kubectl --context aks-genos-prod -n gend exec -i kafka-0 -- kafka-console-producer.sh \
--bootstrap-server localhost:9092 --topic source-db.public.operating_event
# 결과 consume (5초)
kubectl --context aks-genos-prod -n gend exec -it kafka-0 -- kafka-console-consumer.sh \
--bootstrap-server localhost:9092 --topic streaming.anomaly_result \
--from-beginning --max-messages 10 --timeout-ms 5000
기대: 5초 내 streaming.anomaly_result 토픽에 10건 이상 결과 출현. DLQ 토픽 0건 (정상 입력 기준).
5. Prometheus 메트릭 확인
kubectl --context aks-genos-prod -n gend port-forward svc/gend-streaming-anomaly 9090:9090 &
curl -s http://localhost:9090/metrics | grep -E "gend_streaming_(processed|inference_latency|dlq)" | head -10
기대 메트릭:
gend_streaming_processed_total{job="anomaly-detection",sink="kafka"}> 0gend_streaming_inference_latency_seconds_bucket{...}P99 < 0.2gend_streaming_dlq_total{...}0 (정상 입력)
6. 트러블슈팅
Bytewax Pod CrashLoopBackOff
kubectl logs deploy/gend-streaming-anomaly --previous --tail=100
원인 후보:
- ConfigMap env 누락 —
GEND_STREAMING_KSERVE_ENDPOINT,GEND_STREAMING_REDIS_HOST등 - Redis NetworkPolicy 차단 —
kubectl exec -- redis-cli -h redis-feast.gend.svc -p 6379 PING - KServe endpoint 미배포 — Issue #1111 e2e 먼저 실행
- (해소됨) M2 partial 인시던트: 빈
IterableSource([])즉시 exit 0 → CrashLoopBackOff 5회 → M3 (#1084) 에서_build_default_source가KafkaSource(tail=True)또는KeepAliveSource를 반환하도록 수정. 재발 시pytest tests/test_continuous_source.py회귀 가드 먼저 확인.
DLQ 누적률 > 1%
kubectl exec kafka-0 -- kafka-console-consumer.sh \
--bootstrap-server localhost:9092 --topic source-db.public.operating_event.dlq \
--from-beginning --max-messages 5
error_type 필드로 원인 분류:
feast_lookup_miss— feature view 가 비어있음, Feast materialize 필요kserve_5xx— 모델 endpoint OOM 또는 cold startparse_error— Kafka 메시지 schema 변경
처리 lag 폭증 (Kafka consumer lag > 10000)
M1 은 replicas=1 고정 — 일시적 수동 scale:
kubectl scale deploy/gend-streaming-anomaly --replicas=3
M2 의 KEDA ScaledObject 가 자동화 (별도 PR).
7. Continuous-source 동작 (M3 #1084)
M2 partial PR #1278 머지 후 AKS prod 에서 gend-streaming-anomaly Pod 가 즉시 exit 0 → 5회 restart → CrashLoopBackOff 가 발견됐다. 원인은 _build_default_source 가 빈 IterableSource([]) 를 반환해 Bytewax 가 source 완료로 인식한 점.
M3 (#1084 후속) 수정:
- continuous source 계약:
GEND_STREAMING_KAFKA_BOOTSTRAP가 실 broker 면bytewax.connectors.kafka.KafkaSource(brokers, [topic], tail=True)를 사용.tail=True가 partition-end 에서 StopIteration 대신 새 메시지를 영구 대기함. - keepalive 폴백: bootstrap 이 빈 문자열 또는
"keepalive"면KeepAliveSource가 빈 batch 만 영원히 반환 (worker 가 살아있어 /metrics + probe 통과). 운영자는 Kafka 미연동 단계에서도 Pod 상태로 헬스 확인 가능. memory별칭 유지: in-process 단위 테스트의Settings(kafka_bootstrap="memory")는 기존IterableSource유지 (테스트 호환).- SIGTERM 처리: 새 entry point
python -m gend_streaming가 SIGTERM/SIGINT 핸들러 등록 → shutdown event flip →signal.raise_signal(SIGINT)로 Bytewax 런타임에 전파 →KeyboardInterrupt캡처 후 producer flush. K8sterminationGracePeriodSeconds=30s안에서graceful_shutdown_timeout_s=25s로 안전 마진 확보. - preStop hook: deployment.yaml 의
lifecycle.preStop가 SIGTERM 전 2초 sleep 으로 in-flight Kafka poll 마감 시간 확보.
ConfigMap 신규 키 (기본값 그대로 운영):
GEND_STREAMING_KAFKA_CONSUMER_TIMEOUT_MS: "0" # 0=block forever (continuous contract)
GEND_STREAMING_GRACEFUL_SHUTDOWN_TIMEOUT_S: "25"
회귀 가드: apps/streaming/tests/test_continuous_source.py (13 tests + 1 skip when confluent_kafka missing). 향후 source 기본값 변경 시 이 테스트가 먼저 실패해야 함.
8. 다음 단계 (M3 후속 / M4)
- M3 후속: KEDA ScaledObject (Kafka lag trigger) + Redis Sentinel 3-replica +
/api/v1/streaming/jobsAPI/UI + lineage 라우터 확장 (kafka_topic/kserve_endpoint/streaming_job 노드) — M2 본 PR #1241 / #1263 일부 진행 - M4: Flink runtime + gRPC + exactly-once + GPU 노드 핀