Agent Java: instrumenter les message de file d'attente Kafka

L'agent Java New Relic collecte automatiquement les données de la bibliothèque de clients Java de Kafka. Étant donné que Kafka est un système de messagerie hautes performances qui génère beaucoup de données, vous pouvez personnaliser l'agent en fonction du débit et des cas d'utilisation spécifiques de votre application.

Ce document explique comment collecter et afficher ces types de données Kafka :

• Métriques Kafka
• L'événement Kafka
• Transactions Kafka Streams
• Les traces de Kafka distribuées

Exigences

L'instrumentation des clients Kafka est disponible dans les versions d'agent Java 4.12.0 ou supérieures. instrumentation des flux Kafka est disponible dans les versions 8.1.0 de l'agent Java ou supérieur. Pour voir toutes les bibliothèques Kafka prises en charge, consultez la page de compatibilité et d'exigencesJava . Notez que Kafka Streams s'exécute sur les clients Kafka, donc toute l'instrumentation qui s'applique aux clients Kafka s'applique également à Streams.

Afficher les métriques Kafka

Après l'installation, l'agent Java signale automatiquement des métriques Kafka riches avec des informations sur les taux de messagerie, la latence, le décalage, etc. L'agent collecte toutes les métriques Kafka consommateur et producteur (mais pas les métriques connect ou stream).

Pour visualiser ces métriques, créez un dashboard personnalisé :

1. Accédez à l'explorateur métriqueNew Relic .

2. Utilisez l'explorateur de métriques pour localiser vos métriques. Voici quelques dossiers dans lesquels vous pouvez trouver des métriques :

   • Métriques Kafka :

     MessageBroker/Kafka/Internal/KafkaMetricName

     Copier

     Par exemple, la métrique request-rate :

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

     Copier

   • Flux Kafka :

     Kafka/Streams/KafkaStreamsMetricName

     Copier

     Par exemple, la métrique poll-latency-avg :

     Kafka/Streams/stream-thread-metrics/poll-latency-avg

     Copier

   • Kafka Connect :

     Kafka/Connect/KafkaConnectMetricName

     Copier

     Par exemple, la métrique connector-count :

     Kafka/Connect/connect-worker-metrics/connector-count

     Copier

3. Ajoutez les métriques que vous souhaitez monitorer à un dashboard en cliquant sur Add to dashboard.

Voir la collection d'événements Kafka

Vous pouvez configurer l'agent pour collecter des données d'événement au lieu des données d'intervalle de temps métrique (pour la différence entre l'intervalle de temps métrique et les données d'événement, voir collecte de données). Cela vous permet d'utiliser NRQL pour filtrer et facetter les métriques Kafka par défaut. Lorsque cette option est activée, l'agent collecte un événement Kafka toutes les 30 secondes. Cet événement contient toutes les données de Kafka consommateur et produit les métriques capturées depuis l'événement précédent.

Si vous utilisez Kafka Streams, l'agent génère un événement distinct qui contient toutes les données des métriques de flux Kafka capturées depuis l'événement précédent. L'événement est également collecté toutes les 30 secondes.

Pour activer la collection d'événements Kafka :

1. Ajoutez l'élément kafka.metrics.as_events.enabled à votre fichier de configuration newrelic.yml :

   kafka.metrics.as_events.enabled: true

   Copier

2. Redémarrez votre JVM.

3. Utilisez l'explorateur d'événements pour afficher votre événement Kafka, situé dans le type d'événement KafkaMetrics . Ou utilisez NRQL pour requêter directement votre événement. Par exemple:

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

   Copier

   Si vous interrogez les métriques Kafka Streams, utilisez le type d'événement KafkaStreamsMetrics pour accéder aux métriques spécifiques aux flux.

Activer les métriques du nœud Kafka

Il existe un module d'instrumentation alternatif pour les clients Kafka qui fournira plus de granularité pour les métriques Kafka. Ce module d'instrumentation est disponible depuis l'agent 8.6.0 et est désactivé par défaut.

