Monitorear Kafka autogestionado en Kubernetes con OpenTelemetry

Supervise su clúster de Apache Kafka autogestionado que se ejecuta en Kubernetes desplegando el OpenTelemetry Collector para recopilar y reenviar métricas a New Relic.

Arquitectura

New Relic admite dos enfoques para el monitoreo de Kubernetes Kafka autoadministrado: el agente de Java de OpenTelemetry o el Prometheus JMX Exporter. Los siguientes diagramas ilustran el flujo de datos para cada enfoque.

Kubernetes self-managed Kafka monitoring architecture

Pasos de instalación

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

Antes de que empieces

Asegúrese de tener:

Una cuenta de New Relic con un
Clúster de Kubernetes con acceso kubectl
Kafka desplegado como un StatefulSet
Capacidad para modificar y redesplegar el StatefulSet de Kafka

Desplegar OpenTelemetry Collector

Despliegue el recolector OpenTelemetry en su clúster. Este paso también crea el ConfigMap kafka-jmx-config que define qué métricas JMX recopila el agente de Java de cada pod de broker. El recolector debe estar en ejecución antes de reiniciar los brokers de Kafka en el siguiente paso.

Paso 1. Crear secreto de credenciales de 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'

Sugerencia

Para otras configuraciones de endpoints, consulte Configure su endpoint OTLP.

Paso 2. Crear values.yaml con la configuración del recolector

Tanto los recolectores NRDOT como los de OpenTelemetry utilizan una configuración idéntica. Seleccione su imagen de recolector preferida:

NRDOT es la distribución de OpenTelemetry Collector con soporte de New Relic, que ofrece soporte completo de New Relic. Para obtener más información, consulta el repositorio de GitHub de NRDOT Collector.

Cree values.yaml con el siguiente contenido:

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

Parámetro de configuración

La siguiente tabla describe los parámetros de configuración clave:

Parámetro	Descripción
`config.receivers.kafkametrics.brokers`	Reemplace con el DNS de su servicio de arranque de Kafka (por ejemplo, `kafka.kafka.svc.cluster.local:9092`)
`config.processors.resource.attributes[kafka.cluster.name]`	Reemplace con el nombre de su clúster Kafka
`extraObjects[0].metadata.namespace`	Reemplazar con su namespace de Kafka (en el ConfigMap `extraObjects`)
`resources.limits` y `resources.requests`	Ajuste según las necesidades de su carga de trabajo

Utilice el OpenTelemetry Collector de la comunidad para una máxima flexibilidad y una implementación independiente del proveedor.

Crea values.yaml con el mismo contenido que la opción NRDOT anterior, pero cambia la imagen:

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

Toda la demás configuración (receptores, procesadores, pipelines y extraObjects) es idéntica.

Parámetros de configuración: los mismos parámetros que la opción NRDOT anterior. Consulte la tabla de parámetros de configuración para obtener más detalles, incluidos los límites de recursos.

Para opciones de configuración avanzadas, consulte:

Documentación del receptor OTLP

Documentación del receptor de métricas de Kafka

Paso 3. Instalar OpenTelemetry Collector con 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

Paso 4. Verifique el despliegue

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

Paso 1. Crear secreto de credenciales de 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'

Sugerencia

Para otras configuraciones de endpoints, consulte Configure su endpoint OTLP.

Paso 2. Crear archivos de manifiesto

Tanto los recolectores NRDOT como los de OpenTelemetry utilizan una configuración idéntica. Solo difiere la imagen del contenedor. Ambos también requieren el ConfigMap kafka-jmx-config aplicado a su namespace de Kafka.

