Monitorer Kafka auto-hébergé avec OpenTelemetry

Monitorez votre cluster Apache Kafka auto-hébergé en installant le collecteur OpenTelemetry directement sur les hôtes Linux.

Installation et configuration

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 à New Relic.

Avant de commencer

Assurez-vous d'avoir :

Un compte New Relic avec un
Accès réseau du collecteur au port du serveur bootstrap Kafka (généralement 9092)

Télécharger l'agent Java OpenTelemetry

L'agent Java OpenTelemetry s'exécute en tant qu'agent Java attaché à vos brokers Kafka, collectant les métriques Kafka et JMX et les envoyant via OTLP au collecteur :

bash

$# Create directory for OpenTelemetry components
$mkdir -p ~/opentelemetry
$
$# Download OpenTelemetry Java Agent
$curl -L -o ~/opentelemetry/opentelemetry-javaagent.jar \
>  https://github.com/open-telemetry/opentelemetry-java-instrumentation/releases/latest/download/opentelemetry-javaagent.jar

Créer une configuration JMX personnalisée

Créez un fichier de configuration JMX pour l'agent Java OpenTelemetry afin de collecter les métriques Kafka à partir des MBeans JMX.

Créez le fichier ~/opentelemetry/jmx-custom-config.yaml avec la configuration suivante :

---
rules:
  # Per-topic custom metrics using custom MBean commands
  - bean: kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec,topic=*
    metricAttribute:
      topic: param(topic)
    mapping:
      Count:
        metric: kafka.prod.msg.count
        type: counter
        desc: The number of messages per topic
        unit: "{message}"

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

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

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

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

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

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

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

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

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

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

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

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

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

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

  - bean: java.lang:type=MemoryPool,name=*
    type: gauge
    unit: By
    metricAttribute:
      name: param(name)
    mapping:
      Usage.used:
        metric: jvm.memory.pool.used
        desc: Memory pool usage by generation (G1 Old Gen, Eden, Survivor)
      Usage.max:
        metric: jvm.memory.pool.max
        desc: Maximum memory pool size
      CollectionUsage.used:
        metric: jvm.memory.pool.used_after_last_gc
        desc: Memory used after last GC (shows retained memory baseline)
  
  - 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:
      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
      99thPercentile:
        metric: kafka.request.time.99p
        type: gauge
        desc: The 99th 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.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: For KRaft mode, the number of active controllers in the cluster. For ZooKeeper, indicates whether the broker is the controller broker.
        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 - increasing indicates broker failures
        unit: "{election}"

  - 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

Configurer le broker Kafka

Attachez l'agent Java OpenTelemetry à votre broker Kafka en définissant la variable d'environnement KAFKA_OPTS avant de démarrer Kafka.

Exemple de broker unique:

bash

$OTEL_AGENT="$HOME/opentelemetry/opentelemetry-javaagent.jar"
$JMX_CONFIG="$HOME/opentelemetry/jmx-custom-config.yaml"
$
$nohup env KAFKA_OPTS="-javaagent:$OTEL_AGENT \
>    -Dotel.jmx.enabled=true \
>    -Dotel.jmx.config=$JMX_CONFIG \
>    -Dotel.resource.attributes=broker.id=1,kafka.cluster.name=my-kafka-cluster \
>    -Dotel.exporter.otlp.endpoint=http://localhost:4317 \
>    -Dotel.exporter.otlp.protocol=grpc \
>    -Dotel.metrics.exporter=otlp \
>    -Dotel.metric.export.interval=30000" \
>    bin/kafka-server-start.sh config/server.properties &

Important

Clusters multi-brokers: Pour plusieurs brokers, utilisez la même configuration avec des valeurs broker.id uniques (par ex. broker.id=1, broker.id=2, broker.id=3) dans le paramètre -Dotel.resource.attributes pour chaque broker.

nohup - Exécute le broker Kafka en arrière-plan, en continuant même si la session shell se termine
-javaagent - Attache l'agent Java OpenTelemetry à la JVM du broker Kafka
-Dotel.jmx.enabled=true active la collecte de métriques JMX
-Dotel.jmx.config spécifie votre fichier de configuration de métriques JMX personnalisé
-Dotel.resource.attributes ajoute des métadonnées : broker.id unique et kafka.cluster.name
-Dotel.exporter.otlp.endpoint pointe vers votre OpenTelemetry Collector (par défaut : localhost:4317)
-Dotel.exporter.otlp.protocol=grpc utilise le protocole gRPC pour OTLP
-Dotel.metrics.exporter=otlp envoie des métriques via OTLP
-Dotel.metric.export.interval=30000 exporte des métriques toutes les 30 secondes
& - Exécute la commande en arrière-plan
Pour les collecteurs distants (hôte différent) :
bash
```
$-Dotel.exporter.otlp.endpoint=http://collector-host:4317
```
Pour toutes les options de configuration, consultez le guide de configuration de l'agent Java.

