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

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

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

Kafkaメトリクス
Kafkaイベント
Kafka Streams トランザクションを有効にする
Kafkaの分散型トレース

ヒント

また、Kafka との統合もあります。詳細については、 Kafka 監視統合を参照してください。

要件

Kafka クライアントのインスツルメンテーションは、Java エージェントバージョン 4.12.0 以降で使用できます。Kafka ストリームインストルメンテーションは、Java エージェントバージョン 8.1.0 で使用できます。以上。サポートされているすべての Kafka ライブラリを確認するには、 Java の互換性と要件のページを確認してください。Kafka Streams は Kafka クライアント上で実行されるため、Kafka クライアントに適用されるすべてのインストルメンテーションは Streams にも適用されることに注意してください。

Kafkaメトリクスの表示

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

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

New Relic metric explorer にアクセスします。
メトリックエクスプローラーを使用してメトリックを見つけます。メトリクスを見つけることができるいくつかのフォルダーを次に示します。
- Kafka メトリクス:
  MessageBroker/Kafka/Internal/KafkaMetricName
  たとえば、 request-rate 指標は次のようになります。
  MessageBroker/Kafka/Internal/consumer-metrics/request-rate
- カフカストリーム:
  Kafka/Streams/KafkaStreamsMetricName
  たとえば、 poll-latency-avg 指標は次のようになります。
  Kafka/Streams/stream-thread-metrics/poll-latency-avg
- カフカ接続:
  Kafka/Connect/KafkaConnectMetricName
  たとえば、 connector-count 指標は次のようになります。
  Kafka/Connect/connect-worker-metrics/connector-count
Add to dashboard をクリックして、モニターしたいメトリクスをダッシュボードに追加します。

ヒント

Kafka のコンシューマー、プロデューサー、およびストリームのメトリックの完全なリストについては、 Kafka のドキュメントを参照してください。これらのドキュメントのメトリックは、JMX を介して検索できます。ドキュメントに記載されているすべてのメトリクスが New Relic にエクスポートされるわけではないことに注意してください。これは、次のいずれかの理由による可能性があります。

このメトリクスは、実際には Kafka クライアントまたは Kafka ストリームによって生成されるわけではありません。これは、古いバージョンのクライアントまたはストリームを使用しているか、Kafka ライブラリの設定方法と使用方法に基づいていることが原因である可能性があります。
メトリックが数値でないか、その値が NaNです。New Relic は、数値を持つメトリクスのみを受け入れます。

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

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

Kafka ストリームを使用している場合、エージェントは、前のイベント以降にキャプチャされた Kafka ストリームメトリックからのすべてのデータを含む別のイベントを生成します。イベントも 30 秒ごとに収集されます。

重要

エージェントは収集サイクルごとに最大 2000 のイベントを記録しますが、この値はmax_samples_storedで変更できます。Kafka イベントデータは、このプールに含まれています。recordCustomEvent() API 呼び出しを使用してカスタムイベントを New Relic に送信し、2000 を超えるイベントを送信すると、エージェントは一部の Kafka またはカスタムイベントを破棄します。

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

kafka.metrics.as_events.enabled要素をnewrelic.yml構成ファイルに追加します。
```
kafka.metrics.as_events.enabled: true
```
JVMを再起動します。
イベントエクスプローラーを使用して、 KafkaMetricsイベントタイプにある Kafka イベントを表示します。または、 NRQLを使用してイベントを直接クエリします。例えば：
```
SELECT average('producer-metrics.record-send-rate') from KafkaMetrics SINCE 30 minutes ago timeseries
```
Kafka ストリームメトリクスをクエリする場合は、 KafkaStreamsMetrics イベントタイプを使用してストリーム固有のメトリクスにアクセスします。

重要

タイムスライスメトリックとして New Relic に送信できる Kafka メトリックの種類に関する制限は、イベントにも適用されることに注意してください。つまり、非数値および NaN メトリックはイベント属性として含まれません。

Kafka ノードのメトリクスを有効にする

Kafka メトリクスの粒度をさらに高める、Kafka クライアント用の代替インスツルメンテーションモジュールがあります。このインストルメンテーションモジュールはエージェント 8.6.0 以降で利用可能で、デフォルトでは無効になっています。

このインストルメンテーションモジュールを有効にするには、既存のインストルメンテーションモジュールを無効にし、次の内容を newrelic.yml 構成ファイルに追加して新しいインストルメンテーションモジュールを有効にする必要があります。

