OpenTelemetry를 사용하여 Kubernetes(Strimzi)에서 Kafka를 모니터링하세요.

OpenTelemetry Collector 구현하거나 배포하여 Strimzi 연산자를 사용하여 Kubernetes 에서 실행 중인 Kafka 클러스터를 모니터링합니다. 수집기는 Kafka 브로커를 자동으로 발견하고 포괄적인 정보를 수집합니다.

시작하기 전에

다음 사항을 확인하십시오:

뉴렐릭 계정
kubectl 액세스 권한이 있는 쿠버네티스 클러스터
Kafka 구현하다, JMX가 활성화된 Strimzi 연산자를 통해 배포하다

Strimzi Kafka에서 JMX를 활성화합니다.

Strimzi Kafka 리소스에서 Kafka 클러스터에 JMX가 활성화되어 있는지 확인하십시오.

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
  namespace: kafka
spec:
  kafka:
    jmxOptions: {}  # Enables JMX with default settings
    # ...other broker configuration

1단계: 네임스페이스 생성

OpenTelemetry Collector 전용 네임스페이스를 생성하거나 기존 Kafka 네임스페이스를 사용하십시오.

bash

$kubectl create namespace kafka

2단계: 라이선스 키를 사용하여 비밀 키를 생성합니다.

크리에이터 클러스터 키를 Kubernetes 비밀로 저장하세요.

bash

$kubectl create secret generic nr-license-key \
>  --from-literal=NEW_RELIC_LICENSE_KEY=YOUR_LICENSE_KEY \
>  -n kafka

YOUR_LICENSE_KEY 실제 뉴렐릭 키로 바꾸세요.

3단계: 구현하다, 배포하다 OpenTelemetry Collector

3.1 사용자 지정 수집기 이미지 생성

Java 런타임 및 JMX 스크래퍼가 포함된 사용자 지정 OpenTelemetry Collector 이미지를 생성합니다.

중요

버전 호환성: 이 가이드에서는 JMX Scraper 1.52.0 및 OpenTelemetry Collector 0.143.1 버전을 사용합니다. 이전 수집기 버전에는 호환성 목록에 이 스크래퍼의 해시가 포함되어 있지 않을 수 있습니다. 최상의 결과를 얻으려면 이 가이드에 나와 있는 대로 최신 버전을 사용하십시오.

최신 버전을 확인하세요:

OpenTelemetry Collector: OpenTelemetry Collector 릴리스 페이지를방문하세요.
JMX 스크래퍼: OpenTelemetry Java Contrib 릴리스확인
버전 호환성을 확인하세요:

수집기 버전에 맞는 supported_jars.go 파일을 확인하세요.
JMX Scraper 1.52.0이 SHA256 해시와 함께 jmxScraperVersions 맵에 나열되어 있는지 확인하십시오.
목록에 없으면 OpenTelemetry Collector를 최신 버전으로 업데이트하십시오.

타겟, 목표 실행: 시스템에 맞는 올바른 바이너리를 찾으려면 OpenTelemetry Collector 릴리스 페이지를 참조하세요(예: linux_amd64, linux_arm64, darwin_amd64). Dockerfile에서 TARGETARCH 변수를 accordingly 업데이트하세요.

Dockerfile 으로 저장:

# Multi-stage build for OpenTelemetry Collector with Java support for JMX receiver
# This image bundles the OTEL Collector with Java 17 runtime and JMX scraper JAR

FROM alpine:latest as prep

# OpenTelemetry Collector Binary
ARG OTEL_VERSION=0.143.1
ARG TARGETARCH=linux_amd64
ADD "https://github.com/open-telemetry/opentelemetry-collector-releases/releases/download/v${OTEL_VERSION}/otelcol-contrib_${OTEL_VERSION}_${TARGETARCH}.tar.gz" /otelcontribcol
RUN tar -zxvf /otelcontribcol

# JMX Scraper JAR (for JMX receiver with YAML-based configuration)
ARG JMX_SCRAPER_JAR_VERSION=1.52.0
ADD https://github.com/open-telemetry/opentelemetry-java-contrib/releases/download/v${JMX_SCRAPER_JAR_VERSION}/opentelemetry-jmx-scraper.jar /opt/opentelemetry-jmx-scraper.jar

# Set permissions for nonroot user (uid 65532)
ARG USER_UID=65532
RUN chown ${USER_UID} /opt/opentelemetry-jmx-scraper.jar

# Final minimal image with Java runtime
FROM openjdk:17-jre-slim

COPY --from=prep /opt/opentelemetry-jmx-scraper.jar /opt/opentelemetry-jmx-scraper.jar
COPY --from=prep /otelcol-contrib /otelcol-contrib

EXPOSE 4317 4318 8888
ENTRYPOINT ["/otelcol-contrib"]
CMD ["--config", "/conf/otel-agent-config.yaml"]

이미지를 빌드하고 푸시합니다.

bash

