[fix][broker][functions-worker] Ensure prometheus metrics are grouped by type (#8407, #13865)

pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStats.java

@@ -96,7 +96,7 @@ void updateStats(TopicStats stats) {
 
         stats.replicationStats.forEach((n, as) -> {
             AggregatedReplicationStats replStats =
-                    replicationStats.computeIfAbsent(n,  k -> new AggregatedReplicationStats());
+                    replicationStats.computeIfAbsent(n, k -> new AggregatedReplicationStats());
             replStats.msgRateIn += as.msgRateIn;
             replStats.msgRateOut += as.msgRateOut;
             replStats.msgThroughputIn += as.msgThroughputIn;

pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricStreams.java

@@ -0,0 +1,75 @@
+/**
+ * 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 org.apache.pulsar.broker.stats.prometheus;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
+import org.apache.pulsar.common.util.SimpleTextOutputStream;
+
+/**
+ * Helper class to ensure that metrics of the same name are grouped together under the same TYPE header when written.
+ * Those are the requirements of the
+ * <a href="https://github.com/prometheus/docs/blob/master/content/docs/instrumenting/exposition_formats.md#grouping-and-sorting">Prometheus Exposition Format</a>.
+ */
+public class PrometheusMetricStreams {
+    private final Map<String, SimpleTextOutputStream> metricStreamMap = new HashMap<>();
+
+    /**
+     * Write the given metric and sample value to the stream. Will write #TYPE header if metric not seen before.
+     * @param metricName name of the metric.
+     * @param value value of the sample
+     * @param labelsAndValuesArray varargs of label and label value
+     */
+    void writeSample(String metricName, Number value, String... labelsAndValuesArray) {
+        SimpleTextOutputStream stream = initGaugeType(metricName);
+        stream.write(metricName).write('{');
+        for (int i = 0; i < labelsAndValuesArray.length; i += 2) {
+            stream.write(labelsAndValuesArray[i]).write("=\"").write(labelsAndValuesArray[i + 1]).write('\"');
+            if (i + 2 != labelsAndValuesArray.length) {
+                stream.write(',');
+            }
+        }
+        stream.write("} ").write(value).write(' ').write(System.currentTimeMillis()).write('\n');
+    }
+
+    /**
+     * Flush all the stored metrics to the supplied stream.
+     * @param stream the stream to write to.
+     */
+    void flushAllToStream(SimpleTextOutputStream stream) {
+        metricStreamMap.values().forEach(s -> stream.write(s.getBuffer()));
+    }
+
+    /**
+     * Release all the streams to clean up resources.
+     */
+    void releaseAll() {
+        metricStreamMap.values().forEach(s -> s.getBuffer().release());
+        metricStreamMap.clear();
+    }
+
+    private SimpleTextOutputStream initGaugeType(String metricName) {
+        return metricStreamMap.computeIfAbsent(metricName, s -> {
+            SimpleTextOutputStream stream = new SimpleTextOutputStream(PulsarByteBufAllocator.DEFAULT.directBuffer());
+            stream.write("# TYPE ").write(metricName).write(" gauge\n");
+            return stream;
+        });
+    }
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java

@@ -52,7 +52,8 @@
 /**
  * Generate metrics aggregated at the namespace level and optionally at a topic level and formats them out
  * in a text format suitable to be consumed by Prometheus.
- * Format specification can be found at {@link https://prometheus.io/docs/instrumenting/exposition_formats/}
+ * Format specification can be found at <a
+ * href="https://prometheus.io/docs/instrumenting/exposition_formats/">Exposition Formats</a>
  */
 public class PrometheusMetricsGenerator {
 
@@ -86,38 +87,44 @@ public double get() {
     }
 
     public static void generate(PulsarService pulsar, boolean includeTopicMetrics, boolean includeConsumerMetrics,
-        boolean includeProducerMetrics, OutputStream out) throws IOException {
+                                boolean includeProducerMetrics, OutputStream out) throws IOException {
         generate(pulsar, includeTopicMetrics, includeConsumerMetrics, includeProducerMetrics, false, out, null);
     }
 
     public static void generate(PulsarService pulsar, boolean includeTopicMetrics, boolean includeConsumerMetrics,
-        boolean includeProducerMetrics, boolean splitTopicAndPartitionIndexLabel,
-        OutputStream out) throws IOException {
+                                boolean includeProducerMetrics, boolean splitTopicAndPartitionIndexLabel,
+                                OutputStream out) throws IOException {
         generate(pulsar, includeTopicMetrics, includeConsumerMetrics, includeProducerMetrics,
                 splitTopicAndPartitionIndexLabel, out, null);
     }
 
     public static void generate(PulsarService pulsar, boolean includeTopicMetrics, boolean includeConsumerMetrics,
-        boolean includeProducerMetrics, boolean splitTopicAndPartitionIndexLabel, OutputStream out,
-        List<PrometheusRawMetricsProvider> metricsProviders)
-        throws IOException {
+                                boolean includeProducerMetrics, boolean splitTopicAndPartitionIndexLabel,
+                                OutputStream out,
+                                List<PrometheusRawMetricsProvider> metricsProviders)
+            throws IOException {
         ByteBuf buf = ByteBufAllocator.DEFAULT.heapBuffer();
+        boolean exceptionHappens = false;
+        //Used in namespace/topic and transaction aggregators as share metric names
+        PrometheusMetricStreams metricStreams = new PrometheusMetricStreams();
         try {
             SimpleTextOutputStream stream = new SimpleTextOutputStream(buf);
 
             generateSystemMetrics(stream, pulsar.getConfiguration().getClusterName());
 
             NamespaceStatsAggregator.generate(pulsar, includeTopicMetrics, includeConsumerMetrics,
-                    includeProducerMetrics, splitTopicAndPartitionIndexLabel, stream);
+                    includeProducerMetrics, splitTopicAndPartitionIndexLabel, metricStreams);
 
             if (pulsar.getWorkerServiceOpt().isPresent()) {
                 pulsar.getWorkerService().generateFunctionsStats(stream);
             }
 
             if (pulsar.getConfiguration().isTransactionCoordinatorEnabled()) {
-                TransactionAggregator.generate(pulsar, stream, includeTopicMetrics);
+                TransactionAggregator.generate(pulsar, metricStreams, includeTopicMetrics);
             }
 
+            metricStreams.flushAllToStream(stream);
+
             generateBrokerBasicMetrics(pulsar, stream);
 
             generateManagedLedgerBookieClientMetrics(pulsar, stream);
@@ -129,7 +136,12 @@ public static void generate(PulsarService pulsar, boolean includeTopicMetrics, b
             }
             out.write(buf.array(), buf.arrayOffset(), buf.readableBytes());
         } finally {
-            buf.release();
+            //release all the metrics buffers
+            metricStreams.releaseAll();
+            //if exception happens, release buffer
+            if (exceptionHappens) {
+                buf.release();
+            }
         }
     }
 
@@ -142,17 +154,17 @@ private static void generateBrokerBasicMetrics(PulsarService pulsar, SimpleTextO
         if (pulsar.getConfiguration().isExposeManagedLedgerMetricsInPrometheus()) {
             // generate managedLedger metrics
             parseMetricsToPrometheusMetrics(new ManagedLedgerMetrics(pulsar).generate(),
-                clusterName, Collector.Type.GAUGE, stream);
+                    clusterName, Collector.Type.GAUGE, stream);
         }
 
         if (pulsar.getConfiguration().isExposeManagedCursorMetricsInPrometheus()) {
             // generate managedCursor metrics
             parseMetricsToPrometheusMetrics(new ManagedCursorMetrics(pulsar).generate(),
-                clusterName, Collector.Type.GAUGE, stream);
+                    clusterName, Collector.Type.GAUGE, stream);
         }
 
         parseMetricsToPrometheusMetrics(Collections.singletonList(pulsar.getBrokerService()
-                .getPulsarStats().getBrokerOperabilityMetrics().generateConnectionMetrics()),
+                        .getPulsarStats().getBrokerOperabilityMetrics().generateConnectionMetrics()),
                 clusterName, Collector.Type.GAUGE, stream);
 
         // generate loadBalance metrics
@@ -267,17 +279,17 @@ private static void generateSystemMetrics(SimpleTextOutputStream stream, String
 
     static String getTypeStr(Collector.Type type) {
         switch (type) {
-        case COUNTER:
-            return "counter";
-        case GAUGE:
-            return "gauge";
-        case SUMMARY        :
-            return "summary";
-        case HISTOGRAM:
-            return "histogram";
-        case UNTYPED:
-        default:
-            return "untyped";
+            case COUNTER:
+                return "counter";
+            case GAUGE:
+                return "gauge";
+            case SUMMARY:
+                return "summary";
+            case HISTOGRAM:
+                return "histogram";
+            case UNTYPED:
+            default:
+                return "untyped";
         }
     }