OpenTelemetry를 사용하여 Kubernetes에서 자체 관리형 Kafka 모니터링

OpenTelemetry Collector를 배포하여 메트릭을 수집하고 뉴렐릭으로 전달함으로써 Kubernetes에서 실행되는 자체 관리형 아파치 Kafka 클러스터를 모니터링하세요.

아키텍처

뉴렐릭은 자체 관리형 Kubernetes Kafka 모니터링을 위한 두 가지 접근 방식을 지원합니다: OpenTelemetry 자바 에이전트 또는 Prometheus JMX Exporter. 다음 다이어그램은 각 접근 방식에 대한 데이터 흐름을 보여줍니다.

Kubernetes self-managed Kafka monitoring architecture

설치 단계

다음 단계에 따라 브로커에 OpenTelemetry 자바 에이전트를 설치하고 수집기를 배포하여 메트릭과 로그를 수집하고 뉴렐릭으로 전송하여 포괄적인 Kafka 모니터링을 설정하십시오.

시작하기 전에

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

뉴렐릭 계정
kubectl 액세스 권한이 있는 쿠버네티스 클러스터
StatefulSet으로 배포된 Kafka
Kafka StatefulSet을 수정하고 재배포할 수 있는 기능

구현하다, 배포하다 OpenTelemetry Collector

클러스터에 OpenTelemetry 수집기를 배포하십시오. 이 단계에서는 자바 에이전트가 각 브로커 파드에서 수집하는 JMX 메트릭을 정의하는 kafka-jmx-config ConfigMap도 생성합니다. 다음 단계에서 Kafka 브로커를 다시 시작하기 전에 수집기가 실행 중이어야 합니다.

1단계. 뉴렐릭 자격 증명 시크릿 생성

bash

$kubectl create secret generic newrelic-otlp-secret \
>  --namespace newrelic \
>  --from-literal=NEW_RELIC_LICENSE_KEY='your-license-key-here' \
>  --from-literal=NEW_RELIC_OTLP_ENDPOINT='https://otlp.nr-data.net:4317'

bash

$kubectl create secret generic newrelic-otlp-secret \
>  --namespace newrelic \
>  --from-literal=NEW_RELIC_LICENSE_KEY='your-license-key-here' \
>  --from-literal=NEW_RELIC_OTLP_ENDPOINT='https://otlp.eu01.nr-data.net:4317'

팁

다른 엔드포인트 설정에 대해서는 OTLP 엔드포인트 구성을 참조하세요.

2단계. 수집기 설정이 포함된 values.yaml 생성

NRDOT 및 OpenTelemetry 수집기 모두 동일한 설정을 사용합니다. 선호하는 수집기 이미지를 선택하세요:

NRDOT 은 뉴릭에서 지원하는 OpenTelemetry Collector 배포판으로, 뉴릭에 대한 완벽한 지원을 제공합니다. 자세한 내용은 NRDOT Collector GitHub 저장소를 참조하세요.

다음 내용으로 values.yaml 생성하세요:

mode: deployment
replicaCount: 1

image:
  repository: newrelic/nrdot-collector
  tag: "latest"
  pullPolicy: Always

serviceAccount:
  create: true
  name: otel-collector

podSecurityContext:
  runAsNonRoot: true
  runAsUser: 10001

securityContext:
  allowPrivilegeEscalation: false
  readOnlyRootFilesystem: true
  capabilities:
    drop:
      - ALL

resources:
  requests:
    memory: 512Mi
    cpu: 250m
  limits:
    memory: 1Gi
    cpu: 500m

extraEnvsFrom:
  - secretRef:
      name: newrelic-otlp-secret

# Disable unused default ports
ports:
  jaeger-compact:
    enabled: false
  jaeger-thrift:
    enabled: false
  jaeger-grpc:
    enabled: false
  zipkin:
    enabled: false

config:
  receivers:
    # Disable default receivers not needed in NRDOT
    jaeger: null
    zipkin: null

    # OTLP receiver: receives Kafka JMX metrics from broker pods (via Java agent) and app telemetry
    otlp:
      protocols:
        grpc:
          endpoint: "0.0.0.0:4317"

    # Kafka metrics receiver for consumer lag, topic, and partition metrics
    kafkametrics:
      brokers:
        # TODO#1: Replace with your Kafka bootstrap service DNS.
        # Format: <service-name>.<namespace>.svc.cluster.local:<port>
        - "kafka.kafka.svc.cluster.local:9092"
      collection_interval: 30s
      protocol_version: 2.0.0
      scrapers:
        - brokers
        - topics
        - consumers
      topic_match: "^[^_].*$"
      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

  exporters:
    otlp/newrelic:
      endpoint: ${NEW_RELIC_OTLP_ENDPOINT}
      tls:
        insecure: false
      sending_queue:
        num_consumers: 12
        queue_size: 5000
      retry_on_failure:
        enabled: true
      compression: gzip
      timeout: 30s
      headers:
        api-key: ${NEW_RELIC_LICENSE_KEY}

  processors:
    batch/aggregation:
      send_batch_size: 1024
      timeout: 30s

    resource:
      attributes:
        - action: insert
          key: kafka.cluster.name
          # TODO#2: Replace with your Kafka cluster name
          value: my-kafka-cluster

    transform/remove_broker_id:
      metric_statements:
        - context: resource
          statements:
            - delete_key(attributes, "broker.id")

    transform/remove_extra_attributes:
      metric_statements:
        - context: resource
          statements:
            - delete_matching_keys(attributes, "^process\\..*")
            - delete_matching_keys(attributes, "^telemetry\\..*")
            - delete_key(attributes, "host.arch")
            - delete_key(attributes, "os.description")
            - delete_matching_keys(attributes, "^cloud\\..*")
            - delete_key(attributes, "service.instance.id") where IsMatch(attributes["service.name"], "^unknown_service:")
            - delete_key(attributes, "service.name") where IsMatch(attributes["service.name"], "^unknown_service:")

    transform/des_units:
      metric_statements:
        - context: metric
          statements:
            - set(description, "") where description != ""
            - set(unit, "") where unit != ""

    filter/internal_topics:
      metrics:
        datapoint:
          - 'attributes["topic"] != nil and IsMatch(attributes["topic"], "^__.*")'

    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/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"

    cumulativetodelta:

    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
        - include: kafka.partition.replicas
          action: insert
          new_name: kafka.partition.replicas.total
          operations:
            - action: aggregate_labels
              label_set: [topic]
              aggregation_type: sum

    filter/remove_partition_level_replicas:
      metrics:
        exclude:
          match_type: strict
          metric_names:
            - kafka.partition.replicas_in_sync

    groupbyattrs/cluster:
      keys: [kafka.cluster.name]

    metricstransform/cluster_max:
      transforms:
        - include: "kafka\\.partition\\.offline|kafka\\.leader\\.election\\.rate|kafka\\.unclean\\.election\\.rate|kafka\\.partition\\.non_preferred_leader|kafka\\.broker\\.fenced\\.count|kafka\\.cluster\\.partition\\.count|kafka\\.cluster\\.topic\\.count"
          match_type: regexp
          action: update
          operations:
            - action: aggregate_labels
              aggregation_type: max
              label_set: []

  service:
    pipelines:
      # Null out the Helm chart's default pipelines — they reference the jaeger/zipkin
      # receivers we disabled above, which causes a startup error if left enabled.
      traces: null
      logs: null
      metrics: null

      # Broker metrics pipeline (excludes cluster-level metrics)
      metrics/broker:
        receivers: [otlp, kafkametrics]
        processors:
          - resource
          - filter/exclude_cluster_metrics
          - filter/internal_topics
          - transform/remove_extra_attributes
          - transform/des_units
          - cumulativetodelta
          - metricstransform/kafka_topic_sum_aggregation
          - filter/remove_partition_level_replicas
          - batch/aggregation
        exporters: [otlp/newrelic]

      # Cluster metrics pipeline (only cluster-level metrics, no broker.id)
      metrics/cluster:
        receivers: [otlp]
        processors:
          - resource
          - filter/include_cluster_metrics
          - transform/remove_broker_id
          - transform/remove_extra_attributes
          - transform/des_units
          - cumulativetodelta
          - groupbyattrs/cluster
          - metricstransform/cluster_max
          - batch/aggregation
        exporters: [otlp/newrelic]

      # APM traces pipeline (producer + consumer spans via OTel Java agent)
      traces/apps:
        receivers: [otlp]
        processors: [resource, batch/aggregation]
        exporters: [otlp/newrelic]

      # APM logs pipeline (producer + consumer logs via OTel Java agent)
      logs/apps:
        receivers: [otlp]
        processors: [resource, batch/aggregation]
        exporters: [otlp/newrelic]

extraObjects:
  - apiVersion: v1
    kind: ConfigMap
    metadata:
      name: kafka-jmx-config
      namespace: kafka  # TODO#3: Replace with your Kafka namespace
    data:
      kafka-jmx-config.yaml: |
        ---
        rules:
          # Per-topic custom metrics
          - 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 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
          - 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
          - 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}"

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

          - 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)

          - bean: java.lang:type=Memory
            unit: By
            prefix: jvm.memory.
            dropNegativeValues: true
            mapping:
              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

          - bean: java.lang:type=OperatingSystem
            prefix: jvm.
            dropNegativeValues: true
            mapping:
              SystemCpuLoad:
                metric: system.cpu.utilization
                type: gauge
                unit: '1'
                desc: Recent CPU utilization for whole system (0.0 to 1.0)

          - bean: kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec
            mapping:
              Count:
                metric: kafka.message.count
                type: counter
                desc: The number of messages received by the broker
                unit: "{message}"

          - bean: kafka.server:type=BrokerTopicMetrics,name=TotalFetchRequestsPerSec
            metricAttribute:
              type: const(fetch)
            mapping:
              Count:
                metric: &metric kafka.request.count
                type: &type counter
                desc: &desc The number of requests received by the broker
                unit: &unit "{request}"

          - bean: kafka.server:type=BrokerTopicMetrics,name=TotalProduceRequestsPerSec
            metricAttribute:
              type: const(produce)
            mapping:
              Count:
                metric: *metric
                type: *type
                desc: *desc
                unit: *unit

          - bean: kafka.server:type=BrokerTopicMetrics,name=FailedFetchRequestsPerSec
            metricAttribute:
              type: const(fetch)
            mapping:
              Count:
                metric: &metric kafka.request.failed
                type: &type counter
                desc: &desc The number of requests to the broker resulting in a failure
                unit: &unit "{request}"

          - bean: kafka.server:type=BrokerTopicMetrics,name=FailedProduceRequestsPerSec
            metricAttribute:
              type: const(produce)
            mapping:
              Count:
                metric: *metric
                type: *type
                desc: *desc
                unit: *unit

          - beans:
              - kafka.network:type=RequestMetrics,name=TotalTimeMs,request=Produce
              - kafka.network:type=RequestMetrics,name=TotalTimeMs,request=FetchConsumer
              - kafka.network:type=RequestMetrics,name=TotalTimeMs,request=FetchFollower
            metricAttribute:
              type: param(request)
            unit: ms
            mapping:
              99thPercentile:
                metric: kafka.request.time.99p
                type: gauge
                desc: The 99th percentile time the broker has taken to service requests

          - bean: kafka.network:type=RequestChannel,name=RequestQueueSize
            mapping:
              Value:
                metric: kafka.request.queue
                type: gauge
                desc: Size of the request queue
                unit: "{request}"

          - bean: kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec
            metricAttribute:
              direction: const(in)
            mapping:
              Count:
                metric: &metric kafka.network.io
                type: &type counter
                desc: &desc The bytes received or sent by the broker
                unit: &unit By

          - bean: kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec
            metricAttribute:
              direction: const(out)
            mapping:
              Count:
                metric: *metric
                type: *type
                desc: *desc
                unit: *unit

          - beans:
              - kafka.server:type=DelayedOperationPurgatory,name=PurgatorySize,delayedOperation=Produce
              - kafka.server:type=DelayedOperationPurgatory,name=PurgatorySize,delayedOperation=Fetch
            metricAttribute:
              type: param(delayedOperation)
            mapping:
              Value:
                metric: kafka.purgatory.size
                type: gauge
                desc: The number of requests waiting in purgatory
                unit: "{request}"

          - bean: kafka.server:type=ReplicaManager,name=PartitionCount
            mapping:
              Value:
                metric: kafka.partition.count
                type: gauge
                desc: The number of partitions on the broker
                unit: "{partition}"

          - bean: kafka.controller:type=KafkaController,name=OfflinePartitionsCount
            mapping:
              Value:
                metric: kafka.partition.offline
                type: gauge
                desc: The number of partitions offline
                unit: "{partition}"

          - bean: kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions
            mapping:
              Value:
                metric: kafka.partition.under_replicated
                type: gauge
                desc: The number of under replicated partitions
                unit: "{partition}"

          - bean: kafka.server:type=ReplicaManager,name=IsrShrinksPerSec
            metricAttribute:
              operation: const(shrink)
            mapping:
              Count:
                metric: kafka.isr.operation.count
                type: counter
                desc: The number of in-sync replica shrink and expand operations
                unit: "{operation}"

          - bean: kafka.server:type=ReplicaManager,name=IsrExpandsPerSec
            metricAttribute:
              operation: const(expand)
            mapping:
              Count:
                metric: kafka.isr.operation.count
                type: counter
                desc: The number of in-sync replica shrink and expand operations
                unit: "{operation}"

          - bean: kafka.server:type=ReplicaFetcherManager,name=MaxLag,clientId=Replica
            mapping:
              Value:
                metric: kafka.max.lag
                type: gauge
                desc: The max lag in messages between follower and leader replicas
                unit: "{message}"

          - bean: kafka.controller:type=KafkaController,name=ActiveControllerCount
            mapping:
              Value:
                metric: kafka.controller.active.count
                type: gauge
                desc: Number of active controllers in the cluster
                unit: "{controller}"

          - bean: kafka.controller:type=ControllerStats,name=LeaderElectionRateAndTimeMs
            mapping:
              Count:
                metric: kafka.leader.election.rate
                type: counter
                desc: The leader election count
                unit: "{election}"

          - bean: kafka.controller:type=ControllerStats,name=UncleanLeaderElectionsPerSec
            mapping:
              Count:
                metric: kafka.unclean.election.rate
                type: counter
                desc: Unclean leader election count
                unit: "{election}"

          # ── Additional metrics — remove this section to reduce data ingest ───────────

          - beans:
              - kafka.network:type=RequestMetrics,name=TotalTimeMs,request=Produce
              - kafka.network:type=RequestMetrics,name=TotalTimeMs,request=FetchConsumer
              - kafka.network:type=RequestMetrics,name=TotalTimeMs,request=FetchFollower
            metricAttribute:
              type: param(request)
            unit: ms
            mapping:
              Count:
                metric: kafka.request.time.total
                type: counter
                desc: The total time the broker has taken to service requests
              50thPercentile:
                metric: kafka.request.time.50p
                type: gauge
                desc: The 50th percentile time the broker has taken to service requests
              Mean:
                metric: kafka.request.time.avg
                type: gauge
                desc: The average time the broker has taken to service requests

          - bean: kafka.log:type=LogFlushStats,name=LogFlushRateAndTimeMs
            unit: ms
            type: gauge
            prefix: kafka.logs.flush.
            mapping:
              Count:
                metric: count
                unit: '{flush}'
                type: counter
                desc: Log flush count
              50thPercentile:
                metric: time.50p
                desc: Log flush time - 50th percentile
              99thPercentile:
                metric: time.99p
                desc: Log flush time - 99th percentile

          - bean: java.lang:type=GarbageCollector,name=*
            mapping:
              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=ClassLoading
            mapping:
              LoadedClassCount:
                metric: jvm.class.count
                type: gauge
                unit: "{class}"
                desc: Currently loaded class count

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

          - 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)
              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)
              OpenFileDescriptorCount:
                metric: file_descriptor.count
                type: gauge
                unit: "{file_descriptor}"
                desc: Number of open file descriptors

          - 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
              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

