[fix][monitor] topic/subscription with slash breaks the prometheus fo…

…rmat (#187)

(cherry picked from commit 7db320b)

 Conflicts:
	pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java

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

@@ -25,6 +25,7 @@
 import java.io.OutputStream;
 import java.util.Enumeration;
 import java.util.List;
+import java.util.regex.Pattern;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
 import org.apache.pulsar.common.util.SimpleTextOutputStream;
@@ -34,6 +35,7 @@
  * Format specification can be found at {@link https://prometheus.io/docs/instrumenting/exposition_formats/}
  */
 public class PrometheusMetricsGeneratorUtils {
+    private static final Pattern METRIC_LABEL_VALUE_SPECIAL_CHARACTERS = Pattern.compile("[\\\\\"\\n]");
 
     public static void generate(String cluster, OutputStream out,
                                 List<PrometheusRawMetricsProvider> metricsProviders)
@@ -68,17 +70,14 @@ public static void generateSystemMetrics(SimpleTextOutputStream stream, String c
                 stream.write(sample.name);
                 stream.write("{");
                 if (!sample.labelNames.contains("cluster")) {
-                    stream.write("cluster=\"").write(cluster).write('"');
+                    stream.write("cluster=\"").write(writeEscapedLabelValue(cluster)).write('"');
                     // If label is empty, should not append ','.
                     if (!CollectionUtils.isEmpty(sample.labelNames)){
                         stream.write(",");
                     }
                 }
                 for (int j = 0; j < sample.labelNames.size(); j++) {
-                    String labelValue = sample.labelValues.get(j);
-                    if (labelValue != null) {
-                        labelValue = labelValue.replace("\"", "\\\"");
-                    }
+                    String labelValue = writeEscapedLabelValue(sample.labelValues.get(j));
                     if (j > 0) {
                         stream.write(",");
                     }
@@ -119,5 +118,41 @@ static String getTypeStr(Collector.Type type) {
         }
     }
 
+
+    /**
+     * Write a label value to the writer, escaping backslashes, double quotes and newlines.
+     * See Promethues Exporter io.prometheus.client.exporter.common.TextFormat#writeEscapedLabelValue
+     */
+    public static String writeEscapedLabelValue(String s) {
+        if (s == null) {
+            return null;
+        }
+        if (!labelValueNeedsEscape(s)) {
+            return s;
+        }
+        StringBuilder writer = new StringBuilder();
+        for (int i = 0; i < s.length(); i++) {
+            char c = s.charAt(i);
+            switch (c) {
+                case '\\':
+                    writer.append("\\\\");
+                    break;
+                case '\"':
+                    writer.append("\\\"");
+                    break;
+                case '\n':
+                    writer.append("\\n");
+                    break;
+                default:
+                    writer.append(c);
+            }
+        }
+        return writer.toString();
+    }
+
+    static boolean labelValueNeedsEscape(String s) {
+        return METRIC_LABEL_VALUE_SPECIAL_CHARACTERS.matcher(s).find();
+    }
+
 }
 

pulsar-broker-common/src/test/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGeneratorUtilsTest.java

@@ -99,4 +99,26 @@ public void testGenerateSystemMetricsWithoutCustomizedLabel() throws Exception {
     private static String randomString(){
         return UUID.randomUUID().toString().replaceAll("-", "");
     }
+
+
+    @Test
+    public void testWriteEscapedLabelValue() throws Exception {
+        assertEquals(PrometheusMetricsGeneratorUtils.writeEscapedLabelValue(null), null);
+        assertEquals(PrometheusMetricsGeneratorUtils.writeEscapedLabelValue(""), "");
+        assertEquals(PrometheusMetricsGeneratorUtils.writeEscapedLabelValue("ok"), "ok");
+        assertEquals(PrometheusMetricsGeneratorUtils.writeEscapedLabelValue("ok_!234567890!£$%&/()"),
+                "ok_!234567890!£$%&/()");
+        assertEquals(PrometheusMetricsGeneratorUtils.writeEscapedLabelValue("repl\"\\\n"),
+                "repl\\\"\\\\\\n");
+    }
+    @Test
+    public void testWriteEscapedLabelValuePattern() throws Exception {
+        assertFalse(PrometheusMetricsGeneratorUtils.labelValueNeedsEscape(""));
+        assertFalse(PrometheusMetricsGeneratorUtils.labelValueNeedsEscape("ok"));
+        assertFalse(PrometheusMetricsGeneratorUtils.labelValueNeedsEscape("ok_!234567890!£$%&/()"));
+        assertTrue(PrometheusMetricsGeneratorUtils.labelValueNeedsEscape("repl\"\\\n"));
+        assertTrue(PrometheusMetricsGeneratorUtils.labelValueNeedsEscape("repl\""));
+        assertTrue(PrometheusMetricsGeneratorUtils.labelValueNeedsEscape("repl\\"));
+        assertTrue(PrometheusMetricsGeneratorUtils.labelValueNeedsEscape("\nrepl"));
+    }
 }

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

@@ -41,10 +41,7 @@ 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) {
-            String labelValue = labelsAndValuesArray[i + 1];
-            if (labelValue != null) {
-                labelValue = labelValue.replace("\"", "\\\"");
-            }
+            String labelValue = PrometheusMetricsGeneratorUtils.writeEscapedLabelValue(labelsAndValuesArray[i + 1]);
             stream.write(labelsAndValuesArray[i]).write("=\"").write(labelValue).write('\"');
             if (i + 2 != labelsAndValuesArray.length) {
                 stream.write(',');

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

@@ -298,9 +298,12 @@ private static void parseMetricsToPrometheusMetrics(Collection<Metrics> metrics,
                     if (metric.getKey().isEmpty() || "cluster".equals(metric.getKey())) {
                         continue;
                     }
-                    stream.write(", ").write(metric.getKey()).write("=\"").write(metric.getValue()).write('"');
+                    final String metricValue = PrometheusMetricsGeneratorUtils
+                            .writeEscapedLabelValue(metric.getValue());
+                    stream.write(", ").write(metric.getKey()).write("=\"").write(metricValue).write('"');
                     if (value != null && !value.isEmpty() && !appendedQuantile) {
-                        stream.write(", ").write("quantile=\"").write(value).write('"');
+                        stream.write(", ").write("quantile=\"").write(PrometheusMetricsGeneratorUtils
+                                .writeEscapedLabelValue(value)).write('"');
                         appendedQuantile = true;
                     }
                 }

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

@@ -135,4 +135,5 @@ private static void writeSum(Writer w, DataSketchesOpStatsLogger opStat, String
                 .append(success.toString()).append("\"} ")
                 .append(Double.toString(opStat.getSum(success))).append('\n');
     }
+
 }

pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java

@@ -1966,6 +1966,10 @@ public String toString() {
         }
     }
 
+    /**
+     * Test both subscription and topic name with special characters.
+     * @throws Exception
+     */
     @Test
     public void testEscapeLabelValue() throws Exception {
         String ns1 = "prop/ns-abc1";
@@ -1975,7 +1979,7 @@ public void testEscapeLabelValue() throws Exception {
 
         @Cleanup
         final Consumer<?> consumer = pulsarClient.newConsumer()
-                .subscriptionName("sub")
+                .subscriptionName("s\"ub\\")
                 .topic(topic)
                 .subscribe();
         @Cleanup
@@ -1984,12 +1988,13 @@ public void testEscapeLabelValue() throws Exception {
                 false, statsOut);
         String metricsStr = statsOut.toString();
         final List<String> subCountLines = metricsStr.lines()
-                .filter(line -> line.startsWith("pulsar_subscriptions_count"))
+                .filter(line -> line.startsWith("pulsar_subscription_msg_drop_rate"))
                 .collect(Collectors.toList());
         System.out.println(subCountLines);
         assertEquals(subCountLines.size(), 1);
         assertEquals(subCountLines.get(0),
-                "pulsar_subscriptions_count{cluster=\"test\",namespace=\"prop/ns-abc1\",topic=\"persistent://prop/ns-abc1/\\\"mytopic\"} 1");
+                "pulsar_subscription_msg_drop_rate{cluster=\"test\",namespace=\"prop/ns-abc1\","
+                        + "topic=\"persistent://prop/ns-abc1/\\\"mytopic\",subscription=\"s\\\"ub\\\\\"} 0.0");
     }
 
 }

pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchMetrics.java

@@ -1,4 +1,4 @@
-/*
+/**
  * 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

pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClientTests.java

@@ -20,7 +20,6 @@
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
-import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;