OpenTelemetryを使用したKubernetes上のセルフマネージドKafkaのモニター

OpenTelemetry Collectorをデプロイしてメトリクスを収集し、New Relicに転送することで、Kubernetes上で実行されている自己管理型のApache Kafkaクラスタをモニターします。

アーキテクチャー

New Relicは、セルフマネージドKubernetes Kafkaを監視するための2つのアプローチをサポートしています：OpenTelemetry JavaエージェントまたはPrometheus JMX Exporterです。以下の図は、各アプローチのデータフローを示しています。

Kubernetes self-managed Kafka monitoring architecture

インストレーション手順

以下の手順に従って、ブローカーにOpenTelemetry Javaエージェントをインストールし、メトリクスとログを収集してNew Relicに送信するためのコレクターをデプロイして、包括的なKafka監視をセットアップします。

あなたが始める前に

以下のものを用意してください:

New Relicアカウント
kubectlアクセスを持つKubernetesクラスタ
StatefulSetとしてデプロイされたKafka
Kafka StatefulSet を変更および再デプロイする機能

OpenTelemetry Collector をデプロイする

クラスタにOpenTelemetryコレクターをデプロイします。このステップでは、Javaエージェントが各ブローカーポッドからどのJMXメトリクスを収集するかを定義するkafka-jmx-configConfigMapも作成します。次の手順でKafkaブローカーを再起動する前に、コレクターが実行中である必要があります。

ステップ1. New Relicの認証情報シークレットを作成する

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 は、New Relic がサポートする OpenTelemetry Collector のディストリビューションであり、完全な New Relic サポートを提供します。詳細については、 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. New Relicの認証情報シークレットを作成する

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-configConfigMapを適用する必要があります。

kafka-jmx-config.yamlを作成 - Javaエージェントの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

Javaエージェント用にKafka StatefulSetを設定する

コレクターが実行中になったので、Kafka StatefulSetにパッチを適用してOpenTelemetry Javaエージェントの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-configConfigMapはコレクターとともにデプロイされました。otel.exporter.otlp.endpointの値http://otel-collector.newrelic.svc.cluster.local:4317は、コレクターがサービス名otel-collectorでnewrelicネームスペースにデプロイされていることを前提としています。異なる場合は、実際のコレクターサービスのDNSと一致するように更新してください。

パラメータ	説明
`javaagent`	OpenTelemetry Javaエージェントを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秒）

完全な設定オプションについては、Javaエージェント設定ガイドを参照してください。

更新したStatefulSetを適用し、ポッドがロールするのを待ちます：

bash

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

(オプション) 計装プロデューサーまたは消費者アプリケーション

重要

言語サポート: 現在、OpenTelemetry Javaエージェントを使用したKafkaクライアントの計装でサポートされているのはJavaアプリケーションのみです。

Kubernetesで実行されているKafkaプロデューサーおよび消費者アプリケーションからアプリケーションレベルのテレメトリーを収集するには、それらのアプリケーションポッドにOpenTelemetry Javaエージェントを追加します。

アプリケーションのデプロイメントに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

）に置き換えます

Javaエージェントは、コードを変更することなくすぐに使えるKafkaの計装を提供し、リクエストレイテンシ、スループットメトリクス、エラー率、およびディストリビューティッド（分散）トレーシングをキャプチャします。高度な設定については、Kafka計装ドキュメントを参照してください。

ブローカーポッドにPrometheus JMX Exporterをインストールし、メトリクスを収集してNew Relicに送信するコレクターをデプロイすることで、包括的なKafka監視をセットアップするには、以下の手順に従ってください。

あなたが始める前に

以下のものを用意してください:

New Relicアカウント
kubectlアクセスを持つKubernetesクラスタ
ヘッドレスサービス（安定したポッド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. New Relicの認証情報シークレットを作成する

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クラスタ名に置き換えてください（これはNew Relicでメトリクスを識別およびフィルタリングするために使用されます）
`config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	ブローカーポッドと一致するように`targets`リストを更新します — ヘッドレスサービスDNS（`<pod-name>.<headless-service>.<namespace>.svc.cluster.local:9404`）を使用してブローカーごとに1つのエントリ。ブローカー数に合わせてエントリを追加または削除します。`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`）を使用して、ブローカーポッドごとに1つのエントリを追加します。ブローカー数に合わせてエントリを追加または削除してください。`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`	Kafkaクラスタ名に置き換えてください（New Relicでメトリクスを識別するために`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'

高度な設定オプションについては、次の受信機のドキュメントページを参照してください。

Prometheusレシーバーのドキュメント

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. New Relicの認証情報シークレットを作成する

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クラスタ名に置き換えてください（New Relicでメトリクスを識別するために使用されます）
`receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	ブローカーポッドと一致するように`targets`リストを更新します — ヘッドレスサービスDNS（`<pod-name>.<headless-service>.<namespace>.svc.cluster.local:9404`）を使用してブローカーごとに1つのエントリ。`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オプションと同じパラメーター。リソース制限を含む詳細については、設定パラメーター表を参照してください。

