[ISSUE apache#494] support prometheus

metric-exporter/pom.xml

@@ -41,6 +41,14 @@
  <groupId>com.google.guava</groupId>
  <artifactId>guava</artifactId>
  </dependency>
+ <dependency>
+ <groupId>io.prometheus</groupId>
+ <artifactId>simpleclient_dropwizard</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.prometheus</groupId>
+ <artifactId>simpleclient</artifactId>
+ </dependency>
  </dependencies>
 
  <build>

metric-exporter/src/main/java/org/apache/rocketmq/connect/metrics/DropwizardExports.java

@@ -0,0 +1,234 @@
+/*
+ * 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.rocketmq.connect.metrics;
+
+import com.codahale.metrics.Counter;
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.Histogram;
+import com.codahale.metrics.Meter;
+import com.codahale.metrics.Metric;
+import com.codahale.metrics.MetricFilter;
+import com.codahale.metrics.MetricRegistry;
+import com.codahale.metrics.Snapshot;
+import com.codahale.metrics.Timer;
+import io.prometheus.client.dropwizard.samplebuilder.DefaultSampleBuilder;
+import io.prometheus.client.dropwizard.samplebuilder.SampleBuilder;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.SortedMap;
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import org.apache.rocketmq.connect.metrics.stats.Stat;
+
+/**
+ * Collect Dropwizard metrics from a MetricRegistry.
+ */
+public class DropwizardExports extends io.prometheus.client.Collector implements io.prometheus.client.Collector.Describable {
+ private static final Logger LOGGER = Logger.getLogger(DropwizardExports.class.getName());
+ private MetricRegistry registry;
+ private MetricFilter metricFilter;
+ private SampleBuilder sampleBuilder;
+
+ /**
+ * Creates a new DropwizardExports with a {@link DefaultSampleBuilder} and {@link MetricFilter#ALL}.
+ *
+ * @param registry a metric registry to export in prometheus.
+ */
+ public DropwizardExports(MetricRegistry registry) {
+ this.registry = registry;
+ this.metricFilter = MetricFilter.ALL;
+ this.sampleBuilder = new DefaultSampleBuilder();
+ }
+
+ /**
+ * Creates a new DropwizardExports with a {@link DefaultSampleBuilder} and custom {@link MetricFilter}.
+ *
+ * @param registry a metric registry to export in prometheus.
+ * @param metricFilter a custom metric filter.
+ */
+ public DropwizardExports(MetricRegistry registry, MetricFilter metricFilter) {
+ this.registry = registry;
+ this.metricFilter = metricFilter;
+ this.sampleBuilder = new DefaultSampleBuilder();
+ }
+
+ /**
+ * @param registry a metric registry to export in prometheus.
+ * @param sampleBuilder sampleBuilder to use to create prometheus samples.
+ */
+ public DropwizardExports(MetricRegistry registry, SampleBuilder sampleBuilder) {
+ this.registry = registry;
+ this.metricFilter = MetricFilter.ALL;
+ this.sampleBuilder = sampleBuilder;
+ }
+
+ /**
+ * @param registry a metric registry to export in prometheus.
+ * @param metricFilter a custom metric filter.
+ * @param sampleBuilder sampleBuilder to use to create prometheus samples.
+ */
+ public DropwizardExports(MetricRegistry registry, MetricFilter metricFilter, SampleBuilder sampleBuilder) {
+ this.registry = registry;
+ this.metricFilter = metricFilter;
+ this.sampleBuilder = sampleBuilder;
+ }
+
+ private static String getHelpMessage(String metricName, Metric metric) {
+ return String.format("Generated from Dropwizard metric import (metric=%s, type=%s)", metricName, metric.getClass().getName());
+ }
+
+ /**
+ * Export counter as Prometheus <a href="https://prometheus.io/docs/concepts/metric_types/#gauge">Gauge</a>.
+ */
+ MetricFamilySamples fromCounter(String dropwizardName, Counter counter) {
+ MetricFamilySamples.Sample sample = sampleBuilder.createSample(dropwizardName, "", new ArrayList<String>(), new ArrayList<String>(), new Long(counter.getCount()).doubleValue());
+ return new MetricFamilySamples(sample.name, Type.GAUGE, getHelpMessage(dropwizardName, counter), Arrays.asList(sample));
+ }
+
+ /**
+ * Export gauge as a prometheus gauge.
+ */
+ MetricFamilySamples fromGauge(String dropwizardName, Gauge gauge) {
+ Object obj = gauge.getValue();
+ double value;
+ if (obj instanceof Number) {
+ value = ((Number) obj).doubleValue();
+ } else if (obj instanceof Boolean) {
+ value = ((Boolean) obj) ? 1 : 0;
+ } else {
+ LOGGER.log(Level.FINE, String.format("Invalid type for Gauge %s: %s", sanitizeMetricName(dropwizardName), obj == null ? "null" : obj.getClass().getName()));
+ return null;
+ }
+ MetricFamilySamples.Sample sample = sampleBuilder.createSample(dropwizardName, "", new ArrayList<String>(), new ArrayList<String>(), value);
+ return new MetricFamilySamples(sample.name, Type.GAUGE, getHelpMessage(dropwizardName, gauge), Arrays.asList(sample));
+ }
+
+ /**
+ * Export a histogram snapshot as a prometheus SUMMARY.
+ *
+ * @param dropwizardName metric name.
+ * @param snapshot the histogram snapshot.
+ * @param count the total sample count for this snapshot.
+ * @param factor a factor to apply to histogram values.
+ */
+ MetricFamilySamples fromSnapshotAndCount(String dropwizardName, Snapshot snapshot, long count, double factor,
+ String helpMessage) {
+ MetricName metricName = MetricUtils.stringToMetricName(dropwizardName);
+ Stat.HistogramType histogramType = Stat.HistogramType.valueOf(metricName.getType());
+ List<MetricFamilySamples.Sample> samples = new ArrayList<>();
+ switch (histogramType) {
+ case Avg:
+ samples = Arrays.asList(sampleBuilder.createSample(dropwizardName, "", new ArrayList<String>(), new ArrayList<String>(), snapshot.getMean() * factor));
+ break;
+ case Min:
+ samples = Arrays.asList(sampleBuilder.createSample(dropwizardName, "", new ArrayList<String>(), new ArrayList<String>(), snapshot.getMin() * factor));
+ break;
+ case Max:
+ samples = Arrays.asList(sampleBuilder.createSample(dropwizardName, "", new ArrayList<String>(), new ArrayList<String>(), snapshot.getMax() * factor));
+ break;
+ case Percentile_75th:
+ samples = Arrays.asList(sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.75"), snapshot.get75thPercentile() * factor));
+ break;
+ case Percentile_95th:
+ samples = Arrays.asList(sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.95"), snapshot.get95thPercentile() * factor));
+ break;
+ case Percentile_98th:
+ samples = Arrays.asList(sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.98"), snapshot.get98thPercentile() * factor));
+ break;
+ case Percentile_99th:
+ samples = Arrays.asList(sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.99"), snapshot.get99thPercentile() * factor));
+ break;
+ case Percentile_999th:
+ samples = Arrays.asList(sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.999"), snapshot.get999thPercentile() * factor));
+ break;
+ default:
+ samples = Arrays.asList(sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.5"), snapshot.getMedian() * factor), sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.5"), snapshot.getMedian() * factor), sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.75"), snapshot.get75thPercentile() * factor), sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.95"), snapshot.get95thPercentile() * factor), sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.98"), snapshot.get98thPercentile() * factor), sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.99"), snapshot.get99thPercentile() * factor), sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.999"), snapshot.get999thPercentile() * factor), sampleBuilder.createSample(dropwizardName, "_count", new ArrayList<String>(), new ArrayList<String>(), count));
+
+ }
+ return new MetricFamilySamples(samples.get(0).name, Type.SUMMARY, helpMessage, samples);
+ }
+
+ /**
+ * Convert histogram snapshot.
+ */
+ MetricFamilySamples fromHistogram(String dropwizardName, Histogram histogram) {
+ return fromSnapshotAndCount(dropwizardName, histogram.getSnapshot(), histogram.getCount(), 1.0, getHelpMessage(dropwizardName, histogram));
+ }
+
+ /**
+ * Export Dropwizard Timer as a histogram. Use TIME_UNIT as time unit.
+ */
+ MetricFamilySamples fromTimer(String dropwizardName, Timer timer) {
+ return fromSnapshotAndCount(dropwizardName, timer.getSnapshot(), timer.getCount(), 1.0D / TimeUnit.SECONDS.toNanos(1L), getHelpMessage(dropwizardName, timer));
+ }
+
+ /**
+ * Export a Meter as as prometheus COUNTER.
+ */
+ MetricFamilySamples fromMeter(String dropwizardName, Meter meter) {
+ MetricName metricName = MetricUtils.stringToMetricName(dropwizardName);
+ final MetricFamilySamples.Sample sample = sampleBuilder.createSample(dropwizardName, "", new ArrayList<String>(), new ArrayList<String>(), MetricUtils.getMeterValue(metricName, meter));
+ return new MetricFamilySamples(sample.name, Type.COUNTER, getHelpMessage(dropwizardName, meter), Arrays.asList(sample));
+ }
+
+ @Override
+ public List<MetricFamilySamples> collect() {
+ Map<String, MetricFamilySamples> mfSamplesMap = new HashMap<String, MetricFamilySamples>();
+
+ for (SortedMap.Entry<String, Gauge> entry : registry.getGauges(metricFilter).entrySet()) {
+ addToMap(mfSamplesMap, fromGauge(entry.getKey(), entry.getValue()));
+ }
+ for (SortedMap.Entry<String, Counter> entry : registry.getCounters(metricFilter).entrySet()) {
+ addToMap(mfSamplesMap, fromCounter(entry.getKey(), entry.getValue()));
+ }
+ for (SortedMap.Entry<String, Histogram> entry : registry.getHistograms(metricFilter).entrySet()) {
+ addToMap(mfSamplesMap, fromHistogram(entry.getKey(), entry.getValue()));
+ }
+ for (SortedMap.Entry<String, Timer> entry : registry.getTimers(metricFilter).entrySet()) {
+ addToMap(mfSamplesMap, fromTimer(entry.getKey(), entry.getValue()));
+ }
+ for (SortedMap.Entry<String, Meter> entry : registry.getMeters(metricFilter).entrySet()) {
+ addToMap(mfSamplesMap, fromMeter(entry.getKey(), entry.getValue()));
+ }
+ return new ArrayList<MetricFamilySamples>(mfSamplesMap.values());
+ }
+
+ private void addToMap(Map<String, MetricFamilySamples> mfSamplesMap, MetricFamilySamples newMfSamples) {
+ if (newMfSamples != null) {
+ MetricFamilySamples currentMfSamples = mfSamplesMap.get(newMfSamples.name);
+ if (currentMfSamples == null) {
+ mfSamplesMap.put(newMfSamples.name, newMfSamples);
+ } else {
+ Set<MetricFamilySamples.Sample> samples = new HashSet<MetricFamilySamples.Sample>(currentMfSamples.samples);
+ samples.addAll(newMfSamples.samples);
+ List<MetricFamilySamples.Sample> list = new ArrayList<>(samples);
+ mfSamplesMap.put(newMfSamples.name, new MetricFamilySamples(newMfSamples.name, currentMfSamples.type, currentMfSamples.help, list));
+ }
+ }
+ }
+
+ @Override
+ public List<MetricFamilySamples> describe() {
+ return new ArrayList<MetricFamilySamples>();
+ }
+}

metric-exporter/src/main/java/org/apache/rocketmq/connect/metrics/PrometheusSampleBuilder.java

@@ -0,0 +1,59 @@
+/*
+ * 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.rocketmq.connect.metrics;
+
+import io.prometheus.client.Collector;
+import io.prometheus.client.dropwizard.samplebuilder.SampleBuilder;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.commons.lang3.StringUtils;
+
+public class PrometheusSampleBuilder implements SampleBuilder {
+ private static final List<String> SOURCE_TASK_LABEL_NAMES = Arrays.asList("metric_group", "data_type", "connector", "task");
+
+ @Override
+ public Collector.MetricFamilySamples.Sample createSample(String dropwizardName, String nameSuffix,
+ List<String> additionalLabelNames, List<String> additionalLabelValues, double value) {
+ String suffix = nameSuffix == null ? "" : nameSuffix;
+ List<String> labelValues = sanitizeLabelValues(dropwizardName);
+ return new Collector.MetricFamilySamples.Sample(sanitizeMetricName(dropwizardName + suffix), SOURCE_TASK_LABEL_NAMES, labelValues, value);
+ }
+
+ public String sanitizeMetricName(String dropwizardName) {
+ return dropwizardName.split(":")[1].split(",")[1].replaceAll("-", "_");
+ }
+
+ public List<String> sanitizeLabelValues(String dropwizardName) {
+ String[] var = dropwizardName.split(":");
+ String[] split = var[1].split(",");
+
+ String metricGroup = split[0];
+ String metricName = split[1].replaceAll("-", "_");
+ String dateType = split[2];
+ String var3 = split[3];
+ String connectorName = StringUtils.EMPTY;
+ if (!StringUtils.equals(var3, "")) {
+ connectorName = var3.substring(var3.indexOf("=") + 1);
+ }
+ String var4 = split[4];
+ String taskid = StringUtils.EMPTY;
+ if (!StringUtils.equals(var4, "")) {
+ taskid = var4.substring(var4.indexOf("=") + 1);
+ }
+ return Arrays.asList(metricGroup, dateType, connectorName, taskid);
+ }
+}

pom.xml

@@ -66,6 +66,8 @@
  <maven.compiler.source>1.8</maven.compiler.source>
  <maven.compiler.target>1.8</maven.compiler.target>
  <commons-lang3.version>3.12.0</commons-lang3.version>
+ <prometheus.version>0.6.0</prometheus.version>
+
  </properties>
  <dependencyManagement>
  <dependencies>
@@ -202,6 +204,16 @@
  <artifactId>commons-lang3</artifactId>
  <version>${commons-lang3.version}</version>
  </dependency>
+ <dependency>
+ <groupId>io.prometheus</groupId>
+ <artifactId>simpleclient</artifactId>
+ <version>${prometheus.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>io.prometheus</groupId>
+ <artifactId>simpleclient_dropwizard</artifactId>
+ <version>${prometheus.version}</version>
+ </dependency>
  </dependencies>
  </dependencyManagement>
 

rocketmq-connect-runtime/pom.xml

@@ -264,7 +264,10 @@
  <artifactId>metrics-core</artifactId>
  <version>4.2.0</version>
  </dependency>
-
+ <dependency>
+ <groupId>io.prometheus</groupId>
+ <artifactId>simpleclient</artifactId>
+ </dependency>
  <dependency>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>metric-exporter</artifactId>

rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/WorkerConfig.java

@@ -58,6 +58,8 @@ public class WorkerConfig {
  */
  private int httpPort = 8082;
 
+ private int exporterPort = 5557;
+
  /**
  * plugin paths config;
  * Multiple use ',' split
@@ -591,6 +593,14 @@ public void setStateManagementService(String stateManagementService) {
  this.stateManagementService = stateManagementService;
  }
 
+ public int getExporterPort() {
+ return exporterPort;
+ }
+
+ public void setExporterPort(int exporterPort) {
+ this.exporterPort = exporterPort;
+ }
+
  @Override
  public String toString() {
  return "WorkerConfig{" +
@@ -638,6 +648,7 @@ public String toString() {
  ", configManagementService='" + configManagementService + '\'' +
  ", positionManagementService='" + positionManagementService + '\'' +
  ", stateManagementService='" + stateManagementService + '\'' +
+ ", exporterPort=" + exporterPort +
  '}';
  }
 }