Crear kafka-jmx-config.yaml - configuración de métricas JMX para el agente de Java (aplicar a su namespace de 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. Crear collector-configmap.yaml - Configuración de OpenTelemetry Collector:

---
apiVersion: v1
kind: ConfigMap
metadata:
  name: otel-collector-config
  namespace: newrelic
  labels:
    app: otel-collector
data:
  otel-collector-config.yaml: |
    receivers:
      otlp:
        protocols:
          grpc:
            endpoint: "0.0.0.0:4317"

      kafkametrics:
        brokers:
          # TODO#1: Replace with your Kafka bootstrap service DNS
          - "kafka.kafka.svc.cluster.local:9092"
        collection_interval: 30s
        protocol_version: 2.0.0
        scrapers:
          - brokers
          - topics
          - consumers
        topic_match: "^[^_].*$"
        metrics:
          kafka.topic.min_insync_replicas:
            enabled: true
          kafka.topic.replication_factor:
            enabled: true
          kafka.partition.replicas:
            enabled: false
          kafka.partition.oldest_offset:
            enabled: false
          kafka.partition.current_offset:
            enabled: false

    exporters:
      otlp/newrelic:
        endpoint: ${NEW_RELIC_OTLP_ENDPOINT}
        tls:
          insecure: false
        sending_queue:
          num_consumers: 12
          queue_size: 5000
        retry_on_failure:
          enabled: true
        compression: gzip
        timeout: 30s
        headers:
          api-key: ${NEW_RELIC_LICENSE_KEY}

    processors:
      batch/aggregation:
        send_batch_size: 1024
        timeout: 30s
      resource:
        attributes:
        - action: insert
          key: kafka.cluster.name
          # TODO#2: Replace with your Kafka cluster name
          value: my-kafka-cluster
      transform/remove_broker_id:
        metric_statements:
        - context: resource
          statements:
          - delete_key(attributes, "broker.id")
      transform/remove_extra_attributes:
        metric_statements:
        - context: resource
          statements:
          - delete_matching_keys(attributes, "^process\\..*")
          - delete_matching_keys(attributes, "^telemetry\\..*")
          - delete_key(attributes, "host.arch")
          - delete_key(attributes, "os.description")
          - delete_matching_keys(attributes, "^cloud\\..*")
          - delete_key(attributes, "service.instance.id") where IsMatch(attributes["service.name"], "^unknown_service:")
          - delete_key(attributes, "service.name") where IsMatch(attributes["service.name"], "^unknown_service:")
      transform/des_units:
        metric_statements:
        - context: metric
          statements:
          - set(description, "") where description != ""
          - set(unit, "") where unit != ""
      filter/internal_topics:
        metrics:
          datapoint:
            - 'attributes["topic"] != nil and IsMatch(attributes["topic"], "^__.*")'
      filter/include_cluster_metrics:
        metrics:
          include:
            match_type: regexp
            metric_names:
            - "kafka\\.partition\\.offline"
            - "kafka\\.(leader|unclean)\\.election\\.rate"
            - "kafka\\.partition\\.non_preferred_leader"
            - "kafka\\.broker\\.fenced\\.count"
            - "kafka\\.cluster\\.partition\\.count"
            - "kafka\\.cluster\\.topic\\.count"
      filter/exclude_cluster_metrics:
        metrics:
          exclude:
            match_type: regexp
            metric_names:
            - "kafka\\.partition\\.offline"
            - "kafka\\.(leader|unclean)\\.election\\.rate"
            - "kafka\\.partition\\.non_preferred_leader"
            - "kafka\\.broker\\.fenced\\.count"
            - "kafka\\.cluster\\.partition\\.count"
            - "kafka\\.cluster\\.topic\\.count"
      cumulativetodelta:
      metricstransform/kafka_topic_sum_aggregation:
        transforms:
        - include: kafka.partition.replicas_in_sync
          action: insert
          new_name: kafka.partition.replicas_in_sync.total
          operations:
          - action: aggregate_labels
            label_set: [topic]
            aggregation_type: sum
        - include: kafka.partition.replicas
          action: insert
          new_name: kafka.partition.replicas.total
          operations:
          - action: aggregate_labels
            label_set: [topic]
            aggregation_type: sum
      filter/remove_partition_level_replicas:
        metrics:
          exclude:
            match_type: strict
            metric_names:
            - kafka.partition.replicas_in_sync
      groupbyattrs/cluster:
        keys: [kafka.cluster.name]
      metricstransform/cluster_max:
        transforms:
          - include: "kafka\\.partition\\.offline|kafka\\.leader\\.election\\.rate|kafka\\.unclean\\.election\\.rate|kafka\\.partition\\.non_preferred_leader|kafka\\.broker\\.fenced\\.count|kafka\\.cluster\\.partition\\.count|kafka\\.cluster\\.topic\\.count"
            match_type: regexp
            action: update
            operations:
              - action: aggregate_labels
                aggregation_type: max
                label_set: []

    service:
      pipelines:
        metrics/broker:
          receivers: [otlp, kafkametrics]
          processors:
            - resource
            - filter/exclude_cluster_metrics
            - filter/internal_topics
            - transform/remove_extra_attributes
            - transform/des_units
            - cumulativetodelta
            - metricstransform/kafka_topic_sum_aggregation
            - filter/remove_partition_level_replicas
            - batch/aggregation
          exporters: [otlp/newrelic]
        metrics/cluster:
          receivers: [otlp]
          processors:
            - resource
            - filter/include_cluster_metrics
            - transform/remove_broker_id
            - transform/remove_extra_attributes
            - transform/des_units
            - cumulativetodelta
            - groupbyattrs/cluster
            - metricstransform/cluster_max
            - batch/aggregation
          exporters: [otlp/newrelic]
        traces/apps:
          receivers: [otlp]
          processors: [resource, batch/aggregation]
          exporters: [otlp/newrelic]
        logs/apps:
          receivers: [otlp]
          processors: [resource, batch/aggregation]
          exporters: [otlp/newrelic]

2. Crear collector-deployment.yaml - despliegue con ServiceAccount y 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

Parámetro de configuración

La siguiente tabla describe los parámetros de configuración clave:

Parámetro	Descripción
`receivers.kafkametrics.brokers`	Reemplace con el DNS de su servicio de arranque de Kafka
`processors.resource.attributes[kafka.cluster.name]`	Reemplace con el nombre de su clúster Kafka
`resources.limits` y `resources.requests` (en `collector-deployment.yaml`)	Ajuste según las necesidades de su carga de trabajo

Use el OpenTelemetry Collector de la comunidad para una implementación independiente del proveedor.

1. Crear collector-configmap.yaml: igual que la opción NRDOT anterior (la configuración es idéntica)

2. Crear collector-deployment.yaml - Solo la imagen del contenedor y el comando difieren:

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

Paso 3. Despliegue los manifiestos

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

Paso 4. Verifique el despliegue

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

Configurar el StatefulSet de Kafka para el agente de Java

Ahora que el recolector se está ejecutando, aplique un parche a su StatefulSet de Kafka para agregar un contenedor de inicio que descargue el JAR del agente de Java de OpenTelemetry, y luego adjúntelo a la JVM del broker de Kafka a través de KAFKA_OPTS.

Agregue las siguientes secciones a su manifiesto de StatefulSet de Kafka existente:

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

Sugerencia

El ConfigMap kafka-jmx-config se desplegó con el recolector en el paso anterior. El valor otel.exporter.otlp.endpoint http://otel-collector.newrelic.svc.cluster.local:4317 asume que el recolector está desplegado en el namespace newrelic con el nombre de servicio otel-collector. Actualícelo para que coincida con el DNS real de su servicio recolector si es diferente.

Parámetro	Descripción
`javaagent`	Adjunta el agente de Java de OpenTelemetry a la JVM del broker de Kafka
`jmx.enabled=true`	Habilita la recolección de métricas JMX
`jmx.config`	Apunta a su archivo de configuración de métricas JMX personalizado (montado desde ConfigMap)
`resource.attributes`	Agrega metadatos `kafka.cluster.name` a todas las métricas
`otlp.endpoint`	Apunta al servicio OpenTelemetry Collector en su clúster
`otlp.protocol=grpc`	Utiliza el protocolo gRPC para OTLP
`metrics.exporter=otlp`	Envía métricas vía OTLP
`logs.exporter=otlp`	Habilita la recopilación de logs del broker. Establezca en `none` para deshabilitar.
`metric.export.interval`	Establece el intervalo en milisegundos entre los intentos de exportación de métricas, por ejemplo, `30000` (30 segundos)

Para ver las opciones de configuración completas, consulte la guía de configuración del agente de Java.

Aplique su StatefulSet actualizado y espere a que los pods se actualicen:

bash

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

(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 de Java de OpenTelemetry.

Para recopilar telemetría a nivel de aplicación de sus aplicaciones de productor y consumidor de Kafka que se ejecutan en Kubernetes, agregue el agente de Java de OpenTelemetry a esos pods de aplicación.

Agregue un contenedor init y variables de entorno al despliegue de su aplicación:

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

Parámetro de configuración

La siguiente tabla describe los parámetros de configuración clave:

Parámetro	Descripción
`order-process-service`	Reemplace con un nombre único para su aplicación productora o consumidora
`my-kafka-cluster`	Reemplace con el mismo nombre de clúster utilizado en la configuración de su broker
`otel-collector.newrelic.svc.cluster.local`	Reemplace con el nombre DNS real de su servicio recolector ( `<service-name>.<namespace>.svc.cluster.local` )

Parámetro

Descripción

order-process-service

Reemplace con un nombre único para su aplicación productora o consumidora

my-kafka-cluster

Reemplace con el mismo nombre de clúster utilizado en la configuración de su broker

otel-collector.newrelic.svc.cluster.local

Reemplace con el nombre DNS real de su servicio recolector (

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

)

El agente de Java proporciona instrumentación de Kafka lista para usar sin cambios de código, capturando latencias de solicitudes, métricas de rendimiento, tasas de errores y rastreo distribuido. Para una configuración avanzada, consulte la documentación de instrumentación de Kafka.

Siga estos pasos para configurar un monitoreo integral de Kafka instalando Prometheus JMX Exporter en sus pods de broker y desplegando un recolector para recopilar y enviar métricas a New Relic.

Antes de que empieces

Asegúrese de tener:

Una cuenta de New Relic con un
Clúster de Kubernetes con acceso kubectl
Kafka desplegado como un StatefulSet con un servicio headless (para nombres DNS de pod estables)
Capacidad para modificar y redesplegar el StatefulSet de Kafka

Crear ConfigMap de métricas JMX

Cree un ConfigMap que contenga la configuración de JMX Exporter que defina qué métricas de Kafka recolectar. Este ConfigMap se montará en cada pod del broker de Kafka.

Guardar como kafka-jmx-config.yaml. Aplíquelo al namespace donde está desplegado Kafka:

apiVersion: v1
kind: ConfigMap
metadata:
  name: kafka-jmx-metrics
  namespace: kafka  # TODO: Replace with your Kafka namespace
data:
  kafka-metrics-config.yml: |
    startDelaySeconds: 0
    lowercaseOutputName: true
    lowercaseOutputLabelNames: true

    rules:
      # Cluster-level controller metrics
      - pattern: 'kafka.controller<type=KafkaController, name=GlobalTopicCount><>Value'
        name: kafka_cluster_topic_count
        type: GAUGE

      - pattern: 'kafka.controller<type=KafkaController, name=GlobalPartitionCount><>Value'
        name: kafka_cluster_partition_count
        type: GAUGE

      - pattern: 'kafka.controller<type=KafkaController, name=FencedBrokerCount><>Value'
        name: kafka_broker_fenced_count
        type: GAUGE

      - pattern: 'kafka.controller<type=KafkaController, name=PreferredReplicaImbalanceCount><>Value'
        name: kafka_partition_non_preferred_leader
        type: GAUGE

      - pattern: 'kafka.controller<type=KafkaController, name=OfflinePartitionsCount><>Value'
        name: kafka_partition_offline
        type: GAUGE

      - pattern: 'kafka.controller<type=KafkaController, name=ActiveControllerCount><>Value'
        name: kafka_controller_active_count
        type: GAUGE

      # Broker-level replica metrics
      - pattern: 'kafka.server<type=ReplicaManager, name=UnderMinIsrPartitionCount><>Value'
        name: kafka_partition_under_min_isr
        type: GAUGE

      - pattern: 'kafka.server<type=ReplicaManager, name=LeaderCount><>Value'
        name: kafka_broker_leader_count
        type: GAUGE

      - pattern: 'kafka.server<type=ReplicaManager, name=PartitionCount><>Value'
        name: kafka_partition_count
        type: GAUGE

      - pattern: 'kafka.server<type=ReplicaManager, name=UnderReplicatedPartitions><>Value'
        name: kafka_partition_under_replicated
        type: GAUGE

      - pattern: 'kafka.server<type=ReplicaManager, name=IsrShrinksPerSec><>Count'
        name: kafka_isr_operation_count
        type: COUNTER
        labels:
          operation: "shrink"

      - pattern: 'kafka.server<type=ReplicaManager, name=IsrExpandsPerSec><>Count'
        name: kafka_isr_operation_count
        type: COUNTER
        labels:
          operation: "expand"

      - pattern: 'kafka.server<type=ReplicaFetcherManager, name=MaxLag, clientId=Replica><>Value'
        name: kafka_max_lag
        type: GAUGE

      # Broker topic metrics (totals)
      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=MessagesInPerSec><>Count'
        name: kafka_message_count
        type: COUNTER

      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=TotalFetchRequestsPerSec><>Count'
        name: kafka_request_count
        type: COUNTER
        labels:
          type: "fetch"

      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=TotalProduceRequestsPerSec><>Count'
        name: kafka_request_count
        type: COUNTER
        labels:
          type: "produce"

      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=FailedFetchRequestsPerSec><>Count'
        name: kafka_request_failed
        type: COUNTER
        labels:
          type: "fetch"

      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=FailedProduceRequestsPerSec><>Count'
        name: kafka_request_failed
        type: COUNTER
        labels:
          type: "produce"

      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=BytesInPerSec><>Count'
        name: kafka_network_io
        type: COUNTER
        labels:
          direction: "in"

      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=BytesOutPerSec><>Count'
        name: kafka_network_io
        type: COUNTER
        labels:
          direction: "out"

      # Per-topic metrics (only appear after traffic flows)
      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=MessagesInPerSec, topic=(.+)><>Count'
        name: kafka_prod_msg_count
        type: COUNTER
        labels:
          topic: "$1"

      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=BytesInPerSec, topic=(.+)><>Count'
        name: kafka_topic_io
        type: COUNTER
        labels:
          topic: "$1"
          direction: "in"

      - pattern: 'kafka.server<type=BrokerTopicMetrics, name=BytesOutPerSec, topic=(.+)><>Count'
        name: kafka_topic_io
        type: COUNTER
        labels:
          topic: "$1"
          direction: "out"

      # Request metrics
      - pattern: 'kafka.network<type=RequestMetrics, name=TotalTimeMs, request=(Produce|FetchConsumer|FetchFollower)><>99thPercentile'
        name: kafka_request_time_99p
        type: GAUGE
        labels:
          type: "$1"

      - pattern: 'kafka.network<type=RequestChannel, name=RequestQueueSize><>Value'
        name: kafka_request_queue
        type: GAUGE

      - pattern: 'kafka.server<type=DelayedOperationPurgatory, name=PurgatorySize, delayedOperation=(.+)><>Value'
        name: kafka_purgatory_size
        type: GAUGE
        labels:
          type: "$1"

      # Controller stats
      - pattern: 'kafka.controller<type=ControllerStats, name=LeaderElectionRateAndTimeMs><>Count'
        name: kafka_leader_election_rate
        type: COUNTER

      - pattern: 'kafka.controller<type=ControllerStats, name=UncleanLeaderElectionsPerSec><>Count'
        name: kafka_unclean_election_rate
        type: COUNTER

      # JVM Garbage Collection
      - pattern: 'java.lang<name=(.+), type=GarbageCollector><>CollectionCount'
        name: jvm_gc_collections_count
        type: COUNTER
        labels:
          name: "$1"

      # JVM Memory
      - pattern: 'java.lang<type=Memory><HeapMemoryUsage>max'
        name: jvm_memory_heap_max
        type: GAUGE

      - pattern: 'java.lang<type=Memory><HeapMemoryUsage>used'
        name: jvm_memory_heap_used
        type: GAUGE

      # JVM Threading and System
      - pattern: 'java.lang<type=Threading><>ThreadCount'
        name: jvm_thread_count
        type: GAUGE

      - pattern: 'java.lang<type=OperatingSystem><>SystemCpuLoad'
        name: jvm_system_cpu_utilization
        type: GAUGE

      # Broker uptime
      - pattern: 'java.lang<type=Runtime><>Uptime'
        name: kafka_broker_uptime
        type: GAUGE

      # Additional metrics — remove this section to reduce data ingest

      # Request latency: total count, 50th percentile, and average (99p kept above)
      - pattern: 'kafka.network<type=RequestMetrics, name=TotalTimeMs, request=(Produce|FetchConsumer|FetchFollower)><>Count'
        name: kafka_request_time_total
        type: COUNTER
        labels:
          type: "$1"

      - pattern: 'kafka.network<type=RequestMetrics, name=TotalTimeMs, request=(Produce|FetchConsumer|FetchFollower)><>50thPercentile'
        name: kafka_request_time_50p
        type: GAUGE
        labels:
          type: "$1"

      - pattern: 'kafka.network<type=RequestMetrics, name=TotalTimeMs, request=(Produce|FetchConsumer|FetchFollower)><>Mean'
        name: kafka_request_time_avg
        type: GAUGE
        labels:
          type: "$1"

      # Log flush metrics
      - pattern: 'kafka.log<type=LogFlushStats, name=LogFlushRateAndTimeMs><>Count'
        name: kafka_logs_flush_count
        type: COUNTER

      - pattern: 'kafka.log<type=LogFlushStats, name=LogFlushRateAndTimeMs><>50thPercentile'
        name: kafka_logs_flush_time_50p
        type: GAUGE

      - pattern: 'kafka.log<type=LogFlushStats, name=LogFlushRateAndTimeMs><>99thPercentile'
        name: kafka_logs_flush_time_99p
        type: GAUGE

      # JVM GC elapsed time
      - pattern: 'java.lang<name=(.+), type=GarbageCollector><>CollectionTime'
        name: jvm_gc_collections_elapsed
        type: COUNTER
        labels:
          name: "$1"

      # JVM Memory heap committed
      - pattern: 'java.lang<type=Memory><HeapMemoryUsage>committed'
        name: jvm_memory_heap_committed
        type: GAUGE

      # JVM class loading
      - pattern: 'java.lang<type=ClassLoading><>LoadedClassCount'
        name: jvm_class_count
        type: GAUGE

      # Additional JVM OS metrics
      - pattern: 'java.lang<type=OperatingSystem><>SystemLoadAverage'
        name: jvm_system_cpu_load_1m
        type: GAUGE

      - pattern: 'java.lang<type=OperatingSystem><>AvailableProcessors'
        name: jvm_cpu_count
        type: GAUGE

      - pattern: 'java.lang<type=OperatingSystem><>ProcessCpuLoad'
        name: jvm_cpu_recent_utilization
        type: GAUGE

      - pattern: 'java.lang<type=OperatingSystem><>OpenFileDescriptorCount'
        name: jvm_file_descriptor_count
        type: GAUGE

      # JVM Memory Pool
      - pattern: 'java.lang<type=MemoryPool, name=(.+)><Usage>used'
        name: jvm_memory_pool_used
        type: GAUGE
        labels:
          name: "$1"

      - pattern: 'java.lang<type=MemoryPool, name=(.+)><Usage>max'
        name: jvm_memory_pool_max
        type: GAUGE
        labels:
          name: "$1"

      - pattern: 'java.lang<type=MemoryPool, name=(.+)><CollectionUsage>used'
        name: jvm_memory_pool_used_after_last_gc
        type: GAUGE
        labels:
          name: "$1"

Sugerencia

Personalizar métricas: puede agregar o modificar patrones consultando los ejemplos de Prometheus JMX Exporter y la documentación de Kafka MBean.

Aplique el ConfigMap:

bash

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

Configurar el StatefulSet de Kafka para JMX Exporter

Aplique un parche a su StatefulSet de Kafka para agregar un contenedor init que descargue el JAR de Prometheus JMX Exporter, luego adjúntelo a la JVM del broker de Kafka a través de KAFKA_OPTS.

Paso 1. Agregue las siguientes secciones a su manifiesto de StatefulSet de Kafka existente:

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

Paso 2. Aplique su StatefulSet actualizado y espere a que los pod roten:

bash

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

Paso 3. Una vez que se complete el despliegue, verifique que las métricas estén expuestas en cada pod del 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

Importante

Clústeres de múltiples brokers: la configuración del contenedor init y de KAFKA_OPTS se aplica a todos los pods en el StatefulSet automáticamente. Verifique que cada pod del broker exponga métricas después del despliegue.

Desplegar OpenTelemetry Collector

Despliegue el OpenTelemetry Collector en su clúster. El recolector recopila datos de los pods del broker de Kafka utilizando objetivos DNS estáticos y escucha en el puerto 4317 los datos OTLP de las aplicaciones instrumentadas.

El método de instalación con Helm es el enfoque recomendado para desplegar OpenTelemetry Collector en Kubernetes.

Paso 1. Crear secreto de credenciales de 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'

Sugerencia

Para otras configuraciones de endpoints, consulte Configure su endpoint OTLP.

Paso 2. Crear values.yaml con la configuración del recolector

Tanto los recolectores NRDOT como los de OpenTelemetry utilizan una configuración idéntica. Seleccione su imagen de recolector preferida:

Cree values.yaml con el siguiente contenido:

# 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

Parámetro de configuración

La siguiente tabla describe los parámetros de configuración clave:

Parámetro	Descripción
`config.receivers.kafkametrics/cluster.brokers`	Reemplace con su servicio de arranque de Kafka (p. ej., `kafka.kafka.svc.cluster.local:9092`)
`config.receivers.kafkametrics/cluster.brokers` y `config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Reemplace con el namespace donde está desplegado su clúster de Kafka
`config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Reemplace con el nombre de su StatefulSet de Kafka seguido de -headless (por ejemplo, `kafka-headless`)
`config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Reemplace con el nombre de su StatefulSet de Kafka (por ejemplo, `kafka`, aparece en los nombres de pod como `kafka-0`)
`config.processors.resource/cluster-name.attributes[kafka.cluster.name].value`	Reemplace con el nombre de su clúster de Kafka (esto se utilizará para identificar y filtrar sus métricas en New Relic)
`config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Actualice la lista `targets` para que coincida con sus pod de broker —una entrada por broker usando el DNS del servicio headless (`<pod-name>.<headless-service>.<namespace>.svc.cluster.local:9404`). Agregue o elimine entradas para que coincidan con la cantidad de brokers. `broker.id` se extrae automáticamente del nombre del pod mediante `relabel_configs`.
`resources.limits` y `resources.requests`	Ajuste según las necesidades de su carga de trabajo

Sugerencia

Alternativa: autodescubrimiento de pods de Kubernetes

En lugar de objetivos de DNS estáticos, puede usar el descubrimiento de pods de Kubernetes para encontrar automáticamente los pods del broker. Esto es útil para el escalado dinámico sin necesidad de actualizar la lista de objetivos.

Reemplace las secciones clusterRole y prometheus/kafka-jmx en values.yaml con:

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

Utilice el OpenTelemetry Collector de la comunidad para una máxima flexibilidad y una implementación independiente del proveedor.

Cree values.yaml con el siguiente contenido (configuración idéntica, imagen diferente):

# 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

Parámetro de configuración

La siguiente tabla describe los parámetros de configuración clave:

Parámetro	Descripción
`config.receivers.kafkametrics/cluster.brokers`	Reemplace con el DNS de su servicio de arranque de Kafka
`config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Agregue una entrada por pod de broker usando el DNS del servicio headless (`<pod-name>.<headless-service>.<namespace>.svc.cluster.local:9404`). Agregue o elimine entradas para que coincidan con la cantidad de brokers. `broker.id` se extrae automáticamente del nombre del pod mediante `relabel_configs`.
`config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].labels[kafka.cluster.name]`	Reemplace el valor de la etiqueta `kafka.cluster.name` con el nombre de su clúster
`config.processors.resource/cluster-name.attributes[kafka.cluster.name].value`	Reemplace con el nombre de su clúster de Kafka (utilizado en el procesador `resource/cluster-name` para identificar métricas en New Relic)
`resources.limits` y `resources.requests`	Ajuste según las necesidades de su carga de trabajo

Sugerencia

Alternativa: autodescubrimiento de pods de Kubernetes

Reemplace la sección prometheus/kafka-jmx en values.yaml con:

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

Para opciones de configuración avanzadas, consulte estas páginas de documentación del receptor:

Documentación del receptor Prometheus

Documentación del receptor de métricas de Kafka

Paso 3. Instalar OpenTelemetry Collector con 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

Paso 4. Verifique el despliegue:

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

Debería ver logs que indiquen un scraping exitoso de los pods de los brokers de Kafka en el puerto 9404.

El método de instalación de manifiesto proporciona control directo sobre los recursos de Kubernetes sin usar Helm.

Paso 1. Crear secreto de credenciales de 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'

Sugerencia

Para otras configuraciones de endpoints, consulte Configure su endpoint OTLP.

Paso 2. Crear archivos de manifiesto

Tanto los recolectores NRDOT como los de OpenTelemetry utilizan una configuración idéntica. Solo la imagen del contenedor difiere.

1. Crear collector-configmap.yaml - Configuración de OpenTelemetry Collector:

---
apiVersion: v1
kind: ConfigMap
metadata:
  name: otel-collector-config
  namespace: newrelic
  labels:
    app: otel-collector
data:
  otel-collector-config.yaml: |
    receivers:
      otlp:
        protocols:
          grpc:
            endpoint: "0.0.0.0:4317"

      kafkametrics/cluster:
        brokers:
          # TODO#1: Replace with your Kafka bootstrap service DNS
          - "kafka.kafka.svc.cluster.local:9092"
        collection_interval: 30s
        protocol_version: 2.0.0
        scrapers:
          - brokers
          - topics
          - consumers
        topic_match: "^[^_].*$"
        metrics:
          kafka.topic.min_insync_replicas:
            enabled: true
          kafka.topic.replication_factor:
            enabled: true
          kafka.partition.replicas:
            enabled: false
          kafka.partition.oldest_offset:
            enabled: false
          kafka.partition.current_offset:
            enabled: false

      prometheus/kafka-jmx:
        config:
          scrape_configs:
            - job_name: 'kafka-jmx-metrics'
              metrics_path: /metrics
              scrape_interval: 30s
              static_configs:
                # TODO#2: Add one entry per broker pod using headless service DNS
                - targets:
                    - 'kafka-0.kafka-headless.kafka.svc.cluster.local:9404'
                    - 'kafka-1.kafka-headless.kafka.svc.cluster.local:9404'
                    - 'kafka-2.kafka-headless.kafka.svc.cluster.local:9404'
                  labels:
                    kafka.cluster.name: 'my-kafka-cluster'  # TODO#3: Replace with your cluster name
              relabel_configs:
                - source_labels: [__address__]
                  target_label: broker.id
                  regex: '[^-]+-(\d+)\..+:\d+'
                  replacement: '$1'

    exporters:
      otlp/backend:
        endpoint: ${NEW_RELIC_OTLP_ENDPOINT}
        tls:
          insecure: false
        sending_queue:
          num_consumers: 12
          queue_size: 5000
        retry_on_failure:
          enabled: true
        headers:
          api-key: ${NEW_RELIC_LICENSE_KEY}

    processors:
      batch/export:
        send_batch_size: 1024
        timeout: 30s
      memory_limiter:
        limit_percentage: 80
        spike_limit_percentage: 30
        check_interval: 1s
      transform/metric-naming:
        metric_statements:
        - context: metric
          statements:
          - replace_pattern(name, "_", ".")
          - replace_pattern(name, "\\.load\\.1", ".load_1")
          - replace_pattern(name, "\\.recent\\.util", ".recent_util")
          - replace_pattern(name, "file\\.descriptor\\.count", "file_descriptor.count")
          - replace_pattern(name, "\\.memory\\.pool\\.used\\.bytes$", ".memory.pool.used")
          - replace_pattern(name, "\\.memory\\.pool\\.max\\.bytes$", ".memory.pool.max")
          - replace_pattern(name, "\\.memory\\.pool\\.collection\\.used\\.bytes$", ".memory.pool.used_after_last_gc")
          - replace_pattern(name, "\\.non\\.preferred\\.leader", ".non_preferred_leader")
          - replace_pattern(name, "\\.under\\.min\\.isr", ".under_min_isr")
          - replace_pattern(name, "\\.under\\.replicated", ".under_replicated")
          - replace_pattern(name, "\\.total$", "") where name != "kafka.request.time.total"
        - context: datapoint
          statements:
          - set(attributes["name"], attributes["gc"]) where attributes["gc"] != nil
          - delete_key(attributes, "gc") where attributes["gc"] != nil
          - set(attributes["name"], attributes["pool"]) where attributes["pool"] != nil
          - delete_key(attributes, "pool") where attributes["pool"] != nil
      resource/cluster-name:
        attributes:
        - key: kafka.cluster.name
          # TODO#5: Replace with your Kafka cluster name (this will be used to identify and filter your metrics in New Relic)
          value: my-kafka-cluster
          action: upsert
      transform/remove_broker_id:
        metric_statements:
        - context: datapoint
          statements:
          - delete_key(attributes, "broker.id")
      filter/scrape-overhead:
        metrics:
          exclude:
            match_type: regexp
            metric_names:
            - "^jmx_.*"
            - "^process_.*"
            - "^jvm_buffer_pool_.*"
            - "^jvm_threads_.*"
            - "^jvm_classes_.*"
            - "^jvm_memory_(heap|non_heap)_(committed|init|max|used)_bytes$"
            - "^jvm_compilation_.*"
            - "^jvm_(runtime|info).*"
            - "^jvm_memory_pool_(allocated_bytes_total|committed_bytes|init_bytes|collection_(committed|init|max)_bytes)$"
      filter/include_cluster_metrics:
        metrics:
          include:
            match_type: regexp
            metric_names:
            - "^kafka\\.partition\\.offline$"
            - "^kafka\\.(leader|unclean)\\.election\\.rate$"
            - "^kafka\\.partition\\.non_preferred_leader$"
            - "^kafka\\.broker\\.fenced\\.count$"
            - "^kafka\\.cluster\\.partition\\.count$"
            - "^kafka\\.cluster\\.topic\\.count$"
      filter/exclude_cluster_metrics:
        metrics:
          exclude:
            match_type: regexp
            metric_names:
            - "^kafka\\.partition\\.offline$"
            - "^kafka\\.(leader|unclean)\\.election\\.rate$"
            - "^kafka\\.partition\\.non_preferred_leader$"
            - "^kafka\\.broker\\.fenced\\.count$"
            - "^kafka\\.cluster\\.partition\\.count$"
            - "^kafka\\.cluster\\.topic\\.count$"
      transform/remove_attributes:
        metric_statements:
        - context: metric
          statements:
          - set(description, "") where description != ""
          - set(unit, "") where unit != ""
        - context: resource
          statements:
          - delete_key(attributes, "server.address")
          - delete_key(attributes, "server.port")
          - delete_key(attributes, "service.instance.id")
          - delete_key(attributes, "host.name")
          - delete_key(attributes, "k8s.pod.uid")
          - delete_key(attributes, "url.scheme")
      metricstransform/topic-aggregation:
        transforms:
        - include: kafka.partition.replicas_in_sync
          action: insert
          new_name: kafka.partition.replicas_in_sync.total
          operations:
          - action: aggregate_labels
            label_set: [topic]
            aggregation_type: sum
        - include: kafka.partition.replicas
          action: insert
          new_name: kafka.partition.replicas.total
          operations:
          - action: aggregate_labels
            label_set: [topic]
            aggregation_type: sum
      filter/exclude_partition_replicas_metric:
        metrics:
          exclude:
            match_type: strict
            metric_names:
            - kafka.partition.replicas_in_sync
      filter/internal_topics:
        metrics:
          datapoint:
            - 'attributes["topic"] != nil and IsMatch(attributes["topic"], "^__.*")'
      cumulativetodelta:
      groupbyattrs/cluster:
        keys: [kafka.cluster.name]
      metricstransform/cluster_max:
        transforms:
          - include: "kafka\\.partition\\.offline|kafka\\.leader\\.election\\.rate|kafka\\.unclean\\.election\\.rate|kafka\\.partition\\.non_preferred_leader|kafka\\.broker\\.fenced\\.count|kafka\\.cluster\\.partition\\.count|kafka\\.cluster\\.topic\\.count"
            match_type: regexp
            action: update
            operations:
              - action: aggregate_labels
                aggregation_type: max
                label_set: []

    service:
      pipelines:
        traces:
          receivers: [otlp]
          processors: [memory_limiter, batch/export]
          exporters: [otlp/backend]
        metrics:
          receivers: [otlp]
          processors: [memory_limiter, batch/export]
          exporters: [otlp/backend]
        logs:
          receivers: [otlp]
          processors: [memory_limiter, batch/export]
          exporters: [otlp/backend]
        metrics/broker:
          receivers: [prometheus/kafka-jmx]
          processors:
            - resource/cluster-name
            - filter/scrape-overhead
            - transform/metric-naming
            - transform/remove_attributes
            - filter/exclude_cluster_metrics
            - memory_limiter
            - cumulativetodelta
            - batch/export
          exporters: [otlp/backend]
        metrics/cluster/prometheus:
          receivers: [prometheus/kafka-jmx]
          processors:
            - resource/cluster-name
            - filter/scrape-overhead
            - transform/metric-naming
            - transform/remove_attributes
            - filter/include_cluster_metrics
            - transform/remove_broker_id
            - memory_limiter
            - cumulativetodelta
            - groupbyattrs/cluster
            - metricstransform/cluster_max
            - batch/export
          exporters: [otlp/backend]
        metrics/cluster/kafkametrics:
          receivers: [kafkametrics/cluster]
          processors:
            - resource/cluster-name
            - filter/internal_topics
            - transform/remove_attributes
            - metricstransform/topic-aggregation
            - filter/exclude_partition_replicas_metric
            - memory_limiter
            - cumulativetodelta
            - batch/export
          exporters: [otlp/backend]

2. Crear collector-deployment.yaml - Despliegue de OpenTelemetry Collector con 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

Parámetro de configuración

La siguiente tabla describe los parámetros de configuración clave:

Parámetro	Descripción
`receivers.kafkametrics/cluster.brokers`	Reemplace con el DNS de su servicio de arranque de Kafka (por ejemplo, `kafka.kafka.svc.cluster.local:9092`)
`receivers.kafkametrics/cluster.brokers` y `receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Reemplace con el namespace donde está desplegado su clúster de Kafka
`receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Reemplace con el nombre de su StatefulSet de Kafka seguido de `-headless` (por ejemplo, `kafka-headless`)
`receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Reemplace con el nombre de su StatefulSet de Kafka (por ejemplo, `kafka`, aparece en los nombres de pod como `kafka-0`)
`processors.resource/cluster-name.attributes[kafka.cluster.name].value`	Reemplace con el nombre de su clúster de Kafka (utilizado para identificar métricas en New Relic)
`receivers.prometheus/kafka-jmx.config.scrape_configs[0].static_configs[0].targets`	Actualice la lista `targets` para que coincida con sus pod de broker —una entrada por broker usando el DNS del servicio headless (`<pod-name>.<headless-service>.<namespace>.svc.cluster.local:9404`). `broker.id` se extrae automáticamente del nombre del pod mediante `relabel_configs`.
`resources.limits` y `resources.requests` (en `collector-deployment.yaml`)	Ajuste según las necesidades de su carga de trabajo

Use el OpenTelemetry Collector de la comunidad para una implementación independiente del proveedor.

1. Crear collector-configmap.yaml: igual que la opción NRDOT anterior (la configuración es idéntica)

2. Crear collector-deployment.yaml - Solo cambia la imagen del contenedor:

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

Para opciones de configuración avanzadas, consulte estas páginas de documentación del receptor:

Documentación del receptor Prometheus

Documentación del receptor de métricas de Kafka

Paso 3. Despliegue los manifiestos

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

Paso 4. Verifique el despliegue:

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

Debería ver logs que indiquen un scraping exitoso de los pods de los brokers de Kafka en el puerto 9404.

(Opcional) Instrumentar aplicaciones productoras o consumidoras

Importante

Soporte de lenguajes: las aplicaciones Java admiten la instrumentación del cliente Kafka lista para usar mediante el agente de Java de OpenTelemetry.

Para recopilar telemetría a nivel de aplicación de sus aplicaciones de productor y consumidor de Kafka, use el agente de Java de OpenTelemetry con un contenedor 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: {}

Parámetro de configuración

La siguiente tabla describe los parámetros de configuración clave:

Parámetro	Descripción
`service.name`	Reemplace `my-kafka-app` con un nombre único para su aplicación productora o consumidora
`kafka.cluster.name`	Reemplace `my-kafka-cluster` con el mismo nombre de clúster utilizado en la configuración de su colector
`otlp.endpoint`	El extremo `http://otel-collector.newrelic.svc.cluster.local:4317` asume que el recolector está desplegado en el namespace `newrelic` como `otel-collector`

(Opcional) Enviar logs del broker de Kafka

Para recopilar logs del broker de Kafka y enviarlos a New Relic, agregue un receptor filelog a la configuración de su recolector.

Paso 1. Agregar a la sección de receptores:

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}

