Javaエージェントです。Kafkaメッセージキューの計測

New Relic Java エージェントは、 Kafka's Java clients library からデータを自動的に収集します。Kafka は大量のデータを生成する高性能なメッセージングシステムであるため、アプリの特定のスループットやユースケースに合わせてエージェントをカスタマイズすることができます。

このドキュメントでは、3種類のKafkaデータを収集して表示する方法を説明しています。

• Kafkaメトリクス
• Kafkaイベント
• Kafkaの分散型トレース

Kafka のインスツルメンテーションは、Java エージェントのバージョン 4.12.0 以降で利用できます。サポートされているKafkaクライアントのバージョンについては、 Javaの互換性と要件 を参照してください。

Kafkaメトリクスの表示

インストール後 、エージェントは、メッセージングレート、レイテンシー、ラグなどの情報を含むリッチなKafkaメトリクスを自動的にレポートします。Javaエージェントは、 Kafkaのコンシューマーおよびプロデューサーのすべてのメトリクス を収集します（ただし、コネクトおよびストリームのメトリクスは収集しません）。

これらのメトリクスを表示するには、カスタムダッシュボードを作成します。

1. New Relic metric explorer にアクセスします。

2. メトリクスエクスプローラーを使って、メトリクスを探します。Kafkaのメトリクスは、このmetricフォルダにあります。

   MessageBroker/Kafka/Internal/KafkaMetricName

   コピー

   たとえば、 request-rateメトリックは次の場所にあります。

   MessageBroker/Kafka/Internal/consumer-metrics/request-rate

   コピー

   ヒント

   Kafkaのコンシューマーとプロデューサーのメトリクスの全リストについては、 Kafka documentation を参照してください。

3. Add to dashboard をクリックして、モニターしたいメトリクスをダッシュボードに追加します。

Kafkaイベントコレクションの有効化

メトリックのタイムスライスデータではなく、イベントデータを収集するようにエージェントを構成することができます（メトリックのタイムスライスとイベントデータの違いについては、 データ収集 ）。これにより、 NRQL を使用して、デフォルトの Kafka メトリクスをフィルタリングおよびファセットすることができます。有効にすると、エージェントは 30 秒ごとに 1 つの Kafka イベントを収集します。このイベントには、 Kafka コンシューマーからのすべてのデータと、前回のイベント以降に取得したプロダク トメトリクス が含まれます。

Kafkaのイベント収集を有効にするには

1. kafka.metrics.as_events.enabled要素をnewrelic.yml構成ファイルに追加します。

   kafka.metrics.as_events.enabled: true

   コピー

2. JVMを再起動します。

3. イベント エクスプローラーを使用して、 KafkaMetricsイベント タイプにある Kafka イベントを表示します。または、 NRQLを使用してイベントを直接クエリします。例えば：

   SELECT average('producer-metrics.record-send-rate') from KafkaMetrics SINCE 30 minutes ago timeseries

   コピー

Kafkaの分散型トレースの有効化

Java エージェントは、Kafka クライアントから 分散型トレース を収集することもできます。トレースを有効にしても、エージェントの通常の動作には影響しません。Kafkaからのメトリックデータやイベントデータの報告は継続されます。

実現する前に考慮すべき影響と要件

• インスツルメンテーションは、メッセージヘッダーに150～200バイトのペイロードを追加します。 Kafkaメッセージが非常に小さい場合、トレースは大きな処理およびストレージのオーバーヘッドを追加します。このペイロードサイズの追加により、Kafkaのメッセージサイズ制限を超えると、Kafkaがメッセージをドロップする可能性があります。このため、本番環境で有効にする前に、開発環境でKafka分散トレースをテストすることをお勧めします。
• 分散トレーシングは、Kafkaクライアントのバージョン0.11.0.0以降でのみ利用可能です。
• 以前にアプリで distributed tracing を有効にしていない場合は、有効にする前に Transition guide をお読みください。
• Kafkaメッセージヘッダーを介してW3Cトレースコンテキストを伝播するには、Java agent 6.4.0でリリースされたAPIの詳細について、 distributed tracing API usage guide を参照してください。なお、Kafkaメッセージに追加のヘッダを追加すると、ペイロードサイズがさらに大きくなります。これらのAPIの動作を確認するには、 Using Java agent trace APIs with Kafka を参照してください。

この機能を有効にするための完全なプロセスは以下のとおりですが、大まかには以下のような基本的なステップがあります。1) エージェントの設定でトレースを有効にする。2) Java エージェント API を呼び出して、プロデューサ側とコンシューマ側の両方のトランザクションを計測する。

Kafkaから分散型トレースを収集する。

@Trace(dispatcher = true)
public static void createAndSend(KafkaProducer<String, String> producer){
    ProducerRecord<String, String> data = new ProducerRecord<String, String>("topic", "key", "value");
    producer.send(data);
}

@Trace(dispatcher = true)
private static void processMessage(ConsumerRecord<String, String> rec) {
  // create a distributed trace headers map
  Headers dtHeaders = ConcurrentHashMapHeaders.build(HeaderType.MESSAGE);

  // Iterate through each record header and insert the trace headers into the dtHeaders map
  for (Header header : rec.headers()) {
    String headerValue = new String(header.value(), StandardCharsets.UTF_8);

    // using the newrelic key
    if (header.key().equals("newrelic")) {
      dtHeaders.addHeader("newrelic", headerValue);
    }

    // or using the W3C keys
    if (header.key().equals("traceparent")) {
      dtHeaders.addHeader("traceparent", headerValue);
    }

    if (header.key().equals("tracestate")) {
      dtHeaders.addHeader("tracestate", headerValue);
    }
  }

  // Accept distributed tracing headers to link this request to the originating request
  NewRelic.getAgent().getTransaction().acceptDistributedTraceHeaders(TransportType.Kafka, dtHeaders);
}