Pour activer ce module d'instrumentation, vous devez désactiver le module d'instrumentation existant et activer le nouveau en ajoutant ce qui suit à votre fichier de configuration newrelic.yml :

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

Événement de configuration Kafka

Le module instrumentation kafka-clients-config enverra périodiquement un événement avec le contenu de votre configuration client Kafka. Ce module est disponible depuis l'agent 8.6.0 et est désactivé par défaut.

Pour activer kafka-clients-config ajoutez ce qui suit à votre fichier de configuration newrelic.yml :

class_transformer:
  kafka-clients-config:
    enabled: true

Transactions Kafka Streams

Si vous utilisez Kafka Streams, par défaut, nous n'activons pas les transactions. Cela permet d'éviter des frais généraux inutiles, car les applications Kafka ont tendance à avoir un débit élevé.

Contrairement aux transactions JMS, les transactions Kafka Streams ne sont pas traitées par enregistrement. Au lieu de cela, une transaction commence lorsqu'un consommateur Kafka interroge les enregistrements, puis les données résultantes sont traitées.

Si vous souhaitez créer des transactions, vous devez activer un module kafka-streams-spans :

class_transformer:
  kafka-streams-spans:
    enabled: true

Activer la transaction Kafka Connect

Si vous utilisez Kafka Connect, par défaut, nous n'activons pas les transactions. Cela permet d'éviter des frais généraux inutiles, car les applications Kafka ont tendance à avoir un débit élevé.

Les transactions Kafka Connect sont enregistrées pour chaque itération de la tâche sink/source. Pour une tâche sink, une transaction consiste à interroger un consommateur Kafka, à convertir chaque message et à envoyer des données à la cible. Pour une tâche source, une transaction consiste à lire la cible, à convertir les données en messages et à envoyer chaque message avec un producteur Kafka.

Si vous souhaitez collecter ces transactions, vous devez activer un module kafka-connect-spans :

class_transformer:
  kafka-connect-spans:
    enabled: true

Les traces de Kafka distribuées

L'agent Java peut également collecter des traces distribuées à partir des clients Kafka. Étant donné que Kafka Streams s’exécute sur des clients Kafka, les étapes de gestion du tracing distribué s’appliquent également. L'activation de la trace n'affecte pas les opérations normales de l'agent; il continuera à signaler les données métriques ou d'événements de Kafka.

Impacts et exigences à prendre en compte avant d’activer :

• The instrumentation adds a 150 to 200 byte payload to message headers. Si vos messages Kafka sont très petits, la trace peut ajouter une surcharge de traitement et de stockage importante. Cette taille de charge utile supplémentaire peut amener Kafka à supprimer des messages s'ils dépassent votre limite de taille de messagerie Kafka. Pour cette raison, nous recommandons de tester les traces distribuées Kafka dans un environnement de développement avant de les activer en production.
• Le tracing distribué n'est disponible que pour les versions client Kafka 0.11.0.0 ou supérieures.
• Si vous n’avez pas encore activé le tracing distribué pour votre application, lisez le guide de transition avant de l’activer.
• Pour propager le contexte de trace W3C via les en-têtes de message Kafka, consultez le guide d'utilisation d'API de tracing distribué pour plus de détails sur API qui ont été publiées dans agent Java 6.4.0. Notez que l’ajout d’en-têtes supplémentaires aux messages Kafka augmentera encore la taille de la charge utile. Pour voir ces API en action, consultez Utilisation des API de trace d’agent Java avec Kafka.
• Si vous utilisez Kafka Streams, vous devez activer un module d'instrumentation SPAN (reportez-vous à la section Transaction Kafka Streams). Étant donné qu'une transaction n'est pas enregistrée par enregistrement, l'acceptation des en-têtes tracedistribués ne fonctionnera que pour un seul enregistrement.

Le processus complet d'activation de cette fonction est décrit ci-dessous, mais à un niveau élevé, il implique les étapes de base suivantes : 1) activer le tracing via la configuration de l'agent et 2) appeler l'API de l'agent Java pour instrumenter les transactions du côté du producteur et du côté du consommateur.

Pour collecter les traces distribuées à partir de 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);
}