Agente Java: instrumento Kafka fila de mensagens

O agente Java da New Relic coleta automaticamente dados da biblioteca de clientes Java do Kafka. Como o Kafka é um sistema de mensagens de alto desempenho que gera muitos dados, você pode personalizar o agente para as taxas de transferência e casos de uso específicos do seu aplicativo.

Este documento explica como coletar e visualizar três tipos de dados Kafka:

• Métrica Kafka
• Evento Kafka
• Habilitar transação do Kafka Streams
• Rastreamento distribuído Kafka

Requisitos

A instrumentação de clientes Kafka está disponível nas versões 4.12.0 ou superior do agente Java. A instrumentação de fluxos Kafka está disponível no agente Java versões 8.1.0 ou mais alto. Para ver todas as bibliotecas Kafka suportadas, verifique a página de compatibilidade e requisitos do Java. Observe que o Kafka Streams é executado em clientes Kafka, portanto, toda a instrumentação que se aplica aos clientes Kafka também se aplica ao Streams.

Ver métrica Kafka

Após a instalação, o agente reporta automaticamente métricas Kafka ricas com informações sobre taxas de mensagens, latência, atraso e muito mais. O agente Java coleta todas as métricas de consumidor e produtor Kafka (mas não conecta ou transmite métricas).

Para visualizar essas métricas, crie um dashboard personalizado:

1. Vá para o explorador métrico New Relic.

2. Use o explorador métrico para localizar sua métrica. Aqui estão algumas pastas onde você pode encontrar métricas:

   • Métrica Kafka:

     MessageBroker/Kafka/Internal/KafkaMetricName

     Copiar

     Por exemplo, métrica request-rate :

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

     Copiar

   • Fluxos Kafka:

     Kafka/Streams/KafkaStreamsMetricName

     Copiar

     Por exemplo, métrica poll-latency-avg :

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

     Copiar

   • Conexão Kafka:

     Kafka/Connect/KafkaConnectMetricName

     Copiar

     Por exemplo, métrica connector-count :

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

     Copiar

3. Adicione a métrica que deseja monitor a um dashboard clicando em Add to dashboard.

Habilitar coleção de eventos Kafka

Você pode configurar o agente para coletar dados de evento em vez de dados de métrica de fração de tempo (para saber a diferença entre métrica de fração de tempo e dados de evento, consulte coleta de dados). Isso permite que você use NRQL para filtrar e facetar a métrica Kafka padrão. Quando ativado, o agente coleta um evento Kafka a cada 30 segundos. Este evento contém todos os dados do consumidor Kafka e métricas de produção capturadas desde o evento anterior.

Se você estiver usando Kafka Streams, o agente gera um evento separado que contém todos os dados da métrica do fluxo Kafka capturados desde o evento anterior. O evento também é coletado a cada 30 segundos.

Para ativar a coleta de eventos Kafka:

1. Adicione o elemento kafka.metrics.as_events.enabled ao arquivo de configuração newrelic.yml :

   kafka.metrics.as_events.enabled: true

   Copiar

2. Reinicie sua JVM.

3. Use o explorador de eventos para visualizar seus eventos Kafka, localizados no tipo de evento KafkaMetrics . Ou use NRQL para consultar seu evento diretamente. Por exemplo:

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

   Copiar

   Se você estiver consultando a métrica do Kafka Streams, use o tipo de evento KafkaStreamsMetrics para acessar a métrica específica do stream.

Habilitar métrica do nó Kafka

Existe um módulo de instrumentação alternativo para clientes Kafka que fornecerá mais granularidade para a métrica Kafka. Este módulo de instrumentação está disponível desde o agente 8.6.0 e está desabilitado por padrão.

Para ativar este módulo de instrumentação, você deve desativar o módulo de instrumentação existente e ativar o novo adicionando o seguinte ao seu arquivo de configuração newrelic.yml :

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

Habilitar evento de configuração do Kafka

O módulo de instrumentação kafka-clients-config enviará periodicamente eventos com o conteúdo da configuração do cliente Kafka. Este módulo está disponível desde o agente 8.6.0 e está desabilitado por padrão.

Para ativar kafka-clients-config adicione o seguinte ao arquivo de configuração newrelic.yml :

class_transformer:
  kafka-clients-config:
    enabled: true

Habilitar transação do Kafka Streams

Se você estiver usando Kafka Streams, por padrão não habilitamos a transação. Isso evita despesas desnecessárias porque os aplicativos Kafka tendem a ter altas taxas de transferência.

Ao contrário do JMS transação, os Kafka Streams transação não são processados por registro. Em vez disso, uma transação começa quando um consumidor Kafka pesquisa os registros e então os dados resultantes são processados.

Se você deseja criar transações, você precisa ativar um módulo kafka-streams-spans :

class_transformer:
  kafka-streams-spans:
    enabled: true

Habilitar transação Kafka Connect

Se você estiver usando o Kafka Connect, por padrão não habilitamos a transação. Isso evita despesas desnecessárias porque os aplicativos Kafka tendem a ter altas taxas de transferência.

As transações do Kafka Connect são registradas para cada iteração da tarefa sink/source. Para uma tarefa de coletor, uma transação consiste em sondar um consumidor Kafka, converter cada mensagem e enviar dados ao destino. Para uma tarefa de origem, uma transação consiste na leitura do destino, convertendo os dados em mensagens e enviando cada mensagem a um produtor Kafka.

Se você deseja coletar essas transações, você precisa ativar um módulo kafka-connect-spans :

class_transformer:
  kafka-connect-spans:
    enabled: true

Habilitar rastreamento distribuído Kafka

O agente Java também pode coletar rastreamento distribuído de clientes Kafka. Como o Kafka Streams é executado em clientes Kafka, as etapas para gerenciar distributed tracing também se aplicam. A ativação do rastreamento não afeta as operações normais do agente; ele ainda reportará dados métricos ou de eventos do Kafka.

Impactos e requisitos a considerar antes de permitir:

• The instrumentation adds a 150 to 200 byte payload to message headers.

  Se suas mensagens Kafka forem muito pequenas, o rastreamento poderá adicionar sobrecarga significativa de processamento e armazenamento. Esse tamanho de carga útil adicional pode fazer com que o Kafka descarte mensagens se elas excederem o limite de tamanho de mensagens do Kafka. Por esse motivo, recomendamos testar o rastreamento distribuído do Kafka em um ambiente de desenvolvimento antes de habilitá-lo na produção.

• Distributed tracing está disponível apenas para o cliente Kafka versões 0.11.0.0 ou superior.

• Se você ainda não habilitou distributed tracing para seu aplicativo, leia o guia de transição antes de habilitá-lo.

• Para propagar W3C Trace Context por meio de cabeçalhos de mensagens Kafka, consulte o distributed tracing API guia de uso para obter detalhes sobre APIs que foram lançadas no agente Java 6.4.0. Observe que adicionar cabeçalhos adicionais às mensagens Kafka aumentará ainda mais o tamanho da carga útil. Para ver essas API em ação,consulte Usando do agente Java trace API com Kafka.

• Se estiver usando o Kafka Streams, você precisará habilitar um módulo de instrumentação de span (consulte a seção de transação do Kafka Streams). Como uma transação não é registrada por registro, a aceitação de cabeçalhos distributed trace funcionará apenas para um registro.

O processo completo para habilitar isso é descrito abaixo, mas em alto nível envolve estas etapas básicas: 1) habilitar o rastreamento por meio da configuração do agente e 2) chamar a API do agente Java para o instrumento de transação tanto do lado do produtor quanto do consumidor.

Para coletar rastreamento distribuído do 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);
}