Monitorer Kafka autogéré sur Kubernetes avec OpenTelemetry

Monitorez votre cluster Apache Kafka autogéré fonctionnant sur Kubernetes en déployant l'OpenTelemetry Collector pour recueillir et transférer des métriques à New Relic.

Architecture

New Relic prend en charge deux approches pour le monitoring de Kubernetes Kafka autogéré : l’agent Java OpenTelemetry ou l’exportateur JMX Prometheus. Les diagrammes suivants illustrent le flux de données pour chaque approche.

Étapes d'installation

Suivez ces étapes pour configurer un monitoring complet de Kafka en installant l'agent Java OpenTelemetry sur vos brokers et en déployant un collecteur pour récupérer et envoyer des métriques et des logs à New Relic.

Avant de commencer

Assurez-vous d'avoir :

Un compte New Relic avec un
Cluster Kubernetes avec accès kubectl
Kafka déployé en tant que StatefulSet
Capacité à modifier et redéployer le StatefulSet Kafka

Déployer le Collecteur OpenTelemetry

Déployez le collecteur OpenTelemetry dans votre cluster. Cette étape crée également le ConfigMap kafka-jmx-config qui définit quelles métriques JMX l'agent Java collecte à partir de chaque pod de broker. Le collecteur doit être en cours d'exécution avant que vous ne redémarriez les brokers Kafka à l'étape suivante.

Étape 1. Créer un secret d'identifiants 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'

Conseil

Pour d'autres configurations de point de terminaison, consultez Configurer votre point de terminaison OTLP.

Étape 2. Créer values.yaml avec la configuration du collecteur

Les collecteurs NRDOT et OpenTelemetry utilisent une configuration identique. Choisissez votre image de collecteur préférée :

NRDOT est la distribution de l'OpenTelemetry Collector prise en charge par New Relic, offrant un support complet de New Relic. Pour plus d'informations, consultez le dépôt GitHub NRDOT Collector.

Créez values.yaml avec le contenu suivant :

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

Paramètres de configuration

Le tableau suivant décrit les principaux paramètres de configuration :

paramètres	Description
`config.receivers.kafkametrics.brokers`	Remplacer par le DNS de votre service bootstrap Kafka (par exemple, `kafka.kafka.svc.cluster.local:9092`)
`config.processors.resource.attributes[kafka.cluster.name]`	Remplacez par le nom de votre cluster Kafka
`extraObjects[0].metadata.namespace`	Remplacez par votre espace de nommage Kafka (dans le ConfigMap `extraObjects`)
`resources.limits` et `resources.requests`	Ajustez en fonction de vos besoins en workload.

Utilisez l'OpenTelemetry Collector communautaire pour une flexibilité maximale et un déploiement indépendant des fournisseurs.

Créez values.yaml avec le même contenu que l'option NRDOT ci-dessus, mais changez l'image :

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

Toute autre configuration (récepteurs, processeurs, pipelines et extraObjects) est identique.

Paramètres de configuration: mêmes paramètres que l'option NRDOT ci-dessus. Consultez le tableau des paramètres de configuration pour plus de détails, y compris les limites de ressources.

Pour des options de configuration avancées, voir :

Documentation du récepteur OTLP

Documentation du récepteur de métriques Kafka

Étape 3. Installer OpenTelemetry Collector avec Helm

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

Étape 4. Vérifier le déploiement

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

Étape 1. Créer un secret d'identifiants 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'

Conseil

Pour d'autres configurations de point de terminaison, consultez Configurer votre point de terminaison OTLP.

Étape 2. Créer des fichiers manifestes

Les collecteurs NRDOT et OpenTelemetry utilisent une configuration identique. Seule l'image de conteneur diffère. Les deux nécessitent également le ConfigMap kafka-jmx-config appliqué à votre espace de nommage Kafka.

Créer kafka-jmx-config.yaml - Configuration des métriques JMX pour l'agent Java (à appliquer à votre espace de nommage 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. Créer collector-configmap.yaml - Configuration du Collector OpenTelemetry :

---
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. Créer collector-deployment.yaml - déploiement avec ServiceAccount et 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

