OpenTelemetryを使用したKubernetes (Strimzi) 上の Kafka の監視

OpenTelemetry Collectorデプロイして、Strimzi オペレーターを使用してKubernetes上で実行されている Kafka クラスターを監視します。コレクターは、Kafka ブローカー Pod を自動的に検出し、包括的なメトリクスを収集します。

アーキテクチャー

次の図は、監視アーキテクチャーとNew Relicへのデータの流れを示しています。

Kubernetes Strimzi Kafka monitoring architecture with OpenTelemetry

インストレーション手順

Kafka クラスターの監視を設定するには、次の手順に従います。

あなたが始める前に

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

New Relicアカウント
kubectl アクセスを使用したKubernetesクラスター
Strimzi オペレーター経由でデプロイされた Kafka

Kafka JMX メトリクス用に Kafka クラスターを構成する

Prometheus JMX Exporter を介して Kafka JMX メトリクスを公開するように Strimzi Kafka クラスタを構成します。この設定は ConfigMap として展開され、Kafka クラスタによって参照されます。

ステップ 1。JMXメトリクスConfigMapを作成する

どの Kafka メトリクスを収集するかを定義する JMX Exporter パターンを使用して ConfigMap を作成します。 kafka-jmx-config.yamlとして保存:

apiVersion: v1
kind: ConfigMap
metadata:
  name: kafka-jmx-metrics
  namespace: newrelic
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"

ヒント

メトリクスのカスタマイズ: この ConfigMap には、包括的な Kafka ブローカー、トピック、リクエスト、コントローラー、およびJVMメトリクスが含まれています。 Prometheus JMX エクスポーターの例とKafka MBean のドキュメントを参照して、パターンを追加または変更できます。追加の設定については、 JMX エクスポーターのルールのドキュメントを参照してください。

重要

ネームスペース要件: JMX メトリクス ConfigMap と Kafka クラスタは同じネームスペースに存在する必要があります。このガイドでは、両方ともnewrelicネームスペースにデプロイされています。

ConfigMap を適用します。

bash

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

ステップ2。JMX Exporterを使用するようにKafkaクラスタを更新する

メトリクス ConfigMap を参照するように Strimzi Kafka リソースを更新します。

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
  namespace: newrelic
spec:
  kafka:
    version: X.X.X
    metricsConfig:
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          name: kafka-jmx-metrics
          key: kafka-metrics-config.yml
    # ...rest of your Kafka configuration

変更を適用します。Strimzi は Kafka ブローカーのローリング再起動を実行します。

bash

$kubectl apply -f kafka-cluster.yaml

ローリング再起動が完了すると、各 Kafka ブローカーはポート9404で Prometheus メトリクスを公開します。

OpenTelemetry Collector をデプロイする

OpenTelemetry Collectorデプロイして、Kafka クラスタを監視します。ご希望の設置方法を選択してください:

Helmメソッドは、 KubernetesのデプロイOpenTelemetry Collectorに推奨されるアプローチです。

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

New Relicライセンスキーと OTLP エンドポイントを含むKubernetesシークレットを作成します。 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を作成する

完全なOpenTelemetry Collector設定を含むvalues.yamlファイルを作成します。 NRDOT とOpenTelemetry Collector は両方とも同一の設定を使用し、同じ Kafka 監視機能を提供します。ご希望のコレクター画像を選択してください:

NRDOT は、New Relic がサポートする OpenTelemetry Collector のディストリビューションであり、完全な New Relic サポートを提供します。詳細については、 NRDOT Collector GitHub リポジトリを参照してください。

次の内容でvalues.yamlを作成します:

# Deployment mode
mode: deployment
replicaCount: 1

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

# Service account configuration
serviceAccount:
  create: true
  name: otel-collector

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