高度な設定オプションについては、次の受信機のドキュメントページを参照してください。

Prometheusレシーバーのドキュメント

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ブローカーポッドからのスクレイピングが成功したことを示すログが表示されるはずです。

(オプション) 計装プロデューサーまたは消費者アプリケーション

重要

言語サポート: Javaアプリケーションは、OpenTelemetry Javaエージェントを使用したKafkaクライアントの計装を標準でサポートしています。

Kafkaプロデューサーおよび消費者アプリケーションからアプリケーションレベルのテレメトリーを収集するには、initコンテナでOpenTelemetry Javaエージェントを使用します：

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ブローカーのログを収集してNew Relicに送信するには、コレクターの設定に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セクションにログパイプラインを追加します：

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 ブローカーログは次の 2 つの場所に表示されます。

ブローカーエンティティ: New Relicの Kafka ブローカーエンティティに移動して、その特定のブローカーに関連付けられたログを表示します。
ログUI : 次のようなフィルターを備えたログUIを使用して、すべてのKafkaログを書き込みます。 kafka.cluster.name = 'my-cluster'
NRQL を使用してログをクエリすることもできます。
```
FROM Log SELECT * WHERE kafka.cluster.name = 'my-kafka-cluster'
```

データを検索する

数分後、New Relic に Kafka データが表示されるはずです。New Relic UIのさまざまなビューでKafkaデータを探索するための詳細な手順については、データの検索を参照してください。

以下の表は、各シグナルタイプの保存先をまとめています。以下のすべてのクエリで、my-kafka-clusterをKAFKA_CLUSTER_NAMEの値に置き換えます：

シグナル	イベントタイプ	含まれるもの
指標	`Metric`	ブローカー、トピック、パーティション、消費者グループ、およびJVMメトリクス
ログ	`Log`	プロデューサーおよび消費者アプリケーションからのログ（OTel Javaエージェント経由）と、Javaエージェント経由で収集されたブローカーログ
トレース	`Span`	トピックをまたぐメッセージごとの`publish`および`receive`操作を含む、プロデューサーと消費者のスパン

指標

ブローカー、トピック、パーティション、消費者グループ、およびJVMメトリクスは、Metricイベントタイプに格納されます：

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

ログ

OpenTelemetry Javaエージェントでインストゥルメントされたプロデューサーおよび消費者アプリケーションからのログ、およびブローカー上のJavaエージェントを介して収集されたブローカーログは、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の設定、およびサンプルのプロデューサー/消費者アプリケーションを含む完全な動作例は、New Relic OpenTelemetry Examplesリポジトリで利用可能です。

トラブルシューティング

最初にこれらのコマンドを実行してセットアップを確認してください。結果を使用して、どのトラブルシューティングセクションに従うべきかを特定します。

コレクターポッドが実行されているか確認します:

マニフェストによるインストールの場合：

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ブローカーポッドがJavaエージェントで実行されているか確認する:

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

デバッグエクスポーターを追加: New Relicに送信する前に、コレクターログでメトリクスを表示します。プロセッサーとエクスポーターの名前は、監視方法によって異なります：

Javaエージェントメソッド:

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"

コレクターServiceが存在し、正しいポートが設定されていることを確認します:

bash

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

ポート4317がClusterIPサービスとして公開されていることを確認してください。

まず、初期システムチェックを実行して、Javaエージェントがブローカーポッドにアタッチされていることを確認します。

ブローカーポッドのログでJavaエージェントの初期化を確認します:

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"

まず、初期システムチェックを実行して、コレクターServiceがブローカーポッドから到達可能であることを確認します。

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レシーバーに（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

JavaエージェントからのJMXメトリクスについては、StatefulSetのKAFKA_OPTSを更新します：

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

メモリリミッターを追加する:

Javaエージェントのメソッド：

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 Javaエージェントの公式ドキュメント
Prometheus JMX Exporter - Prometheus形式でJMXメトリクスを公開するJavaエージェント
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の使用

Javaエージェント用にKafka StatefulSetを設定する

ヒント

設定パラメーター

(オプション) 計装プロデューサーまたは消費者アプリケーション

重要

設定パラメーター

あなたが始める前に

JMX メトリクス ConfigMap の作成

ヒント

JMX Exporter 用に Kafka StatefulSet を設定する

重要

OpenTelemetry Collector をデプロイする

米国地域

EU地域

ヒント

NRDOT Collector の使用 (推奨)

OpenTelemetry Collectorの使用

米国地域

EU地域

ヒント

NRDOT Collector の使用 (推奨)

OpenTelemetry Collectorの使用

(オプション) 計装プロデューサーまたは消費者アプリケーション

重要

（オプション）Kafkaブローカーログを転送する

ログ収集を構成する

New Relicでログを見つける

データを検索する

指標

ログ

トレース

例

トラブルシューティング

初期システムチェック

デバッグログを有効にする

New Relicにデータが表示されない

Kafka ブローカーから JMX メトリクスが欠落している

ブローカーポッドからのOTLP接続エラー

メモリ使用量が多い

次のステップ

関連資料

アーキテクチャー