--- title: Monitor Kafka on Kubernetes (Strimzi) with OpenTelemetry source: https://docs.newrelic.com/docs/opentelemetry/integrations/kafka/kubernetes-strimzi --- Monitor your Kafka cluster running on Kubernetes with Strimzi operator by deploying the OpenTelemetry Collector. The collector automatically discovers Kafka broker pods and collects comprehensive metrics. ## Architecture [#architecture-overview] The following diagram illustrates the monitoring architecture and data flow to New Relic. ![Kubernetes Strimzi Kafka monitoring architecture with OpenTelemetry](https://docs.newrelic.com/images/otel-kafka-k8s-strimzi-architecture.webp "Kubernetes Strimzi Kafka monitoring architecture with OpenTelemetry") ## Installation steps [#installation-steps] Follow these steps to set up monitoring for your Kafka cluster: ### Before you begin [#prerequisites] Ensure you have: - A [New Relic account](https://newrelic.com/signup) with a license key - Kubernetes cluster with kubectl access - Kafka deployed via [Strimzi operator](https://strimzi.io/) ### Configure Kafka cluster for Kafka JMX metrics Configure your Strimzi Kafka cluster to expose Kafka JMX metrics via the Prometheus JMX Exporter. This configuration will be deployed as a ConfigMap and referenced by your Kafka cluster. **Step 1. Create JMX metrics ConfigMap** Create a ConfigMap with JMX Exporter patterns that define which Kafka metrics to collect. Save as `kafka-jmx-config.yaml`: ```yaml apiVersion: v1 kind: ConfigMap metadata: name: kafka-jmx-metrics namespace: newrelic data: kafka-metrics-config.yml: | startDelaySeconds: 0 lowercaseOutputName: true lowercaseOutputLabelNames: true rules: # Cluster-level controller metrics - pattern: 'kafka.controller<>Value' name: kafka_cluster_topic_count type: GAUGE - pattern: 'kafka.controller<>Value' name: kafka_cluster_partition_count type: GAUGE - pattern: 'kafka.controller<>Value' name: kafka_broker_fenced_count type: GAUGE - pattern: 'kafka.controller<>Value' name: kafka_partition_non_preferred_leader type: GAUGE - pattern: 'kafka.controller<>Value' name: kafka_partition_offline type: GAUGE - pattern: 'kafka.controller<>Value' name: kafka_controller_active_count type: GAUGE # Broker-level replica metrics - pattern: 'kafka.server<>Value' name: kafka_partition_under_min_isr type: GAUGE - pattern: 'kafka.server<>Value' name: kafka_broker_leader_count type: GAUGE - pattern: 'kafka.server<>Value' name: kafka_partition_count type: GAUGE - pattern: 'kafka.server<>Value' name: kafka_partition_under_replicated type: GAUGE - pattern: 'kafka.server<>Count' name: kafka_isr_operation_count type: COUNTER labels: operation: "shrink" - pattern: 'kafka.server<>Count' name: kafka_isr_operation_count type: COUNTER labels: operation: "expand" - pattern: 'kafka.server<>Value' name: kafka_max_lag type: GAUGE # Broker topic metrics (totals) - pattern: 'kafka.server<>Count' name: kafka_message_count type: COUNTER - pattern: 'kafka.server<>Count' name: kafka_request_count type: COUNTER labels: type: "fetch" - pattern: 'kafka.server<>Count' name: kafka_request_count type: COUNTER labels: type: "produce" - pattern: 'kafka.server<>Count' name: kafka_request_failed type: COUNTER labels: type: "fetch" - pattern: 'kafka.server<>Count' name: kafka_request_failed type: COUNTER labels: type: "produce" - pattern: 'kafka.server<>Count' name: kafka_network_io type: COUNTER labels: direction: "in" - pattern: 'kafka.server<>Count' name: kafka_network_io type: COUNTER labels: direction: "out" # Per-topic metrics (only appear after traffic flows) - pattern: 'kafka.server<>Count' name: kafka_prod_msg_count type: COUNTER labels: topic: "$1" - pattern: 'kafka.server<>Count' name: kafka_topic_io type: COUNTER labels: topic: "$1" direction: "in" - pattern: 'kafka.server<>Count' name: kafka_topic_io type: COUNTER labels: topic: "$1" direction: "out" # Request metrics - pattern: 'kafka.network<>99thPercentile' name: kafka_request_time_99p type: GAUGE labels: type: "$1" - pattern: 'kafka.network<>Value' name: kafka_request_queue type: GAUGE - pattern: 'kafka.server<>Value' name: kafka_purgatory_size type: GAUGE labels: type: "$1" # Controller stats - pattern: 'kafka.controller<>Count' name: kafka_leader_election_rate type: COUNTER - pattern: 'kafka.controller<>Count' name: kafka_unclean_election_rate type: COUNTER # JVM Garbage Collection - pattern: 'java.lang<>CollectionCount' name: jvm_gc_collections_count type: COUNTER labels: name: "$1" # JVM Memory - pattern: 'java.langmax' name: jvm_memory_heap_max type: GAUGE - pattern: 'java.langused' name: jvm_memory_heap_used type: GAUGE # JVM Threading and System - pattern: 'java.lang<>ThreadCount' name: jvm_thread_count type: GAUGE - pattern: 'java.lang<>SystemCpuLoad' name: jvm_system_cpu_utilization type: GAUGE # Broker uptime - pattern: 'java.lang<>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<>Count' name: kafka_request_time_total type: COUNTER labels: type: "$1" - pattern: 'kafka.network<>50thPercentile' name: kafka_request_time_50p type: GAUGE labels: type: "$1" - pattern: 'kafka.network<>Mean' name: kafka_request_time_avg type: GAUGE labels: type: "$1" # Log flush metrics - pattern: 'kafka.log<>Count' name: kafka_logs_flush_count type: COUNTER - pattern: 'kafka.log<>50thPercentile' name: kafka_logs_flush_time_50p type: GAUGE - pattern: 'kafka.log<>99thPercentile' name: kafka_logs_flush_time_99p type: GAUGE # JVM GC elapsed time - pattern: 'java.lang<>CollectionTime' name: jvm_gc_collections_elapsed type: COUNTER labels: name: "$1" # JVM Memory heap committed - pattern: 'java.langcommitted' name: jvm_memory_heap_committed type: GAUGE # JVM class loading - pattern: 'java.lang<>LoadedClassCount' name: jvm_class_count type: GAUGE # Additional JVM OS metrics - pattern: 'java.lang<>SystemLoadAverage' name: jvm_system_cpu_load_1m type: GAUGE - pattern: 'java.lang<>AvailableProcessors' name: jvm_cpu_count type: GAUGE - pattern: 'java.lang<>ProcessCpuLoad' name: jvm_cpu_recent_utilization type: GAUGE - pattern: 'java.lang<>OpenFileDescriptorCount' name: jvm_file_descriptor_count type: GAUGE # JVM Memory Pool - pattern: 'java.langused' name: jvm_memory_pool_used type: GAUGE labels: name: "$1" - pattern: 'java.langmax' name: jvm_memory_pool_max type: GAUGE labels: name: "$1" - pattern: 'java.langused' name: jvm_memory_pool_used_after_last_gc type: GAUGE labels: name: "$1" ``` > #### 💡 TIP > > **Customize metrics**: This ConfigMap includes comprehensive Kafka broker, topic, request, controller, and JVM metrics. You can add or modify patterns by referencing the [Prometheus JMX Exporter examples](https://github.com/prometheus/jmx_exporter/tree/main/examples) and [Kafka MBean documentation](https://kafka.apache.org/documentation/#monitoring). Refer to the [JMX Exporter rules documentation](https://github.com/prometheus/jmx_exporter/blob/main/docs/content/1.5.0/http-mode/rules.md) for additional configurations. > #### ⚠️ IMPORTANT > > **Namespace requirement**: The JMX metrics ConfigMap and your Kafka cluster must be in the same namespace. In this guide, both are deployed to the `newrelic` namespace. Apply the ConfigMap: ```bash kubectl apply -f kafka-jmx-config.yaml ``` **Step 2. Update Kafka cluster to use JMX Exporter** Update your Strimzi Kafka resource to reference the metrics ConfigMap: ```yaml apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka metadata: name: my-cluster namespace: newrelic spec: kafka: version: X.X.X metricsConfig: type: jmxPrometheusExporter valueFrom: configMapKeyRef: name: kafka-jmx-metrics key: kafka-metrics-config.yml # ...rest of your Kafka configuration ``` Apply the changes. Strimzi will perform a rolling restart of your Kafka brokers: ```bash kubectl apply -f kafka-cluster.yaml ``` After the rolling restart completes, each Kafka broker will expose Prometheus metrics on port `9404`. ### Deploy OpenTelemetry Collector [#deploy-opentelemetry-collector] Deploy the OpenTelemetry Collector to monitor your Kafka cluster. Choose your preferred installation method: #### Helm install (recommended) The Helm installation method is the recommended approach for deploying OpenTelemetry Collector in Kubernetes. **Step 1. Create New Relic credentials secret** Create a Kubernetes secret containing your New Relic license key and OTLP endpoint. Choose the endpoint for your New Relic region: **US region (default)** ````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' ``` ```` **EU region** ````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' ``` ```` **JP region** ````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.jp.nr-data.net:4317' ``` ```` > #### 💡 TIP > > For other endpoint configurations, see [Configure your OTLP endpoint](https://docs.newrelic.com/docs/opentelemetry/best-practices/opentelemetry-otlp/#configure-endpoint-port-protocol). **Step 2. Create values.yaml with collector configuration** Create a `values.yaml` file that contains the complete OpenTelemetry Collector configuration. Both NRDOT and OpenTelemetry collectors use identical configuration and provide the same Kafka monitoring capabilities. Choose your preferred collector image: **Using NRDOT Collector (recommended)** **NRDOT** is New Relic's supported distribution of the OpenTelemetry Collector, providing full New Relic support. For more information, see the [NRDOT Collector GitHub repository](https://github.com/newrelic/nrdot-collector-releases/tree/main/distributions/nrdot-collector). Create `values.yaml` with the following content: ```yaml # Deployment mode mode: deployment replicaCount: 1 # Use NRDOT collector image image: repository: newrelic/nrdot-collector tag: "latest" pullPolicy: Always # Service account configuration serviceAccount: create: true name: otel-collector # RBAC for Kubernetes service discovery clusterRole: create: true rules: - apiGroups: [""] resources: ["pods", "nodes"] verbs: ["get", "list", "watch"] # Pod security context podSecurityContext: runAsNonRoot: true runAsUser: 10001 # Container security context securityContext: allowPrivilegeEscalation: false readOnlyRootFilesystem: true capabilities: drop: - ALL # Resource limits resources: requests: memory: 512Mi cpu: 250m limits: memory: 1Gi cpu: 500m # Load environment variables from secret extraEnvsFrom: - secretRef: name: newrelic-otlp-secret # Disable unused default ports ports: jaeger-compact: enabled: false jaeger-thrift: enabled: false jaeger-grpc: enabled: false zipkin: enabled: false # OpenTelemetry Collector Configuration config: receivers: # Disable default receivers not needed in NRDOT jaeger: null zipkin: null kafkametrics/cluster: brokers: # TODO#1: Replace with your Kafka bootstrap service - "my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092" collection_interval: 30s protocol_version: 2.0.0 scrapers: - brokers - topics - consumers # Exclude internal Kafka topics (prefixed with __) at the source topic_match: "^[^_].*$" metrics: kafka.topic.min_insync_replicas: enabled: true kafka.topic.replication_factor: enabled: true kafka.partition.replicas: enabled: false kafka.partition.oldest_offset: enabled: false kafka.partition.current_offset: enabled: false prometheus/kafka-jmx: config: scrape_configs: - job_name: 'kafka-jmx-metrics' scrape_interval: 30s kubernetes_sd_configs: - role: pod namespaces: names: # TODO#2: Replace with the namespace where your Kafka cluster is deployed - newrelic relabel_configs: # Filter for Kafka broker pods - source_labels: [__meta_kubernetes_pod_label_strimzi_io_name] action: keep # TODO#3: Replace with your Strimzi Kafka cluster name followed by '-kafka' regex: my-cluster-kafka - source_labels: [__meta_kubernetes_pod_label_strimzi_io_cluster] action: keep # TODO#4: Replace with your Strimzi Kafka cluster name regex: my-cluster # Extract broker ID from pod name - source_labels: [__meta_kubernetes_pod_name] target_label: broker.id regex: '.*-(\\d+)$' replacement: '$1' # Set scrape target to pod IP on port 9404 - source_labels: [__meta_kubernetes_pod_ip] target_label: __address__ replacement: '$1:9404' exporters: # New Relic OTLP exporter otlp/backend: endpoint: ${NEW_RELIC_OTLP_ENDPOINT} tls: insecure: false sending_queue: num_consumers: 12 queue_size: 5000 retry_on_failure: enabled: true headers: api-key: ${NEW_RELIC_LICENSE_KEY} processors: # Batch processor for efficient export batch/export: send_batch_size: 1024 timeout: 30s # Memory limiter to prevent OOM memory_limiter: limit_percentage: 80 spike_limit_percentage: 30 check_interval: 1s # Transform metric naming conventions transform/metric-naming: metric_statements: - context: metric statements: - replace_pattern(name, "_", ".") - replace_pattern(name, "\\.load\\.1", ".load_1") - replace_pattern(name, "\\.recent\\.util", ".recent_util") - replace_pattern(name, "file\\.descriptor\\.count", "file_descriptor.count") - replace_pattern(name, "\\.memory\\.pool\\.used\\.bytes$", ".memory.pool.used") - replace_pattern(name, "\\.memory\\.pool\\.max\\.bytes$", ".memory.pool.max") - replace_pattern(name, "\\.memory\\.pool\\.collection\\.used\\.bytes$", ".memory.pool.used_after_last_gc") - replace_pattern(name, "\\.non\\.preferred\\.leader", ".non_preferred_leader") - replace_pattern(name, "\\.under\\.min\\.isr", ".under_min_isr") - replace_pattern(name, "\\.under\\.replicated", ".under_replicated") - replace_pattern(name, "\\.total$", "") where name != "kafka.request.time.total" - context: datapoint statements: - set(attributes["name"], attributes["gc"]) where attributes["gc"] != nil - delete_key(attributes, "gc") where attributes["gc"] != nil - set(attributes["name"], attributes["pool"]) where attributes["pool"] != nil - delete_key(attributes, "pool") where attributes["pool"] != nil # Add cluster name to all metrics resource/cluster-name: attributes: - key: kafka.cluster.name # TODO#5: Replace with your Kafka cluster name (used to identify and filter metrics in New Relic) value: my-cluster action: upsert # Remove broker.id for cluster-level metrics transform/remove_broker_id: metric_statements: - context: datapoint statements: - delete_key(attributes, "broker.id") # Filter out scrape overhead metrics filter/scrape-overhead: metrics: exclude: match_type: regexp metric_names: - "^jmx_.*" - "^process_.*" - "^jvm_buffer_pool_.*" - "^jvm_threads_.*" - "^jvm_classes_.*" - "^jvm_memory_(heap|non_heap)_(committed|init|max|used)_bytes$" - "^jvm_compilation_.*" - "^jvm_(runtime|info).*" - "^jvm_memory_pool_(allocated_bytes_total|committed_bytes|init_bytes|collection_(committed|init|max)_bytes)$" # Include only cluster-level metrics for cluster pipeline filter/include_cluster_metrics: metrics: include: match_type: regexp metric_names: - "^kafka\\.partition\\.offline$" - "^kafka\\.(leader|unclean)\\.election\\.rate$" - "^kafka\\.partition\\.non_preferred_leader$" - "^kafka\\.broker\\.fenced\\.count$" - "^kafka\\.cluster\\.partition\\.count$" - "^kafka\\.cluster\\.topic\\.count$" # Exclude cluster-level metrics from broker pipeline filter/exclude_cluster_metrics: metrics: exclude: match_type: regexp metric_names: - "^kafka\\.partition\\.offline$" - "^kafka\\.(leader|unclean)\\.election\\.rate$" - "^kafka\\.partition\\.non_preferred_leader$" - "^kafka\\.broker\\.fenced\\.count$" - "^kafka\\.cluster\\.partition\\.count$" - "^kafka\\.cluster\\.topic\\.count$" # Remove unnecessary attributes transform/remove_attributes: metric_statements: - context: metric statements: - set(description, "") where description != "" - set(unit, "") where unit != "" - context: resource statements: - delete_key(attributes, "server.address") - delete_key(attributes, "server.port") - delete_key(attributes, "service.instance.id") - delete_key(attributes, "host.name") - delete_key(attributes, "k8s.pod.uid") - delete_key(attributes, "url.scheme") # Aggregate partition metrics to topic level metricstransform/topic-aggregation: transforms: - include: kafka.partition.replicas_in_sync action: insert new_name: kafka.partition.replicas_in_sync.total operations: - action: aggregate_labels label_set: [topic] aggregation_type: sum - include: kafka.partition.replicas action: insert new_name: kafka.partition.replicas.total operations: - action: aggregate_labels label_set: [topic] aggregation_type: sum # Filter out original partition replicas metric filter/exclude_partition_replicas_metric: metrics: exclude: match_type: strict metric_names: - kafka.partition.replicas_in_sync # Filter internal Kafka topics as a safety net (topic_match handles the receiver side) filter/internal_topics: metrics: datapoint: - 'attributes["topic"] != nil and IsMatch(attributes["topic"], "^__.*")' # Convert cumulative to delta metrics cumulativetodelta: groupbyattrs/cluster: keys: [kafka.cluster.name] metricstransform/cluster_max: transforms: - include: "kafka\\.partition\\.offline|kafka\\.leader\\.election\\.rate|kafka\\.unclean\\.election\\.rate|kafka\\.partition\\.non_preferred_leader|kafka\\.broker\\.fenced\\.count|kafka\\.cluster\\.partition\\.count|kafka\\.cluster\\.topic\\.count" match_type: regexp action: update operations: - action: aggregate_labels aggregation_type: max label_set: [] service: pipelines: # Suppress default pipelines — only custom Kafka metrics pipelines are used traces: null logs: null metrics: null # Broker-level metrics from Prometheus JMX scraping metrics/broker: receivers: - prometheus/kafka-jmx processors: - resource/cluster-name - filter/scrape-overhead - transform/metric-naming - transform/remove_attributes - filter/exclude_cluster_metrics - memory_limiter - cumulativetodelta - batch/export exporters: - otlp/backend # Cluster-level metrics from Prometheus JMX scraping metrics/cluster/prometheus: receivers: - prometheus/kafka-jmx processors: - resource/cluster-name - filter/scrape-overhead - transform/metric-naming - transform/remove_attributes - filter/include_cluster_metrics - transform/remove_broker_id - memory_limiter - cumulativetodelta - groupbyattrs/cluster - metricstransform/cluster_max - batch/export exporters: - otlp/backend # Cluster-level metrics from Kafka metrics receiver metrics/cluster/kafkametrics: receivers: - kafkametrics/cluster processors: - resource/cluster-name - filter/internal_topics - transform/remove_attributes - metricstransform/topic-aggregation - filter/exclude_partition_replicas_metric - memory_limiter - cumulativetodelta - batch/export exporters: - otlp/backend ``` **Configuration parameters** The following table describes the key configuration parameters: | Parameter | Description | | ---------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------- | | `config.receivers.kafkametrics/cluster.brokers` | Replace with your Kafka bootstrap service | | `config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].kubernetes_sd_configs[0].namespaces.names` | Replace with the namespace where your Kafka cluster is deployed | | `config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].relabel_configs[strimzi_io_name].regex` | Replace with your Strimzi Kafka cluster name followed by `-kafka` | | `config.receivers.prometheus/kafka-jmx.config.scrape_configs[0].relabel_configs[strimzi_io_cluster].regex` | Replace with your Strimzi Kafka cluster name | | `config.processors.resource/cluster-name.attributes[kafka.cluster.name].value` | Replace with your Kafka cluster name (this will be used to identify and filter your metrics in New Relic) | | `resources.limits` and `resources.requests` | Adjust according to your workload needs | **Using OpenTelemetry Collector** Use the community **OpenTelemetry Collector** for maximum flexibility and vendor-neutral deployment. Create `values.yaml` with the following content: ```yaml # Deployment mode mode: deployment replicaCount: 1 # Use contrib image for kafkametrics receiver image: repository: otel/opentelemetry-collector-contrib tag: "latest" pullPolicy: Always # Service account configuration serviceAccount: create: true name: otel-collector # RBAC for Kubernetes service discovery clusterRole: create: true rules: - apiGroups: [""] resources: ["pods", "nodes"] verbs: ["get", "list", "watch"] # Pod security context podSecurityContext: runAsNonRoot: true runAsUser: 10001 # Container security context securityContext: allowPrivilegeEscalation: false readOnlyRootFilesystem: true capabilities: drop: - ALL # Resource limits resources: requests: memory: 512Mi cpu: 250m limits: memory: 1Gi cpu: 500m # Load environment variables from secret extraEnvsFrom: - secretRef: name: newrelic-otlp-secret # OpenTelemetry Collector Configuration config: receivers: # Kafka metrics receiver for cluster-level metrics kafkametrics/cluster: brokers: # TODO#1: Replace with your Kafka bootstrap service - "my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092" collection_interval: 30s protocol_version: 2.0.0 scrapers: - brokers - topics - consumers # Exclude internal Kafka topics (prefixed with __) at the source topic_match: "^[^_].*$" metrics: kafka.topic.min_insync_replicas: enabled: true kafka.topic.replication_factor: enabled: true kafka.partition.replicas: enabled: false kafka.partition.oldest_offset: enabled: false kafka.partition.current_offset: enabled: false # Prometheus receiver for JMX metrics from Kafka brokers prometheus/kafka-jmx: config: scrape_configs: - job_name: 'kafka-jmx-metrics' scrape_interval: 30s kubernetes_sd_configs: - role: pod namespaces: names: # TODO#2: Replace with the namespace where your Kafka cluster is deployed - newrelic relabel_configs: # Filter for Kafka broker pods - source_labels: [__meta_kubernetes_pod_label_strimzi_io_name] action: keep # TODO#3: Replace with your Strimzi Kafka cluster name followed by '-kafka' regex: my-cluster-kafka - source_labels: [__meta_kubernetes_pod_label_strimzi_io_cluster] action: keep # TODO#4: Replace with your Strimzi Kafka cluster name regex: my-cluster # Extract broker ID from pod name - source_labels: [__meta_kubernetes_pod_name] target_label: broker.id regex: '.*-(\\d+)$' replacement: '$1' # Set scrape target to pod IP on port 9404 - source_labels: [__meta_kubernetes_pod_ip] target_label: __address__ replacement: '$1:9404' exporters: # New Relic OTLP exporter otlp/backend: endpoint: ${NEW_RELIC_OTLP_ENDPOINT} tls: insecure: false sending_queue: num_consumers: 12 queue_size: 5000 retry_on_failure: enabled: true headers: api-key: ${NEW_RELIC_LICENSE_KEY} processors: # Batch processor for efficient export batch/export: send_batch_size: 1024 timeout: 30s # Memory limiter to prevent OOM memory_limiter: limit_percentage: 80 spike_limit_percentage: 30 check_interval: 1s # Transform metric naming conventions transform/metric-naming: metric_statements: - context: metric statements: - replace_pattern(name, "_", ".") - replace_pattern(name, "\\.load\\.1", ".load_1") - replace_pattern(name, "\\.recent\\.util", ".recent_util") - replace_pattern(name, "file\\.descriptor\\.count", "file_descriptor.count") - replace_pattern(name, "\\.memory\\.pool\\.used\\.bytes$", ".memory.pool.used") - replace_pattern(name, "\\.memory\\.pool\\.max\\.bytes$", ".memory.pool.max") - replace_pattern(name, "\\.memory\\.pool\\.collection\\.used\\.bytes$", ".memory.pool.used_after_last_gc") - replace_pattern(name, "\\.non\\.preferred\\.leader", ".non_preferred_leader") - replace_pattern(name, "\\.under\\.min\\.isr", ".under_min_isr") - replace_pattern(name, "\\.under\\.replicated", ".under_replicated") - replace_pattern(name, "\\.total$", "") where name != "kafka.request.time.total" - context: datapoint statements: - set(attributes["name"], attributes["gc"]) where attributes["gc"] != nil - delete_key(attributes, "gc") where attributes["gc"] != nil - set(attributes["name"], attributes["pool"]) where attributes["pool"] != nil - delete_key(attributes, "pool") where attributes["pool"] != nil # Add cluster name to all metrics resource/cluster-name: attributes: - key: kafka.cluster.name # TODO#5: Replace with your Kafka cluster name (used to identify and filter metrics in New Relic) value: my-cluster action: upsert # Remove broker.id for cluster-level metrics transform/remove_broker_id: metric_statements: - context: datapoint statements: - delete_key(attributes, "broker.id") # Filter out scrape overhead metrics filter/scrape-overhead: metrics: exclude: match_type: regexp metric_names: - "^jmx_.*" - "^process_.*" - "^jvm_buffer_pool_.*" - "^jvm_threads_.*" - "^jvm_classes_.*" - "^jvm_memory_(heap|non_heap)_(committed|init|max|used)_bytes$" - "^jvm_compilation_.*" - "^jvm_(runtime|info).*" - "^jvm_memory_pool_(allocated_bytes_total|committed_bytes|init_bytes|collection_(committed|init|max)_bytes)$" # Include only cluster-level metrics for cluster pipeline filter/include_cluster_metrics: metrics: include: match_type: regexp metric_names: - "^kafka\\.partition\\.offline$" - "^kafka\\.(leader|unclean)\\.election\\.rate$" - "^kafka\\.partition\\.non_preferred_leader$" - "^kafka\\.broker\\.fenced\\.count$" - "^kafka\\.cluster\\.partition\\.count$" - "^kafka\\.cluster\\.topic\\.count$" # Exclude cluster-level metrics from broker pipeline filter/exclude_cluster_metrics: metrics: exclude: match_type: regexp metric_names: - "^kafka\\.partition\\.offline$" - "^kafka\\.(leader|unclean)\\.election\\.rate$" - "^kafka\\.partition\\.non_preferred_leader$" - "^kafka\\.broker\\.fenced\\.count$" - "^kafka\\.cluster\\.partition\\.count$" - "^kafka\\.cluster\\.topic\\.count$" # Remove unnecessary attributes transform/remove_attributes: metric_statements: - context: metric statements: - set(description, "") where description != "" - set(unit, "") where unit != "" - context: resource statements: - delete_key(attributes, "server.address") - delete_key(attributes, "server.port") - delete_key(attributes, "service.instance.id") - delete_key(attributes, "host.name") - delete_key(attributes, "k8s.pod.uid") - delete_key(attributes, "url.scheme") # Aggregate partition metrics to topic level metricstransform/topic-aggregation: transforms: - include: kafka.partition.replicas_in_sync action: insert new_name: kafka.partition.replicas_in_sync.total operations: - action: aggregate_labels label_set: [topic] aggregation_type: sum - include: kafka.partition.replicas action: insert new_name: kafka.partition.replicas.total operations: - action: aggregate_labels label_set: [topic] aggregation_type: sum # Filter out original partition replicas metric filter/exclude_partition_replicas_metric: metrics: exclude: match_type: strict metric_names: - kafka.partition.replicas_in_sync # Filter internal Kafka topics as a safety net (topic_match handles the receiver side) filter/internal_topics: metrics: datapoint: - 'attributes["topic"] != nil and IsMatch(attributes["topic"], "^__.*")' # Convert cumulative to delta metrics cumulativetodelta: groupbyattrs/cluster: keys: [kafka.cluster.name] metricstransform/cluster_max: transforms: - include: "kafka\\.partition\\.offline|kafka\\.leader\\.election\\.rate|kafka\\.unclean\\.election\\.rate|kafka\\.partition\\.non_preferred_leader|kafka\\.broker\\.fenced\\.count|kafka\\.cluster\\.partition\\.count|kafka\\.cluster\\.topic\\.count" match_type: regexp action: update operations: - action: aggregate_labels aggregation_type: max label_set: [] service: pipelines: # Override default pipelines to only use custom Kafka metrics pipelines traces: null logs: null metrics: null # Broker-level metrics from Prometheus JMX scraping metrics/broker: receivers: - prometheus/kafka-jmx processors: - resource/cluster-name - filter/scrape-overhead - transform/metric-naming - transform/remove_attributes - filter/exclude_cluster_metrics - memory_limiter - cumulativetodelta - batch/export exporters: - otlp/backend # Cluster-level metrics from Prometheus JMX scraping metrics/cluster/prometheus: receivers: - prometheus/kafka-jmx processors: - resource/cluster-name - filter/scrape-overhead - transform/metric-naming - transform/remove_attributes - filter/include_cluster_metrics - transform/remove_broker_id - memory_limiter - cumulativetodelta - groupbyattrs/cluster - metricstransform/cluster_max - batch/export exporters: - otlp/backend # Cluster-level metrics from Kafka metrics receiver metrics/cluster/kafkametrics: receivers: - kafkametrics/cluster processors: - resource/cluster-name - filter/internal_topics - transform/remove_attributes - metricstransform/topic-aggregation - filter/exclude_partition_replicas_metric - memory_limiter - cumulativetodelta - batch/export exporters: - otlp/backend ``` **Configuration parameters**: Same parameters as the NRDOT option above. See the configuration parameters table for details, including resource limits. For advanced configuration options, refer to these receiver documentation pages: - [Prometheus receiver documentation](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/receiver/prometheusreceiver) - Additional receiver configuration options - [Kafka metrics receiver documentation](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/receiver/kafkametricsreceiver) - Additional Kafka metrics configuration **Step 3. Install OpenTelemetry Collector with Helm** Add the Helm repository and install the OpenTelemetry Collector using the values.yaml file: ```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 ``` **Step 4. Verify the deployment:** ```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 ``` You should see logs indicating successful scraping from Kafka brokers on port 9404. #### Manifest install The manifest installation method provides direct control over Kubernetes resources without using Helm. **Step 1. Create New Relic credentials secret** Create a Kubernetes secret containing your New Relic license key and OTLP endpoint. Choose the endpoint for your New Relic region: **US region (default)** ````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' ``` ```` **EU region** ````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' ``` ```` **JP region** ````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.jp.nr-data.net:4317' ``` ```` > #### 💡 TIP > > For other endpoint configurations, see [Configure your OTLP endpoint](https://docs.newrelic.com/docs/opentelemetry/best-practices/opentelemetry-otlp/#configure-endpoint-port-protocol). **Step 2. Create manifest files** Create the Kubernetes manifest files for your preferred collector. Both collectors use identical configuration - only the image differs. Choose your collector option and create the three required files: **Using NRDOT Collector (recommended)** **NRDOT** is New Relic's supported distribution of the OpenTelemetry Collector, providing full New Relic support. For more information, see the [NRDOT Collector GitHub repository](https://github.com/newrelic/nrdot-collector-releases/tree/main/distributions/nrdot-collector). **Step 1. Create `collector-rbac.yaml`** - RBAC configuration for Kubernetes API access: ````yaml --- apiVersion: v1 kind: ServiceAccount metadata: name: otel-collector namespace: newrelic labels: app: otel-collector --- apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRole metadata: name: otel-collector labels: app: otel-collector rules: - apiGroups: [""] resources: ["pods", "nodes"] verbs: ["get", "list", "watch"] --- apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding metadata: name: otel-collector labels: app: otel-collector subjects: - kind: ServiceAccount name: otel-collector namespace: newrelic roleRef: kind: ClusterRole name: otel-collector apiGroup: rbac.authorization.k8s.io ``` **Step 2. Create `collector-configmap.yaml`** - OpenTelemetry Collector configuration: ```yaml --- apiVersion: v1 kind: ConfigMap metadata: name: otel-collector-config namespace: newrelic labels: app: otel-collector data: otel-collector-config.yaml: | receivers: kafkametrics/cluster: brokers: # TODO#1: Replace with your Kafka bootstrap service - "my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092" collection_interval: 30s protocol_version: 2.0.0 scrapers: - brokers - topics - consumers # Exclude internal Kafka topics (prefixed with __) at the source topic_match: "^[^_].*$" metrics: kafka.topic.min_insync_replicas: enabled: true kafka.topic.replication_factor: enabled: true kafka.partition.replicas: enabled: false kafka.partition.oldest_offset: enabled: false kafka.partition.current_offset: enabled: false prometheus/kafka-jmx: config: scrape_configs: - job_name: 'kafka-jmx-metrics' scrape_interval: 30s kubernetes_sd_configs: - role: pod namespaces: names: # TODO#2: Replace with the namespace where your Kafka cluster is deployed - kafka relabel_configs: - source_labels: [__meta_kubernetes_pod_label_strimzi_io_name] action: keep # TODO#3: Replace with your Strimzi Kafka cluster name followed by '-kafka' regex: my-cluster-kafka - source_labels: [__meta_kubernetes_pod_label_strimzi_io_cluster] action: keep # TODO#4: Replace with your Strimzi Kafka cluster name regex: my-cluster - source_labels: [__meta_kubernetes_pod_name] target_label: broker.id regex: '.*-(\\d+)$' replacement: '$1' - source_labels: [__meta_kubernetes_pod_ip] target_label: __address__ replacement: '$1:9404' exporters: otlp/backend: endpoint: ${NEW_RELIC_OTLP_ENDPOINT} tls: insecure: false sending_queue: num_consumers: 12 queue_size: 5000 retry_on_failure: enabled: true headers: api-key: ${NEW_RELIC_LICENSE_KEY} processors: batch/export: send_batch_size: 1024 timeout: 30s memory_limiter: limit_percentage: 80 spike_limit_percentage: 30 check_interval: 1s transform/metric-naming: metric_statements: - context: metric statements: - replace_pattern(name, "_", ".") - replace_pattern(name, "\\.load\\.1", ".load_1") - replace_pattern(name, "\\.recent\\.util", ".recent_util") - replace_pattern(name, "file\\.descriptor\\.count", "file_descriptor.count") - replace_pattern(name, "\\.memory\\.pool\\.used\\.bytes$", ".memory.pool.used") - replace_pattern(name, "\\.memory\\.pool\\.max\\.bytes$", ".memory.pool.max") - replace_pattern(name, "\\.memory\\.pool\\.collection\\.used\\.bytes$", ".memory.pool.used_after_last_gc") - replace_pattern(name, "\\.non\\.preferred\\.leader", ".non_preferred_leader") - replace_pattern(name, "\\.under\\.min\\.isr", ".under_min_isr") - replace_pattern(name, "\\.under\\.replicated", ".under_replicated") - replace_pattern(name, "\\.total$", "") where name != "kafka.request.time.total" - context: datapoint statements: - set(attributes["name"], attributes["gc"]) where attributes["gc"] != nil - delete_key(attributes, "gc") where attributes["gc"] != nil - set(attributes["name"], attributes["pool"]) where attributes["pool"] != nil - delete_key(attributes, "pool") where attributes["pool"] != nil resource/cluster-name: attributes: - key: kafka.cluster.name # TODO#5: Replace with your Kafka cluster name (used to identify and filter metrics in New Relic) value: my-cluster action: upsert transform/remove_broker_id: metric_statements: - context: datapoint statements: - delete_key(attributes, "broker.id") filter/scrape-overhead: metrics: exclude: match_type: regexp metric_names: - "^jmx_.*" - "^process_.*" - "^jvm_buffer_pool_.*" - "^jvm_threads_.*" - "^jvm_classes_.*" - "^jvm_memory_(heap|non_heap)_(committed|init|max|used)_bytes$" - "^jvm_compilation_.*" - "^jvm_(runtime|info).*" - "^jvm_memory_pool_(allocated_bytes_total|committed_bytes|init_bytes|collection_(committed|init|max)_bytes)$" filter/include_cluster_metrics: metrics: include: match_type: regexp metric_names: - "^kafka\\.partition\\.offline$" - "^kafka\\.(leader|unclean)\\.election\\.rate$" - "^kafka\\.partition\\.non_preferred_leader$" - "^kafka\\.broker\\.fenced\\.count$" - "^kafka\\.cluster\\.partition\\.count$" - "^kafka\\.cluster\\.topic\\.count$" filter/exclude_cluster_metrics: metrics: exclude: match_type: regexp metric_names: - "^kafka\\.partition\\.offline$" - "^kafka\\.(leader|unclean)\\.election\\.rate$" - "^kafka\\.partition\\.non_preferred_leader$" - "^kafka\\.broker\\.fenced\\.count$" - "^kafka\\.cluster\\.partition\\.count$" - "^kafka\\.cluster\\.topic\\.count$" transform/remove_attributes: metric_statements: - context: metric statements: - set(description, "") where description != "" - set(unit, "") where unit != "" - context: resource statements: - delete_key(attributes, "server.address") - delete_key(attributes, "server.port") - delete_key(attributes, "service.instance.id") - delete_key(attributes, "host.name") - delete_key(attributes, "k8s.pod.uid") - delete_key(attributes, "url.scheme") metricstransform/topic-aggregation: transforms: - include: kafka.partition.replicas_in_sync action: insert new_name: kafka.partition.replicas_in_sync.total operations: - action: aggregate_labels label_set: [topic] aggregation_type: sum - include: kafka.partition.replicas action: insert new_name: kafka.partition.replicas.total operations: - action: aggregate_labels label_set: [topic] aggregation_type: sum filter/exclude_partition_replicas_metric: metrics: exclude: match_type: strict metric_names: - kafka.partition.replicas_in_sync # Filter internal Kafka topics as a safety net (topic_match handles the receiver side) filter/internal_topics: metrics: datapoint: - 'attributes["topic"] != nil and IsMatch(attributes["topic"], "^__.*")' cumulativetodelta: groupbyattrs/cluster: keys: [kafka.cluster.name] metricstransform/cluster_max: transforms: - include: "kafka\\.partition\\.offline|kafka\\.leader\\.election\\.rate|kafka\\.unclean\\.election\\.rate|kafka\\.partition\\.non_preferred_leader|kafka\\.broker\\.fenced\\.count|kafka\\.cluster\\.partition\\.count|kafka\\.cluster\\.topic\\.count" match_type: regexp action: update operations: - action: aggregate_labels aggregation_type: max label_set: [] service: pipelines: metrics/broker: receivers: [prometheus/kafka-jmx] processors: - resource/cluster-name - filter/scrape-overhead - transform/metric-naming - transform/remove_attributes - filter/exclude_cluster_metrics - memory_limiter - cumulativetodelta - batch/export exporters: [otlp/backend] metrics/cluster/prometheus: receivers: [prometheus/kafka-jmx] processors: - resource/cluster-name - filter/scrape-overhead - transform/metric-naming - transform/remove_attributes - filter/include_cluster_metrics - transform/remove_broker_id - memory_limiter - cumulativetodelta - groupbyattrs/cluster - metricstransform/cluster_max - batch/export exporters: [otlp/backend] metrics/cluster/kafkametrics: receivers: [kafkametrics/cluster] processors: - resource/cluster-name - filter/internal_topics - transform/remove_attributes - metricstransform/topic-aggregation - filter/exclude_partition_replicas_metric - memory_limiter - cumulativetodelta - batch/export exporters: [otlp/backend] ```` **Step 3. Create `collector-deployment.yaml`** - OpenTelemetry Collector deployment: ```yaml --- apiVersion: apps/v1 kind: Deployment metadata: name: otel-collector namespace: newrelic labels: app: otel-collector spec: replicas: 1 selector: matchLabels: app: otel-collector template: metadata: labels: app: otel-collector spec: serviceAccountName: otel-collector containers: - name: otel-collector image: newrelic/nrdot-collector:latest command: - "/nrdot-collector" - "--config=/conf/otel-collector-config.yaml" env: - name: NEW_RELIC_LICENSE_KEY valueFrom: secretKeyRef: name: newrelic-otlp-secret key: NEW_RELIC_LICENSE_KEY - name: NEW_RELIC_OTLP_ENDPOINT valueFrom: secretKeyRef: name: newrelic-otlp-secret key: NEW_RELIC_OTLP_ENDPOINT - name: GOGC value: "80" resources: limits: cpu: "1000m" memory: "1Gi" requests: cpu: "200m" memory: "512Mi" volumeMounts: - name: config mountPath: /conf volumes: - name: config configMap: name: otel-collector-config items: - key: otel-collector-config.yaml path: otel-collector-config.yaml ``` **Configuration parameters** The following table describes the key configuration parameters: | Parameter | Description | | --------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------- | | `receivers.kafkametrics/cluster.brokers` | Replace with your Kafka bootstrap service | | `receivers.prometheus/kafka-jmx.config.scrape_configs[0].kubernetes_sd_configs[0].namespaces.names` | Replace with the namespace where your Kafka cluster is deployed | | `receivers.prometheus/kafka-jmx.config.scrape_configs[0].relabel_configs[strimzi_io_name].regex` | Replace with your Strimzi Kafka cluster name followed by `-kafka` | | `receivers.prometheus/kafka-jmx.config.scrape_configs[0].relabel_configs[strimzi_io_cluster].regex` | Replace with your Strimzi Kafka cluster name | | `processors.resource/cluster-name.attributes[kafka.cluster.name].value` | Replace with your Kafka cluster name (this will be used to identify and filter your metrics in New Relic) | **Using OpenTelemetry Collector** Use the community **OpenTelemetry Collector** for vendor-neutral deployment. **Step 1. Create `collector-rbac.yaml`** - Same as NRDOT option above (RBAC configuration is identical) **Step 2. Create `collector-configmap.yaml`** - Same as NRDOT option above (configuration is identical) **Step 3. Create `collector-deployment.yaml`** - OpenTelemetry Collector deployment (only the image differs): ```yaml --- apiVersion: apps/v1 kind: Deployment metadata: name: otel-collector namespace: newrelic labels: app: otel-collector spec: replicas: 1 selector: matchLabels: app: otel-collector template: metadata: labels: app: otel-collector spec: serviceAccountName: otel-collector containers: - name: otel-collector image: otel/opentelemetry-collector-contrib:latest command: - "/otelcol-contrib" - "--config=/conf/otel-collector-config.yaml" env: - name: NEW_RELIC_LICENSE_KEY valueFrom: secretKeyRef: name: newrelic-otlp-secret key: NEW_RELIC_LICENSE_KEY - name: NEW_RELIC_OTLP_ENDPOINT valueFrom: secretKeyRef: name: newrelic-otlp-secret key: NEW_RELIC_OTLP_ENDPOINT - name: GOGC value: "80" resources: limits: cpu: "1000m" memory: "1Gi" requests: cpu: "200m" memory: "512Mi" volumeMounts: - name: config mountPath: /conf volumes: - name: config configMap: name: otel-collector-config items: - key: otel-collector-config.yaml path: otel-collector-config.yaml ``` **Configuration parameters**: Same parameters as the NRDOT option above. See the configuration parameters table for details. For advanced configuration options, refer to these receiver documentation pages: - [Prometheus receiver documentation](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/receiver/prometheusreceiver) - Additional receiver configuration options - [Kafka metrics receiver documentation](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/receiver/kafkametricsreceiver) - Additional Kafka metrics configuration **Step 3. Deploy the manifests** Apply the Kubernetes manifests to deploy the OpenTelemetry Collector: ```bash # Create namespace if it doesn't exist kubectl create namespace newrelic --dry-run=client -o yaml | kubectl apply -f - # Apply RBAC configuration kubectl apply -f collector-rbac.yaml # Apply ConfigMap kubectl apply -f collector-configmap.yaml # Apply Deployment kubectl apply -f collector-deployment.yaml ``` **Step 4. Verify the deployment:** ```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 ``` You should see logs indicating successful scraping from Kafka brokers on port 9404. ### (Optional) Instrument producer or consumer applications [#instrument-apps] > #### ⚠️ IMPORTANT > > **Language support**: Java applications support out-of-the-box Kafka client instrumentation using the OpenTelemetry Java agent. To collect application-level telemetry from your Kafka producer and consumer applications, use the OpenTelemetry Java agent. #### Instrument your Kafka application Use an init container to download the OpenTelemetry Java agent at runtime: ```yaml apiVersion: apps/v1 kind: Deployment metadata: name: kafka-producer-app spec: template: spec: initContainers: - name: download-java-agent image: busybox:latest command: - sh - -c - | wget -O /otel-auto-instrumentation/opentelemetry-javaagent.jar \ https://github.com/open-telemetry/opentelemetry-java-instrumentation/releases/latest/download/opentelemetry-javaagent.jar volumeMounts: - name: otel-auto-instrumentation mountPath: /otel-auto-instrumentation containers: - name: app image: your-kafka-app:latest env: - name: JAVA_TOOL_OPTIONS value: >- -javaagent:/otel-auto-instrumentation/opentelemetry-javaagent.jar -Dotel.service.name=order-process-service -Dotel.resource.attributes=kafka.cluster.name=my-cluster -Dotel.exporter.otlp.endpoint=http://localhost:4317 -Dotel.exporter.otlp.protocol=grpc -Dotel.metrics.exporter=otlp -Dotel.traces.exporter=otlp -Dotel.logs.exporter=otlp -Dotel.instrumentation.kafka.experimental-span-attributes=true -Dotel.instrumentation.messaging.experimental.receive-telemetry.enabled=true -Dotel.instrumentation.kafka.producer-propagation.enabled=true -Dotel.instrumentation.kafka.enabled=true volumeMounts: - name: otel-auto-instrumentation mountPath: /otel-auto-instrumentation volumes: - name: otel-auto-instrumentation emptyDir: {} ``` **Configuration parameters** The following table describes the key configuration parameters: | Parameter | Description | | -------------------- | ------------------------------------------------------------------------------------------------------------------------------ | | `service.name` | Replace `order-process-service` with a unique name for your producer or consumer application | | `kafka.cluster.name` | Replace `my-cluster` with the same cluster name used in your collector configuration | | `otlp.endpoint` | The endpoint `http://localhost:4317` assumes the collector is running as a sidecar in the same pod or accessible via localhost | > #### 💡 TIP > > The configuration above sends telemetry to an OpenTelemetry Collector. If you need to send telemetry to the collector, deploy it as described in [Step 3](#deploy-opentelemetry-collector) with this configuration: > > **Collector configuration for application telemetry** > > Add the OTLP receiver and pipelines to your collector configuration to receive telemetry from instrumented applications: > > **Step 1. Add to receivers section:** > > ````yaml > receivers: > otlp: > protocols: > grpc: > endpoint: "0.0.0.0:4317" > > # ... existing receivers (prometheus/kafka-jmx, kafkametrics/cluster) > ``` > > **Step 2. Add to exporters section:** > ```yaml > exporters: > otlp/backend: > endpoint: ${NEW_RELIC_OTLP_ENDPOINT} > headers: > api-key: ${NEW_RELIC_LICENSE_KEY} > > # ... existing exporters > ``` > > **Step 3. Add to service.pipelines section:** > ```yaml > service: > pipelines: > traces: > receivers: [otlp] > exporters: [otlp/backend] > > metrics: > receivers: [otlp] > exporters: [otlp/backend] > > logs: > receivers: [otlp] > exporters: [otlp/backend] > > # ... existing pipelines (metrics/broker, metrics/cluster/prometheus, metrics/cluster/kafkametrics) > ``` > > This allows the collector to receive application traces, metrics, and logs from your instrumented Kafka clients and forward them to New Relic alongside broker metrics. > > ```` The Java agent provides [out-of-the-box Kafka instrumentation](https://opentelemetry.io/docs/zero-code/java/spring-boot-starter/out-of-the-box-instrumentation/) with zero code changes, capturing: - Request latencies - Throughput metrics - Error rates - Distributed traces For advanced configuration, see the [Kafka instrumentation documentation](https://github.com/open-telemetry/opentelemetry-java-instrumentation/tree/main/instrumentation/kafka). ### (Optional) Forward Kafka broker logs [#forward-broker-logs] To collect Kafka broker logs and send them to New Relic, add a filelog receiver to your collector configuration. **Configure log collection** Add the following to your `values.yaml` collector configuration: **Step 1. Add to receivers section:** ````yaml 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} ``` **Step 2. Add logs pipeline to service section:** ```yaml service: pipelines: # ... existing pipelines ... logs/broker: receivers: [filelog/kafka_broker_0] processors: [memory_limiter, batch/export] exporters: [otlp/backend] ``` **Configuration parameters** The following table describes the key configuration parameters: | Parameter | Description | |-----------|-------------| | `filelog/kafka_broker_0.include` | Update `/var/log/kafka/server.log` to your actual Kafka log path inside the broker pod. For Strimzi, logs are typically written to stdout — use the [Strimzi logging configuration](https://strimzi.io/docs/operators/latest/configuring.html#property-kafka-logging-reference) to redirect them to a file, or use a sidecar log collector instead. | | `filelog/kafka_broker_0.resource.broker.id` | The `broker.id` resource attribute correlates logs with specific broker metrics and entities | | Multiple broker receivers | For multiple brokers, create separate `filelog` receivers (e.g., `filelog/kafka_broker_1`, `filelog/kafka_broker_2`) with their respective broker IDs | | `filelog/kafka_broker_0.multiline.line_start_pattern` | The `multiline` pattern assumes logs start with `[` — adjust if your log format differs | | Log volume | Consider log volume and collection costs before enabling log forwarding | | Reference | For complete configuration options, see the [filelog receiver documentation](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/receiver/filelogreceiver) | **Step 3. Upgrade the Helm release:** ```bash helm upgrade kafka-otel-collector open-telemetry/opentelemetry-collector \ --namespace newrelic \ --values values.yaml ``` ```` **Find your logs in New Relic** Your Kafka broker logs will appear in two places: - **Broker entities**: Navigate to the Kafka broker entity in New Relic to see logs correlated with that specific broker - **Logs UI**: Query all Kafka logs using the [Logs UI](https://docs.newrelic.com/docs/logs/ui-data/use-logs-ui/) with filters like `kafka.cluster.name = 'my-cluster'` You can also query your logs with NRQL: ```sql FROM Log SELECT * WHERE kafka.cluster.name = 'my-kafka-cluster' ``` ## Find your data [#find-data] After a few minutes, your Kafka data should appear in New Relic. See [Find your data](https://docs.newrelic.com/docs/opentelemetry/integrations/kafka/find-and-query-data) for detailed instructions on exploring your Kafka data across different views in the New Relic UI. The following table summarizes where each signal type is stored. Replace `my-kafka-cluster` with your `KAFKA_CLUSTER_NAME` value in all queries below: | Signal | Event type | What's included | | ------- | ---------- | --------------------------------------------------------------------------------------------------------------------------------- | | Metrics | `Metric` | Broker, topic, partition, consumer group, and JVM metrics | | Logs | `Log` | Logs from producer and consumer applications (via OTel Java agent) and broker logs collected via the optional log forwarding step | | Traces | `Span` | Producer and consumer spans, including per-message `publish` and `receive` operations across topics | ### Metrics Broker, topic, partition, consumer group, and JVM metrics are stored in the `Metric` event type. Replace `my-kafka-cluster` with your `KAFKA_CLUSTER_NAME` value: ```sql FROM Metric SELECT * WHERE kafka.cluster.name = 'my-kafka-cluster' SINCE 30 minutes ago ``` ### Logs Logs from producer and consumer applications instrumented with the OpenTelemetry Java agent, and broker logs collected via the optional log forwarding step, are stored in the `Log` event type: ```sql FROM Log SELECT * WHERE kafka.cluster.name = 'my-kafka-cluster' SINCE 30 minutes ago ``` ### Traces If you deploy producer or consumer applications instrumented with the OpenTelemetry Java agent, producer and consumer spans are stored in the `Span` event type: ```sql FROM Span SELECT * WHERE kafka.cluster.name = 'my-kafka-cluster' SINCE 30 minutes ago ``` ## Example [#example] A complete working example with Strimzi Kafka custom resources, JMX Exporter configuration, OTel Collector setup, and sample producer/consumer applications is available in the [New Relic OpenTelemetry Examples repository](https://github.com/newrelic/newrelic-opentelemetry-examples/tree/main/other-examples/collector/kafka/k8s-strimzi). ## Troubleshooting [#troubleshooting] **Enable debug logging** **1. Enable collector debug logs**: Add detailed logging to troubleshoot configuration issues For Helm deployments, update your `values.yaml`: ````yaml config: service: telemetry: logs: level: "debug" # Enable detailed collector internal logs ``` For manifest deployments, edit your collector ConfigMap: ```bash kubectl edit configmap -n newrelic otel-collector-config ``` Add the telemetry section under `service:`: ```yaml service: telemetry: logs: level: "debug" pipelines: # ... existing pipelines ... ``` **2. Add debug exporter**: View metrics in collector logs before sending to New Relic Add to your configuration: ```yaml exporters: debug: verbosity: detailed otlp/backend: endpoint: ${NEW_RELIC_OTLP_ENDPOINT} headers: api-key: ${NEW_RELIC_LICENSE_KEY} service: pipelines: metrics/broker: receivers: [prometheus/kafka-jmx] processors: [resource/cluster-name, filter/scrape-overhead, transform/metric-naming, transform/remove_attributes, filter/exclude_cluster_metrics, memory_limiter, cumulativetodelta, batch/export] exporters: [debug, otlp/backend] # Add debug exporter ``` Then restart the collector and check logs: ```bash # Restart collector kubectl rollout restart deployment -n newrelic otel-collector # View logs with metric output kubectl logs -n newrelic -l app=otel-collector -f ``` **Important**: Remove the debug exporter in production to avoid log overflow. ```` **Collector pod not starting** **1. Check pod status and events**: ````bash # Check pod status # For Helm: kubectl get pods -n newrelic -l app.kubernetes.io/name=opentelemetry-collector # For Manifest: kubectl get pods -n newrelic -l app=otel-collector # View detailed pod description # For Helm: kubectl describe pod -n newrelic -l app.kubernetes.io/name=opentelemetry-collector # For Manifest: kubectl describe pod -n newrelic -l app=otel-collector # Check recent logs # For Helm: kubectl logs -n newrelic -l app.kubernetes.io/name=opentelemetry-collector --previous --tail=50 # For Manifest: kubectl logs -n newrelic -l app=otel-collector --previous --tail=50 ``` **Common issues and solutions**: **2. Invalid configuration**: Validate the ConfigMap YAML syntax ```bash # For manifest deployments, check ConfigMap kubectl get configmap -n newrelic otel-collector-config -o yaml # Validate YAML syntax kubectl get configmap -n newrelic otel-collector-config -o yaml | kubectl apply --dry-run=client -f - # For Helm deployments, check the values helm get values -n newrelic ``` **3. RBAC permissions**: Verify the ServiceAccount has proper ClusterRole bindings ```bash # Check ServiceAccount kubectl get serviceaccount -n newrelic otel-collector # Check ClusterRole and ClusterRoleBinding kubectl get clusterrole otel-collector -o yaml kubectl get clusterrolebinding otel-collector -o yaml ``` **4. Resource constraints**: Check if pod was OOMKilled or is hitting resource limits ```bash # Check resource usage # For Helm: kubectl top pods -n newrelic -l app.kubernetes.io/name=opentelemetry-collector # For Manifest: kubectl top pods -n newrelic -l app=otel-collector # Check for resource limits # For Helm: kubectl describe pod -n newrelic -l app.kubernetes.io/name=opentelemetry-collector | grep -A 5 "Limits\|Requests" # For Manifest: kubectl describe pod -n newrelic -l app=otel-collector | grep -A 5 "Limits\|Requests" ``` **5. Secret not found**: Verify New Relic credentials secret exists ```bash # Check if secret exists kubectl get secret -n newrelic newrelic-otlp-secret # Verify secret has required keys kubectl get secret -n newrelic newrelic-otlp-secret -o jsonpath='{.data}' | jq 'keys' ``` ```` **No Kafka metrics collected** **1. Verify JMX Exporter is enabled**: Check your Strimzi Kafka resource has JMX Exporter configured ````bash # Check Kafka resource configuration for JMX Exporter kubectl get kafka -n kafka -o yaml | grep -A 10 jmxPrometheusExporter # Should show something like: # jmxPrometheusExporter: # lowercaseOutputName: true ``` **2. Check JMX Exporter port**: Verify the exporter is listening on port 9404 ```bash # Check if port 9404 is exposed on Kafka pods kubectl get pods -n kafka -l strimzi.io/name=-kafka -o yaml | grep -A 3 "containerPort: 9404" # Test connectivity from collector pod # For Helm: kubectl exec -it -n newrelic deployment/kafka-monitoring-opentelemetry-collector -- sh -c "curl -s http://:9404/metrics" | head -10 # For Manifest: kubectl exec -it -n newrelic deployment/otel-collector -- sh -c "curl -s http://:9404/metrics" | head -10 ``` **3. Verify Prometheus receiver can scrape metrics**: ```bash # Check collector logs for Prometheus scraping # For Helm: kubectl logs -n newrelic -l app.kubernetes.io/name=opentelemetry-collector | grep -i "prometheus\|scrape" # For Manifest: kubectl logs -n newrelic -l app=otel-collector | grep -i "prometheus\|scrape" # Look for successful scrape messages or errors # For Helm: kubectl logs -n newrelic -l app.kubernetes.io/name=opentelemetry-collector | grep "prometheus/kafka-jmx" # For Manifest: kubectl logs -n newrelic -l app=otel-collector | grep "prometheus/kafka-jmx" ``` **4. Check Kubernetes service discovery**: Ensure pod labels match relabel_configs ```bash # Verify Kafka pod labels kubectl get pods -n kafka -l strimzi.io/name=-kafka --show-labels # Should include labels like: # strimzi.io/cluster= # strimzi.io/name=-kafka ``` **5. Test manual scrape**: Verify metrics are available ```bash # Get Kafka broker pod IP kubectl get pods -n kafka -o wide # Curl metrics endpoint # For Helm: kubectl exec -it -n newrelic deployment/kafka-monitoring-opentelemetry-collector -- curl http://:9404/metrics # For Manifest: kubectl exec -it -n newrelic deployment/otel-collector -- curl http://:9404/metrics ``` **6. Check for kafkametrics receiver errors**: ```bash # Look for kafkametrics connection issues # For Helm: kubectl logs -n newrelic -l app.kubernetes.io/name=opentelemetry-collector | grep "kafkametrics" # For Manifest: kubectl logs -n newrelic -l app=otel-collector | grep "kafkametrics" # Common errors: # - Connection refused: Check broker address is correct # - Timeout: Check network policies allow access # - Authentication failed: Remove TLS configuration if using plaintext ``` ```` **High memory usage** **1. Monitor resource usage**: ````bash # Check current memory usage # For Helm: kubectl top pods -n newrelic -l app.kubernetes.io/name=opentelemetry-collector # For Manifest: kubectl top pods -n newrelic -l app=otel-collector # Watch memory usage over time # For Helm: watch kubectl top pods -n newrelic -l app.kubernetes.io/name=opentelemetry-collector # For Manifest: watch kubectl top pods -n newrelic -l app=otel-collector ``` **2. Reduce monitored topics**: Limit collection to essential topics only ```yaml # In your values.yaml (Helm) or ConfigMap (manifest), add topic filtering: receivers: kafkametrics/cluster: brokers: - "my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092" collection_interval: 30s scrapers: - brokers - topics # Consider removing if not needed - consumers # Consider removing if not needed topic_match: "^(important-topic-1|important-topic-2)$" # Filter specific topics ``` **3. Reduce collection frequency**: Increase intervals to collect less often ```yaml receivers: kafkametrics/cluster: collection_interval: 60s # Increase from 30s to 60s prometheus/kafka-jmx: config: scrape_configs: - job_name: 'kafka-jmx-metrics' scrape_interval: 60s # Increase from 30s to 60s ``` **4. Optimize batch processing**: Adjust batch processor settings ```yaml processors: batch/export: timeout: 60s # Increase from 30s send_batch_size: 512 # Reduce from 1024 ``` **5. Adjust memory limiter**: Tune the memory limiter processor ```yaml processors: memory_limiter: check_interval: 1s limit_percentage: 75 # Reduce from 80 spike_limit_percentage: 20 # Reduce from 30 ``` **6. Update resource limits**: For Helm deployments, update values.yaml ```yaml resources: requests: memory: 512Mi cpu: 250m limits: memory: 1Gi # Adjust as needed cpu: 500m ``` For manifest deployments, update the deployment directly: ```bash kubectl patch deployment -n newrelic otel-collector --patch ' spec: template: spec: containers: - name: otel-collector resources: limits: memory: "1Gi" requests: memory: "512Mi" ' ``` **7. Restart collector after changes**: ```bash # For Helm: kubectl rollout restart deployment -n newrelic kafka-monitoring-opentelemetry-collector # For Manifest: kubectl rollout restart deployment -n newrelic otel-collector ``` ```` **Prometheus scrape errors** **1. Check network connectivity**: Verify collector can reach Kafka broker pods ````bash # Get Kafka broker pod IPs kubectl get pods -n kafka -l strimzi.io/kind=Kafka -o wide # Test connectivity from collector pod # For Helm: kubectl exec -it -n newrelic deployment/kafka-monitoring-opentelemetry-collector -- curl -m 5 http://:9404/metrics | head -20 # For Manifest: kubectl exec -it -n newrelic deployment/otel-collector -- curl -m 5 http://:9404/metrics | head -20 ``` **2. Verify JMX Exporter is running**: Check that Kafka brokers have JMX Exporter enabled ```bash # Check Kafka pod for JMX Exporter container kubectl get pods -n kafka -o yaml | grep -A 5 "jmx-exporter" # Check if port 9404 is listening kubectl exec -n kafka -- netstat -tlnp | grep :9404 # Or test from within the pod kubectl exec -n kafka -- curl -s localhost:9404/metrics | head ``` **3. Check relabel_configs match pod labels**: Ensure Prometheus receiver can discover Kafka pods ```bash # Verify pod labels match the relabel_configs in your configuration kubectl get pods -n kafka -l strimzi.io/kind=Kafka --show-labels # Should show labels like: # strimzi.io/cluster=my-cluster # strimzi.io/name=my-cluster-kafka ``` **4. Verify namespace configuration**: Check Prometheus receiver is looking in the correct namespace ```yaml # In your configuration, verify namespace matches where Kafka is deployed receivers: prometheus/kafka-jmx: config: scrape_configs: - job_name: 'kafka-jmx-metrics' kubernetes_sd_configs: - role: pod namespaces: names: - kafka # Must match your Kafka namespace ``` **5. Check RBAC permissions**: Verify ClusterRole allows pod discovery ```bash # Check ClusterRole has pod list/watch permissions kubectl get clusterrole otel-collector -o yaml | grep -A 3 "pods" # Should include: # - apiGroups: [""] # resources: ["pods", "nodes"] # verbs: ["get", "list", "watch"] ``` **6. Increase scrape timeout**: If metrics endpoint is slow to respond ```yaml receivers: prometheus/kafka-jmx: config: scrape_configs: - job_name: 'kafka-jmx-metrics' scrape_interval: 30s scrape_timeout: 20s # Increase from default 10s ``` **7. Check collector logs for scrape details**: ```bash # View Prometheus receiver logs # For Helm: kubectl logs -n newrelic -l app.kubernetes.io/name=opentelemetry-collector | grep "prometheus/kafka-jmx" # For Manifest: kubectl logs -n newrelic -l app=otel-collector | grep "prometheus/kafka-jmx" # Look for discovered targets # For Helm: kubectl logs -n newrelic -l app.kubernetes.io/name=opentelemetry-collector | grep -i "target\|scrape" # For Manifest: kubectl logs -n newrelic -l app=otel-collector | grep -i "target\|scrape" ``` ```` **Some metrics are missing** **1. Check filter processors**: Verify you're not inadvertently filtering out metrics ````yaml # Review your filter configurations: processors: filter/scrape-overhead: metrics: exclude: match_type: regexp metric_names: - "^jmx_.*" # These are excluded - "^process_.*" - "^jvm_.*" ``` **2. Kafka JMX metric names are transformed**: Underscores are replaced with dots to follow OpenTelemetry standards The configuration automatically transforms metric names from the JMX Exporter (defined in Step 1) to align with OpenTelemetry semantic conventions. For example: - `kafka_topic_io` becomes `kafka.topic.io` - `kafka_broker_leader_count` becomes `kafka.broker.leader.count` When searching for metrics in New Relic, use the transformed names with dots instead of underscores: ```yaml # This transformation is applied automatically: processors: transform/metric-naming: metric_statements: - context: metric statements: - replace_pattern(name, "_", ".") ``` **3. Enable detailed logging**: See exactly what metrics are being processed ```yaml exporters: debug: verbosity: detailed service: pipelines: metrics/broker: exporters: [debug, otlp/backend] ``` **4. Query New Relic for metric names**: Check what metrics are actually being received ```sql FROM Metric SELECT uniques(metricName) WHERE kafka.cluster.name = 'my-cluster' SINCE 1 hour ago ``` ```` **Reduce data ingest** The configuration includes an **Additional metrics** section for comprehensive monitoring. Removing it does not affect core New Relic UI functionality - broker health, consumer lag, cluster overview, and JVM dashboards all continue to work. **1. Remove the Additional metrics section from the JMX ConfigMap** In your `kafka-jmx-config.yaml` ConfigMap, delete everything below this comment (through the end of the `rules:` list): ````yaml # Additional metrics — remove this section to reduce data ingest # Request latency: total count, 50th percentile, and average (99p kept above) - pattern: 'kafka.network-kafka ``` **2. Disable consumer offset metrics in the kafkametrics receiver** In your collector configuration, add to the `kafkametrics/cluster` receiver `metrics` section: ```yaml receivers: kafkametrics/cluster: # ...existing config... metrics: kafka.consumer_group.offset: enabled: false kafka.consumer_group.offset_sum: enabled: false ``` Consumer lag metrics (`kafka.consumer_group.lag`, `kafka.consumer_group.lag_sum`) remain enabled. These are what the New Relic Kafka UI uses for consumer monitoring views. ```` ## Next steps [#next-steps] - [Explore Kafka metrics](https://docs.newrelic.com/docs/opentelemetry/integrations/kafka/metrics-reference) - View the complete metrics reference - [Create custom dashboards](https://docs.newrelic.com/docs/query-your-data/explore-query-data/dashboards/introduction-dashboards) - Build visualizations for your Kafka data - [Set up alerts](https://docs.newrelic.com/docs/opentelemetry/integrations/kafka/find-and-query-data#alerts) - Monitor critical metrics like consumer lag and under-replicated partitions