Kafka メトリクス

概要

OOTB Kafka ダッシュボードの OpenTelemetry Kafka メトリクス

Kafka メトリクスレシーバー、JMX Receiver/ JMX Metrics Gatherer を使用すると、Kafka メトリクスを収集し、すぐに使える Kafka Dashboard]7、“Kafka, Zookeeper and Kafka Consumer Overview” にアクセスできます。

注: JMX Receiver および JMX Metrics Gatherer が代替として検討されます。これらは同じメトリクスのセットを収集します (JMX Receiver は JMX Metrics Gatherer を起動します)。

Kafka メトリクスレシーバー

    receivers:
  kafkametrics:
    brokers: "${env:KAFKA_BROKER_ADDRESS}"
    protocol_version: 2.0.0
    scrapers:
      - brokers
      - topics
      - consumers

    Kafka メトリクスレシーバーを、deployment モードで単一レプリカとして実行されているコレクターで使用します。これにより、同じメトリクスが複数回収集されるのを防ぎます。コレクターは、OTLP HTTP を介してメトリクスを Datadog に直接エクスポートするか、別のコレクターインスタンスに転送できます。

    values.yaml に以下の行を追加します。

    mode: deployment

    Collector 構成に以下を追加します。

    receivers:
  kafkametrics:
    brokers: ${env:KAFKA_BROKER_ADDRESS}
    protocol_version: 2.0.0
    scrapers:
      - brokers
      - topics
      - consumers

    JMX レシーバー

      JMX Receiver には以下の要件があります。

      • コレクターを実行しているホストで JRE が使用可能。
      • コレクターを実行しているホストで JMX Metric Gatherer JAR が使用可能。JMX Metric Gatherer JAR の最新リリースは、opentelemetry-java-contrib リリースページ からダウンロードできます。

      Collector 構成に以下を追加します。

      receivers:
  jmx:
    jar_path: /path/to/opentelemetry-jmx-metrics.jar
    endpoint: ${env:KAFKA_BROKER_JMX_ADDRESS}
    target_system: kafka,jvm
  jmx/consumer:
    jar_path: /path/to/opentelemetry-jmx-metrics.jar
    endpoint: ${env:KAFKA_CONSUMER_JMX_ADDRESS}
    target_system: kafka-consumer
  jmx/producer:
    jar_path: /path/to/opentelemetry-jmx-metrics.jar
    endpoint: ${env:KAFKA_PRODUCER_JMX_ADDRESS}
    target_system: kafka-producer

      JMX Receiver を、deployment モードで単一レプリカとして実行されているコレクターで使用します。これにより、同じメトリクスが複数回収集されるのを防ぎます。コレクターは、OTLP HTTP を介してメトリクスを Datadog に直接エクスポートするか、別のコレクターインスタンスに転送できます。

      JMX Receiver には以下の要件があります。

      • コレクターを実行しているホストで JRE が使用可能。
      • コレクターを実行しているホストで JMX Metric Gatherer JAR が使用可能。JMX Metric Gatherer JAR の最新リリースは こちら からダウンロードできます。

      OTel コレクターのデフォルトイメージは上記の要件を満たしていないため、カスタムイメージをビルドする必要があります。コレクターのバイナリ、JRE、JMX Metrics Gatherer Jar を含むイメージの例については、以下の Dockerfile を参照してください。

      Dockerfile:

      FROM alpine:latest as prep

# OpenTelemetry Collector Binary
ARG OTEL_VERSION=0.92.0
ARG TARGETARCH=linux_amd64
ADD "https://github.com/open-telemetry/opentelemetry-collector-releases/releases/download/v${OTEL_VERSION}/otelcol-contrib_${OTEL_VERSION}_${TARGETARCH}.tar.gz" /otelcontribcol
RUN tar -zxvf /otelcontribcol

# JMX Metrics Gatherer Jar
ARG JMX_GATHERER_JAR_VERSION=1.27.0
ADD https://github.com/open-telemetry/opentelemetry-java-contrib/releases/download/v${JMX_GATHERER_JAR_VERSION}/opentelemetry-jmx-metrics.jar /opt/opentelemetry-jmx-metrics.jar
# nonroot user id (https://groups.google.com/g/distroless-users/c/-DpzCr7xRDY/m/eQqJmJroCgAJ)
ARG USER_UID=65532
RUN chown ${USER_UID} /opt/opentelemetry-jmx-metrics.jar


FROM gcr.io/distroless/java17-debian11:nonroot

COPY --from=prep /opt/opentelemetry-jmx-metrics.jar /opt/opentelemetry-jmx-metrics.jar
COPY --from=prep /otelcol-contrib /otelcol-contrib