Paso 2. Agregar pipeline de logs a la sección de servicio:

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

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

Parámetro de configuración

La siguiente tabla describe los parámetros de configuración clave:

Parámetro	Descripción
`filelog/kafka_broker_0.include`	Actualice `/var/log/kafka/server.log` a la ruta real del log de Kafka dentro del pod del broker
`filelog/kafka_broker_0.resource.broker.id`	El atributo de recurso `broker.id` correlaciona los logs con métricas y entidades específicas del broker
Múltiples receptores de broker	Para múltiples brokers, cree receptores `filelog` separados (p. ej., `filelog/kafka_broker_1`, `filelog/kafka_broker_2`) con sus respectivos ID de broker
`filelog/kafka_broker_0.multiline.line_start_pattern`	El patrón `multiline` asume que los logs comienzan con `[`; ajústelo si su formato de log difiere
Volumen de logs	Considere el volumen de logs y los costos de recopilación antes de habilitar el reenvío de logs
Referencia	Para ver las opciones de configuración completas, consulte la documentación del receptor filelog

Paso 3. Actualice el release de Helm:

bash

$helm upgrade kafka-otel-collector open-telemetry/opentelemetry-collector \
>  --namespace newrelic \
>  --values values.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'
```

Encuentra tus datos

Después de unos minutos, sus datos de Kafka deberían aparecer en New Relic. Consulta Encuentra tus datos para obtener instrucciones detalladas sobre cómo explorar tus datos de Kafka en las diferentes vistas de la UI de New Relic.

La siguiente tabla resume dónde se almacena cada tipo de señal. Reemplace my-kafka-cluster por su valor KAFKA_CLUSTER_NAME en todas las consultas a continuación:

Señal	Tipo de evento	Qué incluye
Métrica	`Metric`	Métricas de broker, topic, partición, grupo de consumidores y JVM
Logs	`Log`	Logs de aplicaciones de productor y consumidor (mediante el agente de Java de OTel) y logs del broker recopilados mediante el agente de Java
Traza	`Span`	Spans de productor y consumidor, incluidas las operaciones `publish` y `receive` por mensaje en todos los temas

Métrica

Las métricas de broker, tema, partición, grupo de consumidores y JVM se almacenan en el tipo de evento Metric:

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

Logs

Los logs de las aplicaciones de productor y consumidor instrumentadas con el agente de Java de OpenTelemetry, y los logs del broker recopilados a través del agente de Java en el broker, se almacenan en el tipo de evento Log:

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

Traza

Los spans de productor y consumidor, incluidas las operaciones publish y receive por mensaje en todos los temas, se almacenan en el tipo de evento Span:

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

Ejemplo

Un ejemplo funcional completo con manifiestos de StatefulSet de Kafka, valores de Helm, configuración de OTel Collector y aplicaciones de productor/consumidor de muestra está disponible en el respositorio de New Relic OpenTelemetry Examples.

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 pod del recolector se está ejecutando:

Para instalaciones con manifiesto:

bash

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

Para instalaciones con 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

Compruebe si los pods del broker de Kafka se están ejecutando con el agente de 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"

Verifique que el volumen otel-agent esté poblado:

bash

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

Probar la conectividad desde el pod del broker al servicio del recolector:

bash

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

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

En su ConfigMap (collector-configmap.yaml), agregue a la sección service:

service:
  telemetry:
    logs:
      level: "debug"

Luego aplique el ConfigMap actualizado y reinicie el despliegue del recolector:

bash

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

Agregar exportador de depuración: ver métricas en los logs del recolector antes de enviarlas a New Relic. Los nombres del procesador y del exportador difieren según el método de monitoreo:

Método del agente de 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étodo de Prometheus JMX Exporter:

exporters:
  debug:
    verbosity: detailed

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

service:
  pipelines:
    metrics/broker:
      receivers: [prometheus/kafka-jmx]
      processors: [resource/cluster-name, filter/scrape-overhead, transform/metric-naming, transform/remove_attributes, filter/exclude_cluster_metrics, memory_limiter, cumulativetodelta, batch/export]
      exporters: [debug, otlp/backend]

    metrics/cluster/prometheus:
      receivers: [prometheus/kafka-jmx]
      processors: [resource/cluster-name, filter/scrape-overhead, transform/metric-naming, transform/remove_attributes, filter/include_cluster_metrics, transform/remove_broker_id, memory_limiter, cumulativetodelta, groupbyattrs/cluster, metricstransform/cluster_max, batch/export]
      exporters: [debug, otlp/backend]

    metrics/cluster/kafkametrics:
      receivers: [kafkametrics/cluster]
      processors: [resource/cluster-name, filter/internal_topics, transform/remove_attributes, metricstransform/topic-aggregation, filter/exclude_partition_replicas_metric, memory_limiter, cumulativetodelta, batch/export]
      exporters: [debug, otlp/backend]

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 pod del recolector y los pod del broker estén en buen estado.

Revisa los registros del recolector en busca de errores (usa la etiqueta que coincida con tu método de instalación — consulta Comprobaciones iniciales del sistema):

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"

Verifique que el servicio del recolector exista y tenga el puerto correcto:

bash

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

Asegúrese de que el puerto 4317 esté expuesto como un servicio ClusterIP.

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

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

bash

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

Verifique que KAFKA_OPTS esté configurado correctamente en los pods del broker:

bash

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

Esto debería mostrar -javaagent:/otel-agent/opentelemetry-javaagent.jar y todos los -Dotel.* parámetros. Verificar:

-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

Verificar que el ConfigMap JMX esté montado:

bash

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

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

bash

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

Primero, ejecute las comprobaciones iniciales del sistema para verificar que el servicio del recolector sea accesible desde los pods del broker.

Verificar la resolución de DNS:

bash

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

Verifique los logs del recolector en busca de errores de OTLP:

bash

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

Verifique que el receptor OTLP esté escuchando en todas las interfaces:

Asegúrese de que el ConfigMap tenga endpoint: "0.0.0.0:4317" (no 127.0.0.1) en el receptor otlp:

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

Monitorear la memoria del pod del recolector:

bash

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

Reducir los temas monitoreados:

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

Reduzca la frecuencia de recolección: aumente los intervalos para recolectar con menos frecuencia

receivers:
  kafkametrics:
    collection_interval: 60s

Para las métricas JMX del agente de Java, actualice KAFKA_OPTS en el StatefulSet:

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

Agregar un limitador de memoria:

Método del agente de 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étodo de Prometheus JMX Exporter:

processors:
  memory_limiter:
    check_interval: 1s
    limit_mib: 512
    spike_limit_mib: 128

service:
  pipelines:
    metrics/broker:
      processors: [memory_limiter, resource/cluster-name, filter/scrape-overhead, transform/metric-naming, transform/remove_attributes, filter/exclude_cluster_metrics, cumulativetodelta, batch/export]
      ...

Después de los cambios, aplique el ConfigMap actualizado y reinicie el recolector:

bash

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

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

Kafka autohospedado - Monitoreo de Kafka para entornos autohospedados (no Kubernetes)
Kubernetes Strimzi - monitoreo de Kafka para Kafka administrado por Strimzi en Kubernetes
Agente de Java de OpenTelemetry - Documentación oficial del agente de Java de OTel
Prometheus JMX Exporter - agente de Java que expone métricas de JMX en formato Prometheus
Receptor de Prometheus - receptor de OTel Collector para recopilar extremos de métricas de Prometheus
receptor kafkametrics - Documentación del receptor de métricas de temas y retraso del consumidor

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

Monitorear Kafka autogestionado en Kubernetes con OpenTelemetry

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

Pasos de instalación

Antes de que empieces

Desplegar OpenTelemetry Collector

región de la UE

Sugerencia

Uso de NRDOT Collector (recomendado)

Uso de OpenTelemetry Collector

Región de EE. UU.

región de la UE

Sugerencia

Uso de NRDOT Collector (recomendado)

Uso de OpenTelemetry Collector

Configurar el StatefulSet de Kafka para el agente de Java

Sugerencia

Parámetro de configuración

(Opcional) Instrumentar aplicaciones productoras o consumidoras

Importante

Parámetro de configuración

Antes de que empieces

Crear ConfigMap de métricas JMX

Sugerencia

Configurar el StatefulSet de Kafka para JMX Exporter

Importante

Desplegar OpenTelemetry Collector

Región de EE. UU.

región de la UE

Sugerencia

Uso de NRDOT Collector (recomendado)

Uso de OpenTelemetry Collector

Región de EE. UU.

región de la UE

Sugerencia

Uso de NRDOT Collector (recomendado)

Uso de OpenTelemetry Collector

(Opcional) Instrumentar aplicaciones productoras o consumidoras

Importante

(Opcional) Enviar logs del broker de Kafka

Configurar la recopilación de logs

Encuentre sus logs en New Relic

Encuentra tus datos

Métrica

Logs

Traza

Ejemplo

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 desde los pods del broker

Alto uso de memoria

Próximos pasos

Recursos relacionados

Arquitectura