Créer une configuration de collecteur

Créez la configuration principale du collecteur OpenTelemetry à ~/opentelemetry/kafka-config.yaml.

receivers:
  # OTLP receiver for Kafka and JMX metrics from Java agents and application telemetry
  otlp:
    protocols:
      grpc:
        endpoint: "0.0.0.0:4317"

  # Kafka metrics receiver for cluster-level metrics
  kafkametrics:
    brokers: ${env:KAFKA_BOOTSTRAP_BROKER_ADDRESSES}
    protocol_version: 2.0.0
    scrapers:
      - brokers
      - topics
      - consumers
    collection_interval: 30s
    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

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

  resourcedetection:
    detectors: [env, ec2, system]
    system:
      resource_attributes:
        host.name:
          enabled: true
        host.id:
          enabled: true

  resource:
    attributes:
      - action: insert
        key: kafka.cluster.name
        value: ${env:KAFKA_CLUSTER_NAME}

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

  transform/remove_extra_attributes:
    metric_statements:
      - context: resource
        statements:
          # Delete all attributes starting with "process."
          - delete_matching_keys(attributes, "^process\\..*")
          # Delete all attributes starting with "telemetry."
          - delete_matching_keys(attributes, "^telemetry\\..*")
          - delete_key(attributes, "host.arch")
          - delete_key(attributes, "os.description")

  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/des_units:
    metric_statements:
      - context: metric
        statements:
          - set(description, "") where description != ""
          - set(unit, "") where unit != ""

  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

exporters:
  otlp/newrelic:
    endpoint: ${env:NEW_RELIC_OTLP_ENDPOINT}
    headers:
      api-key: ${env:NEW_RELIC_LICENSE_KEY}
    compression: gzip
    timeout: 30s