EXPOSE 4317 55680 55679
ENTRYPOINT ["/otelcol-contrib"]
CMD ["--config", "/etc/otelcol-contrib/config.yaml"]

      values.yaml に以下の行を追加します。

      mode: deployment

      Collector 構成に以下を追加します。

      receivers:
  jmx:
    jar_path: /path/to/opentelemetry-jmx-metrics.jar
    endpoint: ${env:KAFKA_BROKER_JMX_ADDRESS}
    target_system: kafka,jvm
  jmx/consumer:
    jar_path: /path/to/opentelemetry-jmx-metrics.jar
    endpoint: ${env:KAFKA_CONSUMER_JMX_ADDRESS}
    target_system: kafka-consumer
  jmx/producer:
    jar_path: /path/to/opentelemetry-jmx-metrics.jar
    endpoint: ${env:KAFKA_PRODUCER_JMX_ADDRESS}
    target_system: kafka-producer

      JMX Metrics Gatherer

        JMX Metric Gatherer は、uber jar として実行し、コマンドラインからプロパティを構成することを目的としています。

        コレクターを実行しているホストで JRE が使用可能であることを確認してください。そうでない場合は、必ずダウンロードしてください。例:

        apt-get update && \
apt-get -y install default-jre-headless

        これが完了したら、JMX Metric Gatherer JAR の最新リリースを こちら でダウンロードして、以下を実行してください。

        // Kafka Broker
java -jar -Dotel.jmx.service.url=service:jmx:rmi:///jndi/rmi://{KAFKA_BROKER_JMX_ADDRESS}/jmxrmi \ -Dotel.jmx.target.system=kafka,jvm \
-Dotel.metrics.exporter=otlp \
-Dotel.exporter.otlp.endpoint=http://localhost:4317 \
-jar /path/to/opentelemetry-jmx-metrics.jar

// Kafka Producer
java -jar -Dotel.jmx.service.url=service:jmx:rmi:///jndi/rmi://{KAFKA_PRODUCER_JMX_ADDRESS}/jmxrmi \ -Dotel.jmx.target.system=kafka-producer \
-Dotel.metrics.exporter=otlp \
-Dotel.exporter.otlp.endpoint=http://localhost:4317 \
-jar /path/to/opentelemetry-jmx-metrics.jar

// Kafka Consumer
java -jar -Dotel.jmx.service.url=service:jmx:rmi:///jndi/rmi://{KAFKA_CONSUMER_JMX_ADDRESS}/jmxrmi \ -Dotel.jmx.target.system=kafka-consumer \
-Dotel.metrics.exporter=otlp \
-Dotel.exporter.otlp.endpoint=http://localhost:4317 \
-jar /path/to/opentelemetry-jmx-metrics.jar

        JMX Metric Gatherer は、uber jar として実行し、コマンドラインからプロパティを構成することを目的としています。

        これを Kubernetes でデプロイするには、JRE と JMX Metrics Gatherer Jar を含むイメージをビルドする必要があります。JRE と JMX Metrics Gatherer Jar を含むイメージの例については、以下の Dockerfile を参照してください。

        Dockerfile:

        FROM alpine:latest as prep

# JMX Metrics Gatherer Jar
ARG JMX_GATHERER_JAR_VERSION=1.27.0
ADD https://github.com/open-telemetry/opentelemetry-java-contrib/releases/download/v${JMX_GATHERER_JAR_VERSION}/opentelemetry-jmx-metrics.jar /opt/opentelemetry-jmx-metrics.jar
# nonroot user id (https://groups.google.com/g/distroless-users/c/-DpzCr7xRDY/m/eQqJmJroCgAJ)
ARG USER_UID=65532
RUN chown ${USER_UID} /opt/opentelemetry-jmx-metrics.jar

FROM gcr.io/distroless/java17-debian11:nonroot

COPY --from=prep /opt/opentelemetry-jmx-metrics.jar /opt/opentelemetry-jmx-metrics.jar

