Monitorea Kafka autohospedado con OpenTelemetry

Supervise su clúster de Apache Kafka autohospedado instalando el recopilador de OpenTelemetry directamente en los hosts de Linux.

Instalación y configuración

Siga estos pasos para configurar un monitoreo integral de Kafka instalando el agente Java de OpenTelemetry en sus brokers e implementando un colector para recopilar y enviar métricas a New Relic.

Antes de que empieces

Asegúrese de tener:

Una cuenta de New Relic con un
Acceso de red desde el colector al puerto del servidor de arranque de Kafka (típicamente 9092)

Descargar el agente Java de OpenTelemetry

El Agente Java de OpenTelemetry se ejecuta como un agente Java adjunto a sus brokers de Kafka, recopilando métricas de Kafka y JMX y enviándolas al colector mediante OTLP:

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

Crear configuración personalizada de JMX

Crea un archivo de configuración JMX del agente Java de OpenTelemetry para recopilar métricas de Kafka de los MBeans de JMX.

Cree el archivo ~/opentelemetry/jmx-custom-config.yaml con la siguiente configuración:

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

Configurar el broker de Kafka

Adjunte el OpenTelemetry Java Agent a su broker de Kafka configurando la variable de entorno KAFKA_OPTS antes de iniciar Kafka.

Ejemplo de broker único:

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 &

Importante

Clústeres de múltiples brokers: Para múltiples brokers, utilice la misma configuración con valores broker.id únicos (p. ej., broker.id=1, broker.id=2, broker.id=3) en el parámetro -Dotel.resource.attributes para cada broker.

nohup - Ejecuta el broker de Kafka en segundo plano, continuando incluso si la sesión de shell finaliza
-javaagent - Adjunta el agente Java de OpenTelemetry a la JVM del broker de Kafka
-Dotel.jmx.enabled=true habilita la recolección de métricas JMX
-Dotel.jmx.config especifica su archivo de configuración de métricas JMX personalizado
-Dotel.resource.attributes agrega metadatos: broker.id único y kafka.cluster.name
-Dotel.exporter.otlp.endpoint apunta a su OpenTelemetry Collector (predeterminado: localhost:4317)
-Dotel.exporter.otlp.protocol=grpc utiliza el protocolo gRPC para OTLP
-Dotel.metrics.exporter=otlp envía métricas vía OTLP
-Dotel.metric.export.interval=30000 exporta métricas cada 30 segundos
& - Ejecuta el comando en segundo plano
Para colectores remotos (host diferente):
bash
```
$-Dotel.exporter.otlp.endpoint=http://collector-host:4317
```
Para ver las opciones de configuración completas, consulte la guía de configuración del agente Java.

Crear configuración del colector

Cree la configuración principal de OpenTelemetry Collector en ~/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]

Aspectos destacados de la arquitectura:

Receptor OTLP: Recibe métricas de Kafka y JMX del Agente Java de OpenTelemetry que se ejecuta en los brokers de Kafka a través de gRPC en el puerto 4317
Enfoque de dos pipelines: Las métricas a nivel de clúster se envían sin broker.id para mapear a la entidad del clúster
Filtrado de métricas: Separa las métricas específicas del broker de las métricas a nivel de clúster para evitar la duplicación
Agregación: Agrega automáticamente las métricas a nivel de partición por tema

Establecer variables de entorno

Configure las variables de entorno requeridas antes de instalar el colector:

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

Reemplazar:

YOUR_LICENSE_KEY con su clave de licencia de New Relic
my-kafka-cluster con un nombre único para su clúster de Kafka
localhost:9092 con la(s) dirección(es) de su broker de arranque de Kafka. Para múltiples brokers, use una lista separada por comas: broker1:9092,broker2:9092,broker3:9092
Endpoint OTLP: Usa https://otlp.nr-data.net:4317 (región de EE. UU.) o https://otlp.eu01.nr-data.net:4317 (región de la UE). Para otras configuraciones de endpoints, consulte Configure su endpoint OTLP

Instale e inicie el recopilador

Elija entre NRDOT Collector (distribución de New Relic) u OpenTelemetry Collector:

Sugerencia

NRDOT Collector es la distribución de New Relic de OpenTelemetry Collector con soporte de New Relic para asistencia.

Descargue e instale el binario

Descargue e instale el binario de NRDOT Collector para su sistema operativo host. El siguiente ejemplo es para la arquitectura 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

Importante

Para otros sistemas operativos y arquitecturas, visita lanzamientos de NRDOT Collector y descarga el binario adecuado para tu sistema.