구성 매개변수

다음 표에서는 주요 설정 시위에 대해 설명합니다.

매개변수	설명
`config.receivers.kafkametrics.brokers`	Kafka 부트스트랩 서비스 DNS(예: `kafka.kafka.svc.cluster.local:9092`)로 교체합니다.
`config.processors.resource.attributes[kafka.cluster.name]`	사용자의 Kafka 클러스터 이름으로 교체하십시오.
`extraObjects[0].metadata.namespace`	Kafka 네임스페이스(`extraObjects` ConfigMap 내)로 교체하십시오
`resources.limits` 그리고 `resources.requests`	워크로드 요구 사항에 따라 조정하십시오.

유연성을 극대화하고 공급업체 비종속 구현, 배포를 위해 커뮤니티 OpenTelemetry Collector 사용하세요.

위의 NRDOT 옵션과 동일한 내용으로 values.yaml 을(를) 생성하되, 이미지를 변경하십시오:

image:
  repository: otel/opentelemetry-collector-contrib
  tag: "latest"
  pullPolicy: Always

다른 모든 설정(수신기, 프로세서, 파이프라인 및 extraObjects)은 동일합니다.

설정 파라미터: 위의 NRDOT 옵션과 동일한 파라미터입니다. 리소스 제한을 포함한 자세한 내용은 설정 파라미터 표를 참조하십시오.

고급 설정 옵션은 다음을 참조하세요:

OTLP 수신기 문서

Kafka 메트릭 수신자 문서

3단계. Helm으로 OpenTelemetry Collector 설치

bash

$helm repo add open-telemetry https://open-telemetry.github.io/opentelemetry-helm-charts
$helm upgrade kafka-monitoring open-telemetry/opentelemetry-collector \
>  --install \
>  --namespace newrelic \
>  --create-namespace \
>  -f values.yaml

4단계. 배포 확인

bash

$# Check pod status
$kubectl get pods -n newrelic -l app.kubernetes.io/name=opentelemetry-collector
$
$# View logs to verify metrics are being received from broker pods
$kubectl logs -n newrelic -l app.kubernetes.io/name=opentelemetry-collector --tail=50

1단계. 뉴렐릭 자격 증명 시크릿 생성

bash

$kubectl create secret generic newrelic-otlp-secret \
>  --namespace newrelic \
>  --from-literal=NEW_RELIC_LICENSE_KEY='your-license-key-here' \
>  --from-literal=NEW_RELIC_OTLP_ENDPOINT='https://otlp.nr-data.net:4317'

bash

$kubectl create secret generic newrelic-otlp-secret \
>  --namespace newrelic \
>  --from-literal=NEW_RELIC_LICENSE_KEY='your-license-key-here' \
>  --from-literal=NEW_RELIC_OTLP_ENDPOINT='https://otlp.eu01.nr-data.net:4317'

팁

다른 엔드포인트 설정에 대해서는 OTLP 엔드포인트 구성을 참조하세요.

2단계. 매니페스트 파일 생성

NRDOT 및 OpenTelemetry 수집기는 모두 동일한 설정을 사용합니다. 컨테이너 이미지만 다릅니다. 또한 둘 다 Kafka 네임스페이스에 적용된 kafka-jmx-config ConfigMap이 필요합니다.

kafka-jmx-config.yaml생성 - 자바 에이전트용 JMX 메트릭 설정(Kafka 네임스페이스에 적용):

apiVersion: v1
kind: ConfigMap
metadata:
  name: kafka-jmx-config
  namespace: kafka  # TODO: Replace with your Kafka namespace
data:
  kafka-jmx-config.yaml: |
    ---
    rules:
      # Per-topic custom metrics
      - 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 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
      - 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
      - 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}"

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

      - 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)

      - bean: java.lang:type=Memory
        unit: By
        prefix: jvm.memory.
        dropNegativeValues: true
        mapping:
          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

      - bean: java.lang:type=OperatingSystem
        prefix: jvm.
        dropNegativeValues: true
        mapping:
          SystemCpuLoad:
            metric: system.cpu.utilization
            type: gauge
            unit: '1'
            desc: Recent CPU utilization for whole system (0.0 to 1.0)

      - bean: kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec
        mapping:
          Count:
            metric: kafka.message.count
            type: counter
            desc: The number of messages received by the broker
            unit: "{message}"

      - bean: kafka.server:type=BrokerTopicMetrics,name=TotalFetchRequestsPerSec
        metricAttribute:
          type: const(fetch)
        mapping:
          Count:
            metric: &metric kafka.request.count
            type: &type counter
            desc: &desc The number of requests received by the broker
            unit: &unit "{request}"

      - bean: kafka.server:type=BrokerTopicMetrics,name=TotalProduceRequestsPerSec
        metricAttribute:
          type: const(produce)
        mapping:
          Count:
            metric: *metric
            type: *type
            desc: *desc
            unit: *unit

      - bean: kafka.server:type=BrokerTopicMetrics,name=FailedFetchRequestsPerSec
        metricAttribute:
          type: const(fetch)
        mapping:
          Count:
            metric: &metric kafka.request.failed
            type: &type counter
            desc: &desc The number of requests to the broker resulting in a failure
            unit: &unit "{request}"

      - bean: kafka.server:type=BrokerTopicMetrics,name=FailedProduceRequestsPerSec
        metricAttribute:
          type: const(produce)
        mapping:
          Count:
            metric: *metric
            type: *type
            desc: *desc
            unit: *unit

      - beans:
          - kafka.network:type=RequestMetrics,name=TotalTimeMs,request=Produce
          - kafka.network:type=RequestMetrics,name=TotalTimeMs,request=FetchConsumer
          - kafka.network:type=RequestMetrics,name=TotalTimeMs,request=FetchFollower
        metricAttribute:
          type: param(request)
        unit: ms
        mapping:
          99thPercentile:
            metric: kafka.request.time.99p
            type: gauge
            desc: The 99th percentile time the broker has taken to service requests

      - bean: kafka.network:type=RequestChannel,name=RequestQueueSize
        mapping:
          Value:
            metric: kafka.request.queue
            type: gauge
            desc: Size of the request queue
            unit: "{request}"

      - bean: kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec
        metricAttribute:
          direction: const(in)
        mapping:
          Count:
            metric: &metric kafka.network.io
            type: &type counter
            desc: &desc The bytes received or sent by the broker
            unit: &unit By

      - bean: kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec
        metricAttribute:
          direction: const(out)
        mapping:
          Count:
            metric: *metric
            type: *type
            desc: *desc
            unit: *unit

      - beans:
          - kafka.server:type=DelayedOperationPurgatory,name=PurgatorySize,delayedOperation=Produce
          - kafka.server:type=DelayedOperationPurgatory,name=PurgatorySize,delayedOperation=Fetch
        metricAttribute:
          type: param(delayedOperation)
        mapping:
          Value:
            metric: kafka.purgatory.size
            type: gauge
            desc: The number of requests waiting in purgatory
            unit: "{request}"

      - bean: kafka.server:type=ReplicaManager,name=PartitionCount
        mapping:
          Value:
            metric: kafka.partition.count
            type: gauge
            desc: The number of partitions on the broker
            unit: "{partition}"

      - bean: kafka.controller:type=KafkaController,name=OfflinePartitionsCount
        mapping:
          Value:
            metric: kafka.partition.offline
            type: gauge
            desc: The number of partitions offline
            unit: "{partition}"

      - bean: kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions
        mapping:
          Value:
            metric: kafka.partition.under_replicated
            type: gauge
            desc: The number of under replicated partitions
            unit: "{partition}"

      - bean: kafka.server:type=ReplicaManager,name=IsrShrinksPerSec
        metricAttribute:
          operation: const(shrink)
        mapping:
          Count:
            metric: kafka.isr.operation.count
            type: counter
            desc: The number of in-sync replica shrink and expand operations
            unit: "{operation}"

      - bean: kafka.server:type=ReplicaManager,name=IsrExpandsPerSec
        metricAttribute:
          operation: const(expand)
        mapping:
          Count:
            metric: kafka.isr.operation.count
            type: counter
            desc: The number of in-sync replica shrink and expand operations
            unit: "{operation}"

      - bean: kafka.server:type=ReplicaFetcherManager,name=MaxLag,clientId=Replica
        mapping:
          Value:
            metric: kafka.max.lag
            type: gauge
            desc: The max lag in messages between follower and leader replicas
            unit: "{message}"

      - bean: kafka.controller:type=KafkaController,name=ActiveControllerCount
        mapping:
          Value:
            metric: kafka.controller.active.count
            type: gauge
            desc: Number of active controllers in the cluster
            unit: "{controller}"

      - bean: kafka.controller:type=ControllerStats,name=LeaderElectionRateAndTimeMs
        mapping:
          Count:
            metric: kafka.leader.election.rate
            type: counter
            desc: The leader election count
            unit: "{election}"

      - bean: kafka.controller:type=ControllerStats,name=UncleanLeaderElectionsPerSec
        mapping:
          Count:
            metric: kafka.unclean.election.rate
            type: counter
            desc: Unclean leader election count
            unit: "{election}"

      # ── Additional metrics — remove this section to reduce data ingest ───────────

      - beans:
          - kafka.network:type=RequestMetrics,name=TotalTimeMs,request=Produce
          - kafka.network:type=RequestMetrics,name=TotalTimeMs,request=FetchConsumer
          - kafka.network:type=RequestMetrics,name=TotalTimeMs,request=FetchFollower
        metricAttribute:
          type: param(request)
        unit: ms
        mapping:
          Count:
            metric: kafka.request.time.total
            type: counter
            desc: The total time the broker has taken to service requests
          50thPercentile:
            metric: kafka.request.time.50p
            type: gauge
            desc: The 50th percentile time the broker has taken to service requests
          Mean:
            metric: kafka.request.time.avg
            type: gauge
            desc: The average time the broker has taken to service requests

      - bean: kafka.log:type=LogFlushStats,name=LogFlushRateAndTimeMs
        unit: ms
        type: gauge
        prefix: kafka.logs.flush.
        mapping:
          Count:
            metric: count
            unit: '{flush}'
            type: counter
            desc: Log flush count
          50thPercentile:
            metric: time.50p
            desc: Log flush time - 50th percentile
          99thPercentile:
            metric: time.99p
            desc: Log flush time - 99th percentile

      - bean: java.lang:type=GarbageCollector,name=*
        mapping:
          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=ClassLoading
        mapping:
          LoadedClassCount:
            metric: jvm.class.count
            type: gauge
            unit: "{class}"
            desc: Currently loaded class count

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

      - 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)
          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)
          OpenFileDescriptorCount:
            metric: file_descriptor.count
            type: gauge
            unit: "{file_descriptor}"
            desc: Number of open file descriptors

      - 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
          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

1. collector-configmap.yaml생성 - OpenTelemetry Collector 설정:

---
apiVersion: v1
kind: ConfigMap
metadata:
  name: otel-collector-config
  namespace: newrelic
  labels:
    app: otel-collector
