[ISSUE#173]Connect Exporter for Prometheus

connectors/rocketmq-connect-metrics-exporter/pom.xml

@@ -0,0 +1,47 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-connect-metrics-exporter</artifactId>
+ <version>1.0-SNAPSHOT</version>
+ <name>rocketmq-connect-metrics-exporter</name>
+
+
+ <properties>
+ <maven.compiler.source>1.8</maven.compiler.source>
+ <maven.compiler.target>1.8</maven.compiler.target>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+ <openmessaging.connector.version>0.1.3</openmessaging.connector.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>io.openmessaging</groupId>
+ <artifactId>openmessaging-connector</artifactId>
+ <version>${openmessaging.connector.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.6.1</version>
+ <configuration>
+ <source>${maven.compiler.source}</source>
+ <target>${maven.compiler.target}</target>
+ <compilerVersion>${maven.compiler.source}</compilerVersion>
+ <showDeprecation>true</showDeprecation>
+ <showWarnings>true</showWarnings>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
+
+</project>

connectors/rocketmq-connect-metrics-exporter/src/main/java/org/apache/rocket/connect/metrics/export/sink/connector/MetricsExportSinkConnector.java

@@ -0,0 +1,41 @@
+package org.apache.rocket.connect.metrics.export.sink.connector;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.Task;
+import io.openmessaging.connector.api.component.task.sink.SinkConnector;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.rocket.connect.metrics.export.sink.util.ServiceProvicerUtil;
+
+public class MetricsExportSinkConnector extends SinkConnector {
+ private KeyValue config;
+
+ private List<MetricsExporter> metricsExporters;
+ {
+ metricsExporters = ServiceProvicerUtil.getMetricsExporterServices();
+ }
+
+ @Override public List<KeyValue> taskConfigs(int maxTasks) {
+ List<KeyValue> configs = new ArrayList<>();
+ configs.add(config);
+ return configs;
+ }
+
+ @Override public Class<? extends Task> taskClass() {
+ return MetricsExportSinkTask.class;
+ }
+
+ @Override public void start(KeyValue config) {
+ this.config = config;
+ }
+
+ @Override public void stop() {
+ this.config = null;
+ }
+
+ @Override public void validate(KeyValue config) {
+ for (MetricsExporter exporter: metricsExporters) {
+ exporter.validate(config);
+ }
+ }
+}

connectors/rocketmq-connect-metrics-exporter/src/main/java/org/apache/rocket/connect/metrics/export/sink/connector/MetricsExportSinkTask.java

@@ -0,0 +1,37 @@
+package org.apache.rocket.connect.metrics.export.sink.connector;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.sink.SinkTask;
+import io.openmessaging.connector.api.component.task.sink.SinkTaskContext;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.errors.ConnectException;
+import java.util.List;
+import org.apache.rocket.connect.metrics.export.sink.util.ServiceProvicerUtil;
+
+
+public class MetricsExportSinkTask extends SinkTask {
+ private List<MetricsExporter> metricsExporters;
+
+ @Override public void put(List<ConnectRecord> sinkRecords) throws ConnectException {
+ for (MetricsExporter exporter : metricsExporters) {
+ exporter.export(sinkRecords);
+ }
+ }
+
+ @Override public void start(KeyValue config) {
+ for (MetricsExporter exporter : metricsExporters) {
+ exporter.start(config);
+ }
+ }
+
+ @Override public void stop() {
+ for (MetricsExporter exporter : metricsExporters) {
+ exporter.stop();
+ }
+ }
+
+ @Override public void init(SinkTaskContext sinkTaskContext) {
+ super.init(sinkTaskContext);
+ metricsExporters = ServiceProvicerUtil.getMetricsExporterServices();
+ }
+}

connectors/rocketmq-connect-metrics-exporter/src/main/java/org/apache/rocket/connect/metrics/export/sink/connector/MetricsExporter.java

@@ -0,0 +1,16 @@
+package org.apache.rocket.connect.metrics.export.sink.connector;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import java.util.List;
+
+public interface MetricsExporter {
+ void export(List<ConnectRecord> sinkRecords);
+
+ void validate(KeyValue config);
+
+ void start(KeyValue config);
+
+ void stop();
+
+}

connectors/rocketmq-connect-metrics-exporter/src/main/java/org/apache/rocket/connect/metrics/export/sink/util/ServiceProvicerUtil.java

@@ -0,0 +1,23 @@
+package org.apache.rocket.connect.metrics.export.sink.util;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.ServiceLoader;
+import org.apache.rocket.connect.metrics.export.sink.connector.MetricsExporter;
+
+/**
+ * @author: ming
+ */
+public class ServiceProvicerUtil {
+
+ public static List<MetricsExporter> getMetricsExporterServices(){
+ List<MetricsExporter> metricsExporterList = new ArrayList<>();
+ ServiceLoader<MetricsExporter> metricsExporters = ServiceLoader.load(MetricsExporter.class);
+ Iterator<MetricsExporter> iterator = metricsExporters.iterator();
+ while (iterator.hasNext()){
+ metricsExporterList.add(iterator.next());
+ }
+ return metricsExporterList;
+ }
+}