Iniciar el colector

Ejecute el colector con su archivo de configuración para comenzar a monitorear:

bash

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

El recopilador comenzará a enviar métricas de Kafka a New Relic en unos minutos.

Descargue e instale el binario

Descargue e instale el binario de OpenTelemetry Collector Contrib para su sistema operativo host. El siguiente ejemplo es para la arquitectura 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

Para otros sistemas operativos, visite la página de lanzamientos de OpenTelemetry Collector.

Iniciar el colector

Ejecute el colector con su archivo de configuración para comenzar a monitorear:

bash

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

El recopilador comenzará a enviar métricas de Kafka a New Relic en unos minutos.

(Opcional) Instrumentar aplicaciones productoras o consumidoras

Importante

Soporte de lenguajes: Actualmente, solo se admiten aplicaciones Java para la instrumentación de clientes Kafka mediante el Agente Java de OpenTelemetry.

Para recopilar telemetría a nivel de aplicación de sus aplicaciones de productor y consumidor de Kafka, utilice el Agente Java de OpenTelemetry que descargó en el Paso 1.

Inicie su aplicación con el agente:

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

Reemplazar:

order-process-service con un nombre único para su aplicación de productor o consumidor
my-kafka-cluster con el mismo nombre de clúster utilizado en la configuración de su recolector

Sugerencia

La configuración anterior envía telemetría a un OpenTelemetry Collector que se ejecuta en 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]

Esto le permite personalizar el procesamiento, agregar filtros o enrutar a múltiples backends. Para otras configuraciones de endpoints, consulte Configure su endpoint OTLP.

El agente Java proporciona instrumentación de Kafka lista para usar sin cambios de código, capturando:

Latencias de solicitud
Métricas de rendimiento
Tasas de error
Rastreo distribuido

Para una configuración avanzada, consulte la documentación de instrumentación de Kafka.

(Opcional) Enviar logs del broker de Kafka

Para recopilar logs del broker de Kafka y enviarlos a New Relic, configure el receptor filelog en su OpenTelemetry Collector.

Actualice la configuración de su recopilador en ~/opentelemetry/kafka-config.yaml para agregar el receptor filelog.

Agregar a la sección de receptores:

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}

Agregar pipeline de logs a la sección de servicio:

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]

Notas de configuración:

