如何聚合来自千分尺计数器数组的值答案

【问题标题】：how can I aggregate the values from an array of micrometer counters如何聚合来自千分尺计数器数组的值
【发布时间】：2021-10-21 11:32:23
【问题描述】：

我正在为一个 kafka 流应用程序收集一系列指标，我遇到的问题是我想要一个特定名称的计量器的综合值。为了让这一点更清楚一点，这些指标显示为一个包含 n 个项目的数组，其中 n 是为 Kafka Streams 应用程序配置的线程数。我必须补充一点，我可以通过添加“FunctionCounter”总数的总和来合并这些值。然而，当我感兴趣的仪表更新时，我没有触发“聚合器方法”的机制。聚合器方法复制如下。

private Double aggregateValues(String idName){
    return meterRegistry.getMeters().stream()
        .filter(meter -> meter.getId().getName().startsWith(idName))
        .filter(FunctionCounter.class::isInstance)
        .map(FunctionCounter.class::cast)
        .mapToDouble(FunctionCounter::count)
        .sum();
}

我创建了一个配置 bean 来尝试同样的方法，没有乐趣

@Configuration
@Component
@Slf4j
public class Metrics {

    @Bean
    public FunctionCounter getAggregateCounter(MeterRegistry registry) {
        List<FunctionCounter> counters = registry.getMeters().stream().
            filter(meter -> meter.getId().getName().
                startsWith("Output_Message_Count"))
            .filter( FunctionCounter.class::isInstance )
            .map(FunctionCounter.class::cast)
            .collect(Collectors.toList());

        FunctionCounter counter = FunctionCounter
                .builder("Combined_Output_Message_Count", counters, state -> state.stream().mapToDouble(FunctionCounter::count).sum())
                .description("a description of what this counter does")
                .tags("region", "test")
                .register(registry);
        return counter;
    }
}

下面列出的执行器/prometheus 端点操作系统的原始数据输出示例

# HELP kafka_stream_thread_task_created_total The total number of newly created tasks
# TYPE kafka_stream_thread_task_created_total counter
kafka_stream_thread_task_created_total{kafka_version="2.7.1",spring_id="stream-builder-process",thread_id="sainsburys.applications.sc-dis.price-specification-acl-e0e5af91-ce55-4e0c-998d-269b9c6bade0-StreamThread-4",} 5.0
kafka_stream_thread_task_created_total{kafka_version="2.7.1",spring_id="stream-builder-process",thread_id="sainsburys.applications.sc-dis.price-specification-acl-e0e5af91-ce55-4e0c-998d-269b9c6bade0-StreamThread-3",} 5.0
kafka_stream_thread_task_created_total{kafka_version="2.7.1",spring_id="stream-builder-process",thread_id="sainsburys.applications.sc-dis.price-specification-acl-e0e5af91-ce55-4e0c-998d-269b9c6bade0-StreamThread-2",} 5.0
kafka_stream_thread_task_created_total{kafka_version="2.7.1",spring_id="stream-builder-process",thread_id="sainsburys.applications.sc-dis.price-specification-acl-e0e5af91-ce55-4e0c-998d-269b9c6bade0-StreamThread-1",} 5.0

最终目标是有一个计量表，在更新总数时合并它们，在上面显示的原始数据的情况下，20 在以下提案中已经讨论过这项工作，但它并没有超出最初的提案阶段该提案可以在以下位置查看。以下网址： https://cwiki.apache.org/confluence/display/KAFKA/KIP-674%3A+Metric+Reporter+to+Aggregate+Metrics+in+Kafka+Streams

【问题讨论】：

标签： java spring-cloud-stream confluent-platform micrometer spring-micrometer

【解决方案1】：

这种聚合应该发生在报告方。例如，在 Prometheus（您似乎正在使用）中，您将编写一个自动聚合它们的查询：

kafka_stream_thread_task_created_total

或者，如果您的最终目标是丢弃 thread_id 这样的标签，那么您可以使用 MeterFilter 来丢弃该标签，以便将所有这些计数器视为同一个计数器。

【讨论】：

【解决方案2】：

我建议写一个Prometheus级别的查询，而不是写java代码。

【讨论】：