service:
  pipelines:
    # Broker metrics pipeline (excludes cluster-level metrics)
    metrics/broker:
      receivers: [otlp, kafkametrics]
      processors: [resourcedetection, resource, filter/exclude_cluster_metrics, 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: [resourcedetection, resource, filter/include_cluster_metrics, transform/remove_broker_id, transform/remove_extra_attributes, transform/des_units, cumulativetodelta, batch/aggregation]
      exporters: [otlp/newrelic]

Points forts de l'architecture :

Récepteur OTLP: Reçoit les métriques Kafka et JMX de l'agent Java OpenTelemetry s'exécutant sur les brokers Kafka via gRPC sur le port 4317
Approche à deux pipelines: les métriques au niveau du cluster sont envoyées sans broker.id pour mapper à l'entité du cluster
Filtrage des métriques: Sépare les métriques spécifiques au broker des métriques au niveau du cluster pour éviter les doublons
Agrégation: Agrège automatiquement les métriques au niveau de la partition par sujet

Définir les variables d'environnement

Définissez les variables d'environnement requises avant d'installer le collecteur :

bash

$export NEW_RELIC_LICENSE_KEY="YOUR_LICENSE_KEY"
$export KAFKA_CLUSTER_NAME="my-kafka-cluster"
$export KAFKA_BOOTSTRAP_BROKER_ADDRESSES="localhost:9092"
$export NEW_RELIC_OTLP_ENDPOINT="https://otlp.nr-data.net:4317" # US region

Remplacer :

YOUR_LICENSE_KEY avec votre clé de licence New Relic
my-kafka-cluster avec un nom unique pour votre cluster Kafka
localhost:9092 avec l'adresse (ou les adresses) de votre broker d'amorçage Kafka. Pour plusieurs courtiers, utilisez une liste séparée par des virgules : broker1:9092,broker2:9092,broker3:9092
Endpoint OTLP : utilise https://otlp.nr-data.net:4317 (région US) ou https://otlp.eu01.nr-data.net:4317 (région UE). Pour d'autres configurations de point de terminaison, consultez Configurer votre point de terminaison OTLP

Installer et démarrer le collecteur

Choisissez entre NRDOT Collector (la distribution de New Relic) ou OpenTelemetry Collector :

Conseil

NRDOT Collector est la distribution New Relic de l'OpenTelemetry Collector avec le support de New Relic pour l'assistance.

Télécharger et installer le binaire

Téléchargez et installez le binaire NRDOT Collector pour votre système d'exploitation hôte. L'exemple ci-dessous concerne l'architecture linux_amd64 :

bash

$# Set version and architecture
$NRDOT_VERSION="1.9.0"
$ARCH="amd64"  # or arm64
$
$# Download and extract
$curl "https://github.com/newrelic/nrdot-collector-releases/releases/download/${NRDOT_VERSION}/nrdot-collector_${NRDOT_VERSION}_linux_${ARCH}.tar.gz" \
>  --location --output collector.tar.gz
$tar -xzf collector.tar.gz
$
$# Move to a location in PATH (optional)
$sudo mv nrdot-collector /usr/local/bin/
$
$# Verify installation
$nrdot-collector --version

Important

Pour les autres systèmes d'exploitation et architectures, consultez les versions de NRDOT Collector et téléchargez le binaire approprié pour votre système.

Démarrer le collecteur

Lancez le collecteur avec votre fichier de configuration pour commencer le monitoring :

bash

$nrdot-collector --config ~/opentelemetry/kafka-config.yaml

Le collecteur commencera à envoyer des métriques Kafka à New Relic en quelques minutes.

Télécharger et installer le binaire

Téléchargez et installez le binaire OpenTelemetry Collector Contrib pour votre système d'exploitation hôte. L'exemple ci-dessous concerne l'architecture linux_amd64 :

bash

$# Set version and architecture
$# Check https://github.com/open-telemetry/opentelemetry-collector-releases/releases/latest for the latest version
$OTEL_VERSION="<collector_version>"
$ARCH="amd64"
$
$# Download the collector
$curl -L -o otelcol-contrib.tar.gz \
>  "https://github.com/open-telemetry/opentelemetry-collector-releases/releases/download/v${OTEL_VERSION}/otelcol-contrib_${OTEL_VERSION}_linux_${ARCH}.tar.gz"
$
$# Extract the binary
$tar -xzf otelcol-contrib.tar.gz
$
$# Move to a location in PATH (optional)
$sudo mv otelcol-contrib /usr/local/bin/
$
$# Verify installation
$otelcol-contrib --version

Pour les autres systèmes d'exploitation, consultez la page des versions du Collecteur OpenTelemetry.

Démarrer le collecteur

Lancez le collecteur avec votre fichier de configuration pour commencer le monitoring :

bash

$otelcol-contrib --config ~/opentelemetry/kafka-config.yaml

Le collecteur commencera à envoyer des métriques Kafka à New Relic en quelques minutes.

(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 applicative de vos applications productrices et consommatrices Kafka, utilisez l'agent Java OpenTelemetry que vous avez téléchargé à l'étape 1.

Démarrez votre application avec l'agent :

bash

$java \
>  -javaagent:$HOME/opentelemetry/opentelemetry-javaagent.jar \
>  -Dotel.service.name="order-process-service" \
>  -Dotel.resource.attributes="kafka.cluster.name=my-kafka-cluster" \
>  -Dotel.exporter.otlp.endpoint=http://localhost:4317 \
>  -Dotel.exporter.otlp.protocol="grpc" \
>  -Dotel.metrics.exporter="otlp" \
>  -Dotel.traces.exporter="otlp" \
>  -Dotel.logs.exporter="otlp" \
>  -Dotel.instrumentation.kafka.experimental-span-attributes="true" \
>  -Dotel.instrumentation.messaging.experimental.receive-telemetry.enabled="true" \
>  -Dotel.instrumentation.kafka.producer-propagation.enabled="true" \
>  -Dotel.instrumentation.kafka.enabled="true" \
>  -jar your-kafka-application.jar

Remplacer :

order-process-service avec un nom unique pour votre application producteur ou consommateur
my-kafka-cluster avec le même nom de cluster utilisé dans votre configuration de collecteur

Conseil

La configuration ci-dessus envoie la télémétrie à un Collector OpenTelemetry s'exécutant sur localhost:4317.

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

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

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

Cela vous permet de personnaliser le traitement, d'ajouter des filtres ou d'acheminer vers plusieurs backends. Pour d'autres configurations de point de terminaison, consultez Configurer votre point de terminaison OTLP.

L'agent Java fournit l'instrumentation Kafka prête à l'emploi sans aucune modification de code, capturant :

Latences des requêtes
Métriques de débit
Taux d'erreur
traces distribuées

Pour une configuration avancée, consultez la documentation d'instrumentation Kafka.

(Facultatif) Transmettre les logs du broker Kafka

Pour collecter les logs des brokers Kafka et les envoyer à New Relic, configurez le récepteur filelog dans votre OpenTelemetry Collector.

Mettez à jour la configuration de votre collecteur à ~/opentelemetry/kafka-config.yaml pour ajouter le récepteur filelog.

Ajouter à la section receivers:

receivers:
  # ... existing receivers (otlp, kafkametrics) ...
  
  # File log receiver for Kafka broker logs
  filelog/kafka_broker_1:
    include:
      - /path/to/kafka/logs/server.log
    start_at: end
    multiline:
      line_start_pattern: '^\['
    resource:
      broker.id: "1"
      kafka.cluster.name: ${env:KAFKA_CLUSTER_NAME}

Ajouter un pipeline de logs à la section Service:

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

Notes de configuration :

Remplacez /path/to/kafka/logs/server.log par le chemin d'accès réel de votre fichier de logs Kafka (par ex. ~/kafka/logs/server.log)
L'attribut de ressource broker.id corrèle les logs avec des métriques et des entités de broker spécifiques.
Pour plusieurs brokers, créez des récepteurs filelog distincts (par ex. filelog/kafka_broker_2, filelog/kafka_broker_3) avec leurs identifiants de broker respectifs
Le modèle multiline suppose que les logs commencent par [ - ajustez si votre format de log est différent
Prenez en compte le volume de logs et les coûts de collecte avant d'activer le transfert de logs
Pour les options de configuration complètes, consultez la documentation du récepteur filelog

Après avoir mis à jour la configuration, redémarrez le collecteur :

bash

$# If running in foreground, stop with Ctrl+C and restart
$nrdot-collector --config ~/opentelemetry/kafka-config.yaml
$# Or for OpenTelemetry Collector
$otelcol-contrib --config ~/opentelemetry/kafka-config.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'
```

Avancé : Personnaliser la collecte de métriques

Vous pouvez ajouter plus de métriques Kafka en étendant les règles dans jmx-custom-config.yaml:

En savoir plus sur la syntaxe de configuration des métriques JMX OpenTelemetry
Trouvez les noms de MBean disponibles dans la documentation de monitoring Kafka

Cela vous permet de collecter n'importe quelle métrique JMX exposée par les brokers Kafka en fonction de vos besoins de monitoring spécifiques.

Trouvez vos données

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

Vous pouvez également interroger vos données avec NRQL :

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

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 collecteur est en cours d'exécution:

bash

$# Check if port 4317 is listening (best indicator collector is running)
$ss -tlnp | grep 4317
$
$# Search for collector process (using bracket trick to exclude grep itself)
$ps aux | grep "[k]afka-config.yaml"
$
$# Or search for common collector names
$ps aux | grep -E "[n]rdot-collector|[o]telcol"

Si aucun résultat ne s'affiche, le collecteur n'est pas en cours d'exécution. Démarrez-le en suivant l'étape 6.

Vérifier si l'agent Java est attaché aux brokers Kafka:

bash

$# Search for Kafka processes with Java agent attached
$ps aux | grep "[o]pentelemetry-javaagent"

Remarque: Cette commande affiche le processus Java complet avec le classpath, ce qui peut être très long (plus de 3 lignes par broker). Ceci est attendu. Recherchez -javaagent:/path/to/opentelemetry-javaagent.jar dans la sortie.

Tester la connectivité du port:

bash

$# Test Kafka bootstrap port (9092)
$timeout 5 bash -c "</dev/tcp/localhost/9092" 2>/dev/null && echo "Port 9092 open" || echo "Port 9092 closed"
$
$# Test OTLP collector port (4317)
$timeout 5 bash -c "</dev/tcp/localhost/4317" 2>/dev/null && echo "Port 4317 open" || echo "Port 4317 closed"

Vérifier les logs du collecteur:

bash

$# View recent collector output
$tail -n 50 ~/logs/collector.log

Vérifier les variables d'environnement:

bash

$echo $NEW_RELIC_LICENSE_KEY
$echo $KAFKA_CLUSTER_NAME
$echo $KAFKA_BOOTSTRAP_BROKER_ADDRESSES

Activer les logs de débogage du collecteur: Ajoutez une logging détaillée pour résoudre les problèmes de configuration

Ajouter à votre configuration du collecteur :

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

Ajouter un exportateur de débogage: Afficher les métriques dans les logs du collecteur avant de les envoyer à New Relic

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

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

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

    metrics/cluster:
      receivers: [otlp]
      processors: [resourcedetection, resource, filter/include_cluster_metrics, transform/remove_broker_id, transform/des_units, cumulativetodelta, batch/aggregation]
      exporters: [debug, otlp/newrelic]  # Add debug exporter

Ensuite, redémarrez le collectteur et vérifiez les logs :

bash

$# Check collector output log
$tail -f ~/logs/collector.log
$
$# Look for metric output in the logs

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 collecteur et l'agent Java fonctionnent.

Vérifiez les logs du collecteur pour détecter des erreurs: recherchez des échecs d'authentification ou de connexion

bash

$# Look for errors in collector output
$tail -n 100 ~/logs/collector.log | grep -i "error\|fail\|refuse"
$
$# Check for OTLP receiver activity
$tail -n 100 ~/logs/collector.log | grep -i "otlp\|metric"

Tout d'abord, exécutez les vérifications initiales du système pour vérifier que l'agent Java est attaché aux processus Kafka.

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

bash

$# Find Kafka log directory (common locations)
$find ~ -name "server.log" -path "*/kafka/logs/*" 2>/dev/null
$
$# Check the log file for OpenTelemetry messages
$# Replace with your actual Kafka log path
$tail -100 ~/kafka/logs/server.log 2>/dev/null | grep -i "otel\|jmx" || echo "Log file not found or no OTel messages"
$
$# Check directory where you started Kafka for nohup.out
$ls -lh nohup.out 2>/dev/null && tail -100 nohup.out | grep -i "otel\|jmx" || echo "No nohup.out file found"

Vérifiez la configuration de l'agent Java: assurez-vous que la commande de démarrage correspond à l'étape 3

bash

$# Check if broker was started with correct Java agent parameters
$ps aux | grep "[o]pentelemetry-javaagent" | grep -o "Dotel\.[^ ]*"

Cela devrait afficher tous les -Dotel.* paramètres. Vérifier :

-Dotel.jmx.enabled=true
-Dotel.jmx.config=<path>
-Dotel.exporter.otlp.endpoint=http://localhost:4317

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

bash

$# Look for metrics coming from brokers
$tail -n 100 ~/logs/collector.log | grep -i "broker.id\|kafka\|jmx"

Tout d'abord, exécutez les vérifications initiales du système pour vérifier que le port 4317 est à l'écoute et accessible.

Vérifiez les logs du collecteur pour des erreurs OTLP spécifiques:

bash

$# Look for connection refusals or timeouts
$tail -n 100 ~/logs/collector.log | grep -i "connection refused\|context deadline exceeded\|failed to connect"

Vérifiez la configuration du récepteur OTLP: Assurez-vous que le collecteur écoute sur 0.0.0.0:4317 (et non 127.0.0.1)

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: "0.0.0.0:4317"  # Accepts connections from any interface

Tester la connectivité à distance (si le collecteur et Kafka sont sur des hôtes différents) :

bash

$# From Kafka broker machine, test connection to collector
$timeout 5 bash -c "</dev/tcp/COLLECTOR_HOST/4317" 2>/dev/null && echo "Can reach collector" || echo "Cannot reach collector"

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 - Monitorez les métriques critiques telles que le retard du consommateur et les partitions sous-répliquées

Cette traduction automatique est fournie pour votre commodité.

Monitorer Kafka auto-hébergé avec OpenTelemetry

Installation et configuration

Avant de commencer

Télécharger l'agent Java OpenTelemetry

Créer une configuration JMX personnalisée

Configurer le broker Kafka

Important

Paramètres de configuration

Créer une configuration de collecteur

Notes de configuration

Documentation supplémentaire du récepteur

Définir les variables d'environnement

Installer et démarrer le collecteur

Conseil

Important

(Facultatif) Instrumenter les applications productrices ou consommatrices

Important

Conseil

Exemple de configuration du collecteur

(Facultatif) Transmettre les logs du broker Kafka

Configurer la collecte des logs

Trouvez vos logs dans New Relic

Avancé : Personnaliser la collecte de métriques

Trouvez vos données

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

Prochaines étapes

Cette traduction automatique est fournie pour votre commodité.

Monitorer Kafka auto-hébergé avec OpenTelemetry

Installation et configuration .css-21sua1{background:none;border:none;width:0;padding:0;}

Avant de commencer

Télécharger l'agent Java OpenTelemetry

Créer une configuration JMX personnalisée

Configurer le broker Kafka

Important

Créer une configuration de collecteur

Notes de configuration

Documentation supplémentaire du récepteur

Définir les variables d'environnement

Installer et démarrer le collecteur

Conseil

Important

(Facultatif) Instrumenter les applications productrices ou consommatrices

Important

Conseil

Exemple de configuration du collecteur

(Facultatif) Transmettre les logs du broker Kafka

Configurer la collecte des logs

Trouvez vos logs dans New Relic

Avancé : Personnaliser la collecte de métriques

Trouvez vos données

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

Prochaines étapes

Installation et configuration