data:
  otel-collector-config.yaml: |
    receivers:
      otlp:
        protocols:
          grpc:
            endpoint: "0.0.0.0:4317"

      kafkametrics:
        brokers:
          # TODO#1: Replace with your Kafka bootstrap service DNS
          - "kafka.kafka.svc.cluster.local:9092"
        collection_interval: 30s
        protocol_version: 2.0.0
        scrapers:
          - brokers
          - topics
          - consumers
        topic_match: "^[^_].*$"
        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

    exporters:
      otlp/newrelic:
        endpoint: ${NEW_RELIC_OTLP_ENDPOINT}
        tls:
          insecure: false
        sending_queue:
          num_consumers: 12
          queue_size: 5000
        retry_on_failure:
          enabled: true
        compression: gzip
        timeout: 30s
        headers:
          api-key: ${NEW_RELIC_LICENSE_KEY}

    processors:
      batch/aggregation:
        send_batch_size: 1024
        timeout: 30s
      resource:
        attributes:
        - action: insert
          key: kafka.cluster.name
          # TODO#2: Replace with your Kafka cluster name
          value: my-kafka-cluster
      transform/remove_broker_id:
        metric_statements:
        - context: resource
          statements:
          - delete_key(attributes, "broker.id")
      transform/remove_extra_attributes:
        metric_statements:
        - context: resource
          statements:
          - delete_matching_keys(attributes, "^process\\..*")
          - delete_matching_keys(attributes, "^telemetry\\..*")
          - delete_key(attributes, "host.arch")
          - delete_key(attributes, "os.description")
          - delete_matching_keys(attributes, "^cloud\\..*")
          - delete_key(attributes, "service.instance.id") where IsMatch(attributes["service.name"], "^unknown_service:")
          - delete_key(attributes, "service.name") where IsMatch(attributes["service.name"], "^unknown_service:")
      transform/des_units:
        metric_statements:
        - context: metric
          statements:
          - set(description, "") where description != ""
          - set(unit, "") where unit != ""
      filter/internal_topics:
        metrics:
          datapoint:
            - 'attributes["topic"] != nil and IsMatch(attributes["topic"], "^__.*")'
      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/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"
      cumulativetodelta:
      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
        - include: kafka.partition.replicas
          action: insert
          new_name: kafka.partition.replicas.total
          operations:
          - action: aggregate_labels
            label_set: [topic]
            aggregation_type: sum
      filter/remove_partition_level_replicas:
        metrics:
          exclude:
            match_type: strict
            metric_names:
            - kafka.partition.replicas_in_sync
      groupbyattrs/cluster:
        keys: [kafka.cluster.name]
      metricstransform/cluster_max:
        transforms:
          - include: "kafka\\.partition\\.offline|kafka\\.leader\\.election\\.rate|kafka\\.unclean\\.election\\.rate|kafka\\.partition\\.non_preferred_leader|kafka\\.broker\\.fenced\\.count|kafka\\.cluster\\.partition\\.count|kafka\\.cluster\\.topic\\.count"
            match_type: regexp
            action: update
            operations:
              - action: aggregate_labels
                aggregation_type: max
                label_set: []

    service:
      pipelines:
        metrics/broker:
          receivers: [otlp, kafkametrics]
          processors:
            - resource
            - filter/exclude_cluster_metrics
            - filter/internal_topics
            - transform/remove_extra_attributes
            - transform/des_units
            - cumulativetodelta
            - metricstransform/kafka_topic_sum_aggregation
            - filter/remove_partition_level_replicas
            - batch/aggregation
          exporters: [otlp/newrelic]
        metrics/cluster:
          receivers: [otlp]
          processors:
            - resource
            - filter/include_cluster_metrics
            - transform/remove_broker_id
            - transform/remove_extra_attributes
            - transform/des_units
            - cumulativetodelta
            - groupbyattrs/cluster
            - metricstransform/cluster_max
            - batch/aggregation
          exporters: [otlp/newrelic]
        traces/apps:
          receivers: [otlp]
          processors: [resource, batch/aggregation]
          exporters: [otlp/newrelic]
        logs/apps:
          receivers: [otlp]
          processors: [resource, batch/aggregation]
          exporters: [otlp/newrelic]

2. collector-deployment.yaml생성 - ServiceAccount 및 Service를 사용한 배포:

---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: otel-collector
  namespace: newrelic
  labels:
    app: otel-collector
---
apiVersion: v1
kind: Service
metadata:
  name: otel-collector
  namespace: newrelic
  labels:
    app: otel-collector
spec:
  selector:
    app: otel-collector
  ports:
  - name: otlp-grpc
    port: 4317
    targetPort: 4317
    protocol: TCP
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: otel-collector
  namespace: newrelic
  labels:
    app: otel-collector
spec:
  replicas: 1
  selector:
    matchLabels:
      app: otel-collector
  template:
    metadata:
      labels:
        app: otel-collector
    spec:
      serviceAccountName: otel-collector
      containers:
      - name: otel-collector
        image: newrelic/nrdot-collector:latest
        command:
        - "/nrdot-collector"
        - "--config=/conf/otel-collector-config.yaml"
        env:
        - name: NEW_RELIC_LICENSE_KEY
          valueFrom:
            secretKeyRef:
              name: newrelic-otlp-secret
              key: NEW_RELIC_LICENSE_KEY
        - name: NEW_RELIC_OTLP_ENDPOINT
          valueFrom:
            secretKeyRef:
              name: newrelic-otlp-secret
              key: NEW_RELIC_OTLP_ENDPOINT
        - name: GOGC
          value: "80"
        ports:
        - name: otlp-grpc
          containerPort: 4317
          protocol: TCP
        resources:
          limits:
            cpu: "1000m"
            memory: "1Gi"
          requests:
            cpu: "200m"
            memory: "512Mi"
        volumeMounts:
        - name: config
          mountPath: /conf
      volumes:
      - name: config
        configMap:
          name: otel-collector-config
          items:
          - key: otel-collector-config.yaml
            path: otel-collector-config.yaml

구성 매개변수

다음 표에서는 주요 설정 시위에 대해 설명합니다.

매개변수	설명
`receivers.kafkametrics.brokers`	사용자의 Kafka 부트스트랩 서비스 DNS로 교체하십시오
`processors.resource.attributes[kafka.cluster.name]`	사용자의 Kafka 클러스터 이름으로 교체하십시오.
`resources.limits` 및 `resources.requests` (`collector-deployment.yaml`에서)	워크로드 요구 사항에 따라 조정하십시오.

공급업체 비종속 구현, 배포를 위해 커뮤니티 OpenTelemetry Collector 사용하세요.

1. collector-configmap.yaml생성 - 위의 NRDOT 옵션과 동일합니다(설정도 동일).

2. collector-deployment.yaml생성 - 컨테이너 이미지와 명령어만 다릅니다:

---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: otel-collector
  namespace: newrelic
  labels:
    app: otel-collector
---
apiVersion: v1
kind: Service
metadata:
  name: otel-collector
  namespace: newrelic
  labels:
    app: otel-collector
spec:
  selector:
    app: otel-collector
  ports:
  - name: otlp-grpc
    port: 4317
    targetPort: 4317
    protocol: TCP
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: otel-collector
  namespace: newrelic
  labels:
    app: otel-collector
spec:
  replicas: 1
  selector:
    matchLabels:
      app: otel-collector
  template:
    metadata:
      labels:
        app: otel-collector
    spec:
      serviceAccountName: otel-collector
      containers:
      - name: otel-collector
        image: otel/opentelemetry-collector-contrib:latest
        command:
        - "/otelcol-contrib"
        - "--config=/conf/otel-collector-config.yaml"
        env:
        - name: NEW_RELIC_LICENSE_KEY
          valueFrom:
            secretKeyRef:
              name: newrelic-otlp-secret
              key: NEW_RELIC_LICENSE_KEY
        - name: NEW_RELIC_OTLP_ENDPOINT
          valueFrom:
            secretKeyRef:
              name: newrelic-otlp-secret
              key: NEW_RELIC_OTLP_ENDPOINT
        - name: GOGC
          value: "80"
        ports:
        - name: otlp-grpc
          containerPort: 4317
          protocol: TCP
        resources:
          limits:
            cpu: "1000m"
            memory: "1Gi"
          requests:
            cpu: "200m"
            memory: "512Mi"
        volumeMounts:
        - name: config
          mountPath: /conf
      volumes:
      - name: config
        configMap:
          name: otel-collector-config
          items:
          - key: otel-collector-config.yaml
            path: otel-collector-config.yaml

설정 파라미터: 위의 NRDOT 옵션과 동일한 파라미터입니다. 리소스 제한을 포함한 자세한 내용은 설정 파라미터 표를 참조하십시오.

3단계. 매니페스트 배포

bash

$# Create namespace if it doesn't exist
$kubectl create namespace newrelic --dry-run=client -o yaml | kubectl apply -f -
$
$# Apply JMX ConfigMap to the Kafka namespace
$kubectl apply -f kafka-jmx-config.yaml
$
$# Apply collector ConfigMap
$kubectl apply -f collector-configmap.yaml
$
$# Apply Deployment and Service
$kubectl apply -f collector-deployment.yaml

4단계. 배포 확인

bash

$# Check pod status
$kubectl get pods -n newrelic -l app=otel-collector
$
$# View logs to verify metrics are being received from broker pods
$kubectl logs -n newrelic -l app=otel-collector --tail=50

자바 에이전트용 Kafka StatefulSet 구성

이제 수집기가 실행 중이므로, Kafka StatefulSet을 패치하여 OpenTelemetry 자바 에이전트 JAR을 다운로드하는 init 컨테이너를 추가한 다음, KAFKA_OPTS을(를) 통해 이를 Kafka 브로커 JVM에 연결합니다.

기존 Kafka StatefulSet 매니페스트에 다음 섹션을 추가합니다:

spec:
  template:
    spec:
      # 1. Init container: downloads OTel Java agent JAR before Kafka starts
      initContainers:
        - name: download-otel-agent
          image: busybox:latest
          command:
            - sh
            - -c
            - |
              wget -O /otel-agent/opentelemetry-javaagent.jar \
                https://github.com/open-telemetry/opentelemetry-java-instrumentation/releases/latest/download/opentelemetry-javaagent.jar
          volumeMounts:
            - name: otel-agent
              mountPath: /otel-agent

      containers:
        - name: kafka  # TODO: Replace with your Kafka container name
          # 2. Attach OTel Java agent to the Kafka broker JVM
          env:
            - name: KAFKA_OPTS
              value: >-
                -javaagent:/otel-agent/opentelemetry-javaagent.jar
                -Dotel.jmx.enabled=true
                -Dotel.jmx.config=/jmx-config/kafka-jmx-config.yaml
                -Dotel.resource.attributes=kafka.cluster.name=my-kafka-cluster
                -Dotel.exporter.otlp.endpoint=http://otel-collector.newrelic.svc.cluster.local:4317
                -Dotel.exporter.otlp.protocol=grpc
                -Dotel.metrics.exporter=otlp
                -Dotel.logs.exporter=otlp
                -Dotel.instrumentation.runtime-telemetry.enabled=false
                -Dotel.metric.export.interval=30000
          volumeMounts:
            - name: otel-agent
              mountPath: /otel-agent
            - name: jmx-config
              mountPath: /jmx-config

      # 3. Volumes: emptyDir for JAR, ConfigMap for JMX rules
      volumes:
        - name: otel-agent
          emptyDir: {}
        - name: jmx-config
          configMap:
            name: kafka-jmx-config  # Deployed with the collector in the previous step

팁

이전 단계에서 수집기와 함께 kafka-jmx-config ConfigMap이 배포되었습니다. otel.exporter.otlp.endpoint 값 http://otel-collector.newrelic.svc.cluster.local:4317 은(는) 수집기가 newrelic 네임스페이스에 서비스 이름 otel-collector(으)로 배포된다고 가정합니다. 다를 경우 실제 수집기 서비스 DNS와 일치하도록 업데이트하십시오.

매개변수	설명
`javaagent`	OpenTelemetry 자바 에이전트를 Kafka 브로커 JVM에 연결합니다.
`jmx.enabled=true`	JMX 메트릭 수집을 활성화합니다.
`jmx.config`	사용자 지정 JMX 메트릭 설정 파일(ConfigMap에서 마운트됨)을 가리킵니다
`resource.attributes`	모든 메트릭에 `kafka.cluster.name` 메타데이터를 추가합니다
`otlp.endpoint`	클러스터의 OpenTelemetry Collector 서비스를 가리킵니다.
`otlp.protocol=grpc`	OTLP에 gRPC 프로토콜을 사용합니다.
`metrics.exporter=otlp`	OTLP를 통해 메트릭을 전송합니다.
`logs.exporter=otlp`	브로커 로그 수집을 활성화합니다. 비활성화하려면 `none` (으)로 설정합니다.
`metric.export.interval`	메트릭 내보내기 시도 간의 간격을 밀리초 단위로 설정합니다(예: `30000` (30초))

전체 설정 옵션은 자바 에이전트 설정 가이드를 참조하세요.

업데이트된 StatefulSet을 적용하고 파드가 롤링될 때까지 기다립니다:

bash

$kubectl apply -f kafka-statefulset.yaml
$kubectl rollout status statefulset/kafka -n kafka  # TODO: Replace with your StatefulSet name and namespace

(선택사항) 제작자 또는 소비자를 위해

중요

언어 지원: 현재 OpenTelemetry 자바 에이전트를 사용한 Kafka 클라이언트 계측에는 자바 애플리케이션만 지원됩니다.

Kubernetes에서 실행되는 Kafka 생산자 및 소비자 애플리케이션에서 애플리케이션 수준의 텔레메트리를 수집하려면 해당 애플리케이션 파드에 OpenTelemetry 자바 에이전트를 추가하세요.

애플리케이션 배포에 init 컨테이너와 환경 변수를 추가합니다:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: kafka-producer-app
spec:
  template:
    spec:
      initContainers:
      - name: download-otel-agent
        image: busybox:latest
        command:
        - sh
        - -c
        - wget -O /otel-agent/opentelemetry-javaagent.jar https://github.com/open-telemetry/opentelemetry-java-instrumentation/releases/latest/download/opentelemetry-javaagent.jar
        volumeMounts:
        - name: otel-agent
          mountPath: /otel-agent

      containers:
      - name: app
        image: your-kafka-app:latest
        env:
        - name: JAVA_TOOL_OPTIONS
          value: >-
            -javaagent:/otel-agent/opentelemetry-javaagent.jar
            -Dotel.service.name=order-process-service
            -Dotel.resource.attributes=kafka.cluster.name=my-kafka-cluster
            -Dotel.exporter.otlp.endpoint=http://otel-collector.newrelic.svc.cluster.local: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
            -Dotel.instrumentation.runtime-telemetry.enabled=false
        volumeMounts:
        - name: otel-agent
          mountPath: /otel-agent

      volumes:
      - name: otel-agent
        emptyDir: {}

구성 매개변수

다음 표에서는 주요 설정 시위에 대해 설명합니다.

매개변수	설명
`order-process-service`	생산자 또는 소비자 애플리케이션의 고유한 이름으로 바꾸십시오.
`my-kafka-cluster`	브로커 설정에 사용된 것과 동일한 클러스터 이름으로 교체하십시오.
`otel-collector.newrelic.svc.cluster.local`	수집기 서비스의 실제 DNS 이름( `<service-name>.<namespace>.svc.cluster.local` )으로 바꿉니다

매개변수

설명

order-process-service

생산자 또는 소비자 애플리케이션의 고유한 이름으로 바꾸십시오.

my-kafka-cluster

브로커 설정에 사용된 것과 동일한 클러스터 이름으로 교체하십시오.

otel-collector.newrelic.svc.cluster.local

수집기 서비스의 실제 DNS 이름(

<service-name>.<namespace>.svc.cluster.local

)으로 바꿉니다

자바 에이전트는 코드 변경 없이 즉시 사용 가능한 Kafka 계측 을 제공하여 요청 지연시간, 처리량 메트릭, 오류율 및 분산 트레이스를 캡처합니다. 고급 설정에 대해서는 Kafka 계측 문서를 참조하세요.

다음 단계에 따라 브로커 파드에 Prometheus JMX Exporter를 설치하고, 메트릭을 수집하여 뉴렐릭으로 전송하는 수집기를 배포하여 포괄적인 Kafka 모니터링을 설정하십시오.

