[SCB-1240] allowed one prometheus cluster support multiple apps

Author: wujimin

core/src/test/java/org/apache/servicecomb/core/executor/TestThreadPoolExecutorEx.java

@@ -21,11 +21,16 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.function.IntSupplier;
 
 import org.junit.Assert;
 import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class TestThreadPoolExecutorEx {
+  private static final Logger LOGGER = LoggerFactory.getLogger(TestThreadPoolExecutorEx.class);
+
   static class TestTask implements Runnable {
     CountDownLatch notify = new CountDownLatch(1);
 
@@ -119,22 +124,36 @@ public void schedule() throws ExecutionException, InterruptedException {
     t3.quit();
     Assert.assertEquals(4, executorEx.getPoolSize());
     Assert.assertEquals(1, executorEx.getRejectedCount());
-    Assert.assertEquals(3, executorEx.getNotFinished());
-    // multi thread, not sure
-    // Assert.assertEquals(0, executorEx.getQueue().size());
+    waitForResult(3, executorEx::getNotFinished);
+    waitForResult(0, executorEx.getQueue()::size);
 
     // reuse thread
     t3 = submitTask();
     Assert.assertEquals(4, executorEx.getPoolSize());
     Assert.assertEquals(1, executorEx.getRejectedCount());
-    Assert.assertEquals(4, executorEx.getNotFinished());
-    // multi thread, not sure
-    // Assert.assertEquals(1, executorEx.getQueue().size());
+    waitForResult(4, executorEx::getNotFinished);
+    waitForResult(0, executorEx.getQueue()::size);
 
     t3.quit();
     t4.quit();
     t5.quit();
     t6.quit();
     executorEx.shutdown();
   }
+
+  private void waitForResult(int expect, IntSupplier supplier) {
+    for (; ; ) {
+      int actual = supplier.getAsInt();
+      if (expect == actual) {
+        return;
+      }
+
+      LOGGER.info("waiting for thread result, expect:{}, actual: {}.", expect, actual);
+      try {
+        TimeUnit.MILLISECONDS.sleep(100);
+      } catch (InterruptedException e) {
+        throw new IllegalStateException(e);
+      }
+    }
+  }
 }

demo/demo-springmvc/springmvc-client/src/main/java/org/apache/servicecomb/demo/springmvc/client/SpringmvcClient.java

@@ -131,13 +131,13 @@ public static void run() {
       String content = restTemplate
           .getForObject("cse://springmvc/codeFirstSpringmvc/prometheusForTest", String.class);
 
-      TestMgr.check(true, content.contains("servicecomb_invocation{operation=\"springmvc.codeFirst.addDate"));
-      TestMgr.check(true, content.contains("servicecomb_invocation{operation=\"springmvc.codeFirst.sayHello"));
-      TestMgr.check(true, content.contains("servicecomb_invocation{operation=\"springmvc.codeFirst.fallbackFromCache"));
-      TestMgr.check(true, content.contains("servicecomb_invocation{operation=\"springmvc.codeFirst.isTrue"));
-      TestMgr.check(true, content.contains("servicecomb_invocation{operation=\"springmvc.codeFirst.add"));
-      TestMgr.check(true, content.contains("servicecomb_invocation{operation=\"springmvc.codeFirst.sayHi2"));
-      TestMgr.check(true, content.contains("servicecomb_invocation{operation=\"springmvc.codeFirst.saySomething"));
+      TestMgr.check(true, content.contains("servicecomb_invocation{appId=\"springmvctest\",operation=\"springmvc.codeFirst.addDate"));
+      TestMgr.check(true, content.contains("servicecomb_invocation{appId=\"springmvctest\",operation=\"springmvc.codeFirst.sayHello"));
+      TestMgr.check(true, content.contains("servicecomb_invocation{appId=\"springmvctest\",operation=\"springmvc.codeFirst.fallbackFromCache"));
+      TestMgr.check(true, content.contains("servicecomb_invocation{appId=\"springmvctest\",operation=\"springmvc.codeFirst.isTrue"));
+      TestMgr.check(true, content.contains("servicecomb_invocation{appId=\"springmvctest\",operation=\"springmvc.codeFirst.add"));
+      TestMgr.check(true, content.contains("servicecomb_invocation{appId=\"springmvctest\",operation=\"springmvc.codeFirst.sayHi2"));
+      TestMgr.check(true, content.contains("servicecomb_invocation{appId=\"springmvctest\",operation=\"springmvc.codeFirst.saySomething"));
 
       String[] metricLines = content.split("\n");
       if (metricLines.length > 0) {

metrics/metrics-integration/metrics-prometheus/src/main/java/org/apache/servicecomb/metrics/prometheus/PrometheusPublisher.java

@@ -25,6 +25,7 @@
 import org.apache.servicecomb.foundation.metrics.MetricsBootstrapConfig;
 import org.apache.servicecomb.foundation.metrics.MetricsInitializer;
 import org.apache.servicecomb.foundation.metrics.registry.GlobalRegistry;
+import org.apache.servicecomb.serviceregistry.RegistryUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -103,6 +104,9 @@ protected Sample convertMeasurementToSample(Measurement measurement) {
     List<String> labelNames = new ArrayList<>();
     List<String> labelValues = new ArrayList<>();
 
+    labelNames.add("appId");
+    labelValues.add(RegistryUtils.getAppId());
+
     for (Tag tag : measurement.id().tags()) {
       labelNames.add(tag.key());
       labelValues.add(tag.value());

metrics/metrics-integration/metrics-prometheus/src/test/java/org/apache/servicecomb/metrics/prometheus/TestPrometheusPublisher.java

@@ -27,11 +27,13 @@
 import org.apache.servicecomb.foundation.common.exceptions.ServiceCombException;
 import org.apache.servicecomb.foundation.metrics.registry.GlobalRegistry;
 import org.apache.servicecomb.foundation.test.scaffolding.config.ArchaiusUtils;
+import org.apache.servicecomb.serviceregistry.RegistryUtils;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
+import org.springframework.util.ResourceUtils;
 
 import com.netflix.spectator.api.Counter;
 import com.netflix.spectator.api.DefaultRegistry;
@@ -40,6 +42,7 @@
 import com.sun.net.httpserver.HttpServer;
 
 import io.prometheus.client.exporter.HTTPServer;
+import mockit.Expectations;
 
 @SuppressWarnings("restriction")
 public class TestPrometheusPublisher {
@@ -81,6 +84,12 @@ public void testBadPublishAddress_TooLargePort() {
 
   @Test
   public void collect() throws IllegalAccessException, IOException {
+    new Expectations(RegistryUtils.class){
+      {
+        RegistryUtils.getAppId();
+        result = "testAppId";
+      }
+    };
     ArchaiusUtils.setProperty(PrometheusPublisher.METRICS_PROMETHEUS_ADDRESS, "localhost:0");
     publisher.init(globalRegistry, null, null);
 
@@ -98,7 +107,7 @@ public void collect() throws IllegalAccessException, IOException {
     try (InputStream is = conn.getInputStream()) {
       Assert.assertEquals("# HELP ServiceComb_Metrics ServiceComb Metrics\n" +
           "# TYPE ServiceComb_Metrics untyped\n" +
-          "count_name{tag1=\"tag1v\",tag2=\"tag2v\",} 1.0\n", IOUtils.toString(is));
+          "count_name{appId=\"testAppId\",tag1=\"tag1v\",tag2=\"tag2v\",} 1.0\n", IOUtils.toString(is));
     }
 
     publisher.destroy();