EXPOSE 4317 55680 55679
ENTRYPOINT ["java"]
CMD ["-Dotel.jmx.service.url=service:jmx:rmi:///jndi/rmi://kafka:1099/jmxrmi", \
"-Dotel.jmx.target.system=kafka,jvm", \
"-Dotel.metrics.exporter=otlp", \
"-Dotel.exporter.otlp.endpoint=http://otelcol:4317", \
"-jar", \
"/opt/opentelemetry-jmx-metrics.jar"]

        ログ収集

        OpenTelemetry Collector を使用してログを収集する方法については、ログ収集 を参照してください。

        すぐに使える Kafka ダッシュボードに表示するには、Kafka ログに source:kafka タグを付ける必要があります。これを行うには、属性プロセッサーを使用します。

        processors:
  attributes:
    actions:
      - key: ddtags
        value: "source:kafka"
        action: insert

        この属性のみが Kafka ログに追加されるようにするには、属性プロセッサーの 包含/除外フィルター を使用します。

        収集データ

        Kafka メトリクスレシーバー

        OTELDESCRIPTION
        kafka.consumer_group.lagCurrent approximate lag of consumer group at partition of topic
        kafka.consumer_group.offsetCurrent offset of the consumer group at partition of topic
        kafka.partition.current_offsetCurrent offset of partition of topic.

        JMX レシーバー / JMX Metrics Gatherer

        Kafka ブローカー

        OTELDESCRIPTIONFILTER
        kafka.controller.active.countController is active on broker
        kafka.isr.operation.countThe number of in-sync replica shrink and expand operationsoperation: shrink
        kafka.isr.operation.countThe number of in-sync replica shrink and expand operationsoperation: expand
        kafka.leader.election.rateLeader election rate - increasing indicates broker failures
        kafka.logs.flush.time.countLog flush count
        kafka.max.lagMax lag in messages between follower and leader replicas
        kafka.message.countThe number of messages received by the broker
        kafka.network.ioThe bytes received or sent by the brokerstate: in
        kafka.network.ioThe bytes received or sent by the brokerstate: out
        kafka.partition.countThe number of partitions on the broker
        kafka.partition.offlineThe number of partitions offline
        kafka.partition.under_replicatedThe number of under replicated partitions
        kafka.purgatory.sizeThe number of requests waiting in purgatorytype: fetch
        kafka.purgatory.sizeThe number of requests waiting in purgatorytype: produce
        kafka.request.failedThe number of requests to the broker resulting in a failuretype: produce
        kafka.request.failedThe number of requests to the broker resulting in a failuretype: fetch
        kafka.request.queueSize of the request queue
        kafka.request.time.99pThe 99th percentile time the broker has taken to service requeststype: produce
        kafka.request.time.99pThe 99th percentile time the broker has taken to service requeststype: fetchfollower
        kafka.request.time.99pThe 99th percentile time the broker has taken to service requeststype: fetchconsumer
        kafka.request.time.avgThe average time the broker has taken to service requeststype: produce
        kafka.request.time.avgThe average time the broker has taken to service requeststype: fetchconsumer
        kafka.request.time.avgThe average time the broker has taken to service requeststype: fetchfollower
        kafka.unclean.election.rateUnclean leader election rate - increasing indicates broker failures

        Kafka プロデューサー

        OTELDESCRIPTION
        kafka.producer.byte-rateThe average number of bytes sent per second for a topic.
        kafka.producer.compression-rateThe average compression rate of record batches for a topic.
        kafka.producer.io-wait-time-ns-avgThe average length of time the I/O thread spent waiting for a socket ready for reads or writes.
        kafka.producer.outgoing-byte-rateThe average number of outgoing bytes sent per second to all servers.
        kafka.producer.record-error-rateThe average per-second number of record sends that resulted in errors for a topic.
        kafka.producer.record-retry-rateThe average per-second number of retried record sends for a topic.
        kafka.producer.record-send-rateThe average number of records sent per second for a topic.
        kafka.producer.request-latency-avgThe average request latency.
        kafka.producer.request-rateThe average number of requests sent per second.
        kafka.producer.response-rateResponses received per second.

        Kafka コンシューマー

        OTELDESCRIPTION
        kafka.consumer.bytes-consumed-rateThe average number of bytes consumed per second
        kafka.consumer.fetch-size-avgThe average number of bytes fetched per request
        kafka.consumer.records-consumed-rateThe average number of records consumed per second
        kafka.consumer.total.bytes-consumed-rateThe average number of bytes consumed for all topics per second.
        kafka.consumer.total.records-consumed-rateThe average number of records consumed for all topics per second.

        注: Datadog では、- は _ に変換されます。たとえば、kafka.producer.request-rate は kafka.producer.request_rate になります。

        OpenTelemetry と Datadog のメトリクス名の完全なマッピングについては、OpenTelemetry メトリクスマッピング を参照してください。

        ログ出力例

        Resource SchemaURL: https://opentelemetry.io/schemas/1.20.0
Resource attributes:
     -> service.name: Str(unknown_service:java)
     -> telemetry.sdk.language: Str(java)
     -> telemetry.sdk.name: Str(opentelemetry)
     -> telemetry.sdk.version: Str(1.27.0)
ScopeMetrics #0
ScopeMetrics SchemaURL: 
InstrumentationScope io.opentelemetry.contrib.jmxmetrics 1.27.0-alpha
Metric #0
Descriptor:
     -> Name: kafka.message.count
     -> Description: The number of messages received by the broker
     -> Unit: {messages}
     -> DataType: Sum
     -> IsMonotonic: true
     -> AggregationTemporality: Cumulative
NumberDataPoints #0
StartTimestamp: 2024-01-22 15:50:24.212 +0000 UTC
Timestamp: 2024-01-22 15:51:24.218 +0000 UTC
Value: 25

        アプリ例

        このドキュメントで説明されている構成を示す以下の サンプルアプリケーション を参照してください。このサンプルアプリケーションは、プロデューサー、コンシューマー、ブローカー、および Zookeeper インスタンスで構成されています。これは、Kafka メトリクスレシーバー、JMX Receiver、および/または JMX Metrics Gatherer の使用を示します。