시작하기 전에

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

뉴렐릭 계정
kubectl 액세스 권한이 있는 쿠버네티스 클러스터
헤드리스 서비스(안정적인 파드 DNS 이름을 위해)와 함께 StatefulSet으로 배포된 Kafka
Kafka StatefulSet을 수정하고 재배포할 수 있는 기능

JMX 메트릭 ConfigMap을 생성합니다.

수집할 Kafka 메트릭을 정의하는 JMX Exporter 설정이 포함된 ConfigMap을 생성합니다. 이 ConfigMap은 각 Kafka 브로커 파드에 마운트됩니다.

kafka-jmx-config.yaml(으)로 저장합니다. Kafka가 배포된 네임스페이스에 적용합니다:

apiVersion: v1
kind: ConfigMap
metadata:
  name: kafka-jmx-metrics
  namespace: kafka  # TODO: Replace with your Kafka namespace
data:
  kafka-metrics-config.yml: |
    startDelaySeconds: 0
    lowercaseOutputName: true
    lowercaseOutputLabelNames: true

    rules:
      # Cluster-level controller metrics
      - pattern: 'kafka.controller<type=KafkaController, name=GlobalTopicCount><>Value'
        name: kafka_cluster_topic_count
        type: GAUGE

      - pattern: 'kafka.controller<type=KafkaController, name=GlobalPartitionCount><>Value'
        name: kafka_cluster_partition_count
        type: GAUGE

      - pattern: 'kafka.controller<type=KafkaController, name=FencedBrokerCount><>Value'
        name: kafka_broker_fenced_count
        type: GAUGE

      - pattern: 'kafka.controller<type=KafkaController, name=PreferredReplicaImbalanceCount><>Value'
        name: kafka_partition_non_preferred_leader
        type: GAUGE

      - pattern: 'kafka.controller<type=KafkaController, name=OfflinePartitionsCount><>Value'
        name: kafka_partition_offline
        type: GAUGE

      - pattern: 'kafka.controller<type=KafkaController, name=ActiveControllerCount><>Value'
        name: kafka_controller_active_count
        type: GAUGE

      # Broker-level replica metrics
      - pattern: 'kafka.server<type=ReplicaManager, name=UnderMinIsrPartitionCount><>Value'
        name: kafka_partition_under_min_isr
        type: GAUGE

      - pattern: 'kafka.server<type=ReplicaManager, name=LeaderCount><>Value'
        name: kafka_broker_leader_count
        type: GAUGE

      - pattern: 'kafka.server<type=ReplicaManager, name=PartitionCount><>Value'
        name: kafka_partition_count
        type: GAUGE

      - pattern: 'kafka.server<type=ReplicaManager, name=UnderReplicatedPartitions><>Value'
        name: kafka_partition_under_replicated
        type: GAUGE

      - pattern: 'kafka.server<type=ReplicaManager, name=IsrShrinksPerSec><>Count'
        name: kafka_isr_operation_count
        type: COUNTER
        labels:
          operation: "shrink"

      - pattern: 'kafka.server<type=ReplicaManager, name=IsrExpandsPerSec><>Count'
        name: kafka_isr_operation_count
        type: COUNTER
        labels:
          operation: "expand"

      - pattern: 'kafka.server<type=ReplicaFetcherManager, name=MaxLag, clientId=Replica><>Value'
        name: kafka_max_lag
        type: GAUGE

      # Broker topic metrics (totals)
      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=MessagesInPerSec><>Count'
        name: kafka_message_count
        type: COUNTER

      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=TotalFetchRequestsPerSec><>Count'
        name: kafka_request_count
        type: COUNTER
        labels:
          type: "fetch"

      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=TotalProduceRequestsPerSec><>Count'
        name: kafka_request_count
        type: COUNTER
        labels:
          type: "produce"

      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=FailedFetchRequestsPerSec><>Count'
        name: kafka_request_failed
        type: COUNTER
        labels:
          type: "fetch"

      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=FailedProduceRequestsPerSec><>Count'
        name: kafka_request_failed
        type: COUNTER
        labels:
          type: "produce"

      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=BytesInPerSec><>Count'
        name: kafka_network_io
        type: COUNTER
        labels:
          direction: "in"

      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=BytesOutPerSec><>Count'
        name: kafka_network_io
        type: COUNTER
        labels:
          direction: "out"

      # Per-topic metrics (only appear after traffic flows)
      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=MessagesInPerSec, topic=(.+)><>Count'
        name: kafka_prod_msg_count
        type: COUNTER
        labels:
          topic: "$1"

      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=BytesInPerSec, topic=(.+)><>Count'
        name: kafka_topic_io
        type: COUNTER
        labels:
          topic: "$1"
          direction: "in"

      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=BytesOutPerSec, topic=(.+)><>Count'
        name: kafka_topic_io
        type: COUNTER
        labels:
          topic: "$1"
          direction: "out"

      # Request metrics
      - pattern: 'kafka.network<type=RequestMetrics, name=TotalTimeMs, request=(Produce|FetchConsumer|FetchFollower)><>99thPercentile'
        name: kafka_request_time_99p
        type: GAUGE
        labels:
          type: "$1"

      - pattern: 'kafka.network<type=RequestChannel, name=RequestQueueSize><>Value'
        name: kafka_request_queue
        type: GAUGE

      - pattern: 'kafka.server<type=DelayedOperationPurgatory, name=PurgatorySize, delayedOperation=(.+)><>Value'
        name: kafka_purgatory_size
        type: GAUGE
        labels:
          type: "$1"

      # Controller stats
      - pattern: 'kafka.controller<type=ControllerStats, name=LeaderElectionRateAndTimeMs><>Count'
        name: kafka_leader_election_rate
        type: COUNTER

      - pattern: 'kafka.controller<type=ControllerStats, name=UncleanLeaderElectionsPerSec><>Count'
        name: kafka_unclean_election_rate
        type: COUNTER

      # JVM Garbage Collection
      - pattern: 'java.lang<name=(.+), type=GarbageCollector><>CollectionCount'
        name: jvm_gc_collections_count
        type: COUNTER
        labels:
          name: "$1"

      # JVM Memory
      - pattern: 'java.lang<type=Memory><HeapMemoryUsage>max'
        name: jvm_memory_heap_max
        type: GAUGE

      - pattern: 'java.lang<type=Memory><HeapMemoryUsage>used'
        name: jvm_memory_heap_used
        type: GAUGE

      # JVM Threading and System
      - pattern: 'java.lang<type=Threading><>ThreadCount'
        name: jvm_thread_count
        type: GAUGE

      - pattern: 'java.lang<type=OperatingSystem><>SystemCpuLoad'
        name: jvm_system_cpu_utilization
        type: GAUGE

      # Broker uptime
      - pattern: 'java.lang<type=Runtime><>Uptime'
        name: kafka_broker_uptime
        type: GAUGE

      # Additional metrics — remove this section to reduce data ingest

      # Request latency: total count, 50th percentile, and average (99p kept above)
      - pattern: 'kafka.network<type=RequestMetrics, name=TotalTimeMs, request=(Produce|FetchConsumer|FetchFollower)><>Count'
        name: kafka_request_time_total
        type: COUNTER
        labels:
          type: "$1"

      - pattern: 'kafka.network<type=RequestMetrics, name=TotalTimeMs, request=(Produce|FetchConsumer|FetchFollower)><>50thPercentile'
        name: kafka_request_time_50p
        type: GAUGE
        labels:
          type: "$1"

      - pattern: 'kafka.network<type=RequestMetrics, name=TotalTimeMs, request=(Produce|FetchConsumer|FetchFollower)><>Mean'
        name: kafka_request_time_avg
        type: GAUGE
        labels:
          type: "$1"

      # Log flush metrics
      - pattern: 'kafka.log<type=LogFlushStats, name=LogFlushRateAndTimeMs><>Count'
        name: kafka_logs_flush_count
        type: COUNTER

      - pattern: 'kafka.log<type=LogFlushStats, name=LogFlushRateAndTimeMs><>50thPercentile'
        name: kafka_logs_flush_time_50p
        type: GAUGE

      - pattern: 'kafka.log<type=LogFlushStats, name=LogFlushRateAndTimeMs><>99thPercentile'
        name: kafka_logs_flush_time_99p
        type: GAUGE

      # JVM GC elapsed time
      - pattern: 'java.lang<name=(.+), type=GarbageCollector><>CollectionTime'
        name: jvm_gc_collections_elapsed
        type: COUNTER
        labels:
          name: "$1"

      # JVM Memory heap committed
      - pattern: 'java.lang<type=Memory><HeapMemoryUsage>committed'
        name: jvm_memory_heap_committed
        type: GAUGE

      # JVM class loading
      - pattern: 'java.lang<type=ClassLoading><>LoadedClassCount'
        name: jvm_class_count
        type: GAUGE

      # Additional JVM OS metrics
      - pattern: 'java.lang<type=OperatingSystem><>SystemLoadAverage'
        name: jvm_system_cpu_load_1m
        type: GAUGE

      - pattern: 'java.lang<type=OperatingSystem><>AvailableProcessors'
        name: jvm_cpu_count
        type: GAUGE

      - pattern: 'java.lang<type=OperatingSystem><>ProcessCpuLoad'
        name: jvm_cpu_recent_utilization
        type: GAUGE

      - pattern: 'java.lang<type=OperatingSystem><>OpenFileDescriptorCount'
        name: jvm_file_descriptor_count
        type: GAUGE

      # JVM Memory Pool
      - pattern: 'java.lang<type=MemoryPool, name=(.+)><Usage>used'
        name: jvm_memory_pool_used
        type: GAUGE
        labels:
          name: "$1"

      - pattern: 'java.lang<type=MemoryPool, name=(.+)><Usage>max'
        name: jvm_memory_pool_max
        type: GAUGE
        labels:
          name: "$1"

      - pattern: 'java.lang<type=MemoryPool, name=(.+)><CollectionUsage>used'
        name: jvm_memory_pool_used_after_last_gc
        type: GAUGE
        labels:
          name: "$1"

팁

메트릭 사용자 지정: Prometheus JMX Exporter 예제 및 Kafka MBean 문서를 참조하여 패턴을 추가하거나 수정할 수 있습니다.

ConfigMap을 적용합니다.

bash

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

JMX Exporter용 Kafka StatefulSet 구성

Kafka StatefulSet을 패치하여 Prometheus JMX Exporter JAR를 다운로드하는 init 컨테이너를 추가한 다음, KAFKA_OPTS을(를) 통해 Kafka 브로커 JVM에 연결하십시오.

1단계. 기존 Kafka StatefulSet 매니페스트에 다음 섹션을 추가합니다:

spec:
  template:
    spec:
      # 1. Init container: downloads JMX Exporter JAR before Kafka starts
      initContainers:
        - name: download-jmx-exporter
          image: busybox:latest
          command:
            - sh
            - -c
            - |
              # Version 1.5.0 is the minimum required version. Check https://github.com/prometheus/jmx_exporter/releases/latest for newer releases.
              JMX_EXPORTER_VERSION="1.5.0"
              wget -O /prometheus-jmx/jmx_prometheus_javaagent.jar \
                "https://github.com/prometheus/jmx_exporter/releases/download/${JMX_EXPORTER_VERSION}/jmx_prometheus_javaagent-${JMX_EXPORTER_VERSION}.jar"
          volumeMounts:
            - name: prometheus-jmx
              mountPath: /prometheus-jmx

      containers:
        - name: kafka  # TODO: Replace with your Kafka container name
          # 2. Attach JMX Exporter as Java agent on port 9404
          env:
            - name: KAFKA_OPTS
              value: "-javaagent:/prometheus-jmx/jmx_prometheus_javaagent.jar=9404:/jmx-config/kafka-metrics-config.yml"
          # 3. Expose port 9404 for Prometheus scraping
          ports:
            - name: jmx-metrics
              containerPort: 9404
              protocol: TCP
          volumeMounts:
            - name: prometheus-jmx
              mountPath: /prometheus-jmx
            - name: jmx-config
              mountPath: /jmx-config

      # 4. Volumes: emptyDir for JAR, ConfigMap for metrics config
      volumes:
        - name: prometheus-jmx
          emptyDir: {}
        - name: jmx-config
          configMap:
            name: kafka-jmx-metrics  # Must match the ConfigMap name from Step 2

2단계. 업데이트된 StatefulSet을 적용하고 파드가 롤링될 때까지 기다립니다:

bash

$kubectl apply -f kafka-statefulset.yaml
$kubectl rollout status statefulset/kafka -n kafka  # TODO: Replace with your StatefulSet name and namespace

3단계. 롤아웃이 완료되면 각 브로커 파드에 메트릭이 노출되는지 확인합니다:

bash

$# Replace kafka-0 and kafka with your pod name and namespace
$kubectl exec -n kafka kafka-0 -- curl -s http://localhost:9404/metrics | grep kafka_ | head -20

중요

다중 브로커 클러스터: init 컨테이너 및 KAFKA_OPTS 설정은 StatefulSet의 모든 파드에 자동으로 적용됩니다. 롤아웃 후 각 브로커 파드가 메트릭을 노출하는지 확인합니다.

구현하다, 배포하다 OpenTelemetry Collector

클러스터에 OpenTelemetry Collector를 배포하십시오. 수집기는 정적 DNS 타겟을 사용하여 Kafka 브로커 파드를 스크랩하고 계측된 애플리케이션의 OTLP 데이터를 포트 4317 에서 수신 대기합니다.

Helm 설치 방법은 Kubernetes 에서 구현하다, 배포하다 OpenTelemetry Collector 에 권장되는 접근 방식입니다.

1단계. 뉴렐릭 자격 증명 시크릿 생성

bash

$kubectl create secret generic newrelic-otlp-secret \
>  --namespace newrelic \
>  --from-literal=NEW_RELIC_LICENSE_KEY='your-license-key-here' \
>  --from-literal=NEW_RELIC_OTLP_ENDPOINT='https://otlp.nr-data.net:4317'

bash

$kubectl create secret generic newrelic-otlp-secret \
>  --namespace newrelic \
>  --from-literal=NEW_RELIC_LICENSE_KEY='your-license-key-here' \
>  --from-literal=NEW_RELIC_OTLP_ENDPOINT='https://otlp.eu01.nr-data.net:4317'

팁

다른 엔드포인트 설정에 대해서는 OTLP 엔드포인트 구성을 참조하세요.