$docker build -t your-registry/otel-collector-kafka:latest .
$docker push your-registry/otel-collector-kafka:latest

3.2 JMX 맞춤형 지표 ConfigMap 생성

먼저, 사용자 정의 JMX 지표 설정을 사용하여 ConfigMap을 만듭니다. jmx-kafka-config.yaml 으로 저장:

apiVersion: v1
kind: ConfigMap
metadata:
  name: jmx-kafka-config
  namespace: kafka
data:
  jmx-kafka-config.yaml: |
    ---
    rules:
      # Per-topic custom metrics using custom MBean commands
      - bean: kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec,topic=*
        metricAttribute:
          topic: param(topic)
        mapping:
          Count:
            metric: kafka.prod.msg.count
            type: counter
            desc: The number of messages in per topic
            unit: "{message}"

      - bean: kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec,topic=*
        metricAttribute:
          topic: param(topic)
          direction: const(in)
        mapping:
          Count:
            metric: kafka.topic.io
            type: counter
            desc: The bytes received or sent per topic
            unit: By

      - bean: kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec,topic=*
        metricAttribute:
          topic: param(topic)
          direction: const(out)
        mapping:
          Count:
            metric: kafka.topic.io
            type: counter
            desc: The bytes received or sent per topic
            unit: By

      # Cluster-level metrics using controller-based MBeans
      - bean: kafka.controller:type=KafkaController,name=GlobalTopicCount
        mapping:
          Value:
            metric: kafka.cluster.topic.count
            type: gauge
            desc: The total number of global topics in the cluster
            unit: "{topic}"

      - bean: kafka.controller:type=KafkaController,name=GlobalPartitionCount
        mapping:
          Value:
            metric: kafka.cluster.partition.count
            type: gauge
            desc: The total number of global partitions in the cluster
            unit: "{partition}"

      - bean: kafka.controller:type=KafkaController,name=FencedBrokerCount
        mapping:
          Value:
            metric: kafka.broker.fenced.count
            type: gauge
            desc: The number of fenced brokers in the cluster
            unit: "{broker}"

      - bean: kafka.controller:type=KafkaController,name=PreferredReplicaImbalanceCount
        mapping:
          Value:
            metric: kafka.partition.non_preferred_leader
            type: gauge
            desc: The count of topic partitions for which the leader is not the preferred leader
            unit: "{partition}"

      # Broker-level metrics using ReplicaManager MBeans
      - bean: kafka.server:type=ReplicaManager,name=UnderMinIsrPartitionCount
        mapping:
          Value:
            metric: kafka.partition.under_min_isr
            type: gauge
            desc: The number of partitions where the number of in-sync replicas is less than the minimum
            unit: "{partition}"

      # Broker uptime metric using JVM Runtime
      - bean: java.lang:type=Runtime
        mapping:
          Uptime:
            metric: kafka.broker.uptime
            type: gauge
            desc: Broker uptime in milliseconds
            unit: ms

      # Leader count per broker
      - bean: kafka.server:type=ReplicaManager,name=LeaderCount
        mapping:
          Value:
            metric: kafka.broker.leader.count
            type: gauge
            desc: Number of partitions for which this broker is the leader
            unit: "{partition}"

      # JVM metrics
      - bean: java.lang:type=GarbageCollector,name=*
        mapping:
          CollectionCount:
            metric: jvm.gc.collections.count
            type: counter
            unit: "{collection}"
            desc: total number of collections that have occurred
            metricAttribute:
              name: param(name)
          CollectionTime:
            metric: jvm.gc.collections.elapsed
            type: counter
            unit: ms
            desc: the approximate accumulated collection elapsed time in milliseconds
            metricAttribute:
              name: param(name)

      - bean: java.lang:type=Memory
        unit: By
        prefix: jvm.memory.
        dropNegativeValues: true
        mapping:
          HeapMemoryUsage.committed:
            metric: heap.committed
            desc: current heap usage
            type: gauge
          HeapMemoryUsage.max:
            metric: heap.max
            desc: current heap usage
            type: gauge
          HeapMemoryUsage.used:
            metric: heap.used
            desc: current heap usage
            type: gauge

      - bean: java.lang:type=Threading
        mapping:
          ThreadCount:
            metric: jvm.thread.count
            type: gauge
            unit: "{thread}"
            desc: Total thread count (Kafka typical range 100-300 threads)

      - bean: java.lang:type=OperatingSystem
        prefix: jvm.
        dropNegativeValues: true
        mapping:
          SystemLoadAverage:
            metric: system.cpu.load_1m
            type: gauge
            unit: "{run_queue_item}"
            desc: System load average (1 minute) - alert if > CPU count
          AvailableProcessors:
            metric: cpu.count
            type: gauge
            unit: "{cpu}"
            desc: Number of processors available
          ProcessCpuLoad:
            metric: cpu.recent_utilization
            type: gauge
            unit: '1'
            desc: Recent CPU utilization for JVM process (0.0 to 1.0)
          SystemCpuLoad:
            metric: system.cpu.utilization
            type: gauge
            unit: '1'
            desc: Recent CPU utilization for whole system (0.0 to 1.0)
          OpenFileDescriptorCount:
            metric: file_descriptor.count
            type: gauge
            unit: "{file_descriptor}"
            desc: Number of open file descriptors - alert if > 80% of ulimit

      - bean: java.lang:type=ClassLoading
        mapping:
          LoadedClassCount:
            metric: jvm.class.count
            type: gauge
            unit: "{class}"
            desc: Currently loaded class count

      - bean: java.lang:type=MemoryPool,name=*
        type: gauge
        unit: By
        metricAttribute:
          name: param(name)
        mapping:
          Usage.used:
            metric: jvm.memory.pool.used
            desc: Memory pool usage by generation (G1 Old Gen, Eden, Survivor)
          Usage.max:
            metric: jvm.memory.pool.max
            desc: Maximum memory pool size
          CollectionUsage.used:
            metric: jvm.memory.pool.used_after_last_gc
            desc: Memory used after last GC (shows retained memory baseline)

