9장: 프로덕션 스트리밍 인프라

upstream sse_backend {
    server backend1:3000;
    server backend2:3000;
    server backend3:3000;
}
 
server {
    listen 443 ssl;
    
    location /api/stream {
        proxy_pass http://sse_backend;
        
        # SSE 핵심 설정
        proxy_http_version 1.1;
        proxy_set_header Connection "";
        
        # 버퍼링 비활성화 — 스트리밍의 핵심
        proxy_buffering off;
        proxy_cache off;
        
        # 청크 전송 인코딩 비활성화
        chunked_transfer_encoding off;
        
        # 연결 타임아웃 (SSE는 장시간 연결)
        proxy_read_timeout 3600s;
        proxy_send_timeout 3600s;
        
        # 헤더 전달
        proxy_set_header X-Accel-Buffering no;
    }
}

upstream ws_backend {
    # 스티키 세션: 같은 클라이언트를 같은 서버로
    ip_hash;
    
    server backend1:8080;
    server backend2:8080;
    server backend3:8080;
}
 
map $http_upgrade $connection_upgrade {
    default upgrade;
    ""      close;
}
 
server {
    listen 443 ssl;
    
    location /ws {
        proxy_pass http://ws_backend;
        proxy_http_version 1.1;
        
        # WebSocket 업그레이드 핵심 헤더
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection $connection_upgrade;
        
        # 유휴 연결 타임아웃
        proxy_read_timeout 86400s;  # 24시간
        proxy_send_timeout 86400s;
        
        # 원본 IP 전달
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    }
}

# CloudFormation 또는 CDK 설정 예시
TargetGroup:
  Type: AWS::ElasticLoadBalancingV2::TargetGroup
  Properties:
    Protocol: HTTP
    Port: 8080
    TargetType: ip
    HealthCheckPath: /health
    # WebSocket 연결 유지를 위한 설정
    TargetGroupAttributes:
      - Key: deregistration_delay.timeout_seconds
        Value: "300"  # 연결 드레이닝 대기 시간
      - Key: stickiness.enabled
        Value: "true"
      - Key: stickiness.type
        Value: "lb_cookie"
      - Key: stickiness.lb_cookie.duration_seconds
        Value: "86400"  # 24시간 스티키 세션

[정적 자산]
클라이언트 ──> CDN ──> 오리진 서버
 
[스트리밍 API]
클라이언트 ──────────> 스트리밍 서버 (CDN 우회)

apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: streaming-ingress
  annotations:
    # Nginx Ingress Controller 설정
    nginx.ingress.kubernetes.io/proxy-read-timeout: "3600"
    nginx.ingress.kubernetes.io/proxy-send-timeout: "3600"
    nginx.ingress.kubernetes.io/proxy-buffering: "off"
    # WebSocket 지원
    nginx.ingress.kubernetes.io/websocket-services: "ws-service"
    # 스티키 세션
    nginx.ingress.kubernetes.io/affinity: "cookie"
    nginx.ingress.kubernetes.io/session-cookie-max-age: "86400"
spec:
  rules:
    - host: api.example.com
      http:
        paths:
          - path: /api/stream
            pathType: Prefix
            backend:
              service:
                name: sse-service
                port:
                  number: 3000
          - path: /ws
            pathType: Prefix
            backend:
              service:
                name: ws-service
                port:
                  number: 8080

apiVersion: apps/v1
kind: Deployment
metadata:
  name: streaming-service