2단계. 수집기 설정이 포함된 values.yaml 생성

NRDOT 및 OpenTelemetry 수집기 모두 동일한 설정을 사용합니다. 선호하는 수집기 이미지를 선택하세요:

다음 내용으로 values.yaml 생성하세요:

# Deployment mode
mode: deployment
replicaCount: 1

# Use NRDOT collector image
image:
  repository: newrelic/nrdot-collector
  tag: "latest"
  pullPolicy: Always

# Service account (no ClusterRole needed for static scraping)
serviceAccount:
  create: true
  name: otel-collector

# Pod security context
podSecurityContext:
  runAsNonRoot: true
  runAsUser: 10001

# Container security context
securityContext:
  allowPrivilegeEscalation: false
  readOnlyRootFilesystem: true
  capabilities:
    drop:
      - ALL

# Resource limits
resources:
  requests:
    memory: 512Mi
    cpu: 250m
  limits:
    memory: 1Gi
    cpu: 500m

# Load environment variables from secret
extraEnvsFrom:
  - secretRef:
      name: newrelic-otlp-secret

# Disable unused default ports
ports:
  jaeger-compact:
    enabled: false
  jaeger-thrift:
    enabled: false
  jaeger-grpc:
    enabled: false
  zipkin:
    enabled: false

# OpenTelemetry Collector Configuration
config:
  receivers:
    # Disable default receivers not needed in NRDOT
    jaeger: null
    zipkin: null

    # OTLP receiver for application traces, metrics, and logs
    otlp:
      protocols:
        grpc:
          endpoint: "0.0.0.0:4317"

    # Kafka metrics receiver for consumer lag, topic, and partition metrics
    kafkametrics/cluster:
      brokers:
        # TODO#1: Replace with your Kafka bootstrap service
        # TODO#2: Replace with the namespace where your Kafka cluster is deployed
        - "kafka.kafka.svc.cluster.local:9092"
      collection_interval: 30s
      protocol_version: 2.0.0
      scrapers:
        - brokers
        - topics
        - consumers
      topic_match: "^[^_].*$"
      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

    # Prometheus receiver scrapes JMX metrics from each broker pod via headless service DNS
    prometheus/kafka-jmx:
      config:
        scrape_configs:
          - job_name: 'kafka-jmx-metrics'
            metrics_path: /metrics
            scrape_interval: 30s
            static_configs:
              # TODO#2: Replace with the namespace where your Kafka cluster is deployed
              # TODO#3: Replace with your Kafka StatefulSet name followed by -headless
              # TODO#4: Replace with your Kafka StatefulSet name
              - targets:
                  - 'kafka-0.kafka-headless.kafka.svc.cluster.local:9404'
                  - 'kafka-1.kafka-headless.kafka.svc.cluster.local:9404'
                  - 'kafka-2.kafka-headless.kafka.svc.cluster.local:9404'
            relabel_configs:
              # Extract broker ordinal from pod DNS name as broker.id
              - source_labels: [__address__]
                target_label: broker.id
                regex: '[^-]+-(\d+)\..+:\d+'
                replacement: '$1'

  exporters:
    otlp/backend:
      endpoint: ${NEW_RELIC_OTLP_ENDPOINT}
      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/export:
      send_batch_size: 1024
      timeout: 30s

    memory_limiter:
      limit_percentage: 80
      spike_limit_percentage: 30
      check_interval: 1s

    transform/metric-naming:
      metric_statements:
        - context: metric
          statements:
            - replace_pattern(name, "_", ".")
            - replace_pattern(name, "\\.load\\.1", ".load_1")
            - replace_pattern(name, "\\.recent\\.util", ".recent_util")
            - replace_pattern(name, "file\\.descriptor\\.count", "file_descriptor.count")
            - replace_pattern(name, "\\.memory\\.pool\\.used\\.bytes$", ".memory.pool.used")
            - replace_pattern(name, "\\.memory\\.pool\\.max\\.bytes$", ".memory.pool.max")
            - replace_pattern(name, "\\.memory\\.pool\\.collection\\.used\\.bytes$", ".memory.pool.used_after_last_gc")
            - replace_pattern(name, "\\.non\\.preferred\\.leader", ".non_preferred_leader")
            - replace_pattern(name, "\\.under\\.min\\.isr", ".under_min_isr")
            - replace_pattern(name, "\\.under\\.replicated", ".under_replicated")
            - replace_pattern(name, "\\.total$", "") where name != "kafka.request.time.total"
        - context: datapoint
          statements:
            - set(attributes["name"], attributes["gc"]) where attributes["gc"] != nil
            - delete_key(attributes, "gc") where attributes["gc"] != nil
            - set(attributes["name"], attributes["pool"]) where attributes["pool"] != nil
            - delete_key(attributes, "pool") where attributes["pool"] != nil

    resource/cluster-name:
      attributes:
        - key: kafka.cluster.name
          # TODO#5: Replace with your Kafka cluster name (this will be used to identify and filter your metrics in New Relic)
          value: my-kafka-cluster
          action: upsert

    transform/remove_broker_id:
      metric_statements:
        - context: datapoint
          statements:
            - delete_key(attributes, "broker.id")

    filter/scrape-overhead:
      metrics:
        exclude:
          match_type: regexp
          metric_names:
            - "^jmx_.*"
            - "^process_.*"
            - "^jvm_buffer_pool_.*"
            - "^jvm_threads_.*"
            - "^jvm_classes_.*"
            - "^jvm_memory_(heap|non_heap)_(committed|init|max|used)_bytes$"
            - "^jvm_compilation_.*"
            - "^jvm_(runtime|info).*"
            - "^jvm_memory_pool_(allocated_bytes_total|committed_bytes|init_bytes|collection_(committed|init|max)_bytes)$"

    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/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$"

    transform/remove_attributes:
      metric_statements:
        - context: metric
          statements:
            - set(description, "") where description != ""
            - set(unit, "") where unit != ""
        - context: resource
          statements:
            - delete_key(attributes, "server.address")
            - delete_key(attributes, "server.port")
            - delete_key(attributes, "service.instance.id")
            - delete_key(attributes, "host.name")
            - delete_key(attributes, "k8s.pod.uid")
            - delete_key(attributes, "url.scheme")

    metricstransform/topic-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
        - include: kafka.partition.replicas
          action: insert
          new_name: kafka.partition.replicas.total
          operations:
            - action: aggregate_labels
              label_set: [topic]
              aggregation_type: sum

    filter/exclude_partition_replicas_metric:
      metrics:
        exclude:
          match_type: strict
          metric_names:
            - kafka.partition.replicas_in_sync

    filter/internal_topics:
      metrics:
        datapoint:
          - 'attributes["topic"] != nil and IsMatch(attributes["topic"], "^__.*")'

    cumulativetodelta:

    groupbyattrs/cluster:
      keys: [kafka.cluster.name]

    metricstransform/cluster_max:
      transforms:
        - include: "kafka\\.partition\\.offline|kafka\\.leader\\.election\\.rate|kafka\\.unclean\\.election\\.rate|kafka\\.partition\\.non_preferred_leader|kafka\\.broker\\.fenced\\.count|kafka\\.cluster\\.partition\\.count|kafka\\.cluster\\.topic\\.count"
          match_type: regexp
          action: update
          operations:
            - action: aggregate_labels
              aggregation_type: max
              label_set: []

  service:
    pipelines:
      # Application traces from instrumented Kafka clients and apps
      traces:
        receivers: [otlp]
        processors: [memory_limiter, batch/export]
        exporters: [otlp/backend]

      # Application metrics from instrumented Kafka clients and apps
      metrics:
        receivers: [otlp]
        processors: [memory_limiter, batch/export]
        exporters: [otlp/backend]

      # Application logs from instrumented Kafka clients and apps
      logs:
        receivers: [otlp]
        processors: [memory_limiter, batch/export]
        exporters: [otlp/backend]

      # Broker-level metrics from Prometheus JMX scraping
      metrics/broker:
        receivers:
          - prometheus/kafka-jmx
        processors:
          - resource/cluster-name
          - filter/scrape-overhead
          - transform/metric-naming
          - transform/remove_attributes
          - filter/exclude_cluster_metrics
          - memory_limiter
          - cumulativetodelta
          - batch/export
        exporters:
          - otlp/backend

      # Cluster-level metrics from Prometheus JMX scraping
      metrics/cluster/prometheus:
        receivers:
          - prometheus/kafka-jmx
        processors:
          - resource/cluster-name
          - filter/scrape-overhead
          - transform/metric-naming
          - transform/remove_attributes
          - filter/include_cluster_metrics
          - transform/remove_broker_id
          - memory_limiter
          - cumulativetodelta
          - groupbyattrs/cluster
          - metricstransform/cluster_max
          - batch/export
        exporters:
          - otlp/backend

      # Cluster-level metrics from Kafka metrics receiver (consumer lag, topics, partitions)
      metrics/cluster/kafkametrics:
        receivers:
          - kafkametrics/cluster
        processors:
          - resource/cluster-name
          - filter/internal_topics
          - transform/remove_attributes
          - metricstransform/topic-aggregation
          - filter/exclude_partition_replicas_metric
          - memory_limiter
          - cumulativetodelta
          - batch/export
        exporters:
          - otlp/backend

구성 매개변수

다음 표에서는 주요 설정 시위에 대해 설명합니다.

매개변수	설명
`config.receivers.kafkametrics/cluster.brokers`	Kafka 부트스트랩 서비스(예: `kafka.kafka.svc.cluster.local:9092`)로 교체합니다
`config.receivers.kafkametrics/cluster.brokers` 그리고 `config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Kafka 클러스터가 배포된 네임스페이스로 교체하십시오.
`config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Kafka StatefulSet 이름 뒤에 -headless를 추가한 값으로 바꿉니다(예: `kafka-headless`)
`config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Kafka StatefulSet 이름으로 바꿉니다(예: `kafka`, `kafka-0`과(와) 같은 파드 이름에 표시됨)
`config.processors.resource/cluster-name.attributes[kafka.cluster.name].value`	Kafka 클러스터 이름으로 바꿉니다(이는 뉴렐릭에서 메트릭을 식별하고 필터링하는 데 사용됩니다)
`config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	브로커 파드와 일치하도록 `targets` 목록을 업데이트하세요 ― 헤드리스 서비스 DNS(`<pod-name>.<headless-service>.<namespace>.svc.cluster.local:9404`)를 사용하여 브로커당 하나의 항목을 지정합니다. 브로커 수에 맞게 항목을 추가하거나 제거하십시오. `broker.id` 은(는) `relabel_configs`을(를) 통해 파드 이름에서 자동으로 추출됩니다.
`resources.limits` 그리고 `resources.requests`	워크로드 요구 사항에 따라 조정하십시오.

팁

대안: Kubernetes 파드 자동 검색

정적 DNS 타겟 대신, Kubernetes 파드 디스커버리를 사용하여 브로커 파드를 자동으로 찾을 수 있습니다. 이는 타겟 목록을 업데이트할 필요 없이 동적 스케일링에 유용합니다.

values.yaml 의 clusterRole 및 prometheus/kafka-jmx 섹션을 다음으로 바꿉니다:

# Add RBAC for Kubernetes pod discovery
clusterRole:
  create: true
  rules:
    - apiGroups: [""]
      resources: ["pods", "nodes"]
      verbs: ["get", "list", "watch"]

# In config.receivers:
prometheus/kafka-jmx:
  config:
    scrape_configs:
      - job_name: 'kafka-jmx-metrics'
        metrics_path: /metrics
        scrape_interval: 30s
        kubernetes_sd_configs:
          - role: pod
            namespaces:
              names:
                # TODO: Replace with your Kafka namespace
                - kafka
        relabel_configs:
          # Filter for Kafka broker pods by app label
          - source_labels: [__meta_kubernetes_pod_label_app]
            action: keep
            # TODO: Replace with your Kafka pod app label value (e.g., "kafka")
            regex: kafka

          # Only scrape running pods
          - source_labels: [__meta_kubernetes_pod_phase]
            action: keep
            regex: Running

          # Extract broker ordinal from pod name as broker.id
          - source_labels: [__meta_kubernetes_pod_name]
            target_label: broker.id
            regex: '.*-(\d+)$'
            replacement: '$1'

          # Set scrape target to pod IP on port 9404
          - source_labels: [__meta_kubernetes_pod_ip]
            target_label: __address__
            replacement: '$1:9404'

유연성을 극대화하고 공급업체 비종속 구현, 배포를 위해 커뮤니티 OpenTelemetry Collector 사용하세요.

다음 내용으로 values.yaml 을(를) 생성합니다(동일한 설정, 다른 이미지):

# Deployment mode
mode: deployment
replicaCount: 1

# Use contrib image for kafkametrics receiver
image:
  repository: otel/opentelemetry-collector-contrib
  tag: "latest"
  pullPolicy: Always

# Service account (no ClusterRole needed for static scraping)
serviceAccount:
  create: true
  name: otel-collector

# Pod security context
podSecurityContext:
  runAsNonRoot: true
  runAsUser: 10001

# Container security context
securityContext:
  allowPrivilegeEscalation: false
  readOnlyRootFilesystem: true
  capabilities:
    drop:
      - ALL

# Resource limits
resources:
  requests:
    memory: 512Mi
    cpu: 250m
  limits:
    memory: 1Gi
    cpu: 500m

# Load environment variables from secret
extraEnvsFrom:
  - secretRef:
      name: newrelic-otlp-secret

# Disable unused default ports
ports:
  jaeger-compact:
    enabled: false
  jaeger-thrift:
    enabled: false
  jaeger-grpc:
    enabled: false
  zipkin:
    enabled: false