Paramètres de configuration

Le tableau suivant décrit les principaux paramètres de configuration :

paramètres	Description
`receivers.kafkametrics.brokers`	Remplacer par le DNS de votre service bootstrap Kafka
`processors.resource.attributes[kafka.cluster.name]`	Remplacez par le nom de votre cluster Kafka
`resources.limits` et `resources.requests` (dans `collector-deployment.yaml`)	Ajustez en fonction de vos besoins en workload.

Utilisez le OpenTelemetry Collector de la communauté pour un déploiement indépendant des fournisseurs.

1. Créer collector-configmap.yaml - Identique à l'option NRDOT ci-dessus (la configuration est identique)

2. Créer collector-deployment.yaml - Seules l'image du conteneur et la commande diffèrent :

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

Paramètres de configuration: mêmes paramètres que l'option NRDOT ci-dessus. Consultez le tableau des paramètres de configuration pour plus de détails, y compris les limites de ressources.

Étape 3. Déployer les manifestes

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

Étape 4. Vérifier le déploiement

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

Configurer le StatefulSet Kafka pour l'agent Java

Maintenant que le collecteur est en cours d'exécution, patchez votre StatefulSet Kafka pour ajouter un conteneur d'initialisation qui télécharge le JAR de l'agent Java OpenTelemetry, puis attachez-le à la JVM du broker Kafka via KAFKA_OPTS.

Ajoutez les sections suivantes à votre manifeste Kafka StatefulSet existant :

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

Conseil

Le ConfigMap kafka-jmx-config a été déployé avec le collecteur à l'étape précédente. La valeur otel.exporter.otlp.endpoint http://otel-collector.newrelic.svc.cluster.local:4317 suppose que le collecteur est déployé dans l'espace de nommage newrelic avec le nom de service otel-collector. Mettez-le à jour pour qu'il corresponde au DNS réel de votre service de collecteur s'il est différent.

paramètres	Description
`javaagent`	Attache l'agent Java OpenTelemetry à la JVM du broker Kafka
`jmx.enabled=true`	Active la collecte de métriques JMX
`jmx.config`	Pointe vers votre fichier de configuration de métriques JMX personnalisé (monté depuis ConfigMap)
`resource.attributes`	Ajoute des métadonnées `kafka.cluster.name` à toutes les métriques
`otlp.endpoint`	Pointe vers le service OpenTelemetry Collector dans votre cluster
`otlp.protocol=grpc`	Utilise le protocole gRPC pour OTLP
`metrics.exporter=otlp`	Envoie des métriques via OTLP
`logs.exporter=otlp`	Active la collecte des logs du broker. Définissez sur `none` pour désactiver.
`metric.export.interval`	Définit l'intervalle en millisecondes entre les tentatives d'exportation de métriques, par exemple `30000` (30 secondes)

Pour toutes les options de configuration, consultez le guide de configuration de l’agent Java.

Appliquez votre StatefulSet mis à jour et attendez que les pod soient déployés :

bash

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

(Facultatif) Instrumenter les applications productrices ou consommatrices

Important

Prise en charge des langages: actuellement, seules les applications Java sont prises en charge pour l'instrumentation des clients Kafka à l'aide de l'agent Java OpenTelemetry.

Pour collecter la télémétrie au niveau de l’application à partir de vos applications producteur et consommateur Kafka s’exécutant dans Kubernetes, ajoutez l’agent Java OpenTelemetry à ces pods d’application.

Ajoutez un conteneur d'initialisation et des variables d'environnement au déploiement de votre application :

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: {}

Paramètres de configuration

Le tableau suivant décrit les principaux paramètres de configuration :

paramètres	Description
`order-process-service`	Remplacez par un nom unique pour votre application producteur ou consommateur
`my-kafka-cluster`	Remplacez par le même nom de cluster utilisé dans la configuration de votre broker
`otel-collector.newrelic.svc.cluster.local`	Remplacez par le nom DNS réel de votre service de collecteur ( `<service-name>.<namespace>.svc.cluster.local` )

paramètres

Description

order-process-service

