Java 에이전트: Kafka 메시지 큐 계측

New Relic Java 에이전트는 Kafka 의 Java 클라이언트 라이브러리에서 자동으로 데이터를 수집합니다. Kafka는 많은 데이터를 생성하는 고성능 메시징 시스템이므로 앱의 특정 처리량 및 사용 사례에 맞게 에이전트를 사용자 지정할 수 있습니다.

이 문서에서는 세 가지 유형의 Kafka 데이터를 수집하고 보는 방법을 설명합니다.

• 카프카 메트릭
• 카프카 이벤트
• Kafka Streams 트랜잭션 활성화
• Kafka 분산 추적

요구 사항

Kafka 클라이언트 계측은 Java 에이전트 버전 4.12.0 이상에서 사용할 수 있습니다. Kafka 스트림 계측은 Java 에이전트 버전 8.1.0에서 사용할 수 있습니다. 또는 더 높게. 지원되는 모든 Kafka 라이브러리를 보려면 Java 호환성 및 요구 사항 페이지를확인하십시오. Kafka Streams는 Kafka 클라이언트 위에서 실행되므로 Kafka 클라이언트에 적용되는 모든 계측이 Streams에도 적용됩니다.

Kafka 측정항목 보기

설치 후 에이전트는 메시징 속도, 대기 시간, 지연 등에 대한 정보와 함께 풍부한 Kafka 메트릭을 자동으로 보고합니다. Java 에이전트는 모든 Kafka 소비자 및 생산자 메트릭 (연결 또는 스트림 메트릭은 아님)을 수집합니다.

이러한 측정항목을 보려면 사용자 정의 대시보드를 만드십시오.

1. New Relic 메트릭 탐색기 로 이동합니다.

2. 측정항목 탐색기를 사용하여 측정항목을 찾습니다. 메트릭을 찾을 수 있는 몇 가지 폴더는 다음과 같습니다.

   • 카프카 지표:

     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초마다 하나의 Kafka 이벤트를 수집합니다. 이 이벤트는 Kafka 소비자의 모든 데이터를 포함하고 이전 이벤트 이후 캡처된 메트릭을 생성 합니다.

Kafka Streams를 사용하는 경우 에이전트는 이전 이벤트 이후에 캡처된 Kafka 스트림 메트릭 의 모든 데이터를 포함하는 별도의 이벤트를 생성합니다. 이벤트도 30초마다 수집됩니다.

Kafka 이벤트 수집을 활성화하려면:

1. newrelic.yml 구성 파일에 kafka.metrics.as_events.enabled 요소를 추가합니다.

   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 Streams 측정항목을 쿼리하는 경우 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 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의 메트릭 또는 이벤트 데이터를 보고합니다.

활성화하기 전에 고려해야 할 영향 및 요구 사항:

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

  Kafka 메시지가 매우 작은 경우 Traces는 상당한 처리 및 저장 오버헤드를 추가할 수 있습니다. 이러한 추가 페이로드 크기로 인해 Kafka 메시징 크기 제한을 초과하는 경우 Kafka가 메시지를 삭제할 수 있습니다. 이러한 이유로 프로덕션 환경에서 활성화하기 전에 개발 환경에서 Kafka 반사 트레이스를 테스트해 보는 것이 좋습니다.

• 분산 추적은 Kafka 클라이언트 버전 0.11.0.0 이상에서만 사용할 수 있습니다.

• 이전에 앱에 대해 분산 추적을 활성화 하지 않은 경우 활성화하기 전에 전환 가이드 를 읽으십시오.

• Kafka 메시지 헤더를 통해 W3C 추적 컨텍스트를 전파하려면 Java 에이전트 6.4.0에서 릴리스된 API에 대한 자세한 내용은 분산 추적 API 사용 안내서 를 참조하십시오. Kafka 메시지에 추가 헤더를 추가하면 페이로드 크기가 더 늘어납니다. 작동 중인 이러한 API를 보려면 Kafka와 함께 Java 에이전트 추적 API 사용 을 참조하십시오.

• Kafka Streams를 사용하는 경우 스팬 계측 모듈을 활성화해야 합니다( Kafka Streams 트랜잭션 섹션참조). 트랜잭션이 레코드별로 기록되지 않기 때문에 분산 추적 헤더 수락은 하나의 레코드에 대해서만 작동합니다.

이를 활성화하는 전체 프로세스는 아래에 설명되어 있지만 높은 수준에서 다음과 같은 기본 단계가 포함됩니다. 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);
}