RocketMQ Promethus Exporter

Introduction

Rocketmq-exporter is a system for monitoring all relevant metrics of the RocketMQ broker and client sides, which packages the metric values obtained from the broker side through mqAdmin into 87 caches.

caution

In previous versions, there were 87 concurrentHashMaps, but since the Map does not delete expired metrics, once there is a label change, a new metric is generated and the old, unused metric cannot be automatically deleted, which eventually causes a memory overflow. However, using the Cache structure can enable expired deletion, and the expiration time can be configured.

The process for Rocketmq-exporter to obtain monitoring metrics is shown in the following figure. The exporter requests data from the MQ cluster through MQAdminExt, and the requested data is standardized into the format required by Prometheus through the MetricService, and then exposed to Promethus through the /metrics interface.

Metric structure

The Metric class is located in the org.apache.rocketmq.expoter.model.metrics package, and is essentially a set of entity classes, with each entity class representing a type of metric, for a total of 14 Metric classes. These classes serve as the keys for the 87 caches and are distinguished by different label values.

The entity classes contain three dimensions of labels：broker、consumer、producer

• Metric classes related to the broker : BrokerRuntimeMetric、BrokerMetric、DLQTopicOffsetMetric、TopicPutNumMetric
• Consumer-related classes : ConsumerRuntimeConsumeFailedMsgsMetric 、ConsumerRuntimeConsumeFailedTPSMetric 、ConsumerRuntimeConsumeOKTPSMetric、ConsumerRuntimeConsumeRTMetric、ConsumerRuntimePullRTMetric、ConsumerRuntimePullTPSMetric、ConsumerCountMetric、ConsumerMetric、ConsumerTopicDiffMetric
• Producer-related metric classes: ProducerMetric

Prometheus pulls metrics

The RocketMQ-exporter project and Prometheus are equivalent to the server-client relationship, where the RocketMQ-exporter project introduces the Prometheus client package, which specifies the type of information to be obtained in the project’s MetricFamilySamples class. Prometheus requests metrics from exporter, and exporter returns the information to Prometheus after packaging it into the corresponding type.

After the rocketmq-exporter project is started, it will collect various metrics from rocketmq into the mfs object. When the browser or Prometheus accesses the corresponding interface, the samples in the mfs object will be generated into the formatted data supported by Prometheus through the service. It mainly includes the following steps:

The browser accesses ip:5557/metrics to call the metrics method in the RMQMetricsController class, where ip is the IP of the host where the rocketmq-exporter project is running.

private void metrics(HttpServletResponse response) throws IOException {
    StringWriter writer = new StringWriter();
    metricsService.metrics(writer);
    response.setHeader("Content-Type", "text/plain; version=0.0.4; charset=utf-8");
    response.getOutputStream().print(writer.toString());
}

By creating a new StringWriter object to collect metric indicators, the metrics in the exporter are collected into the writer object through the metrics method in the MetricsService class, and then the collected indicators are output to the webpage.

The format of the collected metrics is:

<metric name>{<label name>=<label value>, ...} <metric value>

Example：

rocketmq_group_diff{group="rmq_group_test_20220114",topic="fusion_console_tst",countOfOnlineConsumers="0",msgModel="1",} 23.0

The 5 scheduled tasks in the MetricCollectTask class

The MetricCollectTask class has five scheduled tasks: collectTopicOffset, collectConsumerOffset, collectBrokerStatsTopic, collectBrokerStats, and collectBrokerRuntimeStats. They are used to collect consumer offset information and Broker state information, etc. Its cron expression is: cron: 15 0/1 * * * ?, which means it will collect once per minute. Its core function is to obtain information from the broker in the cluster through the mqAdminExt object and then add it to the corresponding 87 monitoring indicators, taking collectTopicOffset as an example:

1. First, initialize the TopicList object and obtain all topic information in the cluster through the mqAdminExt.fetchAllTopicList() method.

   TopicList topicList = null;
   try {  
       topicList = mqAdminExt.fetchAllTopicList();
   } catch (Exception ex) {
           log.error(String.format("collectTopicOffset-exception comes getting topic list from namesrv, address is %s",
               JSON.toJSONString(mqAdminExt.getNameServerAddressList())));
           return;
   }