Remplacez par un nom unique pour votre application producteur ou consommateur

my-kafka-cluster

Remplacez par le même nom de cluster utilisé dans la configuration de votre broker

otel-collector.newrelic.svc.cluster.local

Remplacez par le nom DNS réel de votre service de collecteur (

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

)

L’agent Java fournit l’instrumentation Kafka prête à l’emploi sans aucune modification de code, capturant la latence des requêtes, les métriques de débit, les taux d’erreur et les traces distribuées. Pour une configuration avancée, consultez la documentation d’instrumentation Kafka.

Suivez ces étapes pour configurer un monitoring complet de Kafka en installant l'exportateur JMX Prometheus sur vos pods de broker et en déployant un collecteur pour récupérer et envoyer des métriques à New Relic.

Avant de commencer

Assurez-vous d'avoir :

Un compte New Relic avec un
Cluster Kubernetes avec accès kubectl
Kafka déployé en tant que StatefulSet avec un service headless (pour des noms DNS de pod stables)
Capacité à modifier et redéployer le StatefulSet Kafka

Créer la ConfigMap des métriques JMX

Créez une ConfigMap contenant la configuration de l'exportateur JMX qui définit les métriques Kafka à collecter. Ce ConfigMap sera monté dans chaque pod de broker Kafka.

Enregistrer sous kafka-jmx-config.yaml. Appliquez-le à l'espace de nommage où Kafka est déployé :

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"

Conseil

Personnaliser les métriques: vous pouvez ajouter ou modifier des motifs en consultant les exemples Prometheus JMX Exporter et la documentation des MBeans Kafka.

Appliquer la ConfigMap :

bash

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

Configurer le StatefulSet Kafka pour l'exportateur JMX

Patchez votre StatefulSet Kafka pour ajouter un conteneur d'initialisation qui télécharge le JAR de l'exportateur JMX Prometheus, puis attachez-le à la JVM du broker Kafka via KAFKA_OPTS.

Étape 1. Ajoutez les sections suivantes à votre manifeste StatefulSet Kafka existant :

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

Étape 2. Appliquez votre StatefulSet mis à jour et attendez que les pods soient redéployés :

bash

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

Étape 3. Une fois le déploiement terminé, vérifiez que les métriques sont exposées sur chaque pod de broker :

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

Important

Clusters multi-brokers: le conteneur init et la configuration KAFKA_OPTS s'appliquent automatiquement à tous les pods du StatefulSet. Vérifiez que chaque pod de broker expose des métriques après le déploiement.

Déployer le Collecteur OpenTelemetry

Déployez le Collector OpenTelemetry dans votre cluster. Le collecteur récupère les données des pods de broker Kafka en utilisant des cibles DNS statiques et écoute sur le port 4317 les données OTLP provenant d'applications instrumentées.

La méthode d'installation Helm est l'approche recommandée pour déployer OpenTelemetry Collector dans Kubernetes.

Étape 1. Créer un secret d'identifiants 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'

Conseil

Pour d'autres configurations de point de terminaison, consultez Configurer votre point de terminaison OTLP.

Étape 2. Créer values.yaml avec la configuration du collecteur

Les collecteurs NRDOT et OpenTelemetry utilisent une configuration identique. Choisissez votre image de collecteur préférée :

Créez values.yaml avec le contenu suivant :

# 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

Paramètres de configuration

Le tableau suivant décrit les principaux paramètres de configuration :

paramètres	Description
`config.receivers.kafkametrics/cluster.brokers`	Remplacer par votre service bootstrap Kafka (par ex., `kafka.kafka.svc.cluster.local:9092`)
`config.receivers.kafkametrics/cluster.brokers` et `config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Remplacer par l'espace de nommage où votre cluster Kafka est déployé
`config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Remplacez par le nom de votre StatefulSet Kafka suivi de -headless (par exemple, `kafka-headless`)
`config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Remplacez par le nom de votre StatefulSet Kafka (par exemple, `kafka`, apparaît dans les noms de pod comme `kafka-0`)
`config.processors.resource/cluster-name.attributes[kafka.cluster.name].value`	Remplacez par le nom de votre cluster Kafka (ceci sera utilisé pour identifier et filtrer vos métriques dans New Relic)
`config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Mettez à jour la liste `targets` pour correspondre à vos pods de broker — une entrée par broker utilisant le DNS du service headless (`<pod-name>.<headless-service>.<namespace>.svc.cluster.local:9404`). Ajoutez ou supprimez des entrées pour correspondre à votre nombre de brokers. `broker.id` est automatiquement extrait du nom du pod via `relabel_configs`.
`resources.limits` et `resources.requests`	Ajustez en fonction de vos besoins en workload.

