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

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

このドキュメントでは、次の Kafka データ タイプを収集および表示する方法について説明します。

• Kafkaメトリクス
• Kafkaイベント
• Kafka ストリーム トランザクション
• Kafkaの分散型トレース

要件

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

Kafkaメトリクスの表示

インストール後、 Javaエージェントは、メッセージング レート、レイテンシ、ラグなどに関する情報を含む豊富な Kafka メトリクスを自動的にレポートします。 エージェントは、すべてのKafka 消費者とプロデューサーのメトリクスを収集します (ただし、メトリクスの接続やストリーミングは行いません)。

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

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

2. メトリック エクスプローラーを使用してメトリックを見つけます。メトリクスを見つけることができるいくつかのフォルダーを次に示します。

   • 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

     コピー

3. Add to dashboardをクリックして、監視するメトリックをダッシュボードに追加します。

Kafka イベントコレクションを表示

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

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

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 ストリーム メトリクスをクエリする場合は、 KafkaStreamsMetrics イベント タイプを使用してストリーム固有のメトリクスにアクセスします。

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 ストリーム トランザクション

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 からのメトリックまたはイベント データは引き続きレポートされます。

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

• The instrumentation adds a 150 to 200 byte payload to message headers. 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から分散型トレースを収集する。

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

@Trace(dispatcher = true)
private void consumeMessage(KafkaConsumer<String, String> consumer) {
  final ConsumerRecords<String, String> records = consumer.poll(1000);
  // Process your records ....
}

@KafkaListener(id = "foo", topics = "myTopic", clientIdPrefix = "myClientId")
public void listen(String data) {
  ...
}