class_transformer:
  kafka-clients-metrics:
    enabled: false
  kafka-clients-node-metrics:
    enabled: true

Kafka 構成イベントを有効にする

kafka-clients-config インストルメンテーションモジュールは、Kafka クライアント構成の内容を含むイベントを定期的に送信します。このモジュールはエージェント 8.6.0 以降で利用可能で、デフォルトでは無効になっています。

kafka-clients-config を有効にするには、以下を newrelic.yml 構成ファイルに追加します。

class_transformer:
  kafka-clients-config:
    enabled: true

Kafka Streams トランザクションを有効にする

Kafka Streams を使用している場合、デフォルトではトランザクションは有効になりません。これは、Kafka アプリケーションのスループットが高くなる傾向があるため、不要なオーバーヘッドを防ぐためです。

JMS トランザクションとは異なり、Kafka Streams トランザクションはレコードごとに処理されません。代わりに、Kafka コンシューマーがレコードをポーリングし、結果のデータが処理されるときにトランザクションが開始されます。

トランザクションを作成したい場合は、 kafka-streams-spans モジュールを有効にする必要があります:

class_transformer:
  kafka-streams-spans:
    enabled: true

Kafka Connect トランザクションを有効にしました

Kafka Connect を使用している場合、デフォルトではトランザクションは有効になりません。Kafka アプリケーションはスループットが高い傾向があるため、これは不要なオーバーヘッドを防ぐためです。

Kafka Connect トランザクションは、シンク/ソースタスクの反復ごとに記録されます。シンクタスクの場合、トランザクションは、Kafka コンシューマーのポーリング、各メッセージの変換、およびターゲットへのデータの送信で構成されます。ソースタスクの場合、トランザクションは、ターゲットからの読み取り、データのメッセージへの変換、Kafka プロデューサによる各メッセージの送信で構成されます。

これらのトランザクションを収集したい場合は、 kafka-connect-spans モジュールを有効にする必要があります。

class_transformer:
  kafka-connect-spans:
    enabled: true

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

Java エージェントは、Kafka クライアントから分散トレースを収集することもできます。Kafka Streams は 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 を参照してください。
Kafka Streams を使用している場合は、スパンインストルメンテーションモジュールを有効にする必要があります ( Kafka Streams トランザクションセクションを参照してください)。トランザクションはレコードごとに記録されないため、分散トレースヘッダーの受け入れは 1 つのレコードに対してのみ機能します。

これを有効にする完全なプロセスを以下に説明しますが、大まかに言えば、次の基本的な手順が含まれます。1) エージェント構成を介してトレースを有効にし、2) Java エージェント API を呼び出してプロデューサー側とコンシューマー側の両方でトランザクションを計測します。

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

以前にアプリでdistributed tracingを有効にしていない場合は、有効にする前に distributed tracing transition guide をお読みください。

Kafka 分散トレースを有効にするには、 newrelic.yml構成ファイルで次の 2 つの設定を有効にする必要があります。

distributed_tracing要素が有効になっていることを確認します。
```
distributed_tracing:
  enabled: true
```
以下を設定ファイルに追加して、Kafka固有の分散型トレーシング機能を有効にします。
```
class_transformer:
  com.newrelic.instrumentation.kafka-clients-spans-0.11.0.0:
    enabled: true
```

Kafka プロデューサーを計測するには、 Producer.send(ProducerRecord<K, V> record)を呼び出す前にトランザクションを開始する必要があります。これを行うには、Java エージェント@Trace(dispatcher = true)アノテーションをメソッドに追加します。

例えば：

@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);
}

重要

Kafka Streams を使用している場合、トランザクションを直接開始したり、プロデューサーにデータを送信したりする必要はありません。

Kafka コンシューマーを計測するには、メッセージの処理中にトランザクションを開始する必要があります。エージェントは、分散トレースペイロードヘッダーをnewrelicキーまたは W3C のtraceparentおよびtracestateキーの下に保存します。ヘッダーを取得し、New Relic トランザクション API を呼び出してペイロードを受け入れます。

例えば：

@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);
}

重要

Kafka Streams でトランザクションを有効にした場合 ( Kafka Streams トランザクションセクションを参照)、トランザクションが多くのレコードに適用される場合でも、分散トレースヘッダーの受け入れは 1 つのレコードにのみ適用されます。

この機械翻訳は参考用に提供されます。