Conseil

Alternative : découverte automatique des pods Kubernetes

Au lieu de cibles DNS statiques, vous pouvez utiliser la découverte de pods Kubernetes pour trouver automatiquement les pods broker. Cela est utile pour la mise à l'échelle dynamique sans avoir à mettre à jour la liste des cibles.

Remplacez les sections clusterRole et prometheus/kafka-jmx dans values.yaml par :

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

Utilisez l'OpenTelemetry Collector communautaire pour une flexibilité maximale et un déploiement indépendant des fournisseurs.

Créez values.yaml avec le contenu suivant (configuration identique, image différente) :

# 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

Paramètres de configuration

Le tableau suivant décrit les principaux paramètres de configuration :

paramètres	Description
`config.receivers.kafkametrics/cluster.brokers`	Remplacer par le DNS de votre service bootstrap Kafka
`config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Ajoutez une entrée par pod de broker en utilisant le DNS du service headless (`<pod-name>.<headless-service>.<namespace>.svc.cluster.local:9404`). Ajoutez ou supprimez des entrées pour correspondre à votre nombre de brokers. `broker.id` est automatiquement extrait du nom du pod via `relabel_configs`.
`config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].labels[kafka.cluster.name]`	Remplacez la valeur du label `kafka.cluster.name` par le nom de votre cluster
`config.processors.resource/cluster-name.attributes[kafka.cluster.name].value`	Remplacez par le nom de votre cluster Kafka (utilisé dans le Processeur `resource/cluster-name` pour identifier les métriques dans New Relic)
`resources.limits` et `resources.requests`	Ajustez en fonction de vos besoins en workload.

Conseil

Alternative : découverte automatique des pods Kubernetes

Remplacez la section prometheus/kafka-jmx dans values.yaml par :

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

Pour des options de configuration avancées, reportez-vous aux pages de documentation de ces récepteurs :

Documentation du récepteur Prometheus

Documentation du récepteur de métriques Kafka

Étape 3. Installer OpenTelemetry Collector avec Helm

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

Étape 4. Vérifier le déploiement :

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

Vous devriez voir des logs indiquant un scraping réussi des pods de broker Kafka sur le port 9404.

La méthode d'installation par manifeste offre un contrôle direct sur les ressources Kubernetes sans utiliser Helm.

Étape 1. Créer un secret d'identifiants 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'

Conseil

Pour d'autres configurations de point de terminaison, consultez Configurer votre point de terminaison OTLP.

Étape 2. Créer des fichiers manifestes

Les collecteurs NRDOT et OpenTelemetry utilisent une configuration identique. Seule l'image de conteneur diffère.

1. Créer collector-configmap.yaml - Configuration du Collector OpenTelemetry :

---
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. Créer collector-deployment.yaml - Déploiement de l'OpenTelemetry Collector avec ServiceAccount :

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

Paramètres de configuration

Le tableau suivant décrit les principaux paramètres de configuration :

paramètres	Description
`receivers.kafkametrics/cluster.brokers`	Remplacer par le DNS de votre service bootstrap Kafka (par exemple, `kafka.kafka.svc.cluster.local:9092`)
`receivers.kafkametrics/cluster.brokers` et `receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Remplacer par l'espace de nommage où votre cluster Kafka est déployé
`receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Remplacez par le nom de votre StatefulSet Kafka suivi de `-headless` (par exemple, `kafka-headless`)
`receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Remplacez par le nom de votre StatefulSet Kafka (par exemple, `kafka`, apparaît dans les noms de pod comme `kafka-0`)
`processors.resource/cluster-name.attributes[kafka.cluster.name].value`	Remplacez par le nom de votre cluster Kafka (utilisé pour identifier les métriques dans New Relic)
`receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Mettez à jour la liste `targets` pour correspondre à vos pods de broker — une entrée par broker utilisant le DNS du service headless (`<pod-name>.<headless-service>.<namespace>.svc.cluster.local:9404`). `broker.id` est automatiquement extrait du nom du pod via `relabel_configs`.
`resources.limits` et `resources.requests` (dans `collector-deployment.yaml`)	Ajustez en fonction de vos besoins en workload.

Utilisez le OpenTelemetry Collector de la communauté pour un déploiement indépendant des fournisseurs.

1. Créer collector-configmap.yaml - Identique à l'option NRDOT ci-dessus (la configuration est identique)

2. Créer collector-deployment.yaml - Seule l'image du conteneur diffère :

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

Paramètres de configuration: mêmes paramètres que l'option NRDOT ci-dessus. Consultez le tableau des paramètres de configuration pour plus de détails, y compris les limites de ressources.

Pour des options de configuration avancées, reportez-vous aux pages de documentation de ces récepteurs :

Documentation du récepteur Prometheus

Documentation du récepteur de métriques Kafka

Étape 3. Déployer les manifestes

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

Étape 4. Vérifier le déploiement :

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

Vous devriez voir des logs indiquant un scraping réussi des pods de broker Kafka sur le port 9404.

(Facultatif) Instrumenter les applications productrices ou consommatrices

Important

Prise en charge des langages: les applications Java prennent en charge l’instrumentation client Kafka prête à l’emploi à l’aide de l’agent Java OpenTelemetry.

Pour collecter la télémétrie au niveau de l'application à partir de vos applications producteur et consommateur Kafka, utilisez l'agent Java OpenTelemetry avec un conteneur 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=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: {}

Paramètres de configuration

Le tableau suivant décrit les principaux paramètres de configuration :

paramètres	Description
`service.name`	Remplacez `my-kafka-app` par un nom unique pour votre application producteur ou consommateur
`kafka.cluster.name`	Remplacez `my-kafka-cluster` par le même nom de cluster utilisé dans la configuration de votre collecteur.
`otlp.endpoint`	Le point de terminaison `http://otel-collector.newrelic.svc.cluster.local:4317` suppose que le collecteur est déployé dans l'espace de nommage `newrelic` en tant que `otel-collector`

