Java代码使用Prometheus监控，配置Grafana展示

/* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this file to You under the Apache License, Version 2.0 * (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package wiki.hadoop.common.metric; import static wiki.hadoop.common.metric.AgentMetricItem.*; import static wiki.hadoop.common.metric.MetricRegister.JMX_DOMAIN; import java.io.IOException; import java.lang.management.ManagementFactory; import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicLong; import javax.management.MBeanServer; import javax.management.ObjectName; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import cn.hutool.json.JSONUtil; import io.prometheus.client.Collector; import io.prometheus.client.CounterMetricFamily; import io.prometheus.client.exporter.HTTPServer; import io.prometheus.client.hotspot.DefaultExports; /** * prometheus metric listener */ public class AgentPrometheusMetricListener extends Collector implements MetricListener { public static final String DEFAULT_DIMENSION_LABEL = "dimension"; public static final String HYPHEN_SYMBOL = "-"; private static final Logger LOGGER = LoggerFactory.getLogger(AgentPrometheusMetricListener.class); protected HTTPServer httpServer; private AgentMetricItem metricItem; private Map<String, AtomicLong> metricValueMap = new ConcurrentHashMap<>(); private Map<String, MetricItemValue> dimensionMetricValueMap = new ConcurrentHashMap<>(); private List<String> dimensionKeys = new ArrayList<>();//维度key组成的字段列表，即所有监控实体标记@Dimension的字段 public AgentPrometheusMetricListener() { this.metricItem = new AgentMetricItem(); final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer(); StringBuilder beanName = new StringBuilder(); beanName.append(JMX_DOMAIN).append(DOMAIN_SEPARATOR).append("type=AgentPrometheus"); String strBeanName = beanName.toString(); try { ObjectName objName = new ObjectName(strBeanName); mbs.registerMBean(metricItem, objName); } catch (Exception ex) { LOGGER.error("exception while register mbean:{},error:{}", strBeanName, ex); } // prepare metric value map metricValueMap.put(M_JOB_RUNNING_COUNT, metricItem.jobRunningCount); metricValueMap.put(M_JOB_FATAL_COUNT, metricItem.jobFatalCount); metricValueMap.put(M_TASK_RUNNING_COUNT, metricItem.taskRunningCount); metricValueMap.put(M_TASK_RETRYING_COUNT, metricItem.taskRetryingCount); metricValueMap.put(M_TASK_FATAL_COUNT, metricItem.taskFatalCount); metricValueMap.put(M_SINK_SUCCESS_COUNT, metricItem.sinkSuccessCount); metricValueMap.put(M_SINK_FAIL_COUNT, metricItem.sinkFailCount); metricValueMap.put(M_SOURCE_SUCCESS_COUNT, metricItem.sourceSuccessCount); metricValueMap.put(M_SOURCE_FAIL_COUNT, metricItem.sourceFailCount); metricValueMap.put(M_PLUGIN_READ_COUNT, metricItem.pluginReadCount); metricValueMap.put(M_PLUGIN_SEND_COUNT, metricItem.pluginSendCount); metricValueMap.put(M_PLUGIN_READ_FAIL_COUNT, metricItem.pluginReadFailCount); metricValueMap.put(M_PLUGIN_SEND_FAIL_COUNT, metricItem.pluginSendFailCount); metricValueMap.put(M_PLUGIN_READ_SUCCESS_COUNT, metricItem.pluginReadSuccessCount); metricValueMap.put(M_PLUGIN_SEND_SUCCESS_COUNT, metricItem.pluginSendSuccessCount); int metricsServerPort = 19090; try { this.httpServer = new HTTPServer(metricsServerPort); this.register(); LOGGER.info("Starting prometheus metrics server on port {}", metricsServerPort); } catch (IOException e) { LOGGER.error("exception while register agent prometheus http server,error:{}", e.getMessage()); } } /** * 每次通过Prometheus接口调用时，会触发collect方法 * 测试时可手动调用触发 * * @return java.util.List<io.prometheus.client.Collector.MetricFamilySamples> * @name collect * @date 2023/7/4 4:56 PM * @author Jast */ @Override public List<MetricFamilySamples> collect() { DefaultExports.initialize(); // total /** 返回数据样例 * # HELP agent_total metrics_of_agent_node_total * # TYPE agent_total counter * agent_total{dimension="jobRunningCount",} 0.0 * agent_total{dimension="jobFatalCount",} 0.0 * agent_total{dimension="taskRunningCount",} 0.0 * agent_total{dimension="taskRetryingCount",} 0.0 * agent_total{dimension="taskFatalCount",} 0.0 * agent_total{dimension="sinkSuccessCount",} 0.0 * agent_total{dimension="sinkFailCount",} 0.0 * agent_total{dimension="sourceSuccessCount",} 0.0 * agent_total{dimension="sourceFailCount",} 0.0 * agent_total{dimension="pluginReadCount",} 0.0 * agent_total{dimension="pluginSendCount",} 0.0 * agent_total{dimension="pluginReadFailCount",} 50.0 * agent_total{dimension="pluginSendFailCount",} 0.0 * agent_total{dimension="pluginReadSuccessCount",} 50.0 * agent_total{dimension="pluginSendSuccessCount",} 0.0 */ // 在prometheus中命名为agent_total，（_total是CounterMetricFamily自动添加） CounterMetricFamily totalCounter = new CounterMetricFamily("agent", "metrics_of_agent_node_total", Arrays.asList(DEFAULT_DIMENSION_LABEL)); totalCounter.addMetric(Arrays.asList(M_JOB_RUNNING_COUNT), metricItem.jobRunningCount.get()); totalCounter.addMetric(Arrays.asList(M_JOB_FATAL_COUNT), metricItem.jobFatalCount.get()); totalCounter.addMetric(Arrays.asList(M_TASK_RUNNING_COUNT), metricItem.taskRunningCount.get()); totalCounter.addMetric(Arrays.asList(M_TASK_RETRYING_COUNT), metricItem.taskRetryingCount.get()); totalCounter.addMetric(Arrays.asList(M_TASK_FATAL_COUNT), metricItem.taskFatalCount.get()); totalCounter.addMetric(Arrays.asList(M_SINK_SUCCESS_COUNT), metricItem.sinkSuccessCount.get()); totalCounter.addMetric(Arrays.asList(M_SINK_FAIL_COUNT), metricItem.sinkFailCount.get()); totalCounter.addMetric(Arrays.asList(M_SOURCE_SUCCESS_COUNT), metricItem.sourceSuccessCount.get()); totalCounter.addMetric(Arrays.asList(M_SOURCE_FAIL_COUNT), metricItem.sourceFailCount.get()); totalCounter.addMetric(Arrays.asList(M_PLUGIN_READ_COUNT), metricItem.pluginReadCount.get()); totalCounter.addMetric(Arrays.asList(M_PLUGIN_SEND_COUNT), metricItem.pluginSendCount.get()); totalCounter.addMetric(Arrays.asList(M_PLUGIN_READ_FAIL_COUNT), metricItem.pluginReadFailCount.get()); totalCounter.addMetric(Arrays.asList(M_PLUGIN_SEND_FAIL_COUNT), metricItem.pluginSendFailCount.get()); totalCounter.addMetric(Arrays.asList(M_PLUGIN_READ_SUCCESS_COUNT), metricItem.pluginReadSuccessCount.get()); totalCounter.addMetric(Arrays.asList(M_PLUGIN_SEND_SUCCESS_COUNT), metricItem.pluginSendSuccessCount.get()); List<MetricFamilySamples> mfs = new ArrayList<>(); mfs.add(totalCounter); // 返