spec:
  replicas: 3
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0  # 제로 다운타임
  template:
    spec:
      terminationGracePeriodSeconds: 300  # 연결 드레이닝 대기
      containers:
        - name: streaming
          image: streaming-service:latest
          ports:
            - containerPort: 3000
          resources:
            requests:
              memory: "256Mi"
              cpu: "250m"
            limits:
              memory: "512Mi"
              cpu: "500m"
          # 연결 수 기반 스케일링을 위한 메트릭 노출
          env:
            - name: METRICS_PORT
              value: "9090"
          # 장시간 연결을 유지하는 서비스이므로
          # liveness 프로브를 넉넉하게 설정
          livenessProbe:
            httpGet:
              path: /health
              port: 3000
            initialDelaySeconds: 10
            periodSeconds: 30
            failureThreshold: 5
          readinessProbe:
            httpGet:
              path: /ready
              port: 3000
            initialDelaySeconds: 5
            periodSeconds: 10
          # PreStop hook으로 그레이스풀 셧다운
          lifecycle:
            preStop:
              exec:
                command:
                  - /bin/sh
                  - -c
                  - "curl -X POST localhost:3000/admin/drain && sleep 120"

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: streaming-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: streaming-service
  minReplicas: 2
  maxReplicas: 20
  metrics:
    # 커스텀 메트릭: 활성 연결 수
    - type: Pods
      pods:
        metric:
          name: active_connections
        target:
          type: AverageValue
          averageValue: "500"  # Pod당 평균 500 연결
    # CPU도 보조 지표로
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 70
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 60
      policies:
        - type: Pods
          value: 2
          periodSeconds: 60
    scaleDown:
      stabilizationWindowSeconds: 300  # 스케일 다운은 보수적으로
      policies:
        - type: Pods
          value: 1
          periodSeconds: 120

import { Counter, Histogram, Gauge } from "prom-client";
 
// Prometheus 메트릭 정의
const activeConnections = new Gauge({
  name: "streaming_active_connections",
  help: "Number of active streaming connections",
  labelNames: ["protocol", "endpoint"],
});
 
const ttft = new Histogram({
  name: "streaming_ttft_seconds",
  help: "Time to first token in seconds",
  labelNames: ["model"],
  buckets: [0.1, 0.25, 0.5, 1, 2, 5, 10],
});
 
const tpot = new Histogram({
  name: "streaming_tpot_milliseconds",
  help: "Time per output token in milliseconds",
  labelNames: ["model"],
  buckets: [10, 20, 30, 50, 75, 100, 200],
});
 
const streamCompletions = new Counter({
  name: "streaming_completions_total",
  help: "Total streaming completions",
  labelNames: ["status"], // "success", "cancelled", "error"
});
 
// 사용 예시
function trackStreaming(model: string) {
  const startTime = Date.now();
  let firstTokenTime: number | null = null;
  let lastTokenTime = startTime;
 
  return {
    onFirstToken() {
      firstTokenTime = Date.now();
      ttft.labels(model).observe(
        (firstTokenTime - startTime) / 1000
      );
    },
 
    onToken() {
      const now = Date.now();
      if (firstTokenTime) {
        tpot.labels(model).observe(now - lastTokenTime);
      }
      lastTokenTime = now;
    },
 
    onComplete(status: "success" | "cancelled" | "error") {
      streamCompletions.labels(status).inc();
    },
  };
}

[HTTP/2 (TCP)]
스트림 A: 데이터 ──> 데이터 ──> [손실] ──> 재전송 대기
스트림 B: 데이터 ──> 데이터 ──> [블로킹] ──> 대기  ← 영향 받음
스트림 C: 데이터 ──> 데이터 ──> [블로킹] ──> 대기  ← 영향 받음
 
[HTTP/3 (QUIC)]
스트림 A: 데이터 ──> 데이터 ──> [손실] ──> 재전송 대기
스트림 B: 데이터 ──> 데이터 ──> 데이터 ──> 정상  ← 영향 없음
스트림 C: 데이터 ──> 데이터 ──> 데이터 ──> 정상  ← 영향 없음

// WebTransport 클라이언트 (현재 Chrome/Edge 지원)
const transport = new WebTransport(
  "https://api.example.com/webtransport"
);
 
await transport.ready;
 
// 양방향 스트림 열기
const stream = await transport.createBidirectionalStream();
const writer = stream.writable.getWriter();
const reader = stream.readable.getReader();
 
// 요청 전송
await writer.write(
  new TextEncoder().encode(
    JSON.stringify({ message: "안녕하세요" })
  )
);
 
// 응답 스트리밍 수신
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
 
  const text = new TextDecoder().decode(value);
  console.log("수신:", text);
}