Streaming Inference API Reference

Epic #1084 M2 가 추가한 Streaming Inference 관리 API. M1 (PR #1106) 의 Bytewax 워커 + Feast online + KServe HTTP 파이프라인 위에서 선언적 잡 정의 / 일시정지 / 트리거 를 REST 로 노출한다.

7 endpoint 모두 /api/v1/streaming/jobs 하위. 실 K8s apps/v1 Deployment scale + KEDA ScaledObject 자동 생성은 M3 로 위임 (본 PR 은 row insert + audit emit 까지).

인증 / 권한

• Read (GET): JWT 인증된 모든 역할 (admin / analyst / viewer).
• Mutation (POST / PUT / DELETE): admin 만 — require_admin 데코레이터.
• Workspace fence: tenant_slug JWT claim 기반 — 자기 워크스페이스 + legacy NULL 만 노출. tenant_slug 가 어떤 Workspace.slug 에도 매칭되지 않으면 fail-closed 403.

1. GET /api/v1/streaming/jobs — 리스트

쿼리 파라미터:

name	type	default	range
limit	int	50	1..200
offset	int	0	≥ 0

응답: 200 OK, StreamingJobListResponse (items + total).

curl -H "Authorization: Bearer $JWT" \
  "https://gend.genon.ai/api/v1/streaming/jobs?limit=50"

2. GET /api/v1/streaming/jobs/{id} — 상세

응답:

• 200 OK + StreamingJobRead — 정상.
• 404 Not Found — UUID 미존재.
• 403 Forbidden — 다른 워크스페이스 row.
• 422 Unprocessable Entity — UUID 포맷 오류.

curl -H "Authorization: Bearer $JWT" \
  "https://gend.genon.ai/api/v1/streaming/jobs/<uuid>"

3. GET /api/v1/streaming/jobs/{id}/runs — 잡 별 실행 이력

특정 잡의 streaming_job_run 리스트. created_at DESC 정렬.

쿼리 파라미터:

name	type	default	range
limit	int	50	1..200
offset	int	0	≥ 0

응답: 200 OK, StreamingJobRunListResponse. 잡 미존재 시 404, 다른 워크스페이스 시 403.

4. POST /api/v1/streaming/jobs — 신규 잡 정의 (admin)

요청 본문 (StreamingJobCreate):

필드	타입	필수	비고
name	str (1..128)	✓	UNIQUE(workspace_id, name)
description	str
engine	"bytewax"		M2 는 bytewax 만 (flink M3)
source_topic	str (regex [a-zA-Z0-9._-]+)	✓	Kafka 입력 토픽
sink_topic	str	✓	성공 예측 출력 토픽
dlq_topic	str	✓	DLQ 토픽 (사전 생성 필수)
model_name	str (1..255)	✓	MLflow registered model
model_alias	str		기본 "Production"
consumer_group	str	✓	Kafka offset 커서 식별자
bytewax_image	str		ACR / gend/ allowlist (default genosprodacr.azurecr.io/gend-streaming:latest)

응답:

• 201 Created + StreamingJobRead — 정상.
• 409 Conflict — (workspace_id, name) 중복.
• 403 Forbidden — admin 아님.
• 422 Unprocessable Entity — 필드 검증 실패 (토픽 포맷, 이미지 registry, engine, 등).

workspace_id / created_by 는 서버가 caller 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

5. PUT /api/v1/streaming/jobs/{id} — 수정 (admin)

수정 가능한 필드만 (StreamingJobUpdate, extra='forbid'):

필드	타입	비고
description	str | null
model_alias	str	새 alias — 동작 중인 Pod 는 재시작까지 기존 alias 유지
bytewax_image	str	ACR / gend/ allowlist 재검증
status	"active" | "paused"	"deleted" 는 DELETE 전용

불변 필드 (name / workspace_id / engine / source_topic / sink_topic / dlq_topic / model_name / consumer_group / created_by) 는 422 Unprocessable Entity 로 거부 — 이 중 하나라도 바꿀 필요가 있다면 새 잡을 POST 로 생성하라 (audit trail 보존).

응답:

• 200 OK + StreamingJobRead.
• 404 Not Found / 403 Forbidden / 422 Unprocessable Entity.

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

6. DELETE /api/v1/streaming/jobs/{id} — 소프트 삭제 (admin)

status='deleted' 로 플립. 실행 이력 (streaming_job_run) 은 FK CASCADE 정책으로 보존 (PG 기준 — 본 PR 은 status flip 만).

응답:

• 204 No Content — 정상.
• 404 Not Found.

curl -X DELETE -H "Authorization: Bearer $JWT" \
  https://gend.genon.ai/api/v1/streaming/jobs/<uuid>

M3 에서 deleted_at / deleted_by 컬럼 추가 + 진짜 tombstone 으로 전환.

7. POST /api/v1/streaming/jobs/{id}/trigger — 실행 트리거 (admin)

요청 본문 (StreamingJobTriggerRequest):

필드	타입	default	range
replicas	int	1	1..16

응답:

• 201 Created + StreamingJobRunRead (status="queued").
• 404 Not Found — 잡 미존재.
• 409 Conflict — 잡이 status='deleted'.
• 422 Unprocessable Entity — replicas 범위 초과.

M2 동작: streaming_job_run row 1건 (status='queued') 만 insert. 실제 K8s Deployment scale + KEDA ScaledObject 자동 생성은 M3 — 운영자는 ops 대시보드 / sensor 로 queued row 폴링.

curl -X POST -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -d '{"replicas": 2}' \
  https://gend.genon.ai/api/v1/streaming/jobs/<uuid>/trigger

Audit emit

모든 mutation 은 gend.audit 로거 채널에 구조화 로그 한 줄을 emit:

streaming.job.create job_id=... workspace=... name=... engine=bytewax ...
streaming.job.update job_id=... changed=[...] previous={...} user=...
streaming.job.delete job_id=... previous_status=active deleted_by=...
streaming.job.trigger job_id=... run_id=... replicas=2 triggered_by=...

M3 에서 audit_chain HMAC 파이프라인 통합 + OpenSearch 인덱싱.

에러 응답 (RFC 7807)

{
  "type": "about:blank",
  "title": "Conflict",
  "status": 409,
  "detail": "Streaming job already exists for this (workspace, name) pair"
}

후속 (M3)

• 실 K8s apps/v1 Deployment scale (replicas + image 동기화).
• KEDA ScaledObject 자동 생성 (Kafka consumer_group lag 기반).
• Redis Sentinel 3-replica.
• lineage 라우터 확장 (kafka_topic / kserve_endpoint / streaming_job 노드).
• UI /admin/streaming 페이지.
• Flink engine 옵션.