# 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

    kafkametrics/cluster:
      brokers:
        # TODO#1: Replace with your Kafka bootstrap service
        - "my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092"
      collection_interval: 30s
      protocol_version: 2.0.0
      scrapers:
        - brokers
        - topics
        - consumers
      # Exclude internal Kafka topics (prefixed with __) at the source
      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'
            scrape_interval: 30s
            kubernetes_sd_configs:
              - role: pod
                namespaces:
                  names:
                    # TODO#2: Replace with the namespace where your Kafka cluster is deployed
                    - newrelic
            relabel_configs:
              # Filter for Kafka broker pods
              - source_labels: [__meta_kubernetes_pod_label_strimzi_io_name]
                action: keep
                # TODO#3: Replace with your Strimzi Kafka cluster name followed by '-kafka'
                regex: my-cluster-kafka

              - source_labels: [__meta_kubernetes_pod_label_strimzi_io_cluster]
                action: keep
                # TODO#4: Replace with your Strimzi Kafka cluster name
                regex: my-cluster

              # Extract broker ID from pod name
              - 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'

  exporters:
    # New Relic OTLP exporter
    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 processor for efficient export
    batch/export:
      send_batch_size: 1024
      timeout: 30s

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

    # Transform metric naming conventions
    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

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

    # Remove broker.id for cluster-level metrics
    transform/remove_broker_id:
      metric_statements:
        - context: datapoint
          statements:
            - delete_key(attributes, "broker.id")

    # Filter out scrape overhead metrics
    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)$"

    # Include only cluster-level metrics for cluster pipeline
    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$"

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

    # Remove unnecessary attributes
    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")

    # Aggregate partition metrics to topic level
    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 out original partition replicas metric
    filter/exclude_partition_replicas_metric:
      metrics:
        exclude:
          match_type: strict
          metric_names:
            - kafka.partition.replicas_in_sync

    # Filter internal Kafka topics as a safety net (topic_match handles the receiver side)
    filter/internal_topics:
      metrics:
        datapoint:
          - 'attributes["topic"] != nil and IsMatch(attributes["topic"], "^__.*")'

    # Convert cumulative to delta metrics
    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:
      # Suppress default pipelines — only custom Kafka metrics pipelines are used
      traces: null
      logs: null
      metrics: null

      # 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
      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ブートストラップサービスに置き換えてください
`config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].kubernetes_sd_configs[0].namespaces.names`	Kafka クラスタがデプロイされているネームスペースに置き換えます。
`config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].relabel_configs[strimzi_io_name].regex`	ご使用のStrimzi Kafkaクラスタ名に置き換え、その後に `-kafka`
`config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].relabel_configs[strimzi_io_cluster].regex`	ご使用のStrimzi Kafkaクラスタ名に置き換えてください
`config.processors.resource/cluster-name.attributes[kafka.cluster.name].value`	Kafkaクラスタ名に置き換えてください（これはNew Relicでメトリクスを識別およびフィルタリングするために使用されます）
`resources.limits` および `resources.requests`	ワークロードのニーズに応じて調整してください

最大限の柔軟性とベンダーニュートラルのプロイメントを実現するには、コミュニティ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 configuration
serviceAccount:
  create: true
  name: otel-collector

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

# 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

# OpenTelemetry Collector Configuration
config:
  receivers:
    # Kafka metrics receiver for cluster-level metrics
    kafkametrics/cluster:
      brokers:
        # TODO#1: Replace with your Kafka bootstrap service
        - "my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092"
      collection_interval: 30s
      protocol_version: 2.0.0
      scrapers:
        - brokers
        - topics
        - consumers
      # Exclude internal Kafka topics (prefixed with __) at the source
      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 for JMX metrics from Kafka brokers
    prometheus/kafka-jmx:
      config:
        scrape_configs:
          - job_name: 'kafka-jmx-metrics'
            scrape_interval: 30s
            kubernetes_sd_configs:
              - role: pod
                namespaces:
                  names:
                    # TODO#2: Replace with the namespace where your Kafka cluster is deployed
                    - newrelic
            relabel_configs:
              # Filter for Kafka broker pods
              - source_labels: [__meta_kubernetes_pod_label_strimzi_io_name]
                action: keep
                # TODO#3: Replace with your Strimzi Kafka cluster name followed by '-kafka'
                regex: my-cluster-kafka

              - source_labels: [__meta_kubernetes_pod_label_strimzi_io_cluster]
                action: keep
                # TODO#4: Replace with your Strimzi Kafka cluster name
                regex: my-cluster

              # Extract broker ID from pod name
              - 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'

  exporters:
    # New Relic OTLP exporter
    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 processor for efficient export
    batch/export:
      send_batch_size: 1024
      timeout: 30s

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

    # Transform metric naming conventions
    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

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

    # Remove broker.id for cluster-level metrics
    transform/remove_broker_id:
      metric_statements:
        - context: datapoint
          statements:
            - delete_key(attributes, "broker.id")

    # Filter out scrape overhead metrics
    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)$"

    # Include only cluster-level metrics for cluster pipeline
    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$"

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

    # Remove unnecessary attributes
    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")

    # Aggregate partition metrics to topic level
    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 out original partition replicas metric
    filter/exclude_partition_replicas_metric:
      metrics:
        exclude:
          match_type: strict
          metric_names:
            - kafka.partition.replicas_in_sync

    # Filter internal Kafka topics as a safety net (topic_match handles the receiver side)
    filter/internal_topics:
      metrics:
        datapoint:
          - 'attributes["topic"] != nil and IsMatch(attributes["topic"], "^__.*")'

    # Convert cumulative to delta metrics
    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:
      # Override default pipelines to only use custom Kafka metrics pipelines
      traces: null
      logs: null
      metrics: null

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