# OpenTelemetry Collector Configuration
config:
  receivers:
    # OTLP receiver for application traces, metrics, and logs
    otlp:
      protocols:
        grpc:
          endpoint: "0.0.0.0:4317"

    # Kafka metrics receiver for consumer lag, topic, and partition metrics
    kafkametrics/cluster:
      brokers:
        # TODO#1: Replace with your Kafka bootstrap service
        # TODO#2: Replace with the namespace where your Kafka cluster is deployed
        - "kafka.kafka.svc.cluster.local:9092"
      collection_interval: 30s
      protocol_version: 2.0.0
      scrapers:
        - brokers
        - topics
        - consumers
      topic_match: "^[^_].*$"
      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

    # Prometheus receiver scrapes JMX metrics from each broker pod via headless service DNS
    prometheus/kafka-jmx:
      config:
        scrape_configs:
          - job_name: 'kafka-jmx-metrics'
            metrics_path: /metrics
            scrape_interval: 30s
            static_configs:
              # TODO#2: Replace with the namespace where your Kafka cluster is deployed
              # TODO#3: Replace with your Kafka StatefulSet name followed by -headless
              # TODO#4: Replace with your Kafka StatefulSet name
              - targets:
                  - 'kafka-0.kafka-headless.kafka.svc.cluster.local:9404'
                  - 'kafka-1.kafka-headless.kafka.svc.cluster.local:9404'
                  - 'kafka-2.kafka-headless.kafka.svc.cluster.local:9404'
            relabel_configs:
              - source_labels: [__address__]
                target_label: broker.id
                regex: '[^-]+-(\d+)\..+:\d+'
                replacement: '$1'

  exporters:
    otlp/backend:
      endpoint: ${NEW_RELIC_OTLP_ENDPOINT}
      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/export:
      send_batch_size: 1024
      timeout: 30s
    memory_limiter:
      limit_percentage: 80
      spike_limit_percentage: 30
      check_interval: 1s
    transform/metric-naming:
      metric_statements:
        - context: metric
          statements:
            - replace_pattern(name, "_", ".")
            - replace_pattern(name, "\\.load\\.1", ".load_1")
            - replace_pattern(name, "\\.recent\\.util", ".recent_util")
            - replace_pattern(name, "file\\.descriptor\\.count", "file_descriptor.count")
            - replace_pattern(name, "\\.memory\\.pool\\.used\\.bytes$", ".memory.pool.used")
            - replace_pattern(name, "\\.memory\\.pool\\.max\\.bytes$", ".memory.pool.max")
            - replace_pattern(name, "\\.memory\\.pool\\.collection\\.used\\.bytes$", ".memory.pool.used_after_last_gc")
            - replace_pattern(name, "\\.non\\.preferred\\.leader", ".non_preferred_leader")
            - replace_pattern(name, "\\.under\\.min\\.isr", ".under_min_isr")
            - replace_pattern(name, "\\.under\\.replicated", ".under_replicated")
            - replace_pattern(name, "\\.total$", "") where name != "kafka.request.time.total"
        - context: datapoint
          statements:
            - set(attributes["name"], attributes["gc"]) where attributes["gc"] != nil
            - delete_key(attributes, "gc") where attributes["gc"] != nil
            - set(attributes["name"], attributes["pool"]) where attributes["pool"] != nil
            - delete_key(attributes, "pool") where attributes["pool"] != nil
    resource/cluster-name:
      attributes:
        - key: kafka.cluster.name
          # TODO#5: Replace with your Kafka cluster name (this will be used to identify and filter your metrics in New Relic)
          value: my-kafka-cluster
          action: upsert
    transform/remove_broker_id:
      metric_statements:
        - context: datapoint
          statements:
            - delete_key(attributes, "broker.id")
    filter/scrape-overhead:
      metrics:
        exclude:
          match_type: regexp
          metric_names:
            - "^jmx_.*"
            - "^process_.*"
            - "^jvm_buffer_pool_.*"
            - "^jvm_threads_.*"
            - "^jvm_classes_.*"
            - "^jvm_memory_(heap|non_heap)_(committed|init|max|used)_bytes$"
            - "^jvm_compilation_.*"
            - "^jvm_(runtime|info).*"
            - "^jvm_memory_pool_(allocated_bytes_total|committed_bytes|init_bytes|collection_(committed|init|max)_bytes)$"
    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/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$"
    transform/remove_attributes:
      metric_statements:
        - context: metric
          statements:
            - set(description, "") where description != ""
            - set(unit, "") where unit != ""
        - context: resource
          statements:
            - delete_key(attributes, "server.address")
            - delete_key(attributes, "server.port")
            - delete_key(attributes, "service.instance.id")
            - delete_key(attributes, "host.name")
            - delete_key(attributes, "k8s.pod.uid")
            - delete_key(attributes, "url.scheme")
    metricstransform/topic-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
        - include: kafka.partition.replicas
          action: insert
          new_name: kafka.partition.replicas.total
          operations:
            - action: aggregate_labels
              label_set: [topic]
              aggregation_type: sum
    filter/exclude_partition_replicas_metric:
      metrics:
        exclude:
          match_type: strict
          metric_names:
            - kafka.partition.replicas_in_sync
    filter/internal_topics:
      metrics:
        datapoint:
          - 'attributes["topic"] != nil and IsMatch(attributes["topic"], "^__.*")'
    cumulativetodelta:
    groupbyattrs/cluster:
      keys: [kafka.cluster.name]
    metricstransform/cluster_max:
      transforms:
        - include: "kafka\\.partition\\.offline|kafka\\.leader\\.election\\.rate|kafka\\.unclean\\.election\\.rate|kafka\\.partition\\.non_preferred_leader|kafka\\.broker\\.fenced\\.count|kafka\\.cluster\\.partition\\.count|kafka\\.cluster\\.topic\\.count"
          match_type: regexp
          action: update
          operations:
            - action: aggregate_labels
              aggregation_type: max
              label_set: []

  service:
    pipelines:
      traces:
        receivers: [otlp]
        processors: [memory_limiter, batch/export]
        exporters: [otlp/backend]
      metrics:
        receivers: [otlp]
        processors: [memory_limiter, batch/export]
        exporters: [otlp/backend]
      logs:
        receivers: [otlp]
        processors: [memory_limiter, batch/export]
        exporters: [otlp/backend]
      metrics/broker:
        receivers:
          - prometheus/kafka-jmx
        processors:
          - resource/cluster-name
          - filter/scrape-overhead
          - transform/metric-naming
          - transform/remove_attributes
          - filter/exclude_cluster_metrics
          - memory_limiter
          - cumulativetodelta
          - batch/export
        exporters:
          - otlp/backend
      metrics/cluster/prometheus:
        receivers:
          - prometheus/kafka-jmx
        processors:
          - resource/cluster-name
          - filter/scrape-overhead
          - transform/metric-naming
          - transform/remove_attributes
          - filter/include_cluster_metrics
          - transform/remove_broker_id
          - memory_limiter
          - cumulativetodelta
          - groupbyattrs/cluster
          - metricstransform/cluster_max
          - batch/export
        exporters:
          - otlp/backend
      metrics/cluster/kafkametrics:
        receivers:
          - kafkametrics/cluster
        processors:
          - resource/cluster-name
          - filter/internal_topics
          - transform/remove_attributes
          - metricstransform/topic-aggregation
          - filter/exclude_partition_replicas_metric
          - memory_limiter
          - cumulativetodelta
          - batch/export
        exporters:
          - otlp/backend

구성 매개변수

다음 표에서는 주요 설정 시위에 대해 설명합니다.