2. Add the topic to the topicSet, and iterate through each topic, checking the topic status through the mqAdminExt.examineTopicStats(topic) function.

   Set < String > topicSet = topicList != null ? topicList.getTopicList() : null;
   for (String topic: topicSet) {
        TopicStatsTable topicStats = null;
        try {
            topicStats = mqAdminExt.examineTopicStats(topic);
        } catch (Exception ex) {
            log.error(String.format("collectTopicOffset-getting topic(%s) stats error. the namesrv address is %s",
                topic,
                JSON.toJSONString(mqAdminExt.getNameServerAddressList())));
            continue;}

3. Initialize the topic status set, the hash table brokerOffsetMap for topic information offset divided by broker, and a hash table brokerUpdateTimestampMap with broker name as the key to store the update timestamp.

   Set<Map.Entry<MessageQueue, TopicOffset>> topicStatusEntries = topicStats.getOffsetTable().entrySet();
   HashMap<String, Long> brokerOffsetMap = new HashMap<>();
   HashMap<String, Long> brokerUpdateTimestampMap = new HashMap<>();
   for (Map.Entry<MessageQueue, TopicOffset> topicStatusEntry : topicStatusEntries) {
       MessageQueue q = topicStatusEntry.getKey();
       TopicOffset offset = topicStatusEntry.getValue();
       if (brokerOffsetMap.containsKey(q.getBrokerName())) {
           brokerOffsetMap.put(q.getBrokerName(), brokerOffsetMap.get(q.getBrokerName()) + offset.getMaxOffset());
       } else {
           brokerOffsetMap.put(q.getBrokerName(), offset.getMaxOffset());
       }
       if (brokerUpdateTimestampMap.containsKey(q.getBrokerName())) {
           if (offset.getLastUpdateTimestamp() > brokerUpdateTimestampMap.get(q.getBrokerName())) {
               brokerUpdateTimestampMap.put(q.getBrokerName(), offset.getLastUpdateTimestamp());
           }
       } else {
           brokerUpdateTimestampMap.put(q.getBrokerName(),
           offset.getLastUpdateTimestamp());
       }
   }

4. Finally, by iterating through each item in the brokerOffsetMap, the metricCollector object is obtained through the metricsService and the addTopicOffsetMetric method in the RMQMetricsCollector class is called to add the corresponding value to one of the caches of the 87 metrics in the RMQMetricsCollector class.

    Set<Map.Entry<String, Long>> brokerOffsetEntries = brokerOffsetMap.entrySet();
           for (Map.Entry<String, Long> brokerOffsetEntry : brokerOffsetEntries) {
               metricsService.getCollector().addTopicOffsetMetric(clusterName, brokerOffsetEntry.getKey(), topic,
                   brokerUpdateTimestampMap.get(brokerOffsetEntry.getKey()), brokerOffsetEntry.getValue());
           }
       }
       log.info("topic offset collection task finished...." + (System.currentTimeMillis() - start));
   }

Rocketmq-exporter collects metrics flowchart

Quick star

Configure application.yml

Important configurations in application.yml include:

• server.port sets the port that Prometheus listens to for the rocketmq-exporter, with a default value of 5557.

• rocketmq.config.webTelemetryPath configures the path for Prometheus to obtain metrics, with a default value of /metrics. The default value can be used.

• rocketmq.config.enableACL If the RocketMQ cluster has enabled ACL verification, it needs to be set to true and the corresponding ak and sk need to be configured in accessKey and secretKey.

• rocketmq.config.outOfTimeSeconds is used to configure the expiration time of storing metrics and their values. If it exceeds this time and the key in the cache has not undergone a write change, it will be deleted. Generally, it can be configured as 60s (the time interval for Prometheus to obtain metrics should be reasonably configured according to the expiration time, as long as the expiration time is greater than or equal to the time interval for Prometheus to collect metrics).

• task..cron configures the time interval for the exporter to pull metrics from the broker through a scheduled task, with a default value of “15 0/1 * * ?” which means it will pull metrics every 15s of every minute.

Start exporter application

Start Prometheus according to the configuration on its official website

Configure Prometheus’s static_config: -targets to the exporter’s starting IP and port, such as: localhost:5557.

Access the Prometheus page

If the localhost starts at the default localhost:9090, you can view the collected metric values, as shown in the following figure:

tip

For better visualization effects and to observe the trend of metric value changes, Prometheus is better used with Grafana!

Observability metrics

Observability metrics mainly include two categories: server-side metrics and client-side metrics. Server-side metrics are directly generated by the server, and client-side metrics are generated on the client and obtained by the server through an RPC request to the client. Client-side metrics can be further divided into producer metrics and consumer metrics. All 87 observability metrics and their main meanings are as follows:

Server metrics

Server metrics

Producer metrics

Producer metrics

Consumer metrics

Consumer metrics