(Facultatif) Transmettre les logs du broker Kafka

Pour collecter les logs des brokers Kafka et les envoyer à New Relic, ajoutez un récepteur filelog à la configuration de votre collecteur.

Étape 1. Ajouter à la section des récepteurs :

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}

Étape 2. Ajouter un pipeline de logs à la section Service :

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

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

Paramètres de configuration

Le tableau suivant décrit les principaux paramètres de configuration :

paramètres	Description
`filelog/kafka_broker_0.include`	Remplacez `/var/log/kafka/server.log` par le chemin d'accès réel de vos logs Kafka dans le pod du broker
`filelog/kafka_broker_0.resource.broker.id`	L'attribut de ressource `broker.id` corrèle les logs avec des métriques et des entités de broker spécifiques.
Plusieurs récepteurs de broker	Pour plusieurs brokers, créez des récepteurs `filelog` distincts (par ex. `filelog/kafka_broker_1`, `filelog/kafka_broker_2`) avec leurs identifiants de broker respectifs
`filelog/kafka_broker_0.multiline.line_start_pattern`	Le modèle `multiline` suppose que les logs commencent par `[` — ajustez si votre format de log est différent
Volume de logs	Prenez en compte le volume de logs et les coûts de collecte avant d'activer le transfert de logs
Référence	Pour les options de configuration complètes, consultez la documentation du récepteur filelog

Étape 3. Mettez à niveau la sortie Helm :

bash

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

Vos logs de broker Kafka apparaîtront à deux endroits :

