Kafka Metrics模块解析

2022-07-28  本文已影响0人  tracy_668

[TOC]

背景

Metrics是kafka内部使用的监控模块，主要有以下几个组成部分：

Measurable
Stat
Sensor
Metric

接口分析

1. Measurable

Measurable接口是度量类型最基础的接口，通过measure()方法获取被监控的值。

public interface Measurable extends MetricValueProvider<Double> {
    double measure(MetricConfig config, long now);
}
2. Stat

Stat接口表示需要经过统计计算的度量类型，例如平均值、最大值、最小值等，通过record()方法记录某值并更新度量值。

public interface Stat {
    public void record(MetricConfig config, double value, long timeMs);
}

MeasuleStat继承了Measureable接口和Stat接口，并没有添加新的方法。CompoundStat接口表示多个Stat的组合。
SampledStat是一个比较重要的抽象类，它表示一个抽样的度量值，除了Total外的其他MeasureableStat接口实现都依赖它功能。在SampleStat中可以有多个Sample并通过多个Sample完成对一个值的度量，在每个Sample中都记录了其对应的时间窗口和事件数量，SampledStat在计算最终的结果值时，可以根据这两个值决定是否使用此sample中的数据。SampledStat实现了MeasuleStat接口的record()方法和measure()方法。在record()方法中会根据时间窗口和事件数使用合适的Sample对象进行记录。

public void record(MetricConfig config, double value, long timeMs) {
//      拿到当前时间的sample对象
        Sample sample = current(timeMs);
//      检测当前sample是否已经完成取样        
        if (sample.isComplete(timeMs, config))
            sample = advance(config, timeMs);
//      更新sample对象
        update(sample, config, value, timeMs);
//      smaple对象的事件数加1
        sample.eventCount += 1;
    }

measure()方法首先会将过期的sample重置，之后调用combine方法完成计算。combine方法是抽象方法，不同子类有不同的实现。

public double measure(MetricConfig config, long now) {
//      检查sample是否过期
        purgeObsoleteSamples(config, now);
        return combine(this.samples, config, now);
}
3. Sensor

在实际应用中，对同一个操作需要有多个不同方面的度量，例如需要监控请求的最大长度，同时也需要监控请求的平均长度等。kafka通过将多个相关的度量对象封装在进sensor中实现。

4. Metric

Metrics类，负责统一管理Sensor对象、KafkaMetric对象。

public class Metrics implements Closeable {
//  默认配置信息
    private final MetricConfig config;
//  保存了添加到Metrics中的KafkaMetrics对象
    private final ConcurrentMap<MetricName, KafkaMetric> metrics;
//  保存了添加到Metrics中的Sensor的集合
    private final ConcurrentMap<String, Sensor> sensors;
//  记录了每个Sensor的子Sensor集合
    private final ConcurrentMap<Sensor, List<Sensor>> childrenSensors;
    private final List<MetricsReporter> reporters;
    private final Time time;
    private final ScheduledThreadPoolExecutor metricsScheduler;
    private static final Logger log = LoggerFactory.getLogger(Metrics.class);

//  从sensors集合中获取sensor对象，如果指定的Sensor不存在则创建新Sensor对象，并使用childrenSensors集合记录Sensor的层级关系
    public synchronized Sensor sensor(String name, MetricConfig config, long inactiveSensorExpirationTimeSeconds, Sensor.RecordingLevel recordingLevel, Sensor... parents) {
    //  根据name从sensors集合中获取sensor对象
        Sensor s = getSensor(name);
        if (s == null) {
        //  如果不存在则创建sensor对象
            s = new Sensor(this, name, parents, config == null ? this.config : config, time, inactiveSensorExpirationTimeSeconds, recordingLevel);
            this.sensors.put(name, s);
            if (parents != null) {
            // 通过childrenSensors记录sensor的层级关系
                for (Sensor parent : parents) {
                    List<Sensor> children = childrenSensors.get(parent);
                    if (children == null) {
                        children = new ArrayList<>();
                        childrenSensors.put(parent, children);
                    }
                    children.add(s);
                }
            }
            log.debug("Added sensor with name {}", name);
        }
        return s;
    }
}