Actualice /path/to/kafka/logs/server.log a la ruta real del archivo de log de Kafka (por ejemplo, ~/kafka/logs/server.log)
El atributo de recurso broker.id correlaciona los logs con métricas y entidades específicas del broker
Para múltiples brokers, cree receptores filelog separados (p. ej., filelog/kafka_broker_2, filelog/kafka_broker_3) con sus respectivos ID de broker
El patrón multiline asume que los logs comienzan con [; ajústelo si su formato de log difiere
Considere el volumen de logs y los costos de recopilación antes de habilitar el reenvío de logs
Para ver las opciones de configuración completas, consulte la documentación del receptor filelog

Después de actualizar la configuración, reinicie el recopilador:

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

Los logs de su broker de Kafka aparecerán en dos lugares:

Entidades de broker: Navegue a la entidad de broker de Kafka en New Relic para ver los logs correlacionados con ese broker específico
UI de logs: Consulte todos los logs de Kafka utilizando la UI de logs con filtros como kafka.cluster.name = 'my-cluster'
También puede consultar sus logs con NRQL:
```
FROM Log SELECT * WHERE kafka.cluster.name = 'my-kafka-cluster'
```

Avanzado: Personalizar la recolección de métricas

Puede agregar más métricas de Kafka extendiendo las reglas en jmx-custom-config.yaml:

Obtenga más información sobre la sintaxis de configuración de métricas JMX de OpenTelemetry
Encuentre los nombres de MBean disponibles en la documentación de monitoreo de Kafka

Esto le permite recopilar cualquier métrica JMX expuesta por los brokers de Kafka en función de sus necesidades específicas de monitoreo.

Encuentra tus datos

Después de unos minutos, sus métricas de Kafka deberían aparecer en New Relic. Consulte Encuentre sus datos para obtener instrucciones detalladas sobre cómo explorar sus métricas de Kafka en diferentes vistas en la interfaz de usuario de New Relic.

También puede consultar sus datos con NRQL:

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

Resolución de problemas

Ejecute estos comandos primero para verificar su configuración. Utilice los resultados para identificar qué sección específica de solución de problemas seguir.

Verificar si el recopilador se está ejecutando:

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 no aparecen resultados, el colector no se está ejecutando. Inícielo siguiendo el Paso 6.

Verifique si el agente Java está adjunto a los brokers de Kafka:

bash

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

Nota: Este comando muestra el proceso completo de Java con el classpath, que puede ser muy extenso (más de 3 líneas por broker). Esto es lo esperado. Busque -javaagent:/path/to/opentelemetry-javaagent.jar en la salida.

Probar la conectividad del puerto:

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"

Revisar los logs del colector:

bash

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

Verificar las variables de entorno:

bash

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

Habilitar los logs de depuración del recolector: Agregue logs detallados para solucionar problemas de configuración

Agregue a la configuración de su recolector:

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

Agregar exportador de depuración: Vea las métricas en los logs del recopilador antes de enviarlas a 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

Luego reinicie el recopilador y verifique los logs:

bash

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

Importante: Elimine el exportador de depuración en producción para evitar el desbordamiento de logs.

Primero, ejecute las comprobaciones iniciales del sistema para verificar que el colector y el agente Java se estén ejecutando.

Verifique los logs del colector en busca de errores: Busque fallas de autenticación o conexión

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"

Primero, ejecute las comprobaciones iniciales del sistema para verificar que el agente de Java esté adjunto a los procesos de Kafka.

Verifique los logs del broker para la inicialización del agente Java:

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"

Verifique la configuración del Agente Java: Asegúrese de que el comando de inicio coincida con el Paso 3

bash

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

Esto debería mostrar todos los -Dotel.* parámetros. Verificar:

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

Verifique los logs del colector para métricas JMX entrantes:

bash

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

Primero, ejecute las comprobaciones iniciales del sistema para verificar que el puerto 4317 esté escuchando y sea accesible.

Verifique los logs del recopilador para errores específicos de OTLP:

bash

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

Verifique la configuración del receptor OTLP: Asegúrese de que el colector escuche en 0.0.0.0:4317 (no en 127.0.0.1)

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

Pruebe la conectividad remota (si el colector y Kafka están en hosts diferentes):

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"

Próximos pasos

Explora las métricas de Kafka - Consulta la referencia completa de métricas
Crear dashboards personalizados - Construir visualizaciones para sus datos de Kafka
Configurar alertas - Monitorea métricas críticas como el retraso del consumidor y las particiones subreplicadas

Te ofrecemos esta traducción automática para facilitar la lectura.

Monitorea Kafka autohospedado con OpenTelemetry

Instalación y configuración

Antes de que empieces

Descargar el agente Java de OpenTelemetry

Crear configuración personalizada de JMX

Configurar el broker de Kafka

Importante

Parámetro de configuración

Crear configuración del colector

Notas de configuración

Documentación adicional del receptor

Establecer variables de entorno

Instale e inicie el recopilador

Sugerencia

Importante

(Opcional) Instrumentar aplicaciones productoras o consumidoras

Importante

Sugerencia

Configuración de ejemplo del colector

(Opcional) Enviar logs del broker de Kafka

Configurar la recopilación de logs

Encuentre sus logs en New Relic

Avanzado: Personalizar la recolección de métricas

Encuentra tus datos

Resolución de problemas

Comprobaciones iniciales del sistema

Habilitar el log de depuración

No aparecen datos en New Relic

Faltan métricas JMX de los brokers de Kafka

Errores de conexión OTLP

Próximos pasos

Te ofrecemos esta traducción automática para facilitar la lectura.

Monitorea Kafka autohospedado con OpenTelemetry

Instalación y configuración .css-21sua1{background:none;border:none;width:0;padding:0;}

Antes de que empieces

Descargar el agente Java de OpenTelemetry

Crear configuración personalizada de JMX

Configurar el broker de Kafka

Importante

Crear configuración del colector

Notas de configuración

Documentación adicional del receptor

Establecer variables de entorno

Instale e inicie el recopilador

Sugerencia

Importante

(Opcional) Instrumentar aplicaciones productoras o consumidoras

Importante

Sugerencia

Configuración de ejemplo del colector

(Opcional) Enviar logs del broker de Kafka

Configurar la recopilación de logs

Encuentre sus logs en New Relic

Avanzado: Personalizar la recolección de métricas

Encuentra tus datos

Resolución de problemas

Comprobaciones iniciales del sistema

Habilitar el log de depuración

No aparecen datos en New Relic

Faltan métricas JMX de los brokers de Kafka

Errores de conexión OTLP

Próximos pasos

Instalación y configuración