Entités de broker: Accédez à l'entité de broker Kafka dans New Relic pour voir les logs corrélés à ce broker spécifique
Interface utilisateur des logs: Interrogez tous les logs Kafka à l'aide de l'interface utilisateur des logs avec des filtres tels que kafka.cluster.name = 'my-cluster'
Vous pouvez également interroger vos logs avec NRQL :
```
FROM Log SELECT * WHERE kafka.cluster.name = 'my-kafka-cluster'
```

Trouvez vos données

Après quelques minutes, vos données Kafka devraient apparaître dans New Relic. Consultez Trouver vos données pour obtenir des instructions détaillées sur l’exploration de vos données Kafka dans différentes vues de l’UI de New Relic.

Le tableau suivant résume où chaque type de signal est stocké. Remplacez my-kafka-cluster par votre valeur KAFKA_CLUSTER_NAME dans toutes les requêtes ci-dessous :

Signal	Type d'événement	Ce qui est inclus
Métriques	`Metric`	Métriques de broker, de topic, de partition, de groupe de consommateurs, et de la JVM
Logs	`Log`	Logs des applications de producteur et de consommateur (via l'agent Java OTel) et logs de broker collectés via l'agent Java
Traces	`Span`	Spans de producteur et de consommateur, y compris les opérations `publish` et `receive` par message à travers les topics

Métriques

Les métriques de broker, de topic, de partition, de groupe de consommateurs et de la JVM sont stockées dans le type d'événement Metric :

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

Logs

Les logs des applications productrices et consommatrices instrumentées avec l'agent Java OpenTelemetry, et les logs de broker collectés via l'agent Java sur le broker, sont stockés dans le type d'événement Log :

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

Traces

Les spans de producteur et de consommateur, y compris les opérations publish et receive par message sur les topics, sont stockés dans le type d'événement Span :

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

Exemple

Un exemple fonctionnel complet avec les manifestes Kafka StatefulSet, les valeurs Helm, la configuration du Collecteur Otel, et des exemples d'applications producteur/consommateur est disponible dans le référentiel New Relic OpenTelemetry Examples.

Dépannage

Exécutez d'abord ces commandes pour vérifier votre configuration. Utilisez les résultats pour identifier la section de dépannage spécifique à suivre.

Vérifiez si le pod du collecteur est en cours d’exécution:

Pour les installations via manifeste :

bash

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

Pour les installations 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

Vérifiez si les pods du broker Kafka s'exécutent avec l'agent 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"

Vérifiez que le volume otel-agent est peuplé:

bash

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

Tester la connectivité du pod du broker au service du collecteur:

bash

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

Activer les logs de débogage du collecteur: ajoutez un logging détaillé pour résoudre les problèmes de configuration.

Dans votre ConfigMap (collector-configmap.yaml), ajoutez à la section service :

service:
  telemetry:
    logs:
      level: "debug"

Ensuite, appliquez le ConfigMap mis à jour et redémarrez le déploiement du collecteur :

bash

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

Ajouter un exportateur de débogage: afficher les métriques dans les logs du collecteur avant de les envoyer à New Relic. Les noms du processeur et de l'exportateur diffèrent selon la méthode de monitoring :

Méthode de l'agent 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]

Méthode de l'exportateur JMX Prometheus:

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]

Important: Supprimez l'exportateur de débogage en production pour éviter le débordement des logs.

Tout d'abord, exécutez les vérifications initiales du système pour vérifier que le pod du collecteur et les pod du broker sont sains.

Consultez les logs du collecteur pour détecter des erreurs (utilisez l'étiquette correspondant à votre méthode d'installation — voir Vérifications initiales du système) :

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"

Vérifiez que le service collecteur existe et a le bon port:

bash

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

Assurez-vous que le port 4317 est exposé en tant que service ClusterIP.

Tout d’abord, exécutez les vérifications initiales du système pour vérifier que l’agent Java est attaché aux pods de broker.

Vérifier l'initialisation de l'agent Java dans les logs du pod du broker:

bash

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

Vérifiez que KAFKA_OPTS est correctement défini sur les pods de broker:

bash

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

