Kafka Metric reporter

브로커 그리고 클라이언트는 내부 메트릭 정보를 리포트한다. 메트릭 정보를 수집해서 모니터링 하고자 할 경우 카프카에서 제공하는 MetricReporter를 사용하면 된다.

Producer, Consumer, Kafka Streams는 “metric.reporters”라는 속성을 가지고 있다. 이 속성으로 MetricReporter 구현체를 지정해주면 된다. MetricReporter 구현체는 새로운 메트릭이 생성이 되면 새로운 메트릭 정보를 metricChange(KafkaMetric metric) 메소드를 통해 전달받는다. MetricReporter로 JmxReporter는 기본적으로 항상 추가된다.

MetricsReporter 인터페이스

/**
 * A plugin interface to allow things to listen as new metrics are created so they can be reported.
 * <p>
 * Implement {@link org.apache.kafka.common.ClusterResourceListener} to receive cluster metadata once it's available. Please see the class documentation for ClusterResourceListener for more information.
 */
public interface MetricsReporter extends Configurable {

    /**
     * This is called when the reporter is first registered to initially register all existing metrics
     * @param metrics All currently existing metrics
     */
    public void init(List<KafkaMetric> metrics);

    /**
     * This is called whenever a metric is updated or added
     * @param metric
     */
    public void metricChange(KafkaMetric metric);

    /**
     * This is called whenever a metric is removed
     * @param metric
     */
    public void metricRemoval(KafkaMetric metric);

    /**
     * Called when the metrics repository is closed.
     */
    public void close();

}

init : MetricReporter가 Metrics에 등록될 때 호출된다. 만약 현재 KafkaMetric정보가 Metrics 객체에 존재한다면, 모든 KafkaMetrics 정보를 MetricReporter에게 전달한다.
metricChanged : 새로운 메트릭 정보가 등록될 때 이 메소드를 통해 새로운 메트릭 정보를 전달받는다.
removeMetric : 메트릭 정보가 제거될 때 제거된 메트릭 정보를 이 메소드를 통해 전달받는다.
close : Metrics의 close 메소드가 호출된 경우, Metrics는 등록된 모든 MetricReporter의 close 메소드를 호출한다.

메트릭 정보

브로커 그리고 클라이언트는 많은 메트릭 정보를 리포트한다. 아래는 그 중 일부만을 설명한다.

fetch-size-avg : 요청 당 가져오는 평균 바이트 수
fetch-rate : 초당 fetch 요청을 보내는 수
records-consumed-rate : 초당 컨슘한 레코드의 평균 값
records-consumed-total : 컨슘한 총 레코드 수

KakaMetrics 정보들은 이미 계산된 메트릭 정보이다. 즉 이미 평균을 계산했거나, 최대, 최소값을 계산한 메트릭이 된다. 따라서 카프카 메트릭 정보를 Micrometer를 이용해서 외부 시스템으로 리포팅하고 있다면 이 점을 유의해야 한다. 카프카는 정말 다양한 메트릭 정보를 리포팅한다. 따라서 자세한 것은 공식문서를 참고하길 바란다.

Metrics

Sensor, KafkaMetric, MetricReporter를 가지고 있는 클래스이다.

Sensor : Sensor는 measurement를 기록할 때 사용하는 클래스이다. 또한 Sensor는 measurement와 관련된 메트릭을 가지고 있다. 예를 들어 Sensor가 메시지의 크기를 기록할 수 있으며, 센서에 의해 기록된 메시지 크기의 평균, 최대, 또는 기타 통계 메트릭을 센서에 추가할 수 있다. 그리고 Sensor에 추가된 메트릭 정보는 Metrics 클래스에 추가된다. (그리고 Metrics는 추가된 메트릭을 MetricReporter에게 알린다)

Metrics 클래스는 보통 아래와 같은 방식으로 사용된다.

Metrics metrics = new Metrics(); // this is the global repository of metrics and sensors

// create sensor
Sensor sensor = metrics.sensor(&quot;message-sizes&quot;);
MetricName metricName = new MetricName(&quot;message-size-avg&quot;, &quot;producer-metrics&quot;);
sensor.add(metricName, new Avg());
metricName = new MetricName(&quot;message-size-max&quot;, &quot;producer-metrics&quot;);
sensor.add(metricName, new Max());

Sensor sensor(String name) : 이 메소드를 통해 Sensor 객체를 생성하고, 생성된 Sensor 객체는 Metrics에 등록된다. 만약에 주어진 이름의 Sensor가 이미 존재한다면 Sensor 객체를 생성하지 않고 존재하는 Sensor 객체를 리턴한다.
add(MetricName metricName, MeasurableStat stat) : Sensor에 새로운 메트릭을 등록한다. 등록된 메트릭은 Metrics에도 추가 된다. Metrics는 추가된 메트릭을 MetricReporter에게 알려준다.

Fetcher, KafkaProducer 등은 Metrics 클래스를 통해서 Sensor 객체를 생성하고, Measurement를 기록한다. Fetcher가 Metrics#sensor 메소드를 통해서 Sensor 객체를 생성하면, Sensor 객체는 Metrics에 등록된다. 또한 Sensor에 메트릭을 추가할 때마다 메트릭이 Metrics에도 등록된다. Metrics 클래스에서 특정 메트릭의 현재 값을 알고 싶다면, KafkaMetric#metricValue 메소드를 호출하면 된다. 또한 Metrics에 등록된 모든 메트릭 정보를 가져오고 싶다면, Metrics#metrics() 메소드를 호출하면 된다. Fetcher는 Sensor#record 메소드를 호출해서 Measurement를 기록한다. 그리고 이 때 메트릭이 업데이트 된다.

아래 코드는 Fetcher에서 Sensor recordsFetched를 생성하는 코드이다. recordsFetched는 컨슘한 레코드의 수를 기록하는 Sensor이다.

this.recordsFetched = metrics.sensor("records-fetched");
this.recordsFetched.add(metrics.metricInstance(metricsRegistry.recordsPerRequestAvg), new Avg());
this.recordsFetched.add(new Meter(metrics.metricInstance(metricsRegistry.recordsConsumedRate),
        metrics.metricInstance(metricsRegistry.recordsConsumedTotal)));

아래 코드는 Fetcher가 Sensor에 메트릭을 업데이트하는 부분이다. sensor의 records 메소드를 통해서 메트릭을 업데이트한다.

public void record(TopicPartition partition, int bytes, int records) {
    
    // skip 
    if (this.unrecordedPartitions.isEmpty()) {
        this.sensors.bytesFetched.record(this.fetchMetrics.fetchBytes);
        this.sensors.recordsFetched.record(this.fetchMetrics.fetchRecords);
        // skip
    }
}

Kafka Metric

Kafka Metric reporter

MetricsReporter 인터페이스

관련 속성

메트릭 정보

Metrics

관련 문서