팁

메트릭 수집 사용자 지정: kafka-jmx-config.yaml 파일에 사용자 지정 MBean 규칙을 추가하여 추가 Kafka 메트릭을 수집할 수 있습니다.

JMX 메트릭 규칙의 기본 구문을알아보세요.
Kafka 모니터링 문서에서 사용 가능한 MBean 이름을 찾아보세요.
이를 통해 특정 모니터링 요구 사항에 따라 Kafka 브로커에서 노출하는 모든 JMX 메트릭을 수집할 수 있습니다.

JMX ConfigMap을 적용합니다.

bash

$kubectl apply -f jmx-kafka-config.yaml

3.3 수집기 ConfigMap 생성

OpenTelemetry Collector 설정을 포함하는 ConfigMap을 생성합니다. otel-kafka-config.yaml 으로 저장:

---
apiVersion: v1
kind: ConfigMap
metadata:
  name: otel-collector-config
  namespace: kafka
  labels:
    app: otel-collector
data:
  otel-collector-config.yaml: |
    receivers:
      # Kafka cluster-level metrics (runs once per OTEL collector)
      kafkametrics/cluster:
        brokers:
          - "my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092"
        protocol_version: 2.8.0
        scrapers:
          - brokers
          - topics
          - consumers
        collection_interval: 30s
        metrics:
          kafka.topic.min_insync_replicas:
            enabled: true
          kafka.topic.replication_factor:
            enabled: true
          kafka.partition.replicas:
            enabled: false
          kafka.partition.oldest_offset:
            enabled: false
          kafka.partition.current_offset:
            enabled: false

      # Receiver creator for dynamic per-broker JMX receivers
      receiver_creator:
        watch_observers: [k8s_observer]
        receivers:
          # JMX receiver template (created per discovered broker pod)
          jmx:
            rule: type == "pod" && labels["strimzi.io/kind"] == "Kafka" && labels["strimzi.io/cluster"] == "my-cluster" && labels["strimzi.io/name"] == "my-cluster-kafka"
            config:
              endpoint: 'service:jmx:rmi:///jndi/rmi://`endpoint`:9999/jmxrmi'
              jar_path: /opt/opentelemetry-jmx-scraper.jar
              target_system: kafka
              jmx_configs: /conf-jmx/jmx-kafka-config.yaml
              collection_interval: 30s
              # Set dynamic resource attributes from discovered pod
              resource_attributes:
                broker.endpoint: '`endpoint`'

    exporters:
      otlp:
        endpoint: https://otlp.nr-data.net:4317
        tls:
          insecure: false
        sending_queue:
          num_consumers: 12
          queue_size: 5000
        retry_on_failure:
          enabled: true
        headers:
          api-key: ${NEW_RELIC_LICENSE_KEY}

    processors:
      # Batch processor for efficiency
      batch/aggregation:
        send_batch_size: 1024
        timeout: 30s

      # Memory limiter to prevent OOM
      memory_limiter:
        limit_percentage: 80
        spike_limit_percentage: 30
        check_interval: 1s

      # Detect system resources
      resourcedetection:
        detectors: [env, docker, system]
        timeout: 5s
        override: false

      # Add Kafka cluster metadata
      resource/kafka_metadata:
        attributes:
          - key: kafka.cluster.name
            value: my-cluster
            action: upsert
    

      # Extract Kubernetes attributes
      k8sattributes:
        auth_type: serviceAccount
        passthrough: false
        extract:
          metadata:
            - k8s.pod.name
            - k8s.pod.uid
            - k8s.namespace.name
            - k8s.node.name
          labels:
            - tag_name: strimzi.cluster
              key: strimzi.io/cluster
              from: pod
            - tag_name: strimzi.kind
              key: strimzi.io/kind
              from: pod

      # Transform metrics for New Relic UI
      transform:
        metric_statements:
          - context: metric
            statements:
              # Clean up descriptions and units
              - set(description, "") where description != ""
              - set(unit, "") where unit != ""

          - context: resource
            statements:
              # Extract broker.id from k8s.pod.name: my-cluster-kafka-0 -> 0 (supports multi-digit)
              - set(attributes["broker.id"], ExtractPatterns(attributes["k8s.pod.name"], ".*-(?P<broker_id>\\d+)$")["broker_id"]) where attributes["k8s.pod.name"] != nil

      # Remove broker.id for cluster-level metrics
      transform/remove_broker_id:
        metric_statements:
          - context: resource
            statements:
              - delete_key(attributes, "broker.id")
              - delete_key(attributes, "broker.endpoint")
              - delete_key(attributes, "k8s.pod.name")

      # Topic sum aggregation for replicas_in_sync
      metricstransform/kafka_topic_sum_aggregation:
        transforms:
          - include: kafka.partition.replicas_in_sync
            action: insert
            new_name: kafka.partition.replicas_in_sync.total
            operations:
              - action: aggregate_labels
                label_set: [ topic ]
                aggregation_type: sum

      # Filter to include only cluster-level metrics
      filter/include_cluster_metrics:
        metrics:
          include:
            match_type: regexp
            metric_names:
              - "kafka\\.partition\\.offline"
              - "kafka\\.(leader|unclean)\\.election\\.rate"
              - "kafka\\.partition\\.non_preferred_leader"
              - "kafka\\.broker\\.fenced\\.count"
              - "kafka\\.cluster\\.partition\\.count"
              - "kafka\\.cluster\\.topic\\.count"

      # Filter to exclude cluster-level metrics from broker pipeline
      filter/exclude_cluster_metrics:
        metrics:
          exclude:
            match_type: regexp
            metric_names:
              - "kafka\\.partition\\.offline"
              - "kafka\\.(leader|unclean)\\.election\\.rate"
              - "kafka\\.partition\\.non_preferred_leader"
              - "kafka\\.broker\\.fenced\\.count"
              - "kafka\\.cluster\\.partition\\.count"
              - "kafka\\.cluster\\.topic\\.count"

      # Convert cumulative metrics to delta for New Relic
      cumulativetodelta:

    extensions:
      # K8s observer extension
      k8s_observer:
        auth_type: serviceAccount
        observe_pods: true
        observe_nodes: false

    service:
      extensions: [k8s_observer]

      pipelines:
        # Per-broker metrics pipeline (with broker.id)
        metrics/broker:
          receivers:
            - receiver_creator
            - kafkametrics/cluster
          processors:
            - memory_limiter
            - resourcedetection
            - resource/kafka_metadata
            - k8sattributes
            - filter/exclude_cluster_metrics
            - transform
            - metricstransform/kafka_topic_sum_aggregation
            - cumulativetodelta
            - batch/aggregation
          exporters: [otlp]

        # Cluster-level metrics pipeline (without broker.id, aggregated)
        metrics/cluster:
          receivers:
            - receiver_creator
          processors:
            - memory_limiter
            - resourcedetection
            - resource/kafka_metadata
            - k8sattributes
            - filter/include_cluster_metrics
            - transform/remove_broker_id
            - metricstransform/kafka_topic_sum_aggregation
            - cumulativetodelta
            - batch/aggregation
          exporters: [otlp]