Cela devrait afficher -javaagent:/otel-agent/opentelemetry-javaagent.jar et tous les -Dotel.* paramètres. Vérifier :

-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

Vérifier que le ConfigMap JMX est monté:

bash

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

Vérifiez les logs du collecteur pour les métriques JMX entrantes :

bash

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

Tout d’abord, exécutez les vérifications initiales du système pour vérifier que le Service du collecteur est accessible depuis les pods du broker.

Vérifier la résolution DNS:

bash

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

Vérifiez les logs du collecteur pour les erreurs OTLP:

bash

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

Vérifiez que le récepteur OTLP écoute sur toutes les interfaces:

Assurez-vous que le ConfigMap contient endpoint: "0.0.0.0:4317" (et non 127.0.0.1) dans le Récepteur otlp :

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

Monitorer la mémoire du pod du collecteur:

bash

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

Réduire les sujets monitorés:

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

Réduire la fréquence de collecte: augmenter les intervalles pour collecter moins souvent

receivers:
  kafkametrics:
    collection_interval: 60s

Pour les métriques JMX de l'agent Java, mettez à jour KAFKA_OPTS dans le StatefulSet :

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

Ajouter un limiteur de mémoire:

Méthode de l'agent 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]
      ...

Méthode de l'exportateur JMX Prometheus :

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

Après les modifications, appliquez le ConfigMap mis à jour et redémarrez le collecteur :

bash

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

Prochaines étapes

Explorer les métriques Kafka - Consulter la référence complète des métriques
Créer des dashboards personnalisés - Créez des visualisations pour vos données Kafka
Configurer des alertes - Monitorer les métriques critiques telles que le retard du consommateur et les partitions sous-répliquées

Kafka auto-hébergé - monitoring Kafka pour les environnements auto-hébergés (non-Kubernetes)
Kubernetes Strimzi - monitoring de Kafka pour Kafka géré par Strimzi sur Kubernetes
Agent Java OpenTelemetry - Documentation officielle de l'agent Java OTel
Exportateur JMX Prometheus – agent Java qui expose les métriques JMX au format Prometheus
Récepteur Prometheus – récepteur du Collecteur Otel pour scraper les points de terminaison de métriques Prometheus
Récepteur kafkametrics - Documentation du récepteur de décalage du consommateur et de métriques de topic

Cette traduction automatique est fournie pour votre commodité.

Monitorer Kafka autogéré sur Kubernetes avec OpenTelemetry

Architecture .css-21sua1{background:none;border:none;width:0;padding:0;}

Étapes d'installation

Avant de commencer

Déployer le Collecteur OpenTelemetry

région de l'UE

Conseil

Utilisation de NRDOT Collector (recommandé)

Utiliser le collecteur OpenTelemetry

Région US

région de l'UE

Conseil

Utilisation de NRDOT Collector (recommandé)

Utiliser le collecteur OpenTelemetry

Configurer le StatefulSet Kafka pour l'agent Java

Conseil

Paramètres de configuration

(Facultatif) Instrumenter les applications productrices ou consommatrices

Important

Paramètres de configuration

Avant de commencer

Créer la ConfigMap des métriques JMX

Conseil

Configurer le StatefulSet Kafka pour l'exportateur JMX

Important

Déployer le Collecteur OpenTelemetry

Région US

région de l'UE

Conseil

Utilisation de NRDOT Collector (recommandé)

Utiliser le collecteur OpenTelemetry

Région US

région de l'UE

Conseil

Utilisation de NRDOT Collector (recommandé)

Utiliser le collecteur OpenTelemetry

(Facultatif) Instrumenter les applications productrices ou consommatrices

Important

(Facultatif) Transmettre les logs du broker Kafka

Configurer la collecte des logs

Trouvez vos logs dans New Relic

Trouvez vos données

Métriques

Logs

Traces

Exemple

Dépannage

Vérifications initiales du système

Activer la logging de débogage

Aucune donnée n'apparaît dans New Relic

Métriques JMX manquantes des brokers Kafka

Erreurs de connexion OTLP provenant des pods de broker

Utilisation élevée de la mémoire

Prochaines étapes

Ressources connexes

Architecture