設定パラメーター: 上記のNRDOTオプションと同じパラメーター。リソース制限を含む詳細については、設定パラメーター表を参照してください。

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

Prometheus レシーバーのドキュメント- 追加のレシーバー設定オプション
Kafka メトリクス受信機のドキュメント - 追加の Kafka メトリクス設定
ステップ3. Helmを使用してOpenTelemetry Collectorをインストールする
Helm リポジトリを追加し、values.yaml ファイルを使用して 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の認証情報シークレットを作成する

New Relicライセンスキーと OTLP エンドポイントを含むKubernetesシークレットを作成します。 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. マニフェストファイルを作成する

優先コレクター用の Kubernetes マニフェストファイルを作成します。どちらのコレクターも同じ設定を使用します - 画像のみが異なります。

コレクターオプションを選択し、必要な 3 つのファイルを作成します。

ステップ1. collector-rbac.yamlを作成する - Kubernetes API アクセスの RBAC 設定:

---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: otel-collector
  namespace: newrelic
  labels:
    app: otel-collector
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: otel-collector
  labels:
    app: otel-collector
rules:
- apiGroups: [""]
  resources: ["pods", "nodes"]
  verbs: ["get", "list", "watch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: otel-collector
  labels:
    app: otel-collector
subjects:
- kind: ServiceAccount
  name: otel-collector
  namespace: newrelic
roleRef:
  kind: ClusterRole
  name: otel-collector
  apiGroup: rbac.authorization.k8s.io

ステップ2. 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:
  kafkametrics/cluster:
    brokers:
      # TODO#1: Replace with your Kafka bootstrap service
      - "my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092"
    collection_interval: 30s
    protocol_version: 2.0.0
    scrapers:
      - brokers
      - topics
      - consumers
    # Exclude internal Kafka topics (prefixed with __) at the source
    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'
          scrape_interval: 30s
          kubernetes_sd_configs:
            - role: pod
              namespaces:
                names:
                  # TODO#2: Replace with the namespace where your Kafka cluster is deployed
                  - kafka
          relabel_configs:
            - source_labels: [__meta_kubernetes_pod_label_strimzi_io_name]
              action: keep
              # TODO#3: Replace with your Strimzi Kafka cluster name followed by '-kafka'
              regex: my-cluster-kafka
            - source_labels: [__meta_kubernetes_pod_label_strimzi_io_cluster]
              action: keep
              # TODO#4: Replace with your Strimzi Kafka cluster name
              regex: my-cluster
            - source_labels: [__meta_kubernetes_pod_name]
              target_label: broker.id
              regex: '.*-(\\d+)$'
              replacement: '$1'
            - source_labels: [__meta_kubernetes_pod_ip]
              target_label: __address__
              replacement: '$1:9404'

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 (used to identify and filter metrics in New Relic)
      value: my-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 Kafka topics as a safety net (topic_match handles the receiver side)
  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:
    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]