使用场景

Producer、Consumer、Broker都会用到。下面以Producer举例。
Producer的构造函数中会初始化Metrics。

MetricConfig metricConfig = new MetricConfig().samples(config.getInt(ProducerConfig.METRICS_NUM_SAMPLES_CONFIG))
                    .timeWindow(config.getLong(ProducerConfig.METRICS_SAMPLE_WINDOW_MS_CONFIG), TimeUnit.MILLISECONDS)
                    .recordLevel(Sensor.RecordingLevel.forName(config.getString(ProducerConfig.METRICS_RECORDING_LEVEL_CONFIG)))
                    .tags(metricTags);
List<MetricsReporter> reporters = config.getConfiguredInstances(ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG, MetricsReporter.class);
reporters.add(new JmxReporter(JMX_PREFIX));
this.metrics = new Metrics(metricConfig, reporters, time);

Producer主要用Metrics来度量和统计"produce-throttle-time"的相关指标。

public static Sensor throttleTimeSensor(SenderMetricsRegistry metrics) {
        Sensor produceThrottleTimeSensor = metrics.sensor("produce-throttle-time");
        produceThrottleTimeSensor.add(metrics.produceThrottleTimeAvg, new Avg());
        produceThrottleTimeSensor.add(metrics.produceThrottleTimeMax, new Max());
        return produceThrottleTimeSensor;
    }

如上，metrics首先注册了名为“produce-throttle-time”的sensor。然后给这个sensor加了两个指标，分别是produceThrottleTimeAvg（平均值）和produceThrottleTimeMax（最大值）。这两个指标对应的度量方法分别是Avg的实例对象和Max的实例对象。
什么触发这些指标的统计呢？答案是在客户端收到发送消息的Response后。如下：

throttleTimeSensor.record(responseBody.get(CommonFields.THROTTLE_TIME_MS), now);

这个record方法解析如下：

public void record(double value, long timeMs, boolean checkQuotas) {
        if (shouldRecord()) {
            this.lastRecordTime = timeMs;
//          线程安全
            synchronized (this) {
//          遍历所有stat，这里对应的是上文的Avg和Max
                for (Stat stat : this.stats)
                    stat.record(config, value, timeMs);
                if (checkQuotas)
                    checkQuotas(timeMs);
            }
            for (Sensor parent : parents)
                parent.record(value, timeMs, checkQuotas);
        }
    }

Avg和Max都继承了SampledStat的record()方法。

public void record(MetricConfig config, double value, long timeMs) {
        Sample sample = current(timeMs);
        if (sample.isComplete(timeMs, config))
            sample = advance(config, timeMs);
//      这里的update就由各子类单独实现。
        update(sample, config, value, timeMs);
        sample.eventCount += 1;
    }
// Avg
@Override
    protected void update(Sample sample, MetricConfig config, double value, long now) {
//      很简单，先求和
        sample.value += value;
    }
//  Max
@Override
    protected void update(Sample sample, MetricConfig config, double value, long now) {
//      直接取最大值    
        sample.value = Math.max(sample.value, value);
    }

最后这两个指标的计算会由JmxReporter调用，最终的计算逻辑在SampledStat的combine()方法中。指标值最终会呈现在jmx中。

@Override
    public double measure(MetricConfig config, long now) {
        purgeObsoleteSamples(config, now);
//      measure()方法调用combine()方法
        return combine(this.samples, config, now);
    }

// Avg
@Override
    public double combine(List<Sample> samples, MetricConfig config, long now) {
        double total = 0.0;
        long count = 0;
        for (Sample s : samples) {
            total += s.value;
            count += s.eventCount;
        }
        return count == 0 ? 0 : total / count;
    }

// Max
@Override
    public double combine(List<Sample> samples, MetricConfig config, long now) {
        double max = Double.NEGATIVE_INFINITY;
        for (Sample sample : samples)
            max = Math.max(max, sample.value);
        return max;
    }
上一篇下一篇

猜你喜欢

热点阅读