11장: 프로덕션 모니터링과 운영

[즉시 대응]
- OfflinePartitionsCount > 0 → 파티션에 접근 불가, 데이터 유실 위험
- ActiveControllerCount != 1 → 컨트롤러 장애, 클러스터 관리 불가
 
[주의 관찰]
- UnderReplicatedPartitions > 0 → 레플리케이션 지연, 내구성 저하
- RequestHandlerAvgIdlePercent < 30% → 브로커 과부하
 
[추세 분석]
- BytesInPerSec 증가 추세 → 용량 계획 필요
- LogFlushLatency 점진적 증가 → 디스크 교체 검토

Flink Dashboard (기본: http://jobmanager:8081)
 
├── Overview: 실행 중인 작업, 슬롯 현황
├── Running Jobs
│   ├── Job Graph: DAG 시각화, 오퍼레이터별 처리량
│   ├── Checkpoints: 체크포인트 이력, 크기, 소요 시간
│   ├── Backpressure: 오퍼레이터별 백프레셔 수준
│   └── Metrics: 상세 메트릭 조회
└── Task Managers: 리소스 사용량

lowercaseOutputName: true
lowercaseOutputLabelNames: true
 
rules:
  # 브로커 토픽 메트릭
  - pattern: kafka.server<type=BrokerTopicMetrics, name=(.+), topic=(.+)><>Count
    name: kafka_server_brokertopicmetrics_$1_total
    labels:
      topic: $2
 
  # 파티션 관련 메트릭
  - pattern: kafka.server<type=ReplicaManager, name=(.+)><>Value
    name: kafka_server_replicamanager_$1
 
  # 요청 처리 시간
  - pattern: kafka.network<type=RequestMetrics, name=(.+), request=(.+)><>Count
    name: kafka_network_requestmetrics_$1_total
    labels:
      request: $2
 
  # 컨슈머 그룹 메트릭
  - pattern: kafka.server<type=FetcherLagMetrics, name=(.+), clientId=(.+), topic=(.+), partition=(.+)><>Value
    name: kafka_server_fetcherlagmetrics_$1
    labels:
      client_id: $2
      topic: $3
      partition: $4

# flink-conf.yaml
metrics.reporter.prom.factory.class: org.apache.flink.metrics.prometheus.PrometheusReporterFactory
metrics.reporter.prom.port: 9249

[Kafka 대시보드]
Row 1: 브로커 현황
  - 활성 브로커 수 | UnderReplicated 파티션 | 컨트롤러 상태
 
Row 2: 처리량
  - Messages In/sec (토픽별) | Messages Out/sec (토픽별)
  - Bytes In/sec | Bytes Out/sec
 
Row 3: 지연 시간
  - Produce 요청 P99 지연 | Fetch 요청 P99 지연
  - LogFlush 지연
 
Row 4: 리소스
  - 디스크 사용량 (브로커별) | 네트워크 처리량 | CPU 사용률

[Flink 대시보드]
Row 1: 작업 현황
  - 작업 상태 | 업타임 | 재시작 횟수
 
Row 2: 처리량
  - 오퍼레이터별 Records In/Out
  - 종단간 지연 (이벤트 시간 - 처리 시간)
 
Row 3: 체크포인트
  - 체크포인트 소요 시간 | 체크포인트 크기 | 실패 횟수
 
Row 4: 백프레셔
  - 오퍼레이터별 백프레셔 비율

groups:
  - name: kafka-critical
    rules:
      # P1: 즉시 대응
      - alert: KafkaOfflinePartitions
        expr: kafka_server_replicamanager_offlinepartitionscount > 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "오프라인 파티션 감지"
          description: "{{ $value }}개 파티션에 리더가 없습니다. 데이터 접근 불가."
          runbook: "https://wiki.internal/runbook/kafka-offline-partitions"
 
      - alert: KafkaConsumerLagCritical
        expr: sum(kafka_consumer_lag) by (consumer_group) > 100000
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "컨슈머 랙 임계치 초과"
          description: "컨슈머 그룹 {{ $labels.consumer_group }}의 랙이 {{ $value }}입니다."
 
  - name: kafka-warning
    rules:
      # P2: 업무 시간 대응
      - alert: KafkaUnderReplicatedPartitions
        expr: kafka_server_replicamanager_underreplicatedpartitions > 0
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "언더레플리케이션 파티션 감지"
          description: "{{ $value }}개 파티션의 레플리케이션이 불완전합니다."
 
      - alert: KafkaDiskUsageHigh
        expr: kafka_log_size_bytes / kafka_log_max_size_bytes > 0.8
        for: 30m
        labels:
          severity: warning
        annotations:
          summary: "디스크 사용량 80% 초과"
 
  - name: flink-critical
    rules:
      - alert: FlinkCheckpointFailing
        expr: flink_jobmanager_job_numberOfFailedCheckpoints > 3
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Flink 체크포인트 연속 실패"
          description: "체크포인트가 연속 실패하고 있습니다. 장애 복구가 불가능해질 수 있습니다."
 
      - alert: FlinkJobNotRunning
        expr: flink_jobmanager_job_uptime == 0
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "Flink 작업 중단"

[입력 데이터]
- 일일 메시지 수: 1억 건
- 평균 메시지 크기: 1KB
- 보존 기간: 7일
- 레플리카 팩터: 3
 
[산출]
- 일일 데이터량: 100M x 1KB = 100GB
- 7일 보존: 100GB x 7 = 700GB
- 레플리카: 700GB x 3 = 2.1TB
- 안전 마진 (30%): 2.1TB x 1.3 = 2.73TB
 
→ 브로커 3대, 각 1TB 디스크면 여유 있음
→ 성장률 고려 시 각 2TB 권장

[입력 데이터]
- 초당 메시지: 10,000 msg/sec
- 상태 크기: 윈도우 5분, 키 100만 개
- 체크포인트 간격: 30초
 
[산출]
- TaskManager 당 처리 능력: ~3,000 msg/sec (경험치, 로직 복잡도에 따라 상이)
- 필요 병렬도: 10,000 / 3,000 = 4 (여유 포함 6)
- 상태 메모리: 키 100만 x 상태 크기 100B = 100MB (RocksDB 사용 시 디스크)
- TaskManager 메모리: 4GB~8GB (상태 + 네트워크 버퍼 + 프레임워크 오버헤드)
 
→ TaskManager 3대, 각 2슬롯 (총 6 병렬도)
→ 메모리: 각 8GB

# 토픽별 보존 기간을 용도에 맞게 차등 설정
kafka-configs.sh --alter --topic hot-data \
  --add-config retention.ms=86400000    # 1일 (실시간 처리용)
 
kafka-configs.sh --alter --topic warm-data \
  --add-config retention.ms=604800000   # 7일 (단기 분석용)
 
kafka-configs.sh --alter --topic audit-log \
  --add-config retention.ms=-1          # 무제한 (감사 로그)

# 토픽 수준 압축 설정
kafka-configs.sh --alter --topic orders \
  --add-config compression.type=zstd
 
# Compacted 토픽: 키별 최신 값만 유지
kafka-configs.sh --alter --topic user-profiles \
  --add-config cleanup.policy=compact

StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.hours(1))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .cleanupFullSnapshot()  // 체크포인트 시 만료 상태 정리
    .build();
 
ValueStateDescriptor<UserSession> descriptor =
    new ValueStateDescriptor<>("session", UserSession.class);
descriptor.enableTimeToLive(ttlConfig);