매개변수	설명
`config.receivers.kafkametrics/cluster.brokers`	사용자의 Kafka 부트스트랩 서비스 DNS로 교체하십시오
`config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	헤드리스 서비스 DNS(`<pod-name>.<headless-service>.<namespace>.svc.cluster.local:9404`)를 사용하여 브로커 파드당 하나의 항목을 추가합니다. 브로커 수에 맞게 항목을 추가하거나 제거하십시오. `broker.id` 은(는) `relabel_configs`을(를) 통해 파드 이름에서 자동으로 추출됩니다.
`config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].labels[kafka.cluster.name]`	`kafka.cluster.name` 레이블 값을 클러스터 이름으로 바꿉니다.
`config.processors.resource/cluster-name.attributes[kafka.cluster.name].value`	카프카 클러스터 이름으로 바꾸세요(뉴렐릭에서 메트릭을 식별하기 위해 `resource/cluster-name` 프로세서에서 사용됨)
`resources.limits` 그리고 `resources.requests`	워크로드 요구 사항에 따라 조정하십시오.

팁

대안: Kubernetes 파드 자동 검색

values.yaml 의 prometheus/kafka-jmx 섹션을 다음으로 바꿉니다:

# Add RBAC for Kubernetes pod discovery (add before config:)
clusterRole:
  create: true
  rules:
    - apiGroups: [""]
      resources: ["pods", "nodes"]
      verbs: ["get", "list", "watch"]

# In config.receivers:
prometheus/kafka-jmx:
  config:
    scrape_configs:
      - job_name: 'kafka-jmx-metrics'
        metrics_path: /metrics
        scrape_interval: 30s
        kubernetes_sd_configs:
          - role: pod
            namespaces:
              names:
                # TODO: Replace with your Kafka namespace
                - kafka
        relabel_configs:
          # Filter for Kafka broker pods by app label
          - source_labels: [__meta_kubernetes_pod_label_app]
            action: keep
            # TODO: Replace with your Kafka pod app label value (e.g., "kafka")
            regex: kafka

          # Only scrape running pods
          - source_labels: [__meta_kubernetes_pod_phase]
            action: keep
            regex: Running

          # Extract broker ordinal from pod name as broker.id
          - source_labels: [__meta_kubernetes_pod_name]
            target_label: broker.id
            regex: '.*-(\d+)$'
            replacement: '$1'

          # Set scrape target to pod IP on port 9404
          - source_labels: [__meta_kubernetes_pod_ip]
            target_label: __address__
            replacement: '$1:9404'

고급 설정 옵션에 대해서는 다음 수신기 설명서 페이지를 참조하십시오.

프로메테우스 수신기 문서

Kafka 메트릭 수신자 문서

3단계. Helm으로 OpenTelemetry Collector 설치

bash

$helm repo add open-telemetry https://open-telemetry.github.io/opentelemetry-helm-charts
$helm upgrade kafka-monitoring open-telemetry/opentelemetry-collector \
>  --install \
>  --namespace newrelic \
>  --create-namespace \
>  -f values.yaml

4단계. 배포 확인:

bash

$# Check pod status
$kubectl get pods -n newrelic -l app.kubernetes.io/name=opentelemetry-collector
$
$# View logs to verify metrics collection
$kubectl logs -n newrelic -l app.kubernetes.io/name=opentelemetry-collector --tail=50

포트 9404의 Kafka 브로커 파드에서 성공적으로 스크래핑되었음을 나타내는 로그가 표시되어야 합니다.

매니페스트 설치 방식은 Helm을 사용하지 않고도 Kubernetes 리소스를 직접 제어할 수 있도록 해줍니다.

1단계. 뉴렐릭 자격 증명 시크릿 생성

bash

$kubectl create secret generic newrelic-otlp-secret \
>  --namespace newrelic \
>  --from-literal=NEW_RELIC_LICENSE_KEY='your-license-key-here' \
>  --from-literal=NEW_RELIC_OTLP_ENDPOINT='https://otlp.nr-data.net:4317'

bash

$kubectl create secret generic newrelic-otlp-secret \
>  --namespace newrelic \
>  --from-literal=NEW_RELIC_LICENSE_KEY='your-license-key-here' \
>  --from-literal=NEW_RELIC_OTLP_ENDPOINT='https://otlp.eu01.nr-data.net:4317'

팁

다른 엔드포인트 설정에 대해서는 OTLP 엔드포인트 구성을 참조하세요.

2단계. 매니페스트 파일 생성

NRDOT 및 OpenTelemetry 수집기는 모두 동일한 설정을 사용합니다. 컨테이너 이미지만 다릅니다.

1. collector-configmap.yaml생성 - OpenTelemetry Collector 설정:

---
apiVersion: v1
kind: ConfigMap
metadata:
  name: otel-collector-config
  namespace: newrelic
  labels:
    app: otel-collector
data:
  otel-collector-config.yaml: |
    receivers:
      otlp:
        protocols:
          grpc:
            endpoint: "0.0.0.0:4317"

      kafkametrics/cluster:
        brokers:
          # TODO#1: Replace with your Kafka bootstrap service DNS
          - "kafka.kafka.svc.cluster.local:9092"
        collection_interval: 30s
        protocol_version: 2.0.0
        scrapers:
          - brokers
          - topics
          - consumers
        topic_match: "^[^_].*$"
        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

      prometheus/kafka-jmx:
        config:
          scrape_configs:
            - job_name: 'kafka-jmx-metrics'
              metrics_path: /metrics
              scrape_interval: 30s
              static_configs:
                # TODO#2: Add one entry per broker pod using headless service DNS
                - targets:
                    - 'kafka-0.kafka-headless.kafka.svc.cluster.local:9404'
                    - 'kafka-1.kafka-headless.kafka.svc.cluster.local:9404'
                    - 'kafka-2.kafka-headless.kafka.svc.cluster.local:9404'
                  labels:
                    kafka.cluster.name: 'my-kafka-cluster'  # TODO#3: Replace with your cluster name
              relabel_configs:
                - source_labels: [__address__]
                  target_label: broker.id
                  regex: '[^-]+-(\d+)\..+:\d+'
                  replacement: '$1'

    exporters:
      otlp/backend:
        endpoint: ${NEW_RELIC_OTLP_ENDPOINT}
        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/export:
        send_batch_size: 1024
        timeout: 30s
      memory_limiter:
        limit_percentage: 80
        spike_limit_percentage: 30
        check_interval: 1s
      transform/metric-naming:
        metric_statements:
        - context: metric
          statements:
          - replace_pattern(name, "_", ".")
          - replace_pattern(name, "\\.load\\.1", ".load_1")
          - replace_pattern(name, "\\.recent\\.util", ".recent_util")
          - replace_pattern(name, "file\\.descriptor\\.count", "file_descriptor.count")
          - replace_pattern(name, "\\.memory\\.pool\\.used\\.bytes$", ".memory.pool.used")
          - replace_pattern(name, "\\.memory\\.pool\\.max\\.bytes$", ".memory.pool.max")
          - replace_pattern(name, "\\.memory\\.pool\\.collection\\.used\\.bytes$", ".memory.pool.used_after_last_gc")
          - replace_pattern(name, "\\.non\\.preferred\\.leader", ".non_preferred_leader")
          - replace_pattern(name, "\\.under\\.min\\.isr", ".under_min_isr")
          - replace_pattern(name, "\\.under\\.replicated", ".under_replicated")
          - replace_pattern(name, "\\.total$", "") where name != "kafka.request.time.total"
        - context: datapoint
          statements:
          - set(attributes["name"], attributes["gc"]) where attributes["gc"] != nil
          - delete_key(attributes, "gc") where attributes["gc"] != nil
          - set(attributes["name"], attributes["pool"]) where attributes["pool"] != nil
          - delete_key(attributes, "pool") where attributes["pool"] != nil
      resource/cluster-name:
        attributes:
        - key: kafka.cluster.name
          # TODO#5: Replace with your Kafka cluster name (this will be used to identify and filter your metrics in New Relic)
          value: my-kafka-cluster
          action: upsert
      transform/remove_broker_id:
        metric_statements:
        - context: datapoint
          statements:
          - delete_key(attributes, "broker.id")
      filter/scrape-overhead:
        metrics:
          exclude:
            match_type: regexp
            metric_names:
            - "^jmx_.*"
            - "^process_.*"
            - "^jvm_buffer_pool_.*"
            - "^jvm_threads_.*"
            - "^jvm_classes_.*"
            - "^jvm_memory_(heap|non_heap)_(committed|init|max|used)_bytes$"
            - "^jvm_compilation_.*"
            - "^jvm_(runtime|info).*"
            - "^jvm_memory_pool_(allocated_bytes_total|committed_bytes|init_bytes|collection_(committed|init|max)_bytes)$"
      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/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$"
      transform/remove_attributes:
        metric_statements:
        - context: metric
          statements:
          - set(description, "") where description != ""
          - set(unit, "") where unit != ""
        - context: resource
          statements:
          - delete_key(attributes, "server.address")
          - delete_key(attributes, "server.port")
          - delete_key(attributes, "service.instance.id")
          - delete_key(attributes, "host.name")
          - delete_key(attributes, "k8s.pod.uid")
          - delete_key(attributes, "url.scheme")
      metricstransform/topic-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
        - include: kafka.partition.replicas
          action: insert
          new_name: kafka.partition.replicas.total
          operations:
          - action: aggregate_labels
            label_set: [topic]
            aggregation_type: sum
      filter/exclude_partition_replicas_metric:
        metrics:
          exclude:
            match_type: strict
            metric_names:
            - kafka.partition.replicas_in_sync
      filter/internal_topics:
        metrics:
          datapoint:
            - 'attributes["topic"] != nil and IsMatch(attributes["topic"], "^__.*")'
      cumulativetodelta:
      groupbyattrs/cluster:
        keys: [kafka.cluster.name]
      metricstransform/cluster_max:
        transforms:
          - include: "kafka\\.partition\\.offline|kafka\\.leader\\.election\\.rate|kafka\\.unclean\\.election\\.rate|kafka\\.partition\\.non_preferred_leader|kafka\\.broker\\.fenced\\.count|kafka\\.cluster\\.partition\\.count|kafka\\.cluster\\.topic\\.count"
            match_type: regexp
            action: update
            operations:
              - action: aggregate_labels
                aggregation_type: max
                label_set: []

    service:
      pipelines:
        traces:
          receivers: [otlp]
          processors: [memory_limiter, batch/export]
          exporters: [otlp/backend]
        metrics:
          receivers: [otlp]
          processors: [memory_limiter, batch/export]
          exporters: [otlp/backend]
        logs:
          receivers: [otlp]
          processors: [memory_limiter, batch/export]
          exporters: [otlp/backend]
        metrics/broker:
          receivers: [prometheus/kafka-jmx]
          processors:
            - resource/cluster-name
            - filter/scrape-overhead
            - transform/metric-naming
            - transform/remove_attributes
            - filter/exclude_cluster_metrics
            - memory_limiter
            - cumulativetodelta
            - batch/export
          exporters: [otlp/backend]
        metrics/cluster/prometheus:
          receivers: [prometheus/kafka-jmx]
          processors:
            - resource/cluster-name
            - filter/scrape-overhead
            - transform/metric-naming
            - transform/remove_attributes
            - filter/include_cluster_metrics
            - transform/remove_broker_id
            - memory_limiter
            - cumulativetodelta
            - groupbyattrs/cluster
            - metricstransform/cluster_max
            - batch/export
          exporters: [otlp/backend]
        metrics/cluster/kafkametrics:
          receivers: [kafkametrics/cluster]
          processors:
            - resource/cluster-name
            - filter/internal_topics
            - transform/remove_attributes
            - metricstransform/topic-aggregation
            - filter/exclude_partition_replicas_metric
            - memory_limiter
            - cumulativetodelta
            - batch/export
          exporters: [otlp/backend]

2. collector-deployment.yaml생성 - ServiceAccount를 사용한 OpenTelemetry Collector 배포:

---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: otel-collector
  namespace: newrelic
  labels:
    app: otel-collector
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: otel-collector
  namespace: newrelic
  labels:
    app: otel-collector
spec:
  replicas: 1
  selector:
    matchLabels:
      app: otel-collector
  template:
    metadata:
      labels:
        app: otel-collector
    spec:
      serviceAccountName: otel-collector
      containers:
      - name: otel-collector
        image: newrelic/nrdot-collector:latest
        command:
        - "/nrdot-collector"
        - "--config=/conf/otel-collector-config.yaml"
        env:
        - name: NEW_RELIC_LICENSE_KEY
          valueFrom:
            secretKeyRef:
              name: newrelic-otlp-secret
              key: NEW_RELIC_LICENSE_KEY
        - name: NEW_RELIC_OTLP_ENDPOINT
          valueFrom:
            secretKeyRef:
              name: newrelic-otlp-secret
              key: NEW_RELIC_OTLP_ENDPOINT
        - name: GOGC
          value: "80"
        ports:
        - name: otlp-grpc
          containerPort: 4317
          protocol: TCP
        resources:
          limits:
            cpu: "1000m"
            memory: "1Gi"
          requests:
            cpu: "200m"
            memory: "512Mi"
        volumeMounts:
        - name: config
          mountPath: /conf
      volumes:
      - name: config
        configMap:
          name: otel-collector-config
          items:
          - key: otel-collector-config.yaml
            path: otel-collector-config.yaml
---
apiVersion: v1
kind: Service
metadata:
  name: otel-collector
  namespace: newrelic
  labels:
    app: otel-collector
spec:
  selector:
    app: otel-collector
  ports:
  - name: otlp-grpc
    port: 4317
    targetPort: 4317
    protocol: TCP

구성 매개변수

다음 표에서는 주요 설정 시위에 대해 설명합니다.

매개변수	설명
`receivers.kafkametrics/cluster.brokers`	Kafka 부트스트랩 서비스 DNS(예: `kafka.kafka.svc.cluster.local:9092`)로 교체합니다.
`receivers.kafkametrics/cluster.brokers` 그리고 `receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Kafka 클러스터가 배포된 네임스페이스로 교체하십시오.
`receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Kafka StatefulSet 이름 뒤에 `-headless` 을(를) 추가한 값으로 바꿉니다(예: `kafka-headless`)
`receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Kafka StatefulSet 이름으로 바꿉니다(예: `kafka`, `kafka-0`과(와) 같은 파드 이름에 표시됨)
`processors.resource/cluster-name.attributes[kafka.cluster.name].value`	Kafka 클러스터 이름으로 바꾸세요(뉴렐릭에서 메트릭을 식별하는 데 사용됨)
`receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	브로커 파드와 일치하도록 `targets` 목록을 업데이트하세요 ― 헤드리스 서비스 DNS(`<pod-name>.<headless-service>.<namespace>.svc.cluster.local:9404`)를 사용하여 브로커당 하나의 항목을 지정합니다. `broker.id` 은(는) `relabel_configs`을(를) 통해 파드 이름에서 자동으로 추출됩니다.
`resources.limits` 및 `resources.requests` (`collector-deployment.yaml`에서)	워크로드 요구 사항에 따라 조정하십시오.

공급업체 비종속 구현, 배포를 위해 커뮤니티 OpenTelemetry Collector 사용하세요.

1. collector-configmap.yaml생성 - 위의 NRDOT 옵션과 동일합니다(설정도 동일).

2. collector-deployment.yaml생성 - 컨테이너 이미지만 다름:

---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: otel-collector
  namespace: newrelic
  labels:
    app: otel-collector
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: otel-collector
  namespace: newrelic
  labels:
    app: otel-collector
spec:
  replicas: 1
  selector:
    matchLabels:
      app: otel-collector
  template:
    metadata:
      labels:
        app: otel-collector
    spec:
      serviceAccountName: otel-collector
      containers:
      - name: otel-collector
        image: otel/opentelemetry-collector-contrib:latest
        command:
        - "/otelcol-contrib"
        - "--config=/conf/otel-collector-config.yaml"
        env:
        - name: NEW_RELIC_LICENSE_KEY
          valueFrom:
            secretKeyRef:
              name: newrelic-otlp-secret
              key: NEW_RELIC_LICENSE_KEY
        - name: NEW_RELIC_OTLP_ENDPOINT
          valueFrom:
            secretKeyRef:
              name: newrelic-otlp-secret
              key: NEW_RELIC_OTLP_ENDPOINT
        - name: GOGC
          value: "80"
        ports:
        - name: otlp-grpc
          containerPort: 4317
          protocol: TCP
        resources:
          limits:
            cpu: "1000m"
            memory: "1Gi"
          requests:
            cpu: "200m"
            memory: "512Mi"
        volumeMounts:
        - name: config
          mountPath: /conf
      volumes:
      - name: config
        configMap:
          name: otel-collector-config
          items:
          - key: otel-collector-config.yaml
            path: otel-collector-config.yaml
---
apiVersion: v1
kind: Service
metadata:
  name: otel-collector
  namespace: newrelic
  labels:
    app: otel-collector
spec:
  selector:
    app: otel-collector
  ports:
  - name: otlp-grpc
    port: 4317
    targetPort: 4317
    protocol: TCP

설정 파라미터: 위의 NRDOT 옵션과 동일한 파라미터입니다. 리소스 제한을 포함한 자세한 내용은 설정 파라미터 표를 참조하십시오.

고급 설정 옵션에 대해서는 다음 수신기 설명서 페이지를 참조하십시오.

프로메테우스 수신기 문서

Kafka 메트릭 수신자 문서

3단계. 매니페스트 배포

bash

$# Create namespace if it doesn't exist
$kubectl create namespace newrelic --dry-run=client -o yaml | kubectl apply -f -
$
$# Apply ConfigMap
$kubectl apply -f collector-configmap.yaml
$
$# Apply Deployment (includes ServiceAccount)
$kubectl apply -f collector-deployment.yaml

4단계. 배포 확인:

bash

$# Check pod status
$kubectl get pods -n newrelic -l app=otel-collector
$
$# View logs to verify metrics collection
$kubectl logs -n newrelic -l app=otel-collector --tail=50

포트 9404의 Kafka 브로커 파드에서 성공적으로 스크래핑되었음을 나타내는 로그가 표시되어야 합니다.

(선택사항) 제작자 또는 소비자를 위해

중요

언어 지원: 자바 애플리케이션은 OpenTelemetry 자바 에이전트를 사용하여 기본적으로 Kafka 클라이언트 계측을 지원합니다.

Kafka 프로듀서 및 소비자 애플리케이션에서 애플리케이션 수준 텔레메트리를 수집하려면 init 컨테이너와 함께 OpenTelemetry 자바 에이전트를 사용하십시오:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: kafka-producer-app
spec:
  template:
    spec:
      initContainers:
      - name: download-java-agent
        image: busybox:latest
        command:
        - sh
        - -c
        - |
          wget -O /otel-auto-instrumentation/opentelemetry-javaagent.jar \
          https://github.com/open-telemetry/opentelemetry-java-instrumentation/releases/latest/download/opentelemetry-javaagent.jar
        volumeMounts:
        - name: otel-auto-instrumentation
          mountPath: /otel-auto-instrumentation

      containers:
      - name: app
        image: your-kafka-app:latest
        env:
        - name: JAVA_TOOL_OPTIONS
          value: >-
            -javaagent:/otel-auto-instrumentation/opentelemetry-javaagent.jar
            -Dotel.service.name=my-kafka-app
            -Dotel.resource.attributes=kafka.cluster.name=my-kafka-cluster
            -Dotel.exporter.otlp.endpoint=http://otel-collector.newrelic.svc.cluster.local: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
            -Dotel.instrumentation.runtime-telemetry.enabled=false
        volumeMounts:
        - name: otel-auto-instrumentation
          mountPath: /otel-auto-instrumentation

      volumes:
      - name: otel-auto-instrumentation
        emptyDir: {}

구성 매개변수

다음 표에서는 주요 설정 시위에 대해 설명합니다.

매개변수	설명
`service.name`	`my-kafka-app` 생산자 또는 소비자 애플리케이션에 대한 고유한 이름으로 바꾸십시오.
`kafka.cluster.name`	`my-kafka-cluster` 수집기 설정에 사용된 것과 동일한 클러스터 이름으로 바꾸세요.
`otlp.endpoint`	엔드포인트 `http://otel-collector.newrelic.svc.cluster.local:4317` 은(는) 수집기가 `newrelic` 네임스페이스에 다음과 같이 배포되었다고 가정합니다 `otel-collector`

(선택 사항) Kafka 브로커 로그 전달

Kafka 브로커 로그를 수집하여 뉴렐릭으로 전송하려면, 수집기 설정에 filelog 수신기를 추가하세요.

1단계. receivers 섹션에 추가:

receivers:
  # ... existing receivers ...

  # File log receiver for Kafka broker logs
  filelog/kafka_broker_0:
    include:
      - /var/log/kafka/server.log
    start_at: end
    multiline:
      line_start_pattern: '^\['
    resource:
      broker.id: "0"
      kafka.cluster.name: ${env:KAFKA_CLUSTER_NAME}

2단계. 서비스 섹션에 로그 파이프라인 추가:

service:
  pipelines:
    # ... existing pipelines ...

    logs/broker:
      receivers: [filelog/kafka_broker_0]
      processors: [memory_limiter, batch/export]
      exporters: [otlp/backend]

구성 매개변수

다음 표에서는 주요 설정 시위에 대해 설명합니다.

매개변수	설명
`filelog/kafka_broker_0.include`	`/var/log/kafka/server.log` 을(를) 브로커 파드 내의 실제 Kafka 로그 경로로 업데이트하세요.
`filelog/kafka_broker_0.resource.broker.id`	`broker.id` 리소스 속성은 로그를 특정 브로커 메트릭 및 엔티티와 연관시킵니다.
다중 브로커 수신기	여러 브로커를 사용하는 경우, 각 브로커 ID를 사용하여 별도의 `filelog` 수신기(예: `filelog/kafka_broker_1`, `filelog/kafka_broker_2`)를 생성하십시오.
`filelog/kafka_broker_0.multiline.line_start_pattern`	`multiline` 패턴은 로그가 `[` (으)로 시작한다고 가정합니다 ― 로그 형식이 다르면 조정하세요.
로그 볼륨	로그인 포워딩을 활성화하기 전에 로그 볼륨 및 수집 비용을 고려하십시오.
참조	전체 설정 옵션에 대한 자세한 내용 은 파일로그 수신기 설명서를참조하십시오.