설정 참고 사항:

my-cluster-kafka-bootstrap Strimzi Kafka 서비스 이름으로 바꾸세요.
rule 과 kafka.cluster.name 에서 my-cluster 클러스터 이름으로 바꾸세요.
네임스페이스가 다르면 업데이트하세요. kafka
OTLP 엔드포인트: https://otlp.nr-data.net:4317 (미국 지역) 또는 https://otlp.eu01.nr-data.net:4317 (유럽 지역)을 사용합니다. 다른 지역의 OTLP 엔드포인트 구성 방법을 참조하세요.
receiver_creator 은 Strimzi 레이블을 사용하여 Kafka 브로커를 자동으로 검색합니다.

ConfigMap을 적용합니다.

bash

$kubectl apply -f otel-kafka-config.yaml

3.4 구현하다, 배포하다 수집기

구현, 배포를 만듭니다. otel-collector-deployment.yaml 으로 저장:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: otel-collector
  namespace: kafka
spec:
  replicas: 1
  selector:
    matchLabels:
      app: otel-collector
  template:
    metadata:
      labels:
        app: otel-collector
    spec:
      serviceAccountName: otel-collector
      containers:
      - name: otel-collector
        image: your-registry/otel-collector-kafka:latest
        env:
          - name: NEW_RELIC_LICENSE_KEY
            valueFrom:
              secretKeyRef:
                name: nr-license-key
                key: NEW_RELIC_LICENSE_KEY
        resources:
          limits:
            cpu: "1"
            memory: "2Gi"
          requests:
            cpu: "500m"
            memory: "1Gi"
        volumeMounts:
        - name: vol-kafka-test-cluster
          mountPath: /conf
        - name: jmx-config
          mountPath: /conf-jmx
        ports:
        - containerPort: 4317  # OTLP gRPC
        - containerPort: 4318  # OTLP HTTP
        - containerPort: 8888  # Metrics
      volumes:
      - name: vol-kafka-test-cluster
        configMap:
          name: otel-collector-config
          items:
          - key: otel-collector-config.yaml
            path: otel-agent-config.yaml
      - name: jmx-config
        configMap:
          name: jmx-kafka-config
          items:
          - key: jmx-kafka-config.yaml
            path: jmx-kafka-config.yaml
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: otel-collector
  namespace: kafka
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: otel-collector
rules:
  - apiGroups: [""]
    resources: ["pods", "nodes"]
    verbs: ["get", "list", "watch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: otel-collector
subjects:
  - kind: ServiceAccount
    name: otel-collector
    namespace: kafka
roleRef:
  kind: ClusterRole
  name: otel-collector
  apiGroup: rbac.authorization.k8s.io

자원 설정:

위의 리소스 제한은 중간 규모의 Kafka 클러스터(브로커 5~10개, 토픽 20~100개)에 적합합니다.

구현을 적용합니다.

bash

$kubectl apply -f otel-collector-deployment.yaml

수집기가 실행 중인지 확인하십시오.

bash

$kubectl get pods -n kafka -l app=otel-collector
$kubectl logs -n kafka -l app=otel-collector -f

4단계: (선택 사항) 제작자 또는 소비자 참여

Kubernetes 에서 실행되는 Kafka 제작자 및 소비자 제작으로부터 로그 수준의 텔레메트리를 수집하려면 OpenTelemetry 서버 에이전트를 사용하여 이를 로그합니다.

당신의 Kafka를 편집하세요

Kafka 생산자 또는 소비자 군대를 위해 기존 구현에 OpenTelemetry 동양 에이전트를 추가하고 배포합니다.

에이전트 에이전트 다운로드: 에이전트 JAR을 다운로드하려면 init 컨테이너를 추가하세요.

initContainers:
- name: download-otel-agent
  image: busybox:latest
  command:
    - sh
    - -c
    - |
      wget -O /otel/opentelemetry-javaagent.jar \
        https://github.com/open-telemetry/opentelemetry-java-instrumentation/releases/latest/download/opentelemetry-javaagent.jar
  volumeMounts:
    - name: otel-agent
      mountPath: /otel

군대 에이전트 구성: 견고한 컨테이너에 환경 변수를 추가합니다.

env:
  - name: JAVA_TOOL_OPTIONS
    value: >-
      -javaagent:/otel/opentelemetry-javaagent.jar
      -Dotel.service.name="kafka-producer"
      -Dotel.resource.attributes="kafka.cluster.name=my-cluster"
      -Dotel.exporter.otlp.endpoint="http://localhost:4317"
      -Dotel.exporter.otlp.protocol="grpc"
      -Dotel.metrics.exporter="otlp"
      -Dotel.traces.exporter="otlp"
      -Dotel.logs.exporter="otlp"
      -Dotel.instrumentation.kafka.experimental-span-attributes="true"
      -Dotel.instrumentation.messaging.experimental.receive-telemetry.enabled="true"
      -Dotel.instrumentation.kafka.producer-propagation.enabled="true"
      -Dotel.instrumentation.kafka.enabled="true"
volumeMounts:
  - name: otel-agent
    mountPath: /otel

볼륨 추가: 볼륨 정의를 포함하세요:
```
volumes:
  - name: otel-agent
    emptyDir: {}
```

바꾸다:

kafka-producer 귀하의 독특한 이름으로
my-cluster Kafka 클러스터 이름을 사용하여

팁

위 설정은 텔메트리를 localhost:4317에서 실행되는 OpenTelemetry Collector 로 보냅니다. 이 설정을 사용하여 구현하다, 배포하다:

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: "0.0.0.0:4317"

exporters:
  otlp/newrelic:
    endpoint: https://otlp.nr-data.net:4317
    headers:
      api-key: "${NEW_RELIC_LICENSE_KEY}"
    compression: gzip
    timeout: 30s

service:
  pipelines:
    traces:
      receivers: [otlp]
      exporters: [otlp/newrelic]
    metrics:
      receivers: [otlp]
      exporters: [otlp/newrelic]
    logs:
      receivers: [otlp]
      exporters: [otlp/newrelic]

이를 통해 처리를 사용자 지정하고, 필터를 추가하거나, 여러 백앤드에게 라우팅할 수 있습니다. 다른 엔드포인트 설정에 대해서는 OTLP 엔드포인트 구성을 참조하세요.

잔류 에이전트는 코드 변경이 전혀 없는 기본 Kafka 측정, 캡처 기능을 제공합니다.

요청 지연시간
처리량 지표
오류율
분산 추적

고급 설정에 대해서는 Kafka 측정, 로그 문서를 참조하세요.

5단계: (선택 사항) Kafka 브로커 로그 전달

Kubernetes 환경에서 Kafka 브로커 로그를 수집하여 뉴렐릭으로 전송하려면 OpenTelemetry Collector 에서 파일 로그 수신기를 구성하십시오.

파일 로그 수신기를 추가하도록 수집기 ConfigMap을 업데이트하십시오. receivers 섹션에 다음을 추가하세요:

receivers:
  # ... existing receivers (receiver_creator, kafkametrics/cluster) ...
  
  # File log receiver for Kafka broker logs
  filelog/kafka_broker:
    include:
      - ${env:HOME}/logs/kafka-broker-1.log
    start_at: end
    multiline:
      line_start_pattern: '^\['
    resource:
      broker.id: "1"  # Adjust based on your broker setup

service 섹션에 로그 파이프라인을 추가하세요:

service:
  pipelines:
    # ... existing pipelines (metrics/broker, metrics/cluster) ...
    
    # Logs pipeline for Kafka broker logs
    logs/brokers:
      receivers: [filelog/kafka_broker]
      processors: [batch/aggregation, resourcedetection, resource/kafka_metadata]
      exporters: [otlp]

설정 참고 사항:

include 경로 패턴을 Kafka 로그 파일 위치에 맞게 업데이트하세요.
브로커 식별자에 맞게 broker.id 조정하세요.
multiline 패턴은 로그가 [ 로 시작한다고 가정합니다. 로그 형식이 다르면 조정하십시오.
전체 설정 옵션 및 고급 패턴에 대한 자세한 내용은 파일 로그 수신기 설명서를참조하십시오.

업데이트된 설정을 적용하세요:

bash

$kubectl apply -f otel-collector-config.yaml
$kubectl rollout restart deployment -n kafka otel-collector

Kafka 브로커 로그는 다음 두 곳에서 확인할 수 있습니다.

브로커 부분: 특정 브로커와 상관 관계가 있는 로그를 보려면 뉴렐릭의 Kafka 브로커 부분으로 이동하세요.
로그 UI: 다음과 같은 필터가 포함된 로그 UI 사용하여 모든 Kafka 로그를 쿼리합니다. kafka.cluster.name = 'my-cluster'
NRQL을 사용하여 로그를 쿼리할 수도 있습니다.
```
FROM Log SELECT * WHERE kafka.cluster.name = 'my-kafka-cluster'
```

데이터 찾기

몇 분 후, Kafka 창이 뉴렐릭에 나타날 것입니다. 뉴렐릭 UI 의 다양한 보기에서 Kafka 범위를 탐색하는 방법에 대한 자세한 지침은 "데이터 찾기"를 참조하세요.

NRQL을 사용하여 데이터를 쿼리할 수도 있습니다.

FROM Metric SELECT * WHERE kafka.cluster.name = 'my-kafka-cluster'

문제점 해결

수집기 디버그 로그 활성화: 설정 문제를 해결하기 위해 자세한 로깅을 추가합니다.

수집기 ConfigMap을 편집하세요:

bash

$kubectl edit configmap -n kafka otel-collector-config

otel-collector-config.yaml 데이터의 service: 아래에 텔레메트리 섹션을 추가하세요:

service:
  telemetry:
    logs:
      level: "debug"  # Enable detailed collector internal logs
  extensions: [k8s_observer]
  pipelines:
    # ... existing pipelines ...

저장하고 종료하세요. 수집기는 자동으로 설정을 다시 로드합니다.

디버그 내보내기 추가: 뉴렐릭으로 보내기 전에 수집기 로그에서 지표를 확인하세요.

ConfigMap에 다음을 추가하세요:

exporters:
  debug:
    verbosity: detailed
    sampling_initial: 5        # Log first 5 metrics
    sampling_thereafter: 200   # Then log every 200th metric

  otlp/newrelic:
    endpoint: https://otlp.nr-data.net:4317
    headers:
      api-key: ${env:NEW_RELIC_LICENSE_KEY}
    compression: gzip
    timeout: 30s

service:
  pipelines:
    metrics/brokers-cluster-topics:
      receivers: [receiver_creator, kafkametrics/cluster]
      processors: [resourcedetection, resource, filter/exclude_cluster_metrics, transform/des_units, cumulativetodelta, metricstransform/kafka_topic_sum_aggregation, batch/aggregation]
      exporters: [debug, otlp/newrelic]  # Add debug exporter

그런 다음 수집기를 다시 시작하고 로그를 확인하십시오.

bash

$# Restart collector
$kubectl rollout restart deployment -n kafka otel-collector
$
$# View logs with metric output
$kubectl logs -n kafka -l app=otel-collector -f

중요: 로그 오버플로를 방지하려면 프로덕션 환경에서 디버그 익스포터를 제거하십시오.

파드 상태 및 이벤트 확인:

bash

$# Check pod status
$kubectl get pods -n kafka -l app=otel-collector
$
$# View detailed pod description
$kubectl describe pod -n kafka -l app=otel-collector
$
$# Check recent logs
$kubectl logs -n kafka -l app=otel-collector --previous --tail=50

일반적인 문제점 및 해결책:

JMX 스크래퍼 누락: init 컨테이너가 JAR 파일을 성공적으로 다운로드했는지 확인하십시오.

bash

$# Check init container logs
$kubectl logs -n kafka -l app=otel-collector -c download-jmx-scraper

잘못된 설정: ConfigMap YAML 구문의 유효성을 검사합니다.

bash

$# Check ConfigMap contents
$kubectl get configmap -n kafka otel-kafka-config -o yaml
$
$# Validate YAML syntax
$kubectl get configmap -n kafka otel-kafka-config -o yaml | kubectl apply --dry-run=client -f -

RBAC 권한: 서비스 계정에 적절한 클러스터 역할 바인딩이 있는지 확인하십시오.

bash

$# Check ServiceAccount
$kubectl get serviceaccount -n kafka otel-collector
$
$# Check ClusterRoleBinding
$kubectl get clusterrolebinding otel-collector-binding -o yaml

리소스 제약 조건: 파드가 OOMKilled 오류가 발생하거나 리소스 제한에 걸렸는지 확인합니다.

bash

$# Check resource usage
$kubectl top pods -n kafka -l app=otel-collector
$
$# Check for resource limits
$kubectl describe pod -n kafka -l app=otel-collector | grep -A 5 "Limits\|Requests"

JMX 활성화 여부 확인: Strimzi Kafka 리소스에 JMX가 구성되어 있는지 확인하십시오.

bash

$# Check Kafka resource configuration
$kubectl get kafka -n kafka -o yaml | grep -A 5 jmxOptions

파드 레이블 확인: Kafka 파드에 검색을 위한 올바른 레이블이 지정되어 있는지 확인하십시오.

bash

$# Verify Kafka pod labels
$kubectl get pods -n kafka -l strimzi.io/kind=Kafka --show-labels
$
$# Check if receiver_creator can discover pods
$kubectl logs -n kafka -l app=otel-collector | grep "discovered"

관찰자 로그 검토: 수집기 로그에서 파드 발견 메시지를 찾습니다.

bash

$# Filter for k8s observer logs
$kubectl logs -n kafka -l app=otel-collector | grep -i "observer\|discovery"

JMX 연결 테스트: Kafka 브로커와의 네트워크 연결을 확인합니다.

bash

$# Get Kafka broker pod IPs
$kubectl get pods -n kafka -l strimzi.io/kind=Kafka -o wide
$
$# Test JMX port connectivity from collector pod
$kubectl exec -it -n kafka deployment/otel-collector -- sh -c "nc -zv <kafka-broker-pod-ip> 9999"
$
$# Check if JMX port is listening on Kafka pods
$kubectl exec -it -n kafka <kafka-pod-name> -- netstat -tlnp | grep :9999

수신기 설정 유효성 검사: JMX 수신기가 올바르게 구성되었는지 확인합니다.

bash

$# Check collector configuration
$kubectl logs -n kafka -l app=otel-collector | grep -i "jmx\|kafka"

리소스 사용량 모니터링:

bash

$# Check current memory usage
$kubectl top pods -n kafka -l app=otel-collector
$
$# Watch memory usage over time
$watch kubectl top pods -n kafka -l app=otel-collector

모니터링 주제 축소: 필수적인 주제만 수집하도록 제한

bash

$# Update ConfigMap to filter topics
$kubectl patch configmap -n kafka otel-kafka-config --patch '
$data:
$  config.yaml: |
$    # Add topic filtering to kafkametrics receiver
$    kafkametrics/cluster:
$      topic_match: "^(important-topic-1|important-topic-2)$"
$'

수집 간격 늘리기: 수집 빈도 줄이기

# In your ConfigMap, update intervals:
receivers:
  kafkametrics/cluster:
    collection_interval: 45s  # Increase from 30s to 45s
  receiver_creator:
    receivers:
      jmx:
        config:
          collection_interval: 45s  # Increase from 30s to 45s (max 59s supported)

일괄 처리 최적화: 일괄 처리기 설정을 조정합니다.

# In your ConfigMap:
processors:
  batch/aggregation:
    timeout: 30s
    send_batch_size: 512  # Reduce from 1024

메모리 제한 설정: OOM(메모리 부족)을 방지하기 위해 리소스 제한을 추가하세요.

bash

$# Update deployment with memory limits
$kubectl patch deployment -n kafka otel-collector --patch '
$spec:
$  template:
$    spec:
$      containers:
$      - name: otel-collector
$        resources:
$          limits:
$            memory: "512Mi"
$          requests:
$            memory: "256Mi"
$'

변경 후 수집기를 다시 시작하세요.

bash

$kubectl rollout restart deployment -n kafka otel-collector

로그에 기록된 오류 메시지:

error subprocess/subprocess.go:XXX subprocess died
otelcol.component.id: "jmx/kafka_broker-X"
error: "unexpected shutdown: exit status 1"

JMX 수집 간격을 확인하세요. JMX 스크래퍼가 포함된 JMX 수신기는 최대 59초의 수집 간격만 지원합니다.

ConfigMap을 업데이트하세요:

receivers:
  receiver_creator:
    receivers:
      jmx:
        config:
          jar_path: /opt/opentelemetry/opentelemetry-jmx-scraper.jar
          target_system: kafka
          collection_interval: 59s  # Must be 59s or less, NOT 60s or higher
          jmx_configs: /etc/otel/jmx-kafka-config.yaml

JMX 스크래퍼가 다운로드되었는지 확인하십시오.

bash

$# Check init container logs
$kubectl logs -n kafka -l app=otel-collector -c download-jmx-scraper
$
$# Verify file exists in running pod
$kubectl exec -it -n kafka deployment/otel-collector -- ls -lh /opt/opentelemetry/opentelemetry-jmx-scraper.jar

화면을 사용할 수 있는지 확인하세요. JMX 스크래퍼에는 화면 런타임이 필요합니다.

bash

$# Check Java in collector pod
$kubectl exec -it -n kafka deployment/otel-collector -- java -version

JMX 엔드포인트에 액세스할 수 있는지 확인: 수집기에서 Kafka 브로커까지의 연결성을 테스트합니다.

bash

$# Get Kafka broker pod IP
$kubectl get pods -n kafka -l strimzi.io/kind=Kafka -o wide
$
$# Test JMX port from collector pod
$kubectl exec -it -n kafka deployment/otel-collector -- timeout 5 sh -c "</dev/tcp/<kafka-broker-pod-ip>/9999" && echo "JMX accessible" || echo "JMX not accessible"

자세한 오류 내용은 수집기 로그를 확인하세요.

bash

$# View recent logs
$kubectl logs -n kafka -l app=otel-collector --tail=100 | grep -i "jmx\|error"

다음 단계

Kafka 메트릭 살펴보기 - 전체 메트릭 참조 자료를 확인하세요
맞춤형 대시보드 만들기 - Kafka 데이터에 대한 시각화 구축
알림 설정 - 소비자 지연 및 복제되지 않은 파티션과 같은 중요한 지표를 모니터링합니다.

사용자의 편의를 위해 제공되는 기계 번역입니다.

OpenTelemetry를 사용하여 Kubernetes(Strimzi)에서 Kafka를 모니터링하세요.

시작하기 전에

Strimzi Kafka에서 JMX를 활성화합니다.

1단계: 네임스페이스 생성

2단계: 라이선스 키를 사용하여 비밀 키를 생성합니다.

3단계: 구현하다, 배포하다 OpenTelemetry Collector

3.1 사용자 지정 수집기 이미지 생성

중요

수집기가 이 JMX 스크래퍼 버전을 지원하는지 확인하십시오.

3.2 JMX 맞춤형 지표 ConfigMap 생성

팁

3.3 수집기 ConfigMap 생성

추가 수신자 문서

3.4 구현하다, 배포하다 수집기

4단계: (선택 사항) 제작자 또는 소비자 참여

당신의 Kafka를 편집하세요

팁

5단계: (선택 사항) Kafka 브로커 로그 전달

로그 수집 구성

뉴렐릭에서 내 로그인 찾기

데이터 찾기

문제점 해결

디버그 로깅 활성화

Collector 파드 시작 안 됨

JMX 메트릭이 수집되지 않았습니다.

높은 메모리 사용량

JMX 수신기 하위 프로세스 오류

다음 단계

사용자의 편의를 위해 제공되는 기계 번역입니다.

OpenTelemetry를 사용하여 Kubernetes(Strimzi)에서 Kafka를 모니터링하세요.

시작하기 전에 .css-21sua1{background:none;border:none;width:0;padding:0;}

Strimzi Kafka에서 JMX를 활성화합니다.

1단계: 네임스페이스 생성

2단계: 라이선스 키를 사용하여 비밀 키를 생성합니다.

3단계: 구현하다, 배포하다 OpenTelemetry Collector

3.1 사용자 지정 수집기 이미지 생성

중요

3.2 JMX 맞춤형 지표 ConfigMap 생성

팁

3.3 수집기 ConfigMap 생성

추가 수신자 문서

3.4 구현하다, 배포하다 수집기

4단계: (선택 사항) 제작자 또는 소비자 참여

당신의 Kafka를 편집하세요

팁

5단계: (선택 사항) Kafka 브로커 로그 전달

로그 수집 구성

뉴렐릭에서 내 로그인 찾기

데이터 찾기

문제점 해결

디버그 로깅 활성화

Collector 파드 시작 안 됨

JMX 메트릭이 수집되지 않았습니다.

높은 메모리 사용량

JMX 수신기 하위 프로세스 오류

다음 단계

시작하기 전에