利用 VictoriaMetrics + VictoriaLogs + VictoriaTraces 收集與整理 AWS MSK 全棧實戰指南 ¶

# 01-msk-broker-scrape.yaml
apiVersion: operator.victoriametrics.com/v1beta1
kind: VMStaticScrape
metadata:
  name: aws-msk-production-brokers
  namespace: monitoring
spec:
  jobName: "aws-msk-brokers"
  targetEndpoints:
    # 1. 抓取 MSK Broker JMX 指標 (Throughput, UnderReplicatedPartitions, Controller, JVM)
    - targets:
        - "b-1.prod-msk.xxxxxx.c2.kafka.ap-northeast-1.amazonaws.com:11001"
        - "b-2.prod-msk.xxxxxx.c2.kafka.ap-northeast-1.amazonaws.com:11001"
        - "b-3.prod-msk.xxxxxx.c2.kafka.ap-northeast-1.amazonaws.com:11001"
      interval: "30s"
      scrapeTimeout: "25s"
      labels:
        msk_cluster: "prod-msk-tokyo"
        exporter_type: "jmx"
      relabelConfigs:
        # 自動從目標域名擷取 Broker ID (如 b-1, b-2, b-3) 作為標籤
        - sourceLabels: [__address__]
          regex: "(b-[0-9]+)\\..*"
          targetLabel: broker_id
          replacement: "$1"

    # 2. 抓取 MSK Broker 主機層 Node Exporter 指標 (CPU, Memory, EBS Disk IO & Space)
    - targets:
        - "b-1.prod-msk.xxxxxx.c2.kafka.ap-northeast-1.amazonaws.com:11002"
        - "b-2.prod-msk.xxxxxx.c2.kafka.ap-northeast-1.amazonaws.com:11002"
        - "b-3.prod-msk.xxxxxx.c2.kafka.ap-northeast-1.amazonaws.com:11002"
      interval: "30s"
      labels:
        msk_cluster: "prod-msk-tokyo"
        exporter_type: "node"
      relabelConfigs:
        - sourceLabels: [__address__]
          regex: "(b-[0-9]+)\\..*"
          targetLabel: broker_id
          replacement: "$1"

# 02-otel-msk-kafkametrics.yaml
receivers:
  kafkametrics:
    brokers:
      - "b-1.prod-msk.xxxxxx.c2.kafka.ap-northeast-1.amazonaws.com:9094"
      - "b-2.prod-msk.xxxxxx.c2.kafka.ap-northeast-1.amazonaws.com:9094"
      - "b-3.prod-msk.xxxxxx.c2.kafka.ap-northeast-1.amazonaws.com:9094"
    protocol_version: 3.5.0
    scrapers:
      - brokers
      - topics
      - consumers
    collection_interval: 15s
    auth:
      tls:
        insecure: false

exporters:
  prometheusremotewrite:
    endpoint: "http://vminsert-vm-tokyo-aza.monitoring.svc.cluster.local:8480/insert/0/prometheus/api/v1/write"

# 03-msk-alert-rules.yaml
apiVersion: operator.victoriametrics.com/v1beta1
kind: VMRule
metadata:
  name: aws-msk-critical-alerts
  namespace: monitoring
spec:
  groups:
    - name: aws-msk.rules
      interval: 30s
      rules:
        # 1. 離線分區告警 (最嚴重：代表部分 Partition 無可用 Leader，讀寫中斷)
        - alert: MSKOfflinePartitionsDetected
          expr: sum by (msk_cluster) (kafka_controller_KafkaController_Value{name="OfflinePartitionsCount"}) > 0
          for: 1m
          labels:
            severity: critical
          annotations:
            summary: "AWS MSK 叢集 ({{ $labels.msk_cluster }}) 出現離線分區！"
            description: "目前有 {{ $value }} 個 Partition 無可用 Leader，請立即檢查對應 Broker 日誌。"

        # 2. 副本未同步告警 (代表 ISR 縮小，資料複寫安全降級)
        - alert: MSKUnderReplicatedPartitions
          expr: sum by (msk_cluster, broker_id) (kafka_server_ReplicaManager_Value{name="UnderReplicatedPartitions"}) > 0
          for: 3m
          labels:
            severity: warning
          annotations:
            summary: "MSK Broker ({{ $labels.broker_id }}) 出現未同步副本 (Under-Replicated)"
            description: "Broker {{ $labels.broker_id }} 有 {{ $value }} 個分區落後於 Leader。"

        # 3. Broker 磁碟空間不足告警 (超過 80%)
        - alert: MSKBrokerDiskSpaceHigh
          expr: (1 - (node_filesystem_avail_bytes{msk_cluster!="", mountpoint="/kafka"} / node_filesystem_size_bytes{msk_cluster!="", mountpoint="/kafka"})) * 100 > 80
          for: 5m
          labels:
            severity: warning
          annotations:
            summary: "MSK Broker ({{ $labels.broker_id }}) 磁碟使用率超過 80%"

        # 4. Consumer Group 消費積壓告警 (Lag 持續增長)
        - alert: MSKConsumerGroupLagHigh
          expr: sum by (group, topic) (kafka_consumer_group_lag) > 10000
          for: 5m
          labels:
            severity: warning
          annotations:
            summary: "Kafka 消費群組 ({{ $labels.group }}) 在 Topic ({{ $labels.topic }}) 積壓超過 1 萬筆"

# 04-vector-msk-logs-to-victorialogs.yaml
sources:
  aws_msk_s3_logs:
    type: aws_s3
    region: ap-northeast-1
    sqs:
      queue_url: "https://sqs.ap-northeast-1.amazonaws.com/123456789012/msk-broker-logs-sqs"
    compression: gzip

transforms:
  parse_msk_log4j:
    type: remap
    inputs:
      - aws_msk_s3_logs
    source: |
      # 範例原始日誌: [2026-11-02 09:15:22,410] WARN [ReplicaFetcher replicaId=2, leaderId=1, fetcherId=0] Partition [orders-events,3] reset offset (kafka.server.ReplicaFetcherThread)
      parsed, err = parse_regex(
        string!(.message),
        r'^\[(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3})\]\s+(?P<level>[A-Z]+)\s+(?P<msg>.*)\s+\((?P<logger>[^\)]+)\)$'
      )
      if err == null {
        .level = parsed.level
        .logger = parsed.logger
        ._msg = parsed.msg
      } else {
        ._msg = string!(.message)
      }
      .service = "aws-msk"
      .msk_cluster = "prod-msk-tokyo"

sinks:
  victorialogs_cluster:
    type: elasticsearch
    inputs:
      - parse_msk_log4j
    endpoints:
      - "http://vlinsert-vl-tokyo-aza.logging.svc.cluster.local:9481/insert/elasticsearch/"
      - "http://vlinsert-vl-tokyo-azc.logging.svc.cluster.local:9481/insert/elasticsearch/"
    api_version: "v8"
    compression: gzip
    healthcheck:
      enabled: false
    query:
      _msg_field: "_msg"
      _time_field: "timestamp"
      _stream_fields: "service,msk_cluster,level"