3단계. Helm 릴리스 업그레이드:

bash

$helm upgrade kafka-otel-collector open-telemetry/opentelemetry-collector \
>  --namespace newrelic \
>  --values values.yaml

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

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

데이터 찾기

몇 분 후 Kafka 데이터가 뉴렐릭에 나타납니다. 뉴렐릭 UI의 여러 뷰에서 Kafka 데이터를 탐색하는 방법에 대한 자세한 지침은 데이터 찾기 를 참조하세요.

다음 표는 각 신호 유형이 저장되는 위치를 요약합니다. 아래의 모든 쿼리에서 my-kafka-cluster 을(를) KAFKA_CLUSTER_NAME 값으로 바꾸십시오:

시그널	이벤트 유형	포함 사항
메트릭	`Metric`	브로커, 토픽, 파티션, 소비자 그룹 및 JVM 메트릭
로그	`Log`	생산자 및 소비자 애플리케이션의 로그(OTel 자바 에이전트를 통해) 및 자바 에이전트를 통해 수집된 브로커 로그
트레이스	`Span`	토픽 전반에 걸친 메시지당 `publish` 및 `receive` 작업을 포함하는 생산자 및 소비자 스팬

메트릭

브로커, 토픽, 파티션, 소비자 그룹 및 JVM 메트릭은 Metric 이벤트 유형에 저장됩니다:

FROM Metric SELECT * WHERE kafka.cluster.name = 'my-kafka-cluster' SINCE 30 minutes ago

로그

OpenTelemetry 자바 에이전트로 계측된 생산자 및 소비자 애플리케이션의 로그와 브로커의 자바 에이전트를 통해 수집된 브로커 로그는 Log 이벤트 유형에 저장됩니다:

FROM Log SELECT * WHERE kafka.cluster.name = 'my-kafka-cluster' SINCE 30 minutes ago

트레이스

토픽 전반의 메시지당 publish 및 receive 작업을 포함한 생산자 및 소비자 스팬은 Span 이벤트 유형에 저장됩니다:

FROM Span SELECT * WHERE kafka.cluster.name = 'my-kafka-cluster' SINCE 30 minutes ago

예시

Kafka StatefulSet 매니페스트, Helm 값, OTel Collector 설정 및 샘플 생산자/소비자 애플리케이션이 포함된 완전한 작동 예제는 뉴렐릭 OpenTelemetry 예제 저장소에서 확인할 수 있습니다.

문제점 해결

설정을 확인하려면 먼저 다음 명령어를 실행하세요. 결과를 사용하여 따라야 할 특정 문제 해결, 해결 섹션을 식별합니다.

수집기 파드가 실행 중인지 확인:

매니페스트 설치의 경우:

bash

$kubectl get pods -n newrelic -l app=otel-collector
$kubectl logs -n newrelic -l app=otel-collector --tail=50

Helm 설치의 경우(helm upgrade ... kafka-monitoring):

bash

$kubectl get pods -n newrelic -l app.kubernetes.io/name=opentelemetry-collector
$kubectl logs -n newrelic -l app.kubernetes.io/name=opentelemetry-collector --tail=50

Kafka 브로커 파드가 자바 에이전트와 함께 실행 중인지 확인:

bash

$# List broker pods
$kubectl get pods -n kafka -l app=kafka
$
$# Check env vars on a broker pod (should see KAFKA_OPTS with javaagent)
$kubectl exec -n kafka kafka-0 -- env | grep KAFKA_OPTS
$
$# Check if init container completed successfully
$kubectl describe pod -n kafka kafka-0 | grep -A5 "Init Containers"

otel-agent 볼륨이 채워져 있는지 확인합니다:

bash

$kubectl exec -n kafka kafka-0 -- ls -lh /otel-agent/

브로커 파드에서 수집기 서비스로의 연결 테스트:

bash

$kubectl exec -n kafka kafka-0 -- nc -zv otel-collector.newrelic.svc.cluster.local 4317 && echo "Port reachable" || echo "Cannot reach collector"

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

ConfigMap(collector-configmap.yaml)에서 service 섹션에 추가합니다:

service:
  telemetry:
    logs:
      level: "debug"

그런 다음 업데이트된 ConfigMap을 적용하고 수집기 배포를 다시 시작합니다:

bash

$kubectl apply -f collector-configmap.yaml
$kubectl rollout restart deployment/otel-collector -n newrelic

디버그 익스포터 추가: 뉴렐릭으로 전송하기 전에 수집기 로그에서 메트릭을 확인합니다. 프로세서 및 익스포터 이름은 모니터링 방법에 따라 다릅니다:

자바 에이전트 메서드:

exporters:
  debug:
    verbosity: detailed

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

service:
  pipelines:
    metrics/broker:
      receivers: [otlp, kafkametrics]
      processors: [resource, filter/exclude_cluster_metrics, filter/internal_topics, transform/remove_extra_attributes, transform/des_units, cumulativetodelta, metricstransform/kafka_topic_sum_aggregation, filter/remove_partition_level_replicas, batch/aggregation]
      exporters: [debug, otlp/newrelic]

    metrics/cluster:
      receivers: [otlp]
      processors: [resource, filter/include_cluster_metrics, transform/remove_broker_id, transform/remove_extra_attributes, transform/des_units, cumulativetodelta, groupbyattrs/cluster, metricstransform/cluster_max, batch/aggregation]
      exporters: [debug, otlp/newrelic]

Prometheus JMX Exporter 방식:

exporters:
  debug:
    verbosity: detailed

  otlp/backend:
    endpoint: ${NEW_RELIC_OTLP_ENDPOINT}
    headers:
      api-key: ${NEW_RELIC_LICENSE_KEY}

service:
  pipelines:
    metrics/broker:
      receivers: [prometheus/kafka-jmx]
      processors: [resource/cluster-name, filter/scrape-overhead, transform/metric-naming, transform/remove_attributes, filter/exclude_cluster_metrics, memory_limiter, cumulativetodelta, batch/export]
      exporters: [debug, otlp/backend]

    metrics/cluster/prometheus:
      receivers: [prometheus/kafka-jmx]
      processors: [resource/cluster-name, filter/scrape-overhead, transform/metric-naming, transform/remove_attributes, filter/include_cluster_metrics, transform/remove_broker_id, memory_limiter, cumulativetodelta, groupbyattrs/cluster, metricstransform/cluster_max, batch/export]
      exporters: [debug, otlp/backend]

    metrics/cluster/kafkametrics:
      receivers: [kafkametrics/cluster]
      processors: [resource/cluster-name, filter/internal_topics, transform/remove_attributes, metricstransform/topic-aggregation, filter/exclude_partition_replicas_metric, memory_limiter, cumulativetodelta, batch/export]
      exporters: [debug, otlp/backend]

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

먼저, 초기 시스템 검사를 실행하여 수집기 파드와 브로커 파드가 정상인지 확인하세요.

수집기 로그에서 오류 확인 (설치 방법과 일치하는 레이블 사용 ― 초기 시스템 확인 참조):

bash

$# Manifest
$kubectl logs -n newrelic -l app=otel-collector --tail=100 | grep -i "error\|fail\|refuse"
$
$# Helm
$kubectl logs -n newrelic -l app.kubernetes.io/name=opentelemetry-collector --tail=100 | grep -i "error\|fail\|refuse"

수집기 서비스가 존재하고 올바른 포트를 사용하는지 확인합니다:

bash

$# Manifest
$kubectl get svc otel-collector -n newrelic
$
$# Helm
$kubectl get svc -n newrelic -l app.kubernetes.io/name=opentelemetry-collector

포트 4317 이(가) ClusterIP 서비스로 노출되어 있는지 확인하십시오.

먼저, 초기 시스템 검사를 실행하여 자바 에이전트가 브로커 파드에 연결되어 있는지 확인하세요.

자바 에이전트 초기화를 위한 브로커 파드 로그 확인:

bash

$kubectl logs -n kafka kafka-0 --tail=100 | grep -i "otel\|jmx"

브로커 파드에서 KAFKA_OPTS가 올바르게 설정되어 있는지 확인합니다:

bash

$kubectl exec -n kafka kafka-0 -- env | grep KAFKA_OPTS

-javaagent:/otel-agent/opentelemetry-javaagent.jar 및 모든 -Dotel.* 파라미터가 표시되어야 합니다. 확인:

-Dotel.jmx.enabled=true
-Dotel.jmx.config=/jmx-config/kafka-jmx-config.yaml
-Dotel.exporter.otlp.endpoint=http://otel-collector.newrelic.svc.cluster.local:4317

JMX ConfigMap이 마운트되었는지 확인:

bash

$kubectl exec -n kafka kafka-0 -- ls -lh /jmx-config/
$kubectl exec -n kafka kafka-0 -- cat /jmx-config/kafka-jmx-config.yaml

수집기 로그에서 수신되는 JMX 메트릭을 확인하십시오.

bash

$kubectl logs -n newrelic -l app=otel-collector --tail=100 | grep -i "broker.id\|kafka\|jmx"

브로커 파드에서 수집기 서비스에 연결할 수 있는지 확인하려면 먼저, 초기 시스템 검사를 실행하세요.

DNS 해석 확인:

bash

$kubectl exec -n kafka kafka-0 -- nslookup otel-collector.newrelic.svc.cluster.local

OTLP 오류에 대한 수집기 로그 확인:

bash

$kubectl logs -n newrelic -l app=otel-collector --tail=100 | grep -i "connection refused\|context deadline exceeded\|failed to connect"

OTLP 수신기가 모든 인터페이스에서 수신 대기 중인지 확인:

ConfigMap의 otlp receiver에 127.0.0.1이(가) 아닌 endpoint: "0.0.0.0:4317" 이(가) 있는지 확인하십시오:

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

수집기 파드 메모리 모니터:

bash

$# Manifest
$kubectl top pod -n newrelic -l app=otel-collector
$
$# Helm
$kubectl top pod -n newrelic -l app.kubernetes.io/name=opentelemetry-collector

모니터되는 토픽 줄이기:

receivers:
  kafkametrics:
    brokers: ["kafka-0.kafka-headless.kafka.svc.cluster.local:9092"]
    collection_interval: 30s
    scrapers:
      - brokers
      - topics
      - consumers
    topic_match: "^(important-topic-1|important-topic-2)$"

수집 빈도 감소: 간격을 늘려 수집 횟수를 줄입니다.

receivers:
  kafkametrics:
    collection_interval: 60s

자바 에이전트의 JMX 메트릭의 경우 StatefulSet에서 KAFKA_OPTS 을(를) 업데이트하세요:

- name: KAFKA_OPTS
  value: >-
    ...
    -Dotel.metric.export.interval=60000

메모리 리미터 추가:

자바 에이전트 메서드:

processors:
  memory_limiter:
    check_interval: 1s
    limit_mib: 512
    spike_limit_mib: 128

service:
  pipelines:
    metrics/broker:
      processors: [memory_limiter, resource, filter/exclude_cluster_metrics, filter/internal_topics, transform/remove_extra_attributes, transform/des_units, cumulativetodelta, metricstransform/kafka_topic_sum_aggregation, filter/remove_partition_level_replicas, batch/aggregation]
      ...

Prometheus JMX Exporter 방식:

processors:
  memory_limiter:
    check_interval: 1s
    limit_mib: 512
    spike_limit_mib: 128

service:
  pipelines:
    metrics/broker:
      processors: [memory_limiter, resource/cluster-name, filter/scrape-overhead, transform/metric-naming, transform/remove_attributes, filter/exclude_cluster_metrics, cumulativetodelta, batch/export]
      ...

변경 후, 업데이트된 ConfigMap을 적용하고 수집기를 다시 시작합니다:

bash

$kubectl apply -f collector-configmap.yaml
$kubectl rollout restart deployment/otel-collector -n newrelic

다음 단계

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

자체 호스팅 Kafka - 자체 호스팅(비 Kubernetes) 환경을 위한 Kafka 모니터링
Kubernetes Strimzi - Kubernetes의 Strimzi 관리형 Kafka를 위한 Kafka 모니터링
OpenTelemetry Java 에이전트 - OTel 자바 에이전트 공식 문서
Prometheus JMX Exporter - Prometheus 형식으로 JMX 메트릭을 노출하는 자바 에이전트
Prometheus 리시버 - Prometheus 메트릭 엔드포인트를 스크랩하기 위한 OTel Collector 리시버
kafkametrics 수신기 - 소비자 지연 및 토픽 메트릭 수신기 문서

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

OpenTelemetry를 사용하여 Kubernetes에서 자체 관리형 Kafka 모니터링

아키텍처 .css-21sua1{background:none;border:none;width:0;padding:0;}

설치 단계

시작하기 전에

구현하다, 배포하다 OpenTelemetry Collector

EU 지역

팁

NRDOT Collector 사용(권장)

OpenTelemetry Collector 사용

미국 지역

EU 지역

팁

NRDOT Collector 사용(권장)

OpenTelemetry Collector 사용

자바 에이전트용 Kafka StatefulSet 구성

팁

구성 매개변수

(선택사항) 제작자 또는 소비자를 위해

중요

구성 매개변수

시작하기 전에

JMX 메트릭 ConfigMap을 생성합니다.

팁

JMX Exporter용 Kafka StatefulSet 구성

중요

구현하다, 배포하다 OpenTelemetry Collector

미국 지역

EU 지역

팁

NRDOT Collector 사용(권장)

OpenTelemetry Collector 사용

미국 지역

EU 지역

팁

NRDOT Collector 사용(권장)

OpenTelemetry Collector 사용

(선택사항) 제작자 또는 소비자를 위해

중요

(선택 사항) Kafka 브로커 로그 전달

로그 수집 구성

뉴렐릭에서 내 로그인 찾기

데이터 찾기

메트릭

로그

트레이스

예시

문제점 해결

초기 시스템 점검

디버그 로깅 활성화

뉴릭에 데이터가 나타나지 않습니다

Kafka 브로커에서 JMX 메트릭이 누락되었습니다.

브로커 파드에서 발생하는 OTLP 연결 오류

높은 메모리 사용량

다음 단계

관련 리소스

아키텍처