ステップ3. collector-deployment.yamlを作成 - OpenTelemetry 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"
        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/cluster.brokers`	お使いのKafkaブートストラップサービスに置き換えてください
`receivers.prometheus/kafka-jmx.config.scrape_configs[0].kubernetes_sd_configs[0].namespaces.names`	Kafka クラスタがデプロイされているネームスペースに置き換えます。
`receivers.prometheus/kafka-jmx.config.scrape_configs[0].relabel_configs[strimzi_io_name].regex`	ご使用のStrimzi Kafkaクラスタ名に置き換え、その後に `-kafka`
`receivers.prometheus/kafka-jmx.config.scrape_configs[0].relabel_configs[strimzi_io_cluster].regex`	ご使用のStrimzi Kafkaクラスタ名に置き換えてください
`processors.resource/cluster-name.attributes[kafka.cluster.name].value`	Kafkaクラスタ名に置き換えてください（これはNew Relicでメトリクスを識別およびフィルタリングするために使用されます）

ベンダーニュートラルデプロイメントにはコミュニティOpenTelemetry Collector使用してください。

ステップ1. collector-rbac.yamlを作成 - 上記のNRDOTオプションと同じ（RBAC設定は同一）

ステップ2. collector-configmap.yaml作る- 上記のNRDOTオプションと同じ（設定は同一）

ステップ3. collector-deployment.yaml 作る - OpenTelemetry 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"
        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オプションと同じパラメーター。詳細については、設定パラメーターの表を参照してください。

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

Prometheus レシーバーのドキュメント- 追加のレシーバー設定オプション

Kafka メトリクス受信機のドキュメント - 追加の Kafka メトリクス設定

ステップ3。マニフェストをデプロイする

Kubernetes マニフェストを適用して OpenTelemetry Collector をデプロイします。

bash

$# Create namespace if it doesn't exist
$kubectl create namespace newrelic --dry-run=client -o yaml | kubectl apply -f -
$
$# Apply RBAC configuration
$kubectl apply -f collector-rbac.yaml
$
$# Apply ConfigMap
$kubectl apply -f collector-configmap.yaml
$
$# Apply Deployment
$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プロデューサーおよび消費者アプリケーションからアプリケーションレベルのテレメトリーを収集するには、OpenTelemetry Javaエージェントを使用します。

Kafka アプリケーションを計装する

実行時にOpenTelemetry Javaエージェントをダウンロードするには、initコンテナを使用します：

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=order-process-service
            -Dotel.resource.attributes=kafka.cluster.name=my-cluster
            -Dotel.exporter.otlp.endpoint=http://localhost:4317
            -Dotel.exporter.otlp.protocol=grpc
            -Dotel.metrics.exporter=otlp
            -Dotel.traces.exporter=otlp
            -Dotel.logs.exporter=otlp
            -Dotel.instrumentation.kafka.experimental-span-attributes=true
            -Dotel.instrumentation.messaging.experimental.receive-telemetry.enabled=true
            -Dotel.instrumentation.kafka.producer-propagation.enabled=true
            -Dotel.instrumentation.kafka.enabled=true
        volumeMounts:
        - name: otel-auto-instrumentation
          mountPath: /otel-auto-instrumentation

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

設定パラメーター

次の表では、キー設定について説明します。

パラメータ	説明
`service.name`	`order-process-service`プロデューサーまたは消費者アプリケーションの一意の名前に置き換えます。
`kafka.cluster.name`	`my-cluster`コレクター設定で使用されているのと同じクラスタ名に置き換えます。
`otlp.endpoint`	エンドポイント`http://localhost:4317`は、コレクターが同じポッド内のサイドカーとして実行されているか、ローカルホスト経由でアクセス可能であることを前提としています。

ヒント

上記の設定はテレメトリーをOpenTelemetry Collectorに送信します。テレメトリーをコレクターに送信する必要がある場合は、ステップ 3の説明に従って、次の設定を使用してデプロイします。

インストゥルメント化されたアプリケーションからテレメトリーを受信するには、OTLP レシーバーとパイプラインをコレクター設定に追加します。

ステップ1. receiversセクションに追加します：

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

  # ... existing receivers (prometheus/kafka-jmx, kafkametrics/cluster)

ステップ2。exportersセクションに追加します：

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

  # ... existing exporters

ステップ3. service.pipelinesセクションに追加します：

service:
  pipelines:
    traces:
      receivers: [otlp]
      exporters: [otlp/backend]

    metrics:
      receivers: [otlp]
      exporters: [otlp/backend]

    logs:
      receivers: [otlp]
      exporters: [otlp/backend]

    # ... existing pipelines (metrics/broker, metrics/cluster/prometheus, metrics/cluster/kafkametrics)

これにより、コレクターはインストゥルメントで処理された Kafka クライアントからアプリケーショントレース、メトリクス、ログを受信し、ブローカーメトリクスとともにNew Relicに転送できるようになります。

Javaエージェントは、コードを変更することなくすぐに使えるKafka計装を提供し、以下をキャプチャします：

リクエストのレイテンシ
スループットメトリクス
エラー率
分散型トレース

高度な設定については、 Kafka 計装ドキュメントを参照してください。

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

Kafkaブローカーのログを収集してNew Relicに送信するには、コレクターの設定にfilelogレシーバーを追加します。

values.yamlコレクターの設定に以下を追加します：

