[ISSUE #696] Add metrics plugin

build.gradle

@@ -247,6 +247,7 @@ subprojects {
         Set<String> rootProject = ["eventmesh-admin",
                                    "eventmesh-common",
                                    "eventmesh-connector-api",
+                                   "eventmesh-metrics-api",
                                    "eventmesh-registry-api",
                                    "eventmesh-runtime",
                                    "eventmesh-security-api",

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/CommonConfiguration.java

@@ -21,32 +21,37 @@
 
 import org.apache.commons.lang3.StringUtils;
 
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
 import com.google.common.base.Preconditions;
 
 public class CommonConfiguration {
-    public String eventMeshEnv = "P";
-    public String eventMeshIDC = "FT";
-    public String eventMeshCluster = "LS";
-    public String eventMeshName = "";
-    public String sysID = "5477";
+    public String eventMeshEnv                 = "P";
+    public String eventMeshIDC                 = "FT";
+    public String eventMeshCluster             = "LS";
+    public String eventMeshName                = "";
+    public String sysID                        = "5477";
     public String eventMeshConnectorPluginType = "rocketmq";
-    public String eventMeshSecurityPluginType = "security";
-    public int eventMeshPrometheusPort = 19090;
-    public String eventMeshRegistryPluginType = "namesrv";
-    public String eventMeshTraceExporterType = "Log";
-    public int eventMeshTraceExporterMaxExportSize = 512;
-    public int eventMeshTraceExporterMaxQueueSize = 2048;
-    public int eventMeshTraceExporterExportTimeout = 30;
-    public int eventMeshTraceExporterExportInterval = 5;
-    public String eventMeshTraceExportZipkinIp = "localhost";
-    public int eventMeshTraceExportZipkinPort = 9411;
-
-    public String namesrvAddr = "";
-    public Integer eventMeshRegisterIntervalInMills = 10 * 1000;
-    public Integer eventMeshFetchRegistryAddrInterval = 10 * 1000;
-    public String eventMeshServerIp = null;
-    public boolean eventMeshServerSecurityEnable = false;
-    public boolean eventMeshServerRegistryEnable = false;
+    public String eventMeshSecurityPluginType  = "security";
+    public String eventMeshRegistryPluginType  = "namesrv";
+
+    public List<String> eventMeshMetricsPluginType;
+    public String       eventMeshTraceExporterType           = "Log";
+    public int          eventMeshTraceExporterMaxExportSize  = 512;
+    public int          eventMeshTraceExporterMaxQueueSize   = 2048;
+    public int          eventMeshTraceExporterExportTimeout  = 30;
+    public int          eventMeshTraceExporterExportInterval = 5;
+    public String       eventMeshTraceExportZipkinIp         = "localhost";
+    public int          eventMeshTraceExportZipkinPort       = 9411;
+
+    public    String               namesrvAddr                        = "";
+    public    Integer              eventMeshRegisterIntervalInMills   = 10 * 1000;
+    public    Integer              eventMeshFetchRegistryAddrInterval = 10 * 1000;
+    public    String               eventMeshServerIp                  = null;
+    public    boolean              eventMeshServerSecurityEnable      = false;
+    public    boolean              eventMeshServerRegistryEnable      = false;
     protected ConfigurationWrapper configurationWrapper;
 
     public CommonConfiguration(ConfigurationWrapper configurationWrapper) {
@@ -86,13 +91,6 @@ public void init() {
                     String.format("%s error", ConfKeys.KEYS_EVENTMESH_IDC));
             eventMeshIDC = StringUtils.deleteWhitespace(eventMeshIdcStr);
 
-            String eventMeshPrometheusPortStr =
-                    configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_METRICS_PROMETHEUS_PORT);
-            if (StringUtils.isNotEmpty(eventMeshPrometheusPortStr)) {
-                eventMeshPrometheusPort =
-                        Integer.parseInt(StringUtils.deleteWhitespace(eventMeshPrometheusPortStr));
-            }
-
             eventMeshServerIp =
                     configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_SERVER_HOST_IP);
             if (StringUtils.isBlank(eventMeshServerIp)) {
@@ -120,13 +118,13 @@ public void init() {
                     configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_REGISTRY_ENABLED);
             if (StringUtils.isNotBlank(eventMeshServerRegistryEnableStr)) {
                 eventMeshServerRegistryEnable =
-                        Boolean.parseBoolean(StringUtils.deleteWhitespace(eventMeshServerRegistryEnableStr));
+                    Boolean.parseBoolean(StringUtils.deleteWhitespace(eventMeshServerRegistryEnableStr));
             }
 
             eventMeshRegistryPluginType =
-                    configurationWrapper.getProp(ConfKeys.KEYS_ENENTMESH_REGISTRY_PLUGIN_TYPE);
+                configurationWrapper.getProp(ConfKeys.KEYS_ENENTMESH_REGISTRY_PLUGIN_TYPE);
             Preconditions.checkState(StringUtils.isNotEmpty(eventMeshRegistryPluginType),
-                    String.format("%s error", ConfKeys.KEYS_ENENTMESH_REGISTRY_PLUGIN_TYPE));
+                String.format("%s error", ConfKeys.KEYS_ENENTMESH_REGISTRY_PLUGIN_TYPE));
 
             String eventMeshTraceExporterTypeStr =
                     configurationWrapper.getProp(ConfKeys.KEYS_ENENTMESH_TRACE_EXPORTER_TYPE);
@@ -166,14 +164,22 @@ public void init() {
             String eventMeshTraceExportZipkinIpStr =
                     configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_TRACE_EXPORT_ZIPKIN_IP);
             Preconditions.checkState(StringUtils.isNotEmpty(eventMeshTraceExportZipkinIpStr),
-                    String.format("%s error", ConfKeys.KEYS_EVENTMESH_TRACE_EXPORT_ZIPKIN_IP));
+                String.format("%s error", ConfKeys.KEYS_EVENTMESH_TRACE_EXPORT_ZIPKIN_IP));
             eventMeshTraceExportZipkinIp = StringUtils.deleteWhitespace(eventMeshTraceExportZipkinIpStr);
 
             String eventMeshTraceExportZipkinPortStr =
-                    configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_TRACE_EXPORT_ZIPKIN_PORT);
+                configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_TRACE_EXPORT_ZIPKIN_PORT);
             if (StringUtils.isNotEmpty(eventMeshTraceExportZipkinPortStr)) {
                 eventMeshTraceExportZipkinPort =
-                        Integer.parseInt(StringUtils.deleteWhitespace(eventMeshTraceExportZipkinPortStr));
+                    Integer.parseInt(StringUtils.deleteWhitespace(eventMeshTraceExportZipkinPortStr));
+            }
+
+            String metricsPluginType = configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_METRICS_PLUGIN_TYPE);
+            if (StringUtils.isNotEmpty(metricsPluginType)) {
+                eventMeshMetricsPluginType = Arrays.stream(metricsPluginType.split(","))
+                    .filter(StringUtils::isNotBlank)
+                    .map(String::trim)
+                    .collect(Collectors.toList());
             }
         }
     }
@@ -203,8 +209,6 @@ static class ConfKeys {
 
         public static String KEYS_ENENTMESH_SECURITY_PLUGIN_TYPE = "eventMesh.security.plugin.type";
 
-        public static String KEY_EVENTMESH_METRICS_PROMETHEUS_PORT = "eventMesh.metrics.prometheus.port";
-
         public static String KEYS_EVENTMESH_REGISTRY_ENABLED = "eventMesh.server.registry.enabled";
 
         public static String KEYS_ENENTMESH_REGISTRY_PLUGIN_TYPE = "eventMesh.registry.plugin.type";
@@ -222,5 +226,7 @@ static class ConfKeys {
         public static String KEYS_EVENTMESH_TRACE_EXPORT_ZIPKIN_IP = "eventmesh.trace.export.zipkin.ip";
 
         public static String KEYS_EVENTMESH_TRACE_EXPORT_ZIPKIN_PORT = "eventmesh.trace.export.zipkin.port";
+
+        public static String KEYS_EVENTMESH_METRICS_PLUGIN_TYPE = "eventmesh.metrics.plugin";
     }
 }

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HealthMetrics.java → eventmesh-metrics-plugin/build.gradle

@@ -13,26 +13,4 @@
  * 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.eventmesh.runtime.metrics.http;
-
-import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.codahale.metrics.MetricRegistry;
-
-public class HealthMetrics {
-
-    private EventMeshHTTPServer eventMeshHTTPServer;
-    private MetricRegistry metricRegistry;
-
-    public Logger logger = LoggerFactory.getLogger("httpMonitor");
-
-    public HealthMetrics(EventMeshHTTPServer eventMeshHTTPServer, MetricRegistry metricRegistry) {
-        this.eventMeshHTTPServer = eventMeshHTTPServer;
-        this.metricRegistry = metricRegistry;
-    }
-}
+ */

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/TopicMetrics.java → eventmesh-metrics-plugin/eventmesh-metrics-api/build.gradle

@@ -15,19 +15,14 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.runtime.metrics.http;
+dependencies {
+    api project(":eventmesh-spi")
+    implementation project(":eventmesh-common")
 
-import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
 
-import com.codahale.metrics.MetricRegistry;
+    compileOnly 'org.projectlombok:lombok:1.18.22'
+    annotationProcessor 'org.projectlombok:lombok:1.18.22'
 
-public class TopicMetrics {
-
-    private EventMeshHTTPServer eventMeshHTTPServer;
-    private MetricRegistry metricRegistry;
-
-    public TopicMetrics(EventMeshHTTPServer eventMeshHTTPServer, MetricRegistry metricRegistry) {
-        this.eventMeshHTTPServer = eventMeshHTTPServer;
-        this.metricRegistry = metricRegistry;
-    }
-}
+    testCompileOnly 'org.projectlombok:lombok:1.18.22'
+    testAnnotationProcessor 'org.projectlombok:lombok:1.18.22'
+}

eventmesh-metrics-plugin/eventmesh-metrics-api/gradle.properties

@@ -0,0 +1,16 @@
+#
+# 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.
+#

eventmesh-metrics-plugin/eventmesh-metrics-api/src/main/java/org/apache/eventmesh/metrics/api/MetricsPluginFactory.java

@@ -0,0 +1,41 @@
+/*
+ * 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.eventmesh.metrics.api;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.eventmesh.spi.EventMeshExtensionFactory;
+
+import lombok.experimental.UtilityClass;
+
+@UtilityClass
+public class MetricsPluginFactory {
+
+    /**
+     * Get {@code MetricsRegistry}.
+     *
+     * @param metricsRegistryType
+     * @return
+     */
+    public static MetricsRegistry getMetricsRegistry(String metricsRegistryType) {
+        checkNotNull(metricsRegistryType, "MetricsRegistryType cannot be null");
+
+        MetricsRegistry metricsRegistry = EventMeshExtensionFactory.getExtension(MetricsRegistry.class, metricsRegistryType);
+        return checkNotNull(metricsRegistry, "MetricsRegistryType: " + metricsRegistryType + " is not supported");
+    }
+}

eventmesh-metrics-plugin/eventmesh-metrics-api/src/main/java/org/apache/eventmesh/metrics/api/MetricsRegistry.java

@@ -0,0 +1,56 @@
+/*
+ * 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.eventmesh.metrics.api;
+
+import org.apache.eventmesh.metrics.api.model.Metric;
+import org.apache.eventmesh.spi.EventMeshExtensionType;
+import org.apache.eventmesh.spi.EventMeshSPI;
+
+/**
+ * The top-level interface of metrics registry, used to register different metrics.
+ * It should have multiple sub implementation, e.g. JVM, Prometheus, i.g.
+ *
+ * @since 1.4.0
+ */
+@EventMeshSPI(isSingleton = true, eventMeshExtensionType = EventMeshExtensionType.METRICS)
+public interface MetricsRegistry {
+
+    /**
+     * Start the metrics registry.
+     */
+    void start();
+
+    /**
+     * Close the metrics registry.
+     */
+    void showdown();
+
+    /**
+     * Register a new Metric, if the metric is already exist, it will do nothing.
+     *
+     * @param metric
+     */
+    void register(Metric metric);
+
+    /**
+     * Remove a metric, if the metric is not exist, it will do nothing.
+     *
+     * @param metric
+     */
+    void unRegister(Metric metric);
+}