prometheus-emitter supports sending metrics to pushgateway regularly … (#13034)

* prometheus-emitter supports sending metrics to pushgateway regularly and continuously

* spell check fix

* Optimization variable name and related documents

* Update docs/development/extensions-contrib/prometheus.md

OK, it looks more conspicuous

Co-authored-by: Frank Chen <frankchen@apache.org>

* Update doc

* Update docs/development/extensions-contrib/prometheus.md

Co-authored-by: Frank Chen <frankchen@apache.org>

* When PrometheusEmitter is closed, close the scheduler

* Ensure that registeredMetrics is thread safe.

* Local variable name optimization

* Remove unnecessary white space characters

Co-authored-by: Frank Chen <frankchen@apache.org>

docs/development/extensions-contrib/prometheus.md

@@ -44,6 +44,7 @@ All the configuration parameters for the Prometheus emitter are under `druid.emi
 | `druid.emitter.prometheus.addHostAsLabel`     | Flag to include the hostname as a prometheus label.                                                                                                                                                                                    | no        | false                                |
 | `druid.emitter.prometheus.addServiceAsLabel`  | Flag to include the druid service name (e.g. `druid/broker`, `druid/coordinator`, etc.) as a prometheus label.                                                                                                                         | no        | false                                |
 | `druid.emitter.prometheus.pushGatewayAddress` | Pushgateway address. Required if using `pushgateway` strategy.                                                                                                                                                                         | no        | none                                 |
+|`druid.emitter.prometheus.flushPeriod`|Emit metrics to Pushgateway every `flushPeriod` seconds. Required if `pushgateway` strategy is used.|no|15|
 
 ### Override properties for Peon Tasks
 

extensions-contrib/prometheus-emitter/src/main/java/org/apache/druid/emitter/prometheus/Metrics.java

@@ -36,6 +36,7 @@ import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.SortedSet;
@@ -45,7 +46,7 @@ public class Metrics
 {
 
   private static final Logger log = new Logger(Metrics.class);
-  private final Map<String, DimensionsAndCollector> registeredMetrics = new HashMap<>();
+  private final Map<String, DimensionsAndCollector> registeredMetrics;
   private final ObjectMapper mapper = new ObjectMapper();
   public static final Pattern PATTERN = Pattern.compile("[^a-zA-Z_:][^a-zA-Z0-9_:]*");
 
@@ -63,6 +64,7 @@ public class Metrics
 
   public Metrics(String namespace, String path, boolean isAddHostAsLabel, boolean isAddServiceAsLabel)
   {
+    Map<String, DimensionsAndCollector> registeredMetrics = new HashMap<>();
     Map<String, Metric> metrics = readConfig(path);
     for (Map.Entry<String, Metric> entry : metrics.entrySet()) {
       String name = entry.getKey();
@@ -110,7 +112,7 @@ public class Metrics
         registeredMetrics.put(name, new DimensionsAndCollector(dimensions, collector, metric.conversionFactor));
       }
     }
-
+    this.registeredMetrics = Collections.unmodifiableMap(registeredMetrics);
   }
 
   private Map<String, Metric> readConfig(String path)

extensions-contrib/prometheus-emitter/src/main/java/org/apache/druid/emitter/prometheus/PrometheusEmitter.java

@@ -27,6 +27,7 @@ import io.prometheus.client.Gauge;
 import io.prometheus.client.Histogram;
 import io.prometheus.client.exporter.HTTPServer;
 import io.prometheus.client.exporter.PushGateway;
+import org.apache.druid.java.util.common.concurrent.ScheduledExecutors;
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.java.util.emitter.core.Emitter;
 import org.apache.druid.java.util.emitter.core.Event;
@@ -36,6 +37,8 @@ import java.io.IOException;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.util.Map;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.regex.Pattern;
 
 /**
@@ -55,6 +58,7 @@ public class PrometheusEmitter implements Emitter
   private HTTPServer server;
   private PushGateway pushGateway;
   private String identifier;
+  private ScheduledExecutorService exec;
 
   static PrometheusEmitter of(PrometheusEmitterConfig config)
   {
@@ -91,6 +95,13 @@ public class PrometheusEmitter implements Emitter
       } else {
         pushGateway = new PushGateway(address);
       }
+      exec = ScheduledExecutors.fixed(1, "PrometheusPushGatewayEmitter-%s");
+      exec.scheduleAtFixedRate(
+          () -> flush(),
+          config.getFlushPeriod(),
+          config.getFlushPeriod(),
+          TimeUnit.SECONDS
+      );
     }
   }
 
@@ -190,6 +201,7 @@ public class PrometheusEmitter implements Emitter
         server.stop();
       }
     } else {
+      exec.shutdownNow();
       flush();
     }
   }

extensions-contrib/prometheus-emitter/src/main/java/org/apache/druid/emitter/prometheus/PrometheusEmitterConfig.java

@@ -24,6 +24,7 @@ import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.base.Preconditions;
 
 import javax.annotation.Nullable;
+import java.util.Objects;
 import java.util.regex.Pattern;
 
 /**
@@ -53,6 +54,10 @@ public class PrometheusEmitterConfig
   @Nullable
   private final String pushGatewayAddress;
 
+  @JsonProperty
+  @Nullable
+  private final Integer flushPeriod;
+
   @JsonProperty
   private final boolean addHostAsLabel;
 
@@ -67,7 +72,8 @@ public class PrometheusEmitterConfig
       @JsonProperty("port") @Nullable Integer port,
       @JsonProperty("pushGatewayAddress") @Nullable String pushGatewayAddress,
       @JsonProperty("addHostAsLabel") boolean addHostAsLabel,
-      @JsonProperty("addServiceAsLabel") boolean addServiceAsLabel
+      @JsonProperty("addServiceAsLabel") boolean addServiceAsLabel,
+      @JsonProperty("flushPeriod") Integer flushPeriod
   )
   {
     this.strategy = strategy != null ? strategy : Strategy.exporter;
@@ -75,12 +81,18 @@ public class PrometheusEmitterConfig
     Preconditions.checkArgument(PATTERN.matcher(this.namespace).matches(), "Invalid namespace " + this.namespace);
     if (strategy == Strategy.exporter) {
       Preconditions.checkArgument(port != null, "For `exporter` strategy, port must be specified.");
-    } else if (strategy == Strategy.pushgateway) {
+    } else if (this.strategy == Strategy.pushgateway) {
       Preconditions.checkArgument(pushGatewayAddress != null, "For `pushgateway` strategy, pushGatewayAddress must be specified.");
+      if (Objects.nonNull(flushPeriod)) {
+        Preconditions.checkArgument(flushPeriod > 0, "flushPeriod must be greater than 0.");
+      } else {
+        flushPeriod = 15;
+      }
     }
     this.dimensionMapPath = dimensionMapPath;
     this.port = port;
     this.pushGatewayAddress = pushGatewayAddress;
+    this.flushPeriod = flushPeriod;
     this.addHostAsLabel = addHostAsLabel;
     this.addServiceAsLabel = addServiceAsLabel;
   }
@@ -105,6 +117,12 @@ public class PrometheusEmitterConfig
     return pushGatewayAddress;
   }
 
+  @Nullable
+  public Integer getFlushPeriod()
+  {
+    return flushPeriod;
+  }
+
   public Strategy getStrategy()
   {
     return strategy;

extensions-contrib/prometheus-emitter/src/test/java/org/apache/druid/emitter/prometheus/PrometheusEmitterTest.java

@@ -41,7 +41,7 @@ public class PrometheusEmitterTest
   public void testEmitterWithServiceLabel()
   {
     CollectorRegistry.defaultRegistry.clear();
-    PrometheusEmitterConfig config = new PrometheusEmitterConfig(PrometheusEmitterConfig.Strategy.exporter, null, null, 0, null, false, true);
+    PrometheusEmitterConfig config = new PrometheusEmitterConfig(PrometheusEmitterConfig.Strategy.exporter, null, null, 0, null, false, true, 60);
     PrometheusEmitterModule prometheusEmitterModule = new PrometheusEmitterModule();
     Emitter emitter = prometheusEmitterModule.getEmitter(config);
     ServiceMetricEvent build = ServiceMetricEvent.builder()
@@ -62,7 +62,7 @@ public class PrometheusEmitterTest
   public void testEmitterWithServiceAndHostLabel()
   {
     CollectorRegistry.defaultRegistry.clear();
-    PrometheusEmitterConfig config = new PrometheusEmitterConfig(PrometheusEmitterConfig.Strategy.exporter, null, null, 0, null, true, true);
+    PrometheusEmitterConfig config = new PrometheusEmitterConfig(PrometheusEmitterConfig.Strategy.exporter, null, null, 0, null, true, true, 60);
     PrometheusEmitterModule prometheusEmitterModule = new PrometheusEmitterModule();
     Emitter emitter = prometheusEmitterModule.getEmitter(config);
     ServiceMetricEvent build = ServiceMetricEvent.builder()
@@ -83,7 +83,7 @@ public class PrometheusEmitterTest
   public void testEmitterMetric()
   {
     CollectorRegistry.defaultRegistry.clear();
-    PrometheusEmitterConfig config = new PrometheusEmitterConfig(PrometheusEmitterConfig.Strategy.pushgateway, "namespace", null, 0, "pushgateway", true, true);
+    PrometheusEmitterConfig config = new PrometheusEmitterConfig(PrometheusEmitterConfig.Strategy.pushgateway, "namespace", null, 0, "pushgateway", true, true, 60);
     PrometheusEmitterModule prometheusEmitterModule = new PrometheusEmitterModule();
     Emitter emitter = prometheusEmitterModule.getEmitter(config);
     ServiceMetricEvent build = ServiceMetricEvent.builder()
@@ -104,12 +104,12 @@ public class PrometheusEmitterTest
   @Test
   public void testEmitterStart()
   {
-    PrometheusEmitterConfig exportEmitterConfig = new PrometheusEmitterConfig(PrometheusEmitterConfig.Strategy.exporter, "namespace1", null, 0, null, true, true);
+    PrometheusEmitterConfig exportEmitterConfig = new PrometheusEmitterConfig(PrometheusEmitterConfig.Strategy.exporter, "namespace1", null, 0, null, true, true, 60);
     PrometheusEmitter exportEmitter = new PrometheusEmitter(exportEmitterConfig);
     exportEmitter.start();
     Assert.assertNotNull(exportEmitter.getServer());
 
-    PrometheusEmitterConfig pushEmitterConfig = new PrometheusEmitterConfig(PrometheusEmitterConfig.Strategy.pushgateway, "namespace2", null, 0, "pushgateway", true, true);
+    PrometheusEmitterConfig pushEmitterConfig = new PrometheusEmitterConfig(PrometheusEmitterConfig.Strategy.pushgateway, "namespace2", null, 0, "pushgateway", true, true, 60);
     PrometheusEmitter pushEmitter = new PrometheusEmitter(pushEmitterConfig);
     pushEmitter.start();
     Assert.assertNotNull(pushEmitter.getPushGateway());
@@ -118,7 +118,7 @@ public class PrometheusEmitterTest
   @Test
   public void testEmitterPush() throws IOException
   {
-    PrometheusEmitterConfig emitterConfig = new PrometheusEmitterConfig(PrometheusEmitterConfig.Strategy.pushgateway, "namespace3", null, 0, "pushgateway", true, true);
+    PrometheusEmitterConfig emitterConfig = new PrometheusEmitterConfig(PrometheusEmitterConfig.Strategy.pushgateway, "namespace3", null, 0, "pushgateway", true, true, 60);
 
     PushGateway mockPushGateway = mock(PushGateway.class);
     mockPushGateway.push(anyObject(Collector.class), anyString(), anyObject(ImmutableMap.class));
@@ -145,7 +145,8 @@ public class PrometheusEmitterTest
         1,
         null,
         true,
-        true
+        true,
+        60
     );
 
     Assert.assertThrows(
@@ -158,7 +159,8 @@ public class PrometheusEmitterTest
             null,
             null,
             true,
-            true
+            true,
+            50
         )
     );
   }
@@ -166,7 +168,7 @@ public class PrometheusEmitterTest
   @Test
   public void testEmitterStartWithHttpUrl()
   {
-    PrometheusEmitterConfig pushEmitterConfig = new PrometheusEmitterConfig(PrometheusEmitterConfig.Strategy.pushgateway, "namespace4", null, 0, "http://pushgateway", true, true);
+    PrometheusEmitterConfig pushEmitterConfig = new PrometheusEmitterConfig(PrometheusEmitterConfig.Strategy.pushgateway, "namespace4", null, 0, "http://pushgateway", true, true, 60);
     PrometheusEmitter pushEmitter = new PrometheusEmitter(pushEmitterConfig);
     pushEmitter.start();
     Assert.assertNotNull(pushEmitter.getPushGateway());
@@ -175,7 +177,7 @@ public class PrometheusEmitterTest
   @Test
   public void testEmitterStartWithHttpsUrl()
   {
-    PrometheusEmitterConfig pushEmitterConfig = new PrometheusEmitterConfig(PrometheusEmitterConfig.Strategy.pushgateway, "namespace5", null, 0, "https://pushgateway", true, true);
+    PrometheusEmitterConfig pushEmitterConfig = new PrometheusEmitterConfig(PrometheusEmitterConfig.Strategy.pushgateway, "namespace5", null, 0, "https://pushgateway", true, true, 60);
     PrometheusEmitter pushEmitter = new PrometheusEmitter(pushEmitterConfig);
     pushEmitter.start();
     Assert.assertNotNull(pushEmitter.getPushGateway());
@@ -194,7 +196,8 @@ public class PrometheusEmitterTest
             null,
             "https://pushgateway",
             true,
-            true
+            true,
+            60
         )
     );
 
@@ -206,7 +209,8 @@ public class PrometheusEmitterTest
         null,
         "https://pushgateway",
         true,
-        true
+        true,
+        60
     );
   }
 }

website/.spelling

@@ -862,6 +862,7 @@ HTTPServer
 conversionFactor
 prometheus
 Pushgateway
+flushPeriod
  - ../docs/development/extensions-contrib/tdigestsketch-quantiles.md
 postAggregator
 quantileFromTDigestSketch