ステップ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 ログパスに更新します。Strimziの場合、ログは通常stdoutに書き込まれます。ファイルにリダイレクトするには、Strimziのログ設定を使用するか、代わりにサイドカーログコレクターを使用してください。
`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エージェント経由）、およびオプションのログ転送ステップを介して収集されたブローカーログ
トレース	`Span`	トピックをまたぐメッセージごとの`publish`および`receive`操作を含む、プロデューサーと消費者のスパン

指標

ブローカー、トピック、パーティション、消費者グループ、およびJVMメトリクスは、Metricイベントタイプに保存されます。my-kafka-clusterをKAFKA_CLUSTER_NAME値に置き換えます：

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

ログ

OpenTelemetry Javaエージェントでインストゥルメントされたプロデューサーおよび消費者アプリケーションからのログ、およびオプションのログ転送ステップを介して収集されたブローカーログは、Logイベントタイプに保存されます：

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

トレース

OpenTelemetry Javaエージェントでインストゥルメントされたプロデューサーまたは消費者アプリケーションをデプロイする場合、プロデューサーおよび消費者スパンはSpanイベントタイプに保存されます：

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

例

Strimzi Kafkaカスタムリソース、JMX Exporterの設定、OTel Collectorのセットアップ、およびサンプルのプロデューサー/消費者アプリケーションを含む完全な動作例は、New Relic OpenTelemetry Examplesリポジトリで利用可能です。

トラブルシューティング

1. コレクターデバッグログを有効にする: 設定の問題をトラブルシューティングするための詳細なログを追加します。

Helm デプロイメントの場合は、 values.yamlを更新します。

config:
  service:
    telemetry:
      logs:
        level: "debug"  # Enable detailed collector internal logs

マニフェストのデプロイメントの場合は、コレクター ConfigMap を編集します。

bash

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

service:の下にテレメトリーセクションを追加します。

service:
  telemetry:
    logs:
      level: "debug"
  pipelines:
    # ... existing pipelines ...

2. デバッグエクスポーターを追加: New Relicに送信する前にコレクターログのメトリクスを表示する

設定に追加:

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]  # Add debug exporter

次にコレクターを再起動してログを確認します。

bash

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

重要: ログのオーバーフローを回避するために、本番環境ではデバッグエクスポーターを削除してください。

1. ポッドのステータスとイベントを確認します。

bash

$# Check pod status
$# For Helm:
$kubectl get pods -n newrelic -l app.kubernetes.io/name=opentelemetry-collector
$# For Manifest:
$kubectl get pods -n newrelic -l app=otel-collector
$
$# View detailed pod description
$# For Helm:
$kubectl describe pod -n newrelic -l app.kubernetes.io/name=opentelemetry-collector
$# For Manifest:
$kubectl describe pod -n newrelic -l app=otel-collector
$
$# Check recent logs
$# For Helm:
$kubectl logs -n newrelic -l app.kubernetes.io/name=opentelemetry-collector --previous --tail=50
$# For Manifest:
$kubectl logs -n newrelic -l app=otel-collector --previous --tail=50

よくある問題と解決策:

2. 無効な設定: ConfigMap YAML構文を検証する

bash

$# For manifest deployments, check ConfigMap
$kubectl get configmap -n newrelic otel-collector-config -o yaml
$
$# Validate YAML syntax
$kubectl get configmap -n newrelic otel-collector-config -o yaml | kubectl apply --dry-run=client -f -
$
$# For Helm deployments, check the values
$helm get values <release-name> -n newrelic

3. RBAC権限: ServiceAccountに適切なClusterRoleバインディングがあることを確認する

bash

$# Check ServiceAccount
$kubectl get serviceaccount -n newrelic otel-collector
$
$# Check ClusterRole and ClusterRoleBinding
$kubectl get clusterrole otel-collector -o yaml
$kubectl get clusterrolebinding otel-collector -o yaml

4. リソース制約: ポッドがOOMKilledされたか、リソース制限に達していないか確認する

bash

$# Check resource usage
$# For Helm:
$kubectl top pods -n newrelic -l app.kubernetes.io/name=opentelemetry-collector
$# For Manifest:
$kubectl top pods -n newrelic -l app=otel-collector
$
$# Check for resource limits
$# For Helm:
$kubectl describe pod -n newrelic -l app.kubernetes.io/name=opentelemetry-collector | grep -A 5 "Limits\|Requests"
$# For Manifest:
$kubectl describe pod -n newrelic -l app=otel-collector | grep -A 5 "Limits\|Requests"

5. シークレットが見つかりません: New Relic 認証情報のシークレットが存在することを確認してください

bash

$# Check if secret exists
$kubectl get secret -n newrelic newrelic-otlp-secret
$
$# Verify secret has required keys
$kubectl get secret -n newrelic newrelic-otlp-secret -o jsonpath='{.data}' | jq 'keys'

1. JMX エクスポーターが有効になっていることを確認する: Strimzi Kafka リソースに JMX エクスポーターが設定されていることを確認します。

bash

$# Check Kafka resource configuration for JMX Exporter
$kubectl get kafka -n kafka -o yaml | grep -A 10 jmxPrometheusExporter
$
$# Should show something like:
$# jmxPrometheusExporter:
$#   lowercaseOutputName: true

2. JMXエクスポーターのポートを確認する:エクスポーターがポート9404でリッスンしていることを確認する

bash

$# Check if port 9404 is exposed on Kafka pods
$kubectl get pods -n kafka -l strimzi.io/name=<cluster-name>-kafka -o yaml | grep -A 3 "containerPort: 9404"
$
$# Test connectivity from collector pod
$# For Helm:
$kubectl exec -it -n newrelic deployment/kafka-monitoring-opentelemetry-collector -- sh -c "curl -s http://<kafka-pod-ip>:9404/metrics" | head -10
$# For Manifest:
$kubectl exec -it -n newrelic deployment/otel-collector -- sh -c "curl -s http://<kafka-pod-ip>:9404/metrics" | head -10

3. Prometheusレシーバーがメトリクスを取得できることを確認します。

bash

$# Check collector logs for Prometheus scraping
$# For Helm:
$kubectl logs -n newrelic -l app.kubernetes.io/name=opentelemetry-collector | grep -i "prometheus\|scrape"
$# For Manifest:
$kubectl logs -n newrelic -l app=otel-collector | grep -i "prometheus\|scrape"
$
$# Look for successful scrape messages or errors
$# For Helm:
$kubectl logs -n newrelic -l app.kubernetes.io/name=opentelemetry-collector | grep "prometheus/kafka-jmx"
$# For Manifest:
$kubectl logs -n newrelic -l app=otel-collector | grep "prometheus/kafka-jmx"

4. Kubernetesサービス検出をチェックする:ポッドラベルがrelabel_configsと一致していることを確認する

bash

$# Verify Kafka pod labels
$kubectl get pods -n kafka -l strimzi.io/name=<cluster-name>-kafka --show-labels
$
$# Should include labels like:
$# strimzi.io/cluster=<cluster-name>
$# strimzi.io/name=<cluster-name>-kafka

5. 手動スクレイピングのテスト: メトリクスが利用可能であることを確認する

bash

$# Get Kafka broker pod IP
$kubectl get pods -n kafka -o wide
$
$# Curl metrics endpoint
$# For Helm:
$kubectl exec -it -n newrelic deployment/kafka-monitoring-opentelemetry-collector -- curl http://<kafka-pod-ip>:9404/metrics
$# For Manifest:
$kubectl exec -it -n newrelic deployment/otel-collector -- curl http://<kafka-pod-ip>:9404/metrics

6. kafkametrics レシーバーのエラーを確認します。

bash

$# Look for kafkametrics connection issues
$# For Helm:
$kubectl logs -n newrelic -l app.kubernetes.io/name=opentelemetry-collector | grep "kafkametrics"
$# For Manifest:
$kubectl logs -n newrelic -l app=otel-collector | grep "kafkametrics"
$
$# Common errors:
$# - Connection refused: Check broker address is correct
$# - Timeout: Check network policies allow access
$# - Authentication failed: Remove TLS configuration if using plaintext

1. リソースの使用状況を監視します。

bash

$# Check current memory usage
$# For Helm:
$kubectl top pods -n newrelic -l app.kubernetes.io/name=opentelemetry-collector
$# For Manifest:
$kubectl top pods -n newrelic -l app=otel-collector
$
$# Watch memory usage over time
$# For Helm:
$watch kubectl top pods -n newrelic -l app.kubernetes.io/name=opentelemetry-collector
$# For Manifest:
$watch kubectl top pods -n newrelic -l app=otel-collector

2. モニタートピックの削減：収集を重要なトピックのみに制限します

# In your values.yaml (Helm) or ConfigMap (manifest), add topic filtering:
receivers:
  kafkametrics/cluster:
    brokers:
      - "my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092"
    collection_interval: 30s
    scrapers:
      - brokers
      - topics  # Consider removing if not needed
      - consumers  # Consider removing if not needed
    topic_match: "^(important-topic-1|important-topic-2)$"  # Filter specific topics

3. 収集頻度を減らす：収集間隔を長くして収集頻度を減らす

receivers:
  kafkametrics/cluster:
    collection_interval: 60s  # Increase from 30s to 60s

  prometheus/kafka-jmx:
    config:
      scrape_configs:
        - job_name: 'kafka-jmx-metrics'
          scrape_interval: 60s  # Increase from 30s to 60s

4. バッチ処理の最適化: バッチプロセッサの設定を調整する

processors:
  batch/export:
    timeout: 60s  # Increase from 30s
    send_batch_size: 512  # Reduce from 1024

5. メモリリミッターを調整する: メモリリミッタープロセッサを調整する

processors:
  memory_limiter:
    check_interval: 1s
    limit_percentage: 75  # Reduce from 80
    spike_limit_percentage: 20  # Reduce from 30

6. リソース制限を更新する: Helm デプロイメントの場合は、values.yaml を更新します。

resources:
  requests:
    memory: 512Mi
    cpu: 250m
  limits:
    memory: 1Gi  # Adjust as needed
    cpu: 500m

マニフェストデプロイメントの場合は、デプロイメントを直接更新します。

bash

$kubectl patch deployment -n newrelic otel-collector --patch '
$spec:
$  template:
$    spec:
$      containers:
$      - name: otel-collector
$        resources:
$          limits:
$            memory: "1Gi"
$          requests:
$            memory: "512Mi"
$'

7. 変更後にコレクターを再起動します。

bash

$# For Helm:
$kubectl rollout restart deployment -n newrelic kafka-monitoring-opentelemetry-collector
$# For Manifest:
$kubectl rollout restart deployment -n newrelic otel-collector

1. ネットワーク接続を確認する: コレクターが Kafka ブローカーポッドに到達できることを確認する

bash

$# Get Kafka broker pod IPs
$kubectl get pods -n kafka -l strimzi.io/kind=Kafka -o wide
$
$# Test connectivity from collector pod
$# For Helm:
$kubectl exec -it -n newrelic deployment/kafka-monitoring-opentelemetry-collector -- curl -m 5 http://<kafka-pod-ip>:9404/metrics | head -20
$# For Manifest:
$kubectl exec -it -n newrelic deployment/otel-collector -- curl -m 5 http://<kafka-pod-ip>:9404/metrics | head -20

2. JMXエクスポーターが実行中であることを確認する: KafkaブローカーでJMXエクスポーターが有効になっていることを確認する

bash

$# Check Kafka pod for JMX Exporter container
$kubectl get pods -n kafka -o yaml | grep -A 5 "jmx-exporter"
$
$# Check if port 9404 is listening
$kubectl exec -n kafka <kafka-pod-name> -- netstat -tlnp | grep :9404
$
$# Or test from within the pod
$kubectl exec -n kafka <kafka-pod-name> -- curl -s localhost:9404/metrics | head

3. relabel_configs がポッドラベルと一致することを確認します。Prometheus 受信側が Kafka ポッドを検出できることを確認します。

bash

$# Verify pod labels match the relabel_configs in your configuration
$kubectl get pods -n kafka -l strimzi.io/kind=Kafka --show-labels
$
$# Should show labels like:
$# strimzi.io/cluster=my-cluster
$# strimzi.io/name=my-cluster-kafka

4. ネームスペース設定の確認: Prometheus 受信機が正しいネームスペースを探していることを確認します。

# In your configuration, verify namespace matches where Kafka is deployed
receivers:
  prometheus/kafka-jmx:
    config:
      scrape_configs:
        - job_name: 'kafka-jmx-metrics'
          kubernetes_sd_configs:
            - role: pod
              namespaces:
                names:
                  - kafka  # Must match your Kafka namespace

5. RBAC権限の確認: ClusterRoleがポッド検出を許可していることを確認する

bash

$# Check ClusterRole has pod list/watch permissions
$kubectl get clusterrole otel-collector -o yaml | grep -A 3 "pods"
$
$# Should include:
$# - apiGroups: [""]
$#   resources: ["pods", "nodes"]
$#   verbs: ["get", "list", "watch"]

6. スクレイピングのタイムアウトを増やす: メトリクスエンドポイントの応答が遅い場合

receivers:
  prometheus/kafka-jmx:
    config:
      scrape_configs:
        - job_name: 'kafka-jmx-metrics'
          scrape_interval: 30s
          scrape_timeout: 20s  # Increase from default 10s

7. スクレイプの詳細についてはコレクターログを確認してください。

bash

$# View Prometheus receiver logs
$# For Helm:
$kubectl logs -n newrelic -l app.kubernetes.io/name=opentelemetry-collector | grep "prometheus/kafka-jmx"
$# For Manifest:
$kubectl logs -n newrelic -l app=otel-collector | grep "prometheus/kafka-jmx"
$
$# Look for discovered targets
$# For Helm:
$kubectl logs -n newrelic -l app.kubernetes.io/name=opentelemetry-collector | grep -i "target\|scrape"
$# For Manifest:
$kubectl logs -n newrelic -l app=otel-collector | grep -i "target\|scrape"

1. フィルタプロセッサを確認します。メトリクスを誤ってフィルタリングしていないことを確認します。

# Review your filter configurations:
processors:
  filter/scrape-overhead:
    metrics:
      exclude:
        match_type: regexp
        metric_names:
          - "^jmx_.*"  # These are excluded
          - "^process_.*"
          - "^jvm_.*"

2. Kafka JMX メトリクス名が変換されます。OpenTelemetry 標準に従って、アンダースコアがドットに置き換えられますOpenTelemetry

この設定は、 OpenTelemetryセマンティック規則に合わせて、JMX Exporter (ステップ 1 で定義) からのメトリクス名を自動的に変換します。例えば：

kafka_topic_io なる kafka.topic.io
kafka_broker_leader_count なる kafka.broker.leader.count

New Relicでメトリクスを検索する場合は、アンダースコアの代わりにドットを使用して変換された名前を使用します。

# This transformation is applied automatically:
processors:
  transform/metric-naming:
    metric_statements:
      - context: metric
        statements:
          - replace_pattern(name, "_", ".")

3. 詳細ログを有効にする: メトリクスが処理されている内容を正確に確認します

exporters:
  debug:
    verbosity: detailed

service:
  pipelines:
    metrics/broker:
      exporters: [debug, otlp/backend]

4. メトリクス名に書き込むNew Relic : 実際に受信されているメトリクスを確認する

FROM Metric SELECT uniques(metricName)
WHERE kafka.cluster.name = 'my-cluster'
SINCE 1 hour ago

設定には、包括的な監視のためのAdditional metrics [追加メトリクス]セクションが含まれています。これを削除しても、コアのNew Relic UI機能には影響しません。ブローカーの健全性、消費者のラグ、クラスタの概要、およびJVMダッシュボードはすべて引き続き機能します。

1. JMX ConfigMap から追加のメトリクスセクションを削除します。

kafka-jmx-config.yaml ConfigMap で、このコメントより下の部分（ rules:リストの末尾まで）をすべて削除してください。

# 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...'
  ...

ConfigMapを編集したら、それを適用してKafkaブローカーを再起動し、変更を反映させてください。

bash

$kubectl apply -f kafka-jmx-config.yaml
$kubectl rollout restart statefulset -n kafka <kafka-cluster-name>-kafka

2. kafkametrics レシーバーで消費者オフセットメトリクスを無効にする

コレクター設定で、 kafkametrics/clusterレシーバーmetricsセクションに次の内容を追加します。

receivers:
  kafkametrics/cluster:
    # ...existing config...
    metrics:
      kafka.consumer_group.offset:
        enabled: false
      kafka.consumer_group.offset_sum:
        enabled: false

消費者ラグメトリクス (kafka.consumer_group.lag 、 kafka.consumer_group.lag_sum) は有効のままです。これらは、 New Relic Kafka UI消費者監視ビューに使用するものです。

次のステップ

Kafka メトリクスを調べる- 完全なメトリクスリファレンスを見る
カスタムダッシュボードの作成- Kafka データの視覚化を構築します
アラートの設定 — 消費者ラグやレプリカ不足のパーティションなどの重要なメトリクスをモニターします

この機械翻訳は、参考として提供されています。

OpenTelemetryを使用したKubernetes (Strimzi) 上の Kafka の監視

アーキテクチャー .css-21sua1{background:none;border:none;width:0;padding:0;}

インストレーション手順

あなたが始める前に

Kafka JMX メトリクス用に Kafka クラスターを構成する

ヒント

重要

OpenTelemetry Collector をデプロイする

EU地域

ヒント

NRDOT Collector の使用 (推奨)

OpenTelemetry Collectorの使用

米国地域

EU地域

ヒント

NRDOT Collector の使用 (推奨)

OpenTelemetry Collectorの使用

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

重要

Kafka アプリケーションを計装する

ヒント

アプリケーションテレメトリー用Collector設定

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

ログ収集を構成する

New Relicでログを見つける

データを検索する

指標

ログ

トレース

例

トラブルシューティング

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

Collectorポッドが起動しない

Kafka メトリクスは収集されませんでした

メモリ使用量が多い

プロメテウスのスクレイピングエラー

一部のメトリクスが欠落しています

データ取り込みを減らす

次のステップ

アーキテクチャー