You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by mi...@apache.org on 2022/01/13 07:12:50 UTC
[incubator-eventmesh] branch master updated: [ISSUE #696] Add metrics plugin (#709)
This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 1cb0023 [ISSUE #696] Add metrics plugin (#709)
1cb0023 is described below
commit 1cb00234b27d879e50aa658d1c4261d20e7e8b0d
Author: Wenjun Ruan <we...@apache.org>
AuthorDate: Thu Jan 13 15:10:33 2022 +0800
[ISSUE #696] Add metrics plugin (#709)
1. Add Metrics plugin module
2. Implement opentelemetry metrics module
related issue #696
---
build.gradle | 1 +
.../common/config/CommonConfiguration.java | 80 ++--
.../build.gradle | 24 +-
.../eventmesh-metrics-api/build.gradle | 21 +-
.../eventmesh-metrics-api/gradle.properties | 32 +-
.../metrics/api/MetricsPluginFactory.java | 29 +-
.../eventmesh/metrics/api/MetricsRegistry.java | 56 +++
.../metrics/api/model/HttpSummaryMetrics.java | 104 +++---
.../apache/eventmesh/metrics/api/model/Metric.java | 20 +-
.../metrics/api/model/TcpSummaryMetrics.java | 123 +++++++
.../eventmesh-metrics-opentelemetry/build.gradle | 38 ++
.../gradle.properties | 34 +-
.../OpenTelemetryMetricsRegistry.java | 87 +++++
.../config/OpenTelemetryConfiguration.java | 81 +++++
.../metrics/OpenTelemetryHttpExporter.java | 298 +++++++++++++++
.../metrics/OpenTelemetryTcpExporter.java | 89 +++++
...rg.apache.eventmesh.metrics.api.MetricsRegistry | 34 +-
.../src/main/resources/opentelemetry.properties | 33 +-
.../src/main/resources}/prometheus.yml | 1 +
.../gradle.properties | 32 +-
eventmesh-runtime/build.gradle | 25 +-
eventmesh-runtime/conf/eventmesh.properties | 4 +-
.../eventmesh/runtime/boot/AbstractHTTPServer.java | 10 +-
.../runtime/boot/EventMeshHTTPServer.java | 20 +-
.../eventmesh/runtime/boot/EventMeshTCPServer.java | 16 +-
.../http/processor/BatchSendMessageProcessor.java | 13 +-
.../processor/BatchSendMessageV2Processor.java | 34 +-
.../http/processor/HeartBeatProcessor.java | 6 +-
.../http/processor/ReplyMessageProcessor.java | 79 ++--
.../http/processor/SendAsyncMessageProcessor.java | 16 +-
.../http/processor/SendSyncMessageProcessor.java | 34 +-
.../http/processor/SubscribeProcessor.java | 26 +-
.../http/processor/UnSubscribeProcessor.java | 11 +-
.../protocol/http/producer/SendMessageContext.java | 2 +-
.../protocol/http/push/AsyncHTTPPushRequest.java | 6 +-
.../core/protocol/http/retry/HttpRetryer.java | 9 +-
.../protocol/tcp/client/EventMeshTcp2Client.java | 2 +-
.../tcp/client/EventMeshTcpMessageDispatcher.java | 2 +-
.../tcp/client/group/ClientGroupWrapper.java | 136 ++++---
.../core/protocol/tcp/client/session/Session.java | 3 +-
.../tcp/client/session/push/SessionPusher.java | 2 +-
.../tcp/client/session/send/SessionSender.java | 4 +-
.../runtime/metrics/http/HTTPMetricsServer.java | 158 ++++----
.../runtime/metrics/http/TcpSummaryMetrics.java | 403 ---------------------
.../OpenTelemetryHTTPMetricsExporter.java | 310 ----------------
.../OpenTelemetryPrometheusExporter.java | 63 ----
.../OpenTelemetryTCPMetricsExporter.java | 102 ------
.../runtime/metrics/tcp/EventMeshTcpMonitor.java | 192 +++++-----
.../org/apache/eventmesh/runtime/util/Utils.java | 4 +-
.../eventmesh/spi/EventMeshExtensionType.java | 1 +
settings.gradle | 3 +
51 files changed, 1307 insertions(+), 1606 deletions(-)
diff --git a/build.gradle b/build.gradle
index 4f7e7f6..faa7209 100644
--- a/build.gradle
+++ b/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",
diff --git a/eventmesh-common/src/main/java/org/apache/eventmesh/common/config/CommonConfiguration.java b/eventmesh-common/src/main/java/org/apache/eventmesh/common/config/CommonConfiguration.java
index 01b5688..a20d427 100644
--- a/eventmesh-common/src/main/java/org/apache/eventmesh/common/config/CommonConfiguration.java
+++ b/eventmesh-common/src/main/java/org/apache/eventmesh/common/config/CommonConfiguration.java
@@ -21,32 +21,37 @@ import org.apache.eventmesh.common.utils.IPUtils;
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 class CommonConfiguration {
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 class CommonConfiguration {
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 class CommonConfiguration {
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 @@ public class CommonConfiguration {
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 @@ public class CommonConfiguration {
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";
}
}
\ No newline at end of file
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HealthMetrics.java b/eventmesh-metrics-plugin/build.gradle
similarity index 56%
copy from eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HealthMetrics.java
copy to eventmesh-metrics-plugin/build.gradle
index 7451d99..d973dce 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HealthMetrics.java
+++ b/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;
- }
-}
+ */
\ No newline at end of file
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/TopicMetrics.java b/eventmesh-metrics-plugin/eventmesh-metrics-api/build.gradle
similarity index 62%
rename from eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/TopicMetrics.java
rename to eventmesh-metrics-plugin/eventmesh-metrics-api/build.gradle
index 06f97ca..c64ce2f 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/TopicMetrics.java
+++ b/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'
+}
\ No newline at end of file
diff --git a/eventmesh-runtime/conf/prometheus.yml b/eventmesh-metrics-plugin/eventmesh-metrics-api/gradle.properties
similarity index 56%
copy from eventmesh-runtime/conf/prometheus.yml
copy to eventmesh-metrics-plugin/eventmesh-metrics-api/gradle.properties
index ea48021..a9fd83f 100644
--- a/eventmesh-runtime/conf/prometheus.yml
+++ b/eventmesh-metrics-plugin/eventmesh-metrics-api/gradle.properties
@@ -13,34 +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.
-#
-global:
- scrape_interval: 15s
- scrape_timeout: 10s
- evaluation_interval: 15s
-alerting:
- alertmanagers:
- - static_configs:
- - targets: []
- scheme: http
- timeout: 10s
- api_version: v1
-scrape_configs:
-- job_name: prometheus
- honor_timestamps: true
- scrape_interval: 15s
- scrape_timeout: 10s
- metrics_path: /metrics
- scheme: http
- static_configs:
- - targets:
- - localhost:9090
-- job_name: EventMesh_HTTP_export_test
- honor_timestamps: true
- scrape_interval: 15s
- scrape_timeout: 10s
- metrics_path: /metrics
- scheme: http
- static_configs:
- - targets:
- - 127.0.0.1:19090
+#
\ No newline at end of file
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HealthMetrics.java b/eventmesh-metrics-plugin/eventmesh-metrics-api/src/main/java/org/apache/eventmesh/metrics/api/MetricsPluginFactory.java
similarity index 51%
rename from eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HealthMetrics.java
rename to eventmesh-metrics-plugin/eventmesh-metrics-api/src/main/java/org/apache/eventmesh/metrics/api/MetricsPluginFactory.java
index 7451d99..309a88c 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HealthMetrics.java
+++ b/eventmesh-metrics-plugin/eventmesh-metrics-api/src/main/java/org/apache/eventmesh/metrics/api/MetricsPluginFactory.java
@@ -15,24 +15,27 @@
* limitations under the License.
*/
-package org.apache.eventmesh.runtime.metrics.http;
+package org.apache.eventmesh.metrics.api;
-import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
+import static com.google.common.base.Preconditions.checkNotNull;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.eventmesh.spi.EventMeshExtensionFactory;
-import com.codahale.metrics.MetricRegistry;
+import lombok.experimental.UtilityClass;
-public class HealthMetrics {
+@UtilityClass
+public class MetricsPluginFactory {
- private EventMeshHTTPServer eventMeshHTTPServer;
- private MetricRegistry metricRegistry;
+ /**
+ * Get {@code MetricsRegistry}.
+ *
+ * @param metricsRegistryType
+ * @return
+ */
+ public static MetricsRegistry getMetricsRegistry(String metricsRegistryType) {
+ checkNotNull(metricsRegistryType, "MetricsRegistryType cannot be null");
- public Logger logger = LoggerFactory.getLogger("httpMonitor");
-
- public HealthMetrics(EventMeshHTTPServer eventMeshHTTPServer, MetricRegistry metricRegistry) {
- this.eventMeshHTTPServer = eventMeshHTTPServer;
- this.metricRegistry = metricRegistry;
+ MetricsRegistry metricsRegistry = EventMeshExtensionFactory.getExtension(MetricsRegistry.class, metricsRegistryType);
+ return checkNotNull(metricsRegistry, "MetricsRegistryType: " + metricsRegistryType + " is not supported");
}
}
diff --git a/eventmesh-metrics-plugin/eventmesh-metrics-api/src/main/java/org/apache/eventmesh/metrics/api/MetricsRegistry.java b/eventmesh-metrics-plugin/eventmesh-metrics-api/src/main/java/org/apache/eventmesh/metrics/api/MetricsRegistry.java
new file mode 100644
index 0000000..6a1bf2e
--- /dev/null
+++ b/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);
+}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/SummaryMetrics.java b/eventmesh-metrics-plugin/eventmesh-metrics-api/src/main/java/org/apache/eventmesh/metrics/api/model/HttpSummaryMetrics.java
similarity index 84%
rename from eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/SummaryMetrics.java
rename to eventmesh-metrics-plugin/eventmesh-metrics-api/src/main/java/org/apache/eventmesh/metrics/api/model/HttpSummaryMetrics.java
index 3bf4fe9..c8c90f6 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/SummaryMetrics.java
+++ b/eventmesh-metrics-plugin/eventmesh-metrics-api/src/main/java/org/apache/eventmesh/metrics/api/model/HttpSummaryMetrics.java
@@ -15,54 +15,36 @@
* limitations under the License.
*/
-package org.apache.eventmesh.runtime.metrics.http;
-
-import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
+package org.apache.eventmesh.metrics.api.model;
import java.util.Collections;
-import java.util.Iterator;
import java.util.LinkedList;
+import java.util.concurrent.DelayQueue;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicLong;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.codahale.metrics.MetricRegistry;
-
-public class SummaryMetrics {
+import lombok.extern.slf4j.Slf4j;
- public Logger logger = LoggerFactory.getLogger("httpMonitor");
+// todo: split this class
+@Slf4j
+public class HttpSummaryMetrics implements Metric {
- private EventMeshHTTPServer eventMeshHTTPServer;
+ private static final int STATIC_PERIOD = 30 * 1000;
- private MetricRegistry metricRegistry;
-
- public SummaryMetrics(EventMeshHTTPServer eventMeshHTTPServer, MetricRegistry metricRegistry) {
- this.eventMeshHTTPServer = eventMeshHTTPServer;
- this.metricRegistry = metricRegistry;
- }
-
- public static final int STATIC_PERIOD = 30 * 1000;
-
- private float avg(LinkedList<Integer> linkedList) {
- float sum = 0.0f;
- if (linkedList.isEmpty()) {
- return sum;
- }
-
- Iterator<Integer> it = linkedList.iterator();
- while (it.hasNext()) {
- float tps = (float) it.next();
- sum += tps;
- }
-
- return sum / linkedList.size();
+ public HttpSummaryMetrics(final ThreadPoolExecutor batchMsgExecutor,
+ final ThreadPoolExecutor sendMsgExecutor,
+ final ThreadPoolExecutor pushMsgExecutor,
+ final DelayQueue<?> httpFailedQueue) {
+ this.batchMsgExecutor = batchMsgExecutor;
+ this.sendMsgExecutor = sendMsgExecutor;
+ this.pushMsgExecutor = pushMsgExecutor;
+ this.httpFailedQueue = httpFailedQueue;
}
public static final String EVENTMESH_MONITOR_FORMAT_HTTP = "{\"maxHTTPTPS\":\"%.1f\",\"avgHTTPTPS\":\"%.1f\","
- //EVENTMESH tps related to accepting external http requests
- + "\"maxHTTPCOST\":\"%s\",\"avgHTTPCOST\":\"%.1f\",\"avgHTTPBodyDecodeCost\":\"%.1f\", "
- + "\"httpDiscard\":\"%s\"}";
+ //EVENTMESH tps related to accepting external http requests
+ + "\"maxHTTPCOST\":\"%s\",\"avgHTTPCOST\":\"%.1f\",\"avgHTTPBodyDecodeCost\":\"%.1f\", "
+ + "\"httpDiscard\":\"%s\"}";
private float wholeCost = 0f;
@@ -75,7 +57,7 @@ public class SummaryMetrics {
private AtomicLong httpRequestPerSecond = new AtomicLong(0);
- private LinkedList<Integer> httpRequestTPSSnapshots = new LinkedList<Integer>();
+ private LinkedList<Integer> httpRequestTPSSnapshots = new LinkedList<>();
public float avgHTTPCost() {
float cost = (wholeRequestNum.longValue() == 0L) ? 0f : wholeCost / wholeRequestNum.longValue();
@@ -151,7 +133,7 @@ public class SummaryMetrics {
//////////////////////////////////////////////////////////////////////////
public static final String EVENTMESH_MONITOR_FORMAT_BATCHSENDMSG = "{\"maxBatchSendMsgTPS\":\"%.1f\",\"avgBatchSendMsgTPS\":\"%.1f\","
- + " \"sum\":\"%s\", \"sumFail\":\"%s\", \"sumFailRate\":\"%.2f\", \"discard\":\"%s\"}";
+ + " \"sum\":\"%s\", \"sumFail\":\"%s\", \"sumFailRate\":\"%.2f\", \"discard\":\"%s\"}";
private AtomicLong sendBatchMsgNumPerSecond = new AtomicLong(0);
@@ -219,7 +201,7 @@ public class SummaryMetrics {
//////////////////////////////////////////////////////////////////////////
public static final String EVENTMESH_MONITOR_FORMAT_SENDMSG = "{\"maxSendMsgTPS\":\"%.1f\",\"avgSendMsgTPS\":\"%.1f\","
- + " \"sum\":\"%s\", \"sumFail\":\"%s\", \"sumFailRate\":\"%.2f\", \"replyMsg\":\"%s\", \"replyFail\":\"%s\"}";
+ + " \"sum\":\"%s\", \"sumFail\":\"%s\", \"sumFailRate\":\"%.2f\", \"replyMsg\":\"%s\", \"replyFail\":\"%s\"}";
private AtomicLong sendMsgNumSum = new AtomicLong(0);
@@ -298,7 +280,7 @@ public class SummaryMetrics {
////////////////////////////////////////////////////////////////////////////
public static final String EVENTMESH_MONITOR_FORMAT_PUSHMSG = "{\"maxPushMsgTPS\":\"%.1f\",\"avgPushMsgTPS\":\"%.1f\","
- + " \"sum\":\"%s\", \"sumFail\":\"%s\", \"sumFailRate\":\"%.1f\", \"maxClientLatency\":\"%.1f\", \"avgClientLatency\":\"%.1f\"}";
+ + " \"sum\":\"%s\", \"sumFail\":\"%s\", \"sumFailRate\":\"%.1f\", \"maxClientLatency\":\"%.1f\", \"avgClientLatency\":\"%.1f\"}";
private float wholePushCost = 0f;
@@ -380,11 +362,11 @@ public class SummaryMetrics {
///////////////////////////////////////////////////////////////////////////////////////////////////////////////
public static final String EVENTMESH_MONITOR_FORMAT_BLOCKQ = "{\"batchMsgQ\":\"%s\",\"sendMsgQ\":\"%s\","
- + "\"pushMsgQ\":\"%s\",\"httpRetryQ\":\"%s\"}";
+ + "\"pushMsgQ\":\"%s\",\"httpRetryQ\":\"%s\"}";
///////////////////////////////////////////////////////////////////////////
public static final String EVENTMESH_MONITOR_FORMAT_MQ_CLIENT = "{\"batchAvgSend2MQCost\":\"%.1f\", "
- + "\"avgSend2MQCost\":\"%.1f\", \"avgReply2MQCost\":\"%.1f\"}";
+ + "\"avgSend2MQCost\":\"%.1f\", \"avgReply2MQCost\":\"%.1f\"}";
private float batchSend2MQWholeCost = 0f;
@@ -436,4 +418,40 @@ public class SummaryMetrics {
reply2MQWholeCost = 0f;
reply2MQNum.set(0L);
}
+
+ // execute metrics
+ private final ThreadPoolExecutor batchMsgExecutor;
+
+ private final ThreadPoolExecutor sendMsgExecutor;
+
+ private final ThreadPoolExecutor pushMsgExecutor;
+
+ private final DelayQueue<?> httpFailedQueue;
+
+ public int getBatchMsgQueueSize() {
+ return batchMsgExecutor.getQueue().size();
+ }
+
+ public int getSendMsgQueueSize() {
+ return sendMsgExecutor.getQueue().size();
+ }
+
+ public int getPushMsgQueueSize() {
+ return pushMsgExecutor.getQueue().size();
+ }
+
+ public int getHttpRetryQueueSize() {
+ return httpFailedQueue.size();
+ }
+
+
+ private float avg(LinkedList<Integer> linkedList) {
+ if (linkedList.isEmpty()) {
+ return 0.0f;
+ }
+
+ int sum = linkedList.stream().reduce(Integer::sum).get();
+ return (float) sum / linkedList.size();
+ }
+
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/GroupMetrics.java b/eventmesh-metrics-plugin/eventmesh-metrics-api/src/main/java/org/apache/eventmesh/metrics/api/model/Metric.java
similarity index 62%
rename from eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/GroupMetrics.java
rename to eventmesh-metrics-plugin/eventmesh-metrics-api/src/main/java/org/apache/eventmesh/metrics/api/model/Metric.java
index 85ded0b..ab8903b 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/GroupMetrics.java
+++ b/eventmesh-metrics-plugin/eventmesh-metrics-api/src/main/java/org/apache/eventmesh/metrics/api/model/Metric.java
@@ -15,20 +15,10 @@
* limitations under the License.
*/
-package org.apache.eventmesh.runtime.metrics.http;
+package org.apache.eventmesh.metrics.api.model;
-import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
-
-import com.codahale.metrics.MetricRegistry;
-
-public class GroupMetrics {
-
- private EventMeshHTTPServer eventMeshHTTPServer;
-
- private MetricRegistry metricRegistry;
-
- public GroupMetrics(EventMeshHTTPServer eventMeshHTTPServer, MetricRegistry metricRegistry) {
- this.eventMeshHTTPServer = eventMeshHTTPServer;
- this.metricRegistry = metricRegistry;
- }
+/**
+ * Top-level metric, all metrics should implement this interface.
+ */
+public interface Metric {
}
diff --git a/eventmesh-metrics-plugin/eventmesh-metrics-api/src/main/java/org/apache/eventmesh/metrics/api/model/TcpSummaryMetrics.java b/eventmesh-metrics-plugin/eventmesh-metrics-api/src/main/java/org/apache/eventmesh/metrics/api/model/TcpSummaryMetrics.java
new file mode 100644
index 0000000..674e542
--- /dev/null
+++ b/eventmesh-metrics-plugin/eventmesh-metrics-api/src/main/java/org/apache/eventmesh/metrics/api/model/TcpSummaryMetrics.java
@@ -0,0 +1,123 @@
+/*
+ * 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.model;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import lombok.Data;
+
+@Data
+public class TcpSummaryMetrics implements Metric {
+ private final AtomicInteger client2eventMeshMsgNum;
+ private final AtomicInteger eventMesh2mqMsgNum;
+ private final AtomicInteger mq2eventMeshMsgNum;
+ private final AtomicInteger eventMesh2clientMsgNum;
+
+ private int client2eventMeshTPS;
+ private int eventMesh2clientTPS;
+ private int eventMesh2mqTPS;
+ private int mq2eventMeshTPS;
+ private int subTopicNum;
+
+ private int allConnections;
+
+ private int retrySize;
+
+ public TcpSummaryMetrics() {
+ this.client2eventMeshMsgNum = new AtomicInteger(0);
+ this.eventMesh2mqMsgNum = new AtomicInteger(0);
+ this.mq2eventMeshMsgNum = new AtomicInteger(0);
+ this.eventMesh2clientMsgNum = new AtomicInteger(0);
+ }
+
+ public int client2eventMeshMsgNum() {
+ return client2eventMeshMsgNum.get();
+ }
+
+ public int eventMesh2mqMsgNum() {
+ return eventMesh2mqMsgNum.get();
+ }
+
+ public int mq2eventMeshMsgNum() {
+ return mq2eventMeshMsgNum.get();
+ }
+
+ public int eventMesh2clientMsgNum() {
+ return eventMesh2clientMsgNum.get();
+ }
+
+ public int getClient2eventMeshTPS() {
+ return client2eventMeshTPS;
+ }
+
+ public void setClient2eventMeshTPS(int client2eventMeshTPS) {
+ this.client2eventMeshTPS = client2eventMeshTPS;
+ }
+
+ public int getEventMesh2clientTPS() {
+ return eventMesh2clientTPS;
+ }
+
+ public void setEventMesh2clientTPS(int eventMesh2clientTPS) {
+ this.eventMesh2clientTPS = eventMesh2clientTPS;
+ }
+
+ public int getEventMesh2mqTPS() {
+ return eventMesh2mqTPS;
+ }
+
+ public void setEventMesh2mqTPS(int eventMesh2mqTPS) {
+ this.eventMesh2mqTPS = eventMesh2mqTPS;
+ }
+
+ public int getMq2eventMeshTPS() {
+ return mq2eventMeshTPS;
+ }
+
+ public void setMq2eventMeshTPS(int mq2eventMeshTPS) {
+ this.mq2eventMeshTPS = mq2eventMeshTPS;
+ }
+
+ public int getAllTPS() {
+ return client2eventMeshTPS + eventMesh2clientTPS;
+ }
+
+ public int getSubTopicNum() {
+ return subTopicNum;
+ }
+
+ public void setSubTopicNum(int subTopicNum) {
+ this.subTopicNum = subTopicNum;
+ }
+
+ public int getAllConnections() {
+ return allConnections;
+ }
+
+ public void setAllConnections(int allConnections) {
+ this.allConnections = allConnections;
+ }
+
+ public void setRetrySize(int retrySize) {
+ this.retrySize = retrySize;
+ }
+
+ public int getRetrySize() {
+ return retrySize;
+ }
+}
diff --git a/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/build.gradle b/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/build.gradle
new file mode 100644
index 0000000..2257341
--- /dev/null
+++ b/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/build.gradle
@@ -0,0 +1,38 @@
+/*
+ * 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.
+ */
+
+dependencies {
+ implementation project(":eventmesh-metrics-plugin:eventmesh-metrics-api")
+ implementation project(":eventmesh-common")
+ implementation 'org.slf4j:slf4j-api'
+ implementation 'org.apache.commons:commons-lang3'
+ implementation 'com.google.guava:guava'
+
+ // todoļ¼Can we remove some dependency?
+ implementation 'io.opentelemetry:opentelemetry-api'
+ implementation 'io.opentelemetry:opentelemetry-sdk'
+ implementation 'io.opentelemetry:opentelemetry-sdk-metrics'
+ implementation 'io.opentelemetry:opentelemetry-exporter-prometheus'
+ implementation 'io.prometheus:simpleclient'
+ implementation 'io.prometheus:simpleclient_httpserver'
+
+ compileOnly 'org.projectlombok:lombok:1.18.22'
+ annotationProcessor 'org.projectlombok:lombok:1.18.22'
+
+ testCompileOnly 'org.projectlombok:lombok:1.18.22'
+ testAnnotationProcessor 'org.projectlombok:lombok:1.18.22'
+}
\ No newline at end of file
diff --git a/eventmesh-runtime/conf/prometheus.yml b/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/gradle.properties
similarity index 56%
copy from eventmesh-runtime/conf/prometheus.yml
copy to eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/gradle.properties
index ea48021..55ebd70 100644
--- a/eventmesh-runtime/conf/prometheus.yml
+++ b/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/gradle.properties
@@ -1,4 +1,3 @@
-#
# 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.
@@ -14,33 +13,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
-global:
- scrape_interval: 15s
- scrape_timeout: 10s
- evaluation_interval: 15s
-alerting:
- alertmanagers:
- - static_configs:
- - targets: []
- scheme: http
- timeout: 10s
- api_version: v1
-scrape_configs:
-- job_name: prometheus
- honor_timestamps: true
- scrape_interval: 15s
- scrape_timeout: 10s
- metrics_path: /metrics
- scheme: http
- static_configs:
- - targets:
- - localhost:9090
-- job_name: EventMesh_HTTP_export_test
- honor_timestamps: true
- scrape_interval: 15s
- scrape_timeout: 10s
- metrics_path: /metrics
- scheme: http
- static_configs:
- - targets:
- - 127.0.0.1:19090
+
+pluginType=metrics
+pluginName=opentelemetry
\ No newline at end of file
diff --git a/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/java/org/apache/eventmesh/metrics/opentelemetry/OpenTelemetryMetricsRegistry.java b/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/java/org/apache/eventmesh/metrics/opentelemetry/OpenTelemetryMetricsRegistry.java
new file mode 100644
index 0000000..b35d510
--- /dev/null
+++ b/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/java/org/apache/eventmesh/metrics/opentelemetry/OpenTelemetryMetricsRegistry.java
@@ -0,0 +1,87 @@
+/*
+ * 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.opentelemetry;
+
+import org.apache.eventmesh.metrics.api.MetricsRegistry;
+import org.apache.eventmesh.metrics.api.model.HttpSummaryMetrics;
+import org.apache.eventmesh.metrics.api.model.Metric;
+import org.apache.eventmesh.metrics.api.model.TcpSummaryMetrics;
+import org.apache.eventmesh.metrics.opentelemetry.config.OpenTelemetryConfiguration;
+import org.apache.eventmesh.metrics.opentelemetry.metrics.OpenTelemetryHttpExporter;
+import org.apache.eventmesh.metrics.opentelemetry.metrics.OpenTelemetryTcpExporter;
+
+import java.io.IOException;
+
+import io.opentelemetry.exporter.prometheus.PrometheusCollector;
+import io.opentelemetry.sdk.metrics.SdkMeterProvider;
+import io.prometheus.client.exporter.HTTPServer;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class OpenTelemetryMetricsRegistry implements MetricsRegistry {
+
+ private volatile HTTPServer prometheusHttpServer;
+
+ @Override
+ public void start() {
+ if (prometheusHttpServer == null) {
+ synchronized (OpenTelemetryMetricsRegistry.class) {
+ if (prometheusHttpServer == null) {
+ SdkMeterProvider sdkMeterProvider = SdkMeterProvider.builder().buildAndRegisterGlobal();
+ PrometheusCollector
+ .builder().setMetricProducer(sdkMeterProvider).buildAndRegister();
+ int port = OpenTelemetryConfiguration.getEventMeshPrometheusPort();
+ try {
+ //Use the daemon thread to start an HTTP server to serve the default Prometheus registry.
+ prometheusHttpServer = new HTTPServer(port, true);
+ } catch (IOException e) {
+ log.error("failed to start prometheus server, port: {} due to {}", port, e.getMessage());
+ }
+ }
+ }
+ }
+
+ }
+
+ @Override
+ public void showdown() {
+ if (prometheusHttpServer != null) {
+ prometheusHttpServer.stop();
+ }
+ }
+
+ @Override
+ public void register(Metric metric) {
+ if (metric == null) {
+ throw new IllegalArgumentException("Metric cannot be null");
+ }
+ if (metric instanceof HttpSummaryMetrics) {
+ OpenTelemetryHttpExporter.export("apache-eventmesh", (HttpSummaryMetrics) metric);
+ }
+
+ if (metric instanceof TcpSummaryMetrics) {
+ OpenTelemetryTcpExporter.export("apache-eventmesh", (TcpSummaryMetrics) metric);
+ }
+ }
+
+ @Override
+ public void unRegister(Metric metric) {
+ // todo: need to split the current metrics
+ }
+}
diff --git a/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/java/org/apache/eventmesh/metrics/opentelemetry/config/OpenTelemetryConfiguration.java b/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/java/org/apache/eventmesh/metrics/opentelemetry/config/OpenTelemetryConfiguration.java
new file mode 100644
index 0000000..8c31951
--- /dev/null
+++ b/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/java/org/apache/eventmesh/metrics/opentelemetry/config/OpenTelemetryConfiguration.java
@@ -0,0 +1,81 @@
+/*
+ * 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.opentelemetry.config;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.net.URL;
+import java.util.Properties;
+
+import lombok.experimental.UtilityClass;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@UtilityClass
+public class OpenTelemetryConfiguration {
+
+ private static final String CONFIG_FILE = "opentelemetry.properties";
+ private static final Properties properties = new Properties();
+
+ private int eventMeshPrometheusPort = 19090;
+
+ static {
+ loadProperties();
+ initializeConfig();
+ }
+
+ public static int getEventMeshPrometheusPort() {
+ return eventMeshPrometheusPort;
+ }
+
+ private void initializeConfig() {
+ String eventMeshPrometheusPortStr = properties.getProperty("eventMesh.metrics.prometheus.port");
+ if (StringUtils.isNotEmpty(eventMeshPrometheusPortStr)) {
+ eventMeshPrometheusPort = Integer.parseInt(StringUtils.deleteWhitespace(eventMeshPrometheusPortStr));
+ }
+ }
+
+ private void loadProperties() {
+ URL resource = OpenTelemetryConfiguration.class.getClassLoader().getResource(CONFIG_FILE);
+ if (resource != null) {
+ try (InputStream inputStream = resource.openStream()) {
+ if (inputStream.available() > 0) {
+ properties.load(new BufferedReader(new InputStreamReader(inputStream)));
+ }
+ } catch (IOException e) {
+ throw new RuntimeException("Load opentelemetry.properties file from classpath error");
+ }
+ }
+ // get from config home
+ try {
+ String configPath = System.getProperty("confPath", System.getenv("confPath")) + File.separator + CONFIG_FILE;
+ if (new File(configPath).exists()) {
+ properties.load(new BufferedReader(new FileReader(configPath)));
+ }
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Cannot load opentelemetry.properties file from conf");
+ }
+ }
+
+}
diff --git a/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/java/org/apache/eventmesh/metrics/opentelemetry/metrics/OpenTelemetryHttpExporter.java b/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/java/org/apache/eventmesh/metrics/opentelemetry/metrics/OpenTelemetryHttpExporter.java
new file mode 100644
index 0000000..8402043
--- /dev/null
+++ b/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/java/org/apache/eventmesh/metrics/opentelemetry/metrics/OpenTelemetryHttpExporter.java
@@ -0,0 +1,298 @@
+/*
+ * 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.opentelemetry.metrics;
+
+import org.apache.eventmesh.metrics.api.model.HttpSummaryMetrics;
+
+import io.opentelemetry.api.metrics.GlobalMeterProvider;
+import io.opentelemetry.api.metrics.Meter;
+import io.opentelemetry.api.metrics.common.Labels;
+
+import lombok.experimental.UtilityClass;
+
+@UtilityClass
+public class OpenTelemetryHttpExporter {
+
+ public static void export(String name, HttpSummaryMetrics summaryMetrics) {
+ Meter meter = GlobalMeterProvider.getMeter(name);
+ //maxHTTPTPS
+ meter
+ .doubleValueObserverBuilder("eventmesh.http.request.tps.max")
+ .setDescription("max TPS of HTTP.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.maxHTTPTPS(), Labels.empty()))
+ .build();
+
+ //avgHTTPTPS
+ meter
+ .doubleValueObserverBuilder("eventmesh.http.request.tps.avg")
+ .setDescription("avg TPS of HTTP.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.avgHTTPTPS(), Labels.empty()))
+ .build();
+
+ //maxHTTPCost
+ meter
+ .longValueObserverBuilder("eventmesh.http.request.cost.max")
+ .setDescription("max cost of HTTP.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.maxHTTPCost(), Labels.empty()))
+ .build();
+
+ //avgHTTPCost
+ meter
+ .doubleValueObserverBuilder("eventmesh.http.request.cost.avg")
+ .setDescription("avg cost of HTTP.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.avgHTTPCost(), Labels.empty()))
+ .build();
+
+ //avgHTTPBodyDecodeCost
+ meter
+ .doubleValueObserverBuilder("eventmesh.http.body.decode.cost.avg")
+ .setDescription("avg body decode cost of HTTP.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.avgHTTPBodyDecodeCost(), Labels.empty()))
+ .build();
+
+ //httpDiscard
+ meter
+ .longValueObserverBuilder("eventmesh.http.request.discard.num")
+ .setDescription("http request discard num.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.getHttpDiscard(), Labels.empty()))
+ .build();
+
+ //maxBatchSendMsgTPS
+ meter
+ .doubleValueObserverBuilder("eventmesh.batch.send.message.tps.max")
+ .setDescription("max of batch send message tps.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.maxSendBatchMsgTPS(), Labels.empty()))
+ .build();
+
+ //avgBatchSendMsgTPS
+ meter
+ .doubleValueObserverBuilder("eventmesh.batch.send.message.tps.avg")
+ .setDescription("avg of batch send message tps.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.avgSendBatchMsgTPS(), Labels.empty()))
+ .build();
+
+ //sum
+ meter
+ .doubleValueObserverBuilder("eventmesh.batch.send.message.num")
+ .setDescription("sum of batch send message number.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.getSendBatchMsgNumSum(), Labels.empty()))
+ .build();
+
+ //sumFail
+ meter
+ .doubleValueObserverBuilder("eventmesh.batch.send.message.fail.num")
+ .setDescription("sum of batch send message fail message number.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.getSendBatchMsgFailNumSum(), Labels.empty()))
+ .build();
+
+ //sumFailRate
+ meter
+ .doubleValueObserverBuilder("eventmesh.batch.send.message.fail.rate")
+ .setDescription("send batch message fail rate.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.getSendBatchMsgFailRate(), Labels.empty()))
+ .build();
+
+ //discard
+ meter
+ .doubleValueObserverBuilder("eventmesh.batch.send.message.discard.num")
+ .setDescription("sum of send batch message discard number.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.getSendBatchMsgDiscardNumSum(), Labels.empty()))
+ .build();
+
+ //maxSendMsgTPS
+ meter
+ .doubleValueObserverBuilder("eventmesh.send.message.tps.max")
+ .setDescription("max of send message tps.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.maxSendMsgTPS(), Labels.empty()))
+ .build();
+
+ //avgSendMsgTPS
+ meter
+ .doubleValueObserverBuilder("eventmesh.send.message.tps.avg")
+ .setDescription("avg of send message tps.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.avgSendMsgTPS(), Labels.empty()))
+ .build();
+
+ //sum
+ meter
+ .doubleValueObserverBuilder("eventmesh.send.message.num")
+ .setDescription("sum of send message number.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.getSendMsgNumSum(), Labels.empty()))
+ .build();
+
+ //sumFail
+ meter
+ .doubleValueObserverBuilder("eventmesh.send.message.fail.num")
+ .setDescription("sum of send message fail number.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.getSendMsgFailNumSum(), Labels.empty()))
+ .build();
+
+ //sumFailRate
+ meter
+ .doubleValueObserverBuilder("eventmesh.send.message.fail.rate")
+ .setDescription("send message fail rate.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.getSendMsgFailRate(), Labels.empty()))
+ .build();
+
+ //replyMsg
+ meter
+ .doubleValueObserverBuilder("eventmesh.reply.message.num")
+ .setDescription("sum of reply message number.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.getReplyMsgNumSum(), Labels.empty()))
+ .build();
+
+ //replyFail
+ meter
+ .doubleValueObserverBuilder("eventmesh.reply.message.fail.num")
+ .setDescription("sum of reply message fail number.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.getReplyMsgFailNumSum(), Labels.empty()))
+ .build();
+
+ //maxPushMsgTPS
+ meter
+ .doubleValueObserverBuilder("eventmesh.push.message.tps.max")
+ .setDescription("max of push message tps.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.maxPushMsgTPS(), Labels.empty()))
+ .build();
+
+ //avgPushMsgTPS
+ meter
+ .doubleValueObserverBuilder("eventmesh.push.message.tps.avg")
+ .setDescription("avg of push message tps.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.avgPushMsgTPS(), Labels.empty()))
+ .build();
+
+ //sum
+ meter
+ .doubleValueObserverBuilder("eventmesh.http.push.message.num")
+ .setDescription("sum of http push message number.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.getHttpPushMsgNumSum(), Labels.empty()))
+ .build();
+
+ //sumFail
+ meter
+ .doubleValueObserverBuilder("eventmesh.http.push.message.fail.num")
+ .setDescription("sum of http push message fail number.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.getHttpPushFailNumSum(), Labels.empty()))
+ .build();
+
+ //sumFailRate
+ meter
+ .doubleValueObserverBuilder("eventmesh.http.push.message.fail.rate")
+ .setDescription("http push message fail rate.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.getHttpPushMsgFailRate(), Labels.empty()))
+ .build();
+
+ //maxClientLatency
+ meter
+ .doubleValueObserverBuilder("eventmesh.http.push.latency.max")
+ .setDescription("max of http push latency.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.maxHTTPPushLatency(), Labels.empty()))
+ .build();
+
+ //avgClientLatency
+ meter
+ .doubleValueObserverBuilder("eventmesh.http.push.latency.avg")
+ .setDescription("avg of http push latency.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.avgHTTPPushLatency(), Labels.empty()))
+ .build();
+
+ //batchMsgQ
+ meter
+ .longValueObserverBuilder("eventmesh.batch.message.queue.size")
+ .setDescription("size of batch message queue.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.getBatchMsgQueueSize(), Labels.empty()))
+ .build();
+
+ //sendMsgQ
+ meter
+ .longValueObserverBuilder("eventmesh.send.message.queue.size")
+ .setDescription("size of send message queue.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.getSendMsgQueueSize(), Labels.empty()))
+ .build();
+
+ //pushMsgQ
+ meter
+ .longValueObserverBuilder("eventmesh.push.message.queue.size")
+ .setDescription("size of push message queue.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.getPushMsgQueueSize(), Labels.empty()))
+ .build();
+
+ //httpRetryQ
+ meter
+ .longValueObserverBuilder("eventmesh.http.retry.queue.size")
+ .setDescription("size of http retry queue.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.getHttpRetryQueueSize(), Labels.empty()))
+ .build();
+
+ //batchAvgSend2MQCost
+ meter
+ .doubleValueObserverBuilder("eventmesh.batch.send.message.cost.avg")
+ .setDescription("avg of batch send message cost.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.avgBatchSendMsgCost(), Labels.empty()))
+ .build();
+
+ //avgSend2MQCost
+ meter
+ .doubleValueObserverBuilder("eventmesh.send.message.cost.avg")
+ .setDescription("avg of send message cost.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.avgSendMsgCost(), Labels.empty()))
+ .build();
+
+ //avgReply2MQCost
+ meter
+ .doubleValueObserverBuilder("eventmesh.reply.message.cost.avg")
+ .setDescription("avg of reply message cost.")
+ .setUnit("HTTP")
+ .setUpdater(result -> result.observe(summaryMetrics.avgReplyMsgCost(), Labels.empty()))
+ .build();
+ }
+
+}
diff --git a/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/java/org/apache/eventmesh/metrics/opentelemetry/metrics/OpenTelemetryTcpExporter.java b/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/java/org/apache/eventmesh/metrics/opentelemetry/metrics/OpenTelemetryTcpExporter.java
new file mode 100644
index 0000000..1d3db5b
--- /dev/null
+++ b/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/java/org/apache/eventmesh/metrics/opentelemetry/metrics/OpenTelemetryTcpExporter.java
@@ -0,0 +1,89 @@
+/*
+ * 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.opentelemetry.metrics;
+
+import org.apache.eventmesh.metrics.api.model.TcpSummaryMetrics;
+
+import io.opentelemetry.api.metrics.GlobalMeterProvider;
+import io.opentelemetry.api.metrics.Meter;
+import io.opentelemetry.api.metrics.common.Labels;
+
+import lombok.experimental.UtilityClass;
+
+@UtilityClass
+public class OpenTelemetryTcpExporter {
+
+ public static void export(final String meterName, final TcpSummaryMetrics summaryMetrics) {
+ final Meter meter = GlobalMeterProvider.getMeter(meterName);
+ //retryQueueSize
+ meter.doubleValueObserverBuilder("eventmesh.tcp.retry.queue.size")
+ .setDescription("get size of retry queue.")
+ .setUnit("TCP")
+ .setUpdater(result -> result.observe(summaryMetrics.getRetrySize(), Labels.empty()))
+ .build();
+
+ //client2eventMeshTPS
+ meter.doubleValueObserverBuilder("eventmesh.tcp.server.tps")
+ .setDescription("get tps of client to eventMesh.")
+ .setUnit("TCP")
+ .setUpdater(result -> result.observe(summaryMetrics.getClient2eventMeshTPS(), Labels.empty()))
+ .build();
+
+ //eventMesh2mqTPS
+ meter.doubleValueObserverBuilder("eventmesh.tcp.mq.provider.tps")
+ .setDescription("get tps of eventMesh to mq.")
+ .setUnit("TCP")
+ .setUpdater(result -> result.observe(summaryMetrics.getEventMesh2mqTPS(), Labels.empty()))
+ .build();
+
+ //mq2eventMeshTPS
+ meter.doubleValueObserverBuilder("eventmesh.tcp.mq.consumer.tps")
+ .setDescription("get tps of mq to eventMesh.")
+ .setUnit("TCP")
+ .setUpdater(result -> result.observe(summaryMetrics.getMq2eventMeshTPS(), Labels.empty()))
+ .build();
+
+ //eventMesh2clientTPS
+ meter.doubleValueObserverBuilder("eventmesh.tcp.client.tps")
+ .setDescription("get tps of eventMesh to client.")
+ .setUnit("TCP")
+ .setUpdater(result -> result.observe(summaryMetrics.getEventMesh2clientTPS(), Labels.empty()))
+ .build();
+
+ //allTPS
+ meter.doubleValueObserverBuilder("eventmesh.tcp.all.tps")
+ .setDescription("get all TPS.")
+ .setUnit("TCP")
+ .setUpdater(result -> result.observe(summaryMetrics.getAllTPS(), Labels.empty()))
+ .build();
+
+ //EventMeshTcpConnectionHandler.connections
+ meter.doubleValueObserverBuilder("eventmesh.tcp.connection.num")
+ .setDescription("EventMeshTcpConnectionHandler.connections.")
+ .setUnit("TCP")
+ .setUpdater(result -> result.observe(summaryMetrics.getAllConnections(), Labels.empty()))
+ .build();
+
+ //subTopicNum
+ meter.doubleValueObserverBuilder("eventmesh.tcp.sub.topic.num")
+ .setDescription("get sub topic num.")
+ .setUnit("TCP")
+ .setUpdater(result -> result.observe(summaryMetrics.getSubTopicNum(), Labels.empty()))
+ .build();
+ }
+}
diff --git a/eventmesh-runtime/conf/prometheus.yml b/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.metrics.api.MetricsRegistry
similarity index 56%
copy from eventmesh-runtime/conf/prometheus.yml
copy to eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.metrics.api.MetricsRegistry
index ea48021..a7b76d4 100644
--- a/eventmesh-runtime/conf/prometheus.yml
+++ b/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.metrics.api.MetricsRegistry
@@ -1,4 +1,3 @@
-#
# 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.
@@ -13,34 +12,5 @@
# 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.
-#
-global:
- scrape_interval: 15s
- scrape_timeout: 10s
- evaluation_interval: 15s
-alerting:
- alertmanagers:
- - static_configs:
- - targets: []
- scheme: http
- timeout: 10s
- api_version: v1
-scrape_configs:
-- job_name: prometheus
- honor_timestamps: true
- scrape_interval: 15s
- scrape_timeout: 10s
- metrics_path: /metrics
- scheme: http
- static_configs:
- - targets:
- - localhost:9090
-- job_name: EventMesh_HTTP_export_test
- honor_timestamps: true
- scrape_interval: 15s
- scrape_timeout: 10s
- metrics_path: /metrics
- scheme: http
- static_configs:
- - targets:
- - 127.0.0.1:19090
+
+opentelemetry=org.apache.eventmesh.metrics.opentelemetry.OpenTelemetryMetricsRegistry
\ No newline at end of file
diff --git a/eventmesh-runtime/conf/prometheus.yml b/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/resources/opentelemetry.properties
similarity index 56%
copy from eventmesh-runtime/conf/prometheus.yml
copy to eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/resources/opentelemetry.properties
index ea48021..4a8da07 100644
--- a/eventmesh-runtime/conf/prometheus.yml
+++ b/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/resources/opentelemetry.properties
@@ -14,33 +14,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
-global:
- scrape_interval: 15s
- scrape_timeout: 10s
- evaluation_interval: 15s
-alerting:
- alertmanagers:
- - static_configs:
- - targets: []
- scheme: http
- timeout: 10s
- api_version: v1
-scrape_configs:
-- job_name: prometheus
- honor_timestamps: true
- scrape_interval: 15s
- scrape_timeout: 10s
- metrics_path: /metrics
- scheme: http
- static_configs:
- - targets:
- - localhost:9090
-- job_name: EventMesh_HTTP_export_test
- honor_timestamps: true
- scrape_interval: 15s
- scrape_timeout: 10s
- metrics_path: /metrics
- scheme: http
- static_configs:
- - targets:
- - 127.0.0.1:19090
+
+#prometheusPort
+eventMesh.metrics.prometheus.port=19090
\ No newline at end of file
diff --git a/eventmesh-runtime/conf/prometheus.yml b/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/resources/prometheus.yml
similarity index 99%
copy from eventmesh-runtime/conf/prometheus.yml
copy to eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/resources/prometheus.yml
index ea48021..8703ea2 100644
--- a/eventmesh-runtime/conf/prometheus.yml
+++ b/eventmesh-metrics-plugin/eventmesh-metrics-opentelemetry/src/main/resources/prometheus.yml
@@ -14,6 +14,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
+
global:
scrape_interval: 15s
scrape_timeout: 10s
diff --git a/eventmesh-runtime/conf/prometheus.yml b/eventmesh-metrics-plugin/gradle.properties
similarity index 56%
rename from eventmesh-runtime/conf/prometheus.yml
rename to eventmesh-metrics-plugin/gradle.properties
index ea48021..a9fd83f 100644
--- a/eventmesh-runtime/conf/prometheus.yml
+++ b/eventmesh-metrics-plugin/gradle.properties
@@ -13,34 +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.
-#
-global:
- scrape_interval: 15s
- scrape_timeout: 10s
- evaluation_interval: 15s
-alerting:
- alertmanagers:
- - static_configs:
- - targets: []
- scheme: http
- timeout: 10s
- api_version: v1
-scrape_configs:
-- job_name: prometheus
- honor_timestamps: true
- scrape_interval: 15s
- scrape_timeout: 10s
- metrics_path: /metrics
- scheme: http
- static_configs:
- - targets:
- - localhost:9090
-- job_name: EventMesh_HTTP_export_test
- honor_timestamps: true
- scrape_interval: 15s
- scrape_timeout: 10s
- metrics_path: /metrics
- scheme: http
- static_configs:
- - targets:
- - 127.0.0.1:19090
+#
\ No newline at end of file
diff --git a/eventmesh-runtime/build.gradle b/eventmesh-runtime/build.gradle
index 6c34c51..6db36c2 100644
--- a/eventmesh-runtime/build.gradle
+++ b/eventmesh-runtime/build.gradle
@@ -16,17 +16,13 @@
*/
dependencies {
+ implementation 'io.cloudevents:cloudevents-core'
+
implementation 'io.opentelemetry:opentelemetry-api'
implementation 'io.opentelemetry:opentelemetry-sdk'
- implementation 'io.opentelemetry:opentelemetry-sdk-metrics'
- implementation 'io.opentelemetry:opentelemetry-exporter-prometheus'
- implementation 'io.prometheus:simpleclient'
- implementation 'io.prometheus:simpleclient_httpserver'
- implementation 'io.cloudevents:cloudevents-core'
implementation 'io.opentelemetry:opentelemetry-exporter-zipkin'
implementation 'io.opentelemetry:opentelemetry-semconv'
-
implementation "org.apache.httpcomponents:httpclient"
implementation 'io.netty:netty-all'
@@ -40,25 +36,12 @@ dependencies {
implementation project(":eventmesh-admin:eventmesh-admin-rocketmq")
implementation project(":eventmesh-protocol-plugin:eventmesh-protocol-api")
-
- // for debug only, can be removed
implementation project(":eventmesh-protocol-plugin:eventmesh-protocol-cloudevents")
implementation project(":eventmesh-protocol-plugin:eventmesh-protocol-meshmessage")
implementation project(":eventmesh-protocol-plugin:eventmesh-protocol-openmessage")
- testImplementation project(":eventmesh-common")
- testImplementation project(":eventmesh-spi")
- testImplementation project(":eventmesh-connector-plugin:eventmesh-connector-api")
- testImplementation project(":eventmesh-connector-plugin:eventmesh-connector-standalone")
- testImplementation project(":eventmesh-security-plugin:eventmesh-security-api")
- testImplementation project(":eventmesh-security-plugin:eventmesh-security-acl")
- testImplementation project(":eventmesh-registry-plugin:eventmesh-registry-api")
- testImplementation project(":eventmesh-admin:eventmesh-admin-rocketmq")
-
- testImplementation project(":eventmesh-protocol-plugin:eventmesh-protocol-api")
-
- testImplementation "org.apache.httpcomponents:httpclient"
- testImplementation "io.netty:netty-all"
+ implementation project(":eventmesh-metrics-plugin:eventmesh-metrics-api")
+ implementation project(":eventmesh-metrics-plugin:eventmesh-metrics-opentelemetry")
testImplementation "org.mockito:mockito-core"
testImplementation "org.powermock:powermock-module-junit4"
diff --git a/eventmesh-runtime/conf/eventmesh.properties b/eventmesh-runtime/conf/eventmesh.properties
index 755eeb3..000cd5d 100644
--- a/eventmesh-runtime/conf/eventmesh.properties
+++ b/eventmesh-runtime/conf/eventmesh.properties
@@ -71,8 +71,8 @@ eventMesh.security.plugin.type=security
eventMesh.server.registry.enabled=false
eventMesh.registry.plugin.type=namesrv
-#prometheusPort
-eventMesh.metrics.prometheus.port=19090
+# metrics plugin, if you have multiple plugin, you can use ',' to split
+eventmesh.metrics.plugin=opentelemetry
#trace exporter
eventmesh.trace.exporter.type=Zipkin
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
index 6a012e7..592854d 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
@@ -296,7 +296,7 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
}
decoder.destroy();
}
- metrics.summaryMetrics.recordDecodeTimeCost(System.currentTimeMillis() - bodyDecodeStart);
+ metrics.getSummaryMetrics().recordDecodeTimeCost(System.currentTimeMillis() - bodyDecodeStart);
return httpRequestBody;
}
@@ -336,7 +336,7 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
sendError(ctx, errorStatus);
return;
}
- metrics.summaryMetrics.recordHTTPRequest();
+ metrics.getSummaryMetrics().recordHTTPRequest();
final HttpCommand requestCommand = new HttpCommand();
@@ -422,7 +422,7 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
return;
}
- metrics.summaryMetrics
+ metrics.getSummaryMetrics()
.recordHTTPReqResTimeCost(System.currentTimeMillis() - request.getReqTime());
if (httpLogger.isDebugEnabled()) {
@@ -437,8 +437,8 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
} catch (RejectedExecutionException re) {
HttpCommand responseCommand = request.createHttpCommandResponse(EventMeshRetCode.OVERLOAD);
asyncContext.onComplete(responseCommand);
- metrics.summaryMetrics.recordHTTPDiscard();
- metrics.summaryMetrics.recordHTTPReqResTimeCost(System.currentTimeMillis() - request.getReqTime());
+ metrics.getSummaryMetrics().recordHTTPDiscard();
+ metrics.getSummaryMetrics().recordHTTPReqResTimeCost(System.currentTimeMillis() - request.getReqTime());
try {
sendResponse(ctx, asyncContext.getResponse().httpResponse());
} catch (Exception e) {
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
index 06e6ab6..590138a 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
@@ -19,6 +19,8 @@ package org.apache.eventmesh.runtime.boot;
import org.apache.eventmesh.common.ThreadPoolFactory;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
+import org.apache.eventmesh.metrics.api.MetricsPluginFactory;
+import org.apache.eventmesh.metrics.api.MetricsRegistry;
import org.apache.eventmesh.runtime.common.ServiceState;
import org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupConf;
@@ -40,11 +42,14 @@ import org.apache.eventmesh.runtime.metrics.http.HTTPMetricsServer;
import org.apache.eventmesh.runtime.trace.OpenTelemetryTraceFactory;
import java.util.List;
+import java.util.Optional;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
+import org.assertj.core.util.Lists;
+
import com.google.common.eventbus.EventBus;
import com.google.common.util.concurrent.RateLimiter;
@@ -190,7 +195,17 @@ public class EventMeshHTTPServer extends AbstractHTTPServer {
msgRateLimiter = RateLimiter.create(eventMeshHttpConfiguration.eventMeshHttpMsgReqNumPerSecond);
batchRateLimiter = RateLimiter.create(eventMeshHttpConfiguration.eventMeshBatchMsgRequestNumPerSecond);
- metrics = new HTTPMetricsServer(this);
+ // The MetricsRegistry is singleton, so we can use factory method to get.
+ final List<MetricsRegistry> metricsRegistries = Lists.newArrayList();
+ Optional.ofNullable(eventMeshHttpConfiguration.eventMeshMetricsPluginType)
+ .ifPresent(
+ metricsPlugins -> metricsPlugins.forEach(
+ pluginType -> metricsRegistries.add(MetricsPluginFactory.getMetricsRegistry(pluginType))));
+
+ httpRetryer = new HttpRetryer(this);
+ httpRetryer.init();
+
+ metrics = new HTTPMetricsServer(this, metricsRegistries);
metrics.init();
consumerManager = new ConsumerManager(this);
@@ -199,9 +214,6 @@ public class EventMeshHTTPServer extends AbstractHTTPServer {
producerManager = new ProducerManager(this);
producerManager.init();
- httpRetryer = new HttpRetryer(this);
- httpRetryer.init();
-
registerHTTPRequestProcessor();
super.openTelemetryTraceFactory = new OpenTelemetryTraceFactory(eventMeshHttpConfiguration);
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTCPServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTCPServer.java
index fe471d2..c94f824 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTCPServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTCPServer.java
@@ -23,6 +23,8 @@ import org.apache.eventmesh.common.ThreadPoolFactory;
import org.apache.eventmesh.common.exception.EventMeshException;
import org.apache.eventmesh.common.protocol.tcp.codec.Codec;
import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.metrics.api.MetricsPluginFactory;
+import org.apache.eventmesh.metrics.api.MetricsRegistry;
import org.apache.eventmesh.runtime.admin.controller.ClientManageController;
import org.apache.eventmesh.runtime.configuration.EventMeshTCPConfiguration;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
@@ -37,12 +39,16 @@ import org.apache.eventmesh.runtime.metrics.tcp.EventMeshTcpMonitor;
import org.apache.eventmesh.runtime.registry.Registry;
import org.apache.eventmesh.runtime.util.EventMeshThreadFactoryImpl;
+import java.util.List;
+import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
+import org.assertj.core.util.Lists;
+
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.AdaptiveRecvByteBufAllocator;
@@ -210,12 +216,18 @@ public class EventMeshTCPServer extends AbstractRemotingServer {
eventMeshTcpRetryer = new EventMeshTcpRetryer(this);
eventMeshTcpRetryer.init();
- eventMeshTcpMonitor = new EventMeshTcpMonitor(this);
+ // The MetricsRegistry is singleton, so we can use factory method to get.
+ final List<MetricsRegistry> metricsRegistries = Lists.newArrayList();
+ Optional.ofNullable(eventMeshTCPConfiguration.eventMeshMetricsPluginType)
+ .ifPresent(
+ metricsPlugins -> metricsPlugins.forEach(
+ pluginType -> metricsRegistries.add(MetricsPluginFactory.getMetricsRegistry(pluginType))));
+ eventMeshTcpMonitor = new EventMeshTcpMonitor(this, metricsRegistries);
eventMeshTcpMonitor.init();
if (eventMeshTCPConfiguration.eventMeshServerRegistryEnable) {
eventMeshRebalanceService = new EventMeshRebalanceService(this,
- new EventmeshRebalanceImpl(this));
+ new EventmeshRebalanceImpl(this));
eventMeshRebalanceService.init();
}
logger.info("--------------------------EventMeshTCPServer Inited");
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageProcessor.java
index afdd365..2f2103b 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageProcessor.java
@@ -166,11 +166,10 @@ public class BatchSendMessageProcessor implements HttpRequestProcessor {
if (!eventMeshHTTPServer.getBatchRateLimiter()
.tryAcquire(eventSize, EventMeshConstants.DEFAULT_FASTFAIL_TIMEOUT_IN_MILLISECONDS, TimeUnit.MILLISECONDS)) {
responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchResponseHeader,
- SendMessageBatchResponseBody.buildBody(EventMeshRetCode.EVENTMESH_BATCH_SPEED_OVER_LIMIT_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_BATCH_SPEED_OVER_LIMIT_ERR.getErrMsg()));
- eventMeshHTTPServer.metrics.summaryMetrics
- .recordSendBatchMsgDiscard(eventSize);
+ sendMessageBatchResponseHeader,
+ SendMessageBatchResponseBody.buildBody(EventMeshRetCode.EVENTMESH_BATCH_SPEED_OVER_LIMIT_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_BATCH_SPEED_OVER_LIMIT_ERR.getErrMsg()));
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendBatchMsgDiscard(eventSize);
asyncContext.onComplete(responseEventMeshCommand);
return;
}
@@ -258,7 +257,7 @@ public class BatchSendMessageProcessor implements HttpRequestProcessor {
}
long delta = eventSize;
- eventMeshHTTPServer.metrics.summaryMetrics.recordSendBatchMsg(delta);
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendBatchMsg(delta);
if (eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshServerBatchMsgBatchEnabled) {
for (List<CloudEvent> eventlist : topicBatchMessageMappings.values()) {
@@ -314,7 +313,7 @@ public class BatchSendMessageProcessor implements HttpRequestProcessor {
}
long batchEndTime = System.currentTimeMillis();
- eventMeshHTTPServer.metrics.summaryMetrics.recordBatchSendMsgCost(batchEndTime - batchStartTime);
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordBatchSendMsgCost(batchEndTime - batchStartTime);
batchMessageLogger.debug("batchMessage|eventMesh2mq|REQ|ASYNC|batchId={}|send2MQCost={}ms|msgNum={}|topics={}",
batchId, batchEndTime - batchStartTime, eventSize, topicBatchMessageMappings.keySet());
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java
index a1485e6..9b9f44e 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java
@@ -185,12 +185,11 @@ public class BatchSendMessageV2Processor implements HttpRequestProcessor {
.tryAcquire(EventMeshConstants.DEFAULT_FASTFAIL_TIMEOUT_IN_MILLISECONDS,
TimeUnit.MILLISECONDS)) {
responseEventMeshCommand = request.createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
- SendMessageBatchV2ResponseBody
- .buildBody(EventMeshRetCode.EVENTMESH_BATCH_SPEED_OVER_LIMIT_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_BATCH_SPEED_OVER_LIMIT_ERR.getErrMsg()));
- eventMeshHTTPServer.metrics.summaryMetrics
- .recordSendBatchMsgDiscard(1);
+ sendMessageBatchV2ResponseHeader,
+ SendMessageBatchV2ResponseBody
+ .buildBody(EventMeshRetCode.EVENTMESH_BATCH_SPEED_OVER_LIMIT_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_BATCH_SPEED_OVER_LIMIT_ERR.getErrMsg()));
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendBatchMsgDiscard(1);
asyncContext.onComplete(responseEventMeshCommand);
return;
}
@@ -244,7 +243,7 @@ public class BatchSendMessageV2Processor implements HttpRequestProcessor {
return;
}
- eventMeshHTTPServer.metrics.summaryMetrics.recordSendBatchMsg(1);
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendBatchMsg(1);
final SendMessageContext sendMessageContext =
new SendMessageContext(bizNo, event, batchEventMeshProducer, eventMeshHTTPServer);
@@ -254,22 +253,20 @@ public class BatchSendMessageV2Processor implements HttpRequestProcessor {
@Override
public void onSuccess(SendResult sendResult) {
long batchEndTime = System.currentTimeMillis();
- eventMeshHTTPServer.metrics.summaryMetrics
- .recordBatchSendMsgCost(batchEndTime - batchStartTime);
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordBatchSendMsgCost(batchEndTime - batchStartTime);
batchMessageLogger.debug(
- "batchMessageV2|eventMesh2mq|REQ|ASYNC|bizSeqNo={}|send2MQCost={}ms|topic={}",
- bizNo, batchEndTime - batchStartTime, topic);
+ "batchMessageV2|eventMesh2mq|REQ|ASYNC|bizSeqNo={}|send2MQCost={}ms|topic={}",
+ bizNo, batchEndTime - batchStartTime, topic);
}
@Override
public void onException(OnExceptionContext context) {
long batchEndTime = System.currentTimeMillis();
eventMeshHTTPServer.getHttpRetryer().pushRetry(sendMessageContext.delay(10000));
- eventMeshHTTPServer.metrics.summaryMetrics
- .recordBatchSendMsgCost(batchEndTime - batchStartTime);
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordBatchSendMsgCost(batchEndTime - batchStartTime);
batchMessageLogger.error(
- "batchMessageV2|eventMesh2mq|REQ|ASYNC|bizSeqNo={}|send2MQCost={}ms|topic={}",
- bizNo, batchEndTime - batchStartTime, topic, context.getException());
+ "batchMessageV2|eventMesh2mq|REQ|ASYNC|bizSeqNo={}|send2MQCost={}ms|topic={}",
+ bizNo, batchEndTime - batchStartTime, topic, context.getException());
}
});
@@ -284,11 +281,10 @@ public class BatchSendMessageV2Processor implements HttpRequestProcessor {
asyncContext.onComplete(responseEventMeshCommand);
long batchEndTime = System.currentTimeMillis();
eventMeshHTTPServer.getHttpRetryer().pushRetry(sendMessageContext.delay(10000));
- eventMeshHTTPServer.metrics.summaryMetrics
- .recordBatchSendMsgCost(batchEndTime - batchStartTime);
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordBatchSendMsgCost(batchEndTime - batchStartTime);
batchMessageLogger.error(
- "batchMessageV2|eventMesh2mq|REQ|ASYNC|bizSeqNo={}|send2MQCost={}ms|topic={}",
- bizNo, batchEndTime - batchStartTime, topic, e);
+ "batchMessageV2|eventMesh2mq|REQ|ASYNC|bizSeqNo={}|send2MQCost={}ms|topic={}",
+ bizNo, batchEndTime - batchStartTime, topic, e);
}
responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HeartBeatProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HeartBeatProcessor.java
index a2e4e70..0051afd 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HeartBeatProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HeartBeatProcessor.java
@@ -192,7 +192,7 @@ public class HeartBeatProcessor implements HttpRequestProcessor {
httpLogger.debug("{}", httpCommand);
}
eventMeshHTTPServer.sendResponse(ctx, httpCommand.httpResponse());
- eventMeshHTTPServer.metrics.summaryMetrics.recordHTTPReqResTimeCost(
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordHTTPReqResTimeCost(
System.currentTimeMillis() - asyncContext.getRequest().getReqTime());
} catch (Exception ex) {
//ignore
@@ -211,8 +211,8 @@ public class HeartBeatProcessor implements HttpRequestProcessor {
long endTime = System.currentTimeMillis();
httpLogger.error("message|eventMesh2mq|REQ|ASYNC|heartBeatMessageCost={}ms",
endTime - startTime, e);
- eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsgFailed();
- eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsgCost(endTime - startTime);
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgFailed();
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgCost(endTime - startTime);
}
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
index cb5e1d4..45ec636 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
@@ -144,10 +144,10 @@ public class ReplyMessageProcessor implements HttpRequestProcessor {
if (!eventMeshHTTPServer.getMsgRateLimiter()
.tryAcquire(EventMeshConstants.DEFAULT_FASTFAIL_TIMEOUT_IN_MILLISECONDS, TimeUnit.MILLISECONDS)) {
responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- replyMessageResponseHeader,
- ReplyMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getErrMsg()));
- eventMeshHTTPServer.metrics.summaryMetrics.recordHTTPDiscard();
+ replyMessageResponseHeader,
+ ReplyMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getErrMsg()));
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordHTTPDiscard();
asyncContext.onComplete(responseEventMeshCommand);
return;
}
@@ -229,37 +229,34 @@ public class ReplyMessageProcessor implements HttpRequestProcessor {
} catch (Exception e) {
messageLogger.error("msg2MQMsg err, bizSeqNo={}, topic={}", bizNo, replyTopic, e);
responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- replyMessageResponseHeader,
- ReplyMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getErrMsg() + EventMeshUtil.stackTrace(e, 2)));
+ replyMessageResponseHeader,
+ ReplyMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getErrMsg() + EventMeshUtil.stackTrace(e, 2)));
asyncContext.onComplete(responseEventMeshCommand);
return;
}
final SendMessageContext sendMessageContext = new SendMessageContext(bizNo, event, eventMeshProducer, eventMeshHTTPServer);
- eventMeshHTTPServer.metrics.summaryMetrics.recordReplyMsg();
-
- CompleteHandler<HttpCommand> handler = new CompleteHandler<HttpCommand>() {
- @Override
- public void onResponse(HttpCommand httpCommand) {
- try {
- if (httpLogger.isDebugEnabled()) {
- httpLogger.debug("{}", httpCommand);
- }
- eventMeshHTTPServer.sendResponse(ctx, httpCommand.httpResponse());
- eventMeshHTTPServer.metrics.summaryMetrics.recordHTTPReqResTimeCost(
- System.currentTimeMillis() - asyncContext.getRequest().getReqTime());
- } catch (Exception ex) {
- //ignore
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordReplyMsg();
+
+ CompleteHandler<HttpCommand> handler = httpCommand -> {
+ try {
+ if (httpLogger.isDebugEnabled()) {
+ httpLogger.debug("{}", httpCommand);
}
+ eventMeshHTTPServer.sendResponse(ctx, httpCommand.httpResponse());
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordHTTPReqResTimeCost(
+ System.currentTimeMillis() - asyncContext.getRequest().getReqTime());
+ } catch (Exception ex) {
+ //ignore
}
};
try {
CloudEvent clone = CloudEventBuilder.from(sendMessageContext.getEvent())
- .withExtension(EventMeshConstants.REQ_EVENTMESH2MQ_TIMESTAMP, String.valueOf(System.currentTimeMillis()))
- .build();
+ .withExtension(EventMeshConstants.REQ_EVENTMESH2MQ_TIMESTAMP, String.valueOf(System.currentTimeMillis()))
+ .build();
sendMessageContext.setEvent(clone);
eventMeshProducer.reply(sendMessageContext, new SendCallback() {
@Override
@@ -269,26 +266,26 @@ public class ReplyMessageProcessor implements HttpRequestProcessor {
SendMessageResponseBody.buildBody(EventMeshRetCode.SUCCESS.getRetCode(), EventMeshRetCode.SUCCESS.getErrMsg()));
asyncContext.onComplete(succ, handler);
long endTime = System.currentTimeMillis();
- eventMeshHTTPServer.metrics.summaryMetrics.recordReplyMsgCost(endTime - startTime);
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordReplyMsgCost(endTime - startTime);
messageLogger.info("message|eventMesh2mq|RSP|SYNC|reply2MQCost={}|topic={}|origTopic={}|bizSeqNo={}|uniqueId={}",
- endTime - startTime, replyMQCluster + "-" + EventMeshConstants.RR_REPLY_TOPIC,
- origTopic, bizNo, uniqueId);
+ endTime - startTime, replyMQCluster + "-" + EventMeshConstants.RR_REPLY_TOPIC,
+ origTopic, bizNo, uniqueId);
}
@Override
public void onException(OnExceptionContext context) {
HttpCommand err = asyncContext.getRequest().createHttpCommandResponse(
- replyMessageResponseHeader,
- SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_REPLY_MSG_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_REPLY_MSG_ERR.getErrMsg()
- + EventMeshUtil.stackTrace(context.getException(), 2)));
+ replyMessageResponseHeader,
+ SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_REPLY_MSG_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_REPLY_MSG_ERR.getErrMsg()
+ + EventMeshUtil.stackTrace(context.getException(), 2)));
asyncContext.onComplete(err, handler);
long endTime = System.currentTimeMillis();
- eventMeshHTTPServer.metrics.summaryMetrics.recordReplyMsgFailed();
- eventMeshHTTPServer.metrics.summaryMetrics.recordReplyMsgCost(endTime - startTime);
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordReplyMsgFailed();
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordReplyMsgCost(endTime - startTime);
messageLogger.error("message|eventMesh2mq|RSP|SYNC|reply2MQCost={}|topic={}|origTopic={}|bizSeqNo={}|uniqueId={}",
- endTime - startTime, replyMQCluster + "-" + EventMeshConstants.RR_REPLY_TOPIC,
- origTopic, bizNo, uniqueId, context.getException());
+ endTime - startTime, replyMQCluster + "-" + EventMeshConstants.RR_REPLY_TOPIC,
+ origTopic, bizNo, uniqueId, context.getException());
}
//@Override
@@ -311,18 +308,16 @@ public class ReplyMessageProcessor implements HttpRequestProcessor {
});
} catch (Exception ex) {
HttpCommand err = asyncContext.getRequest().createHttpCommandResponse(
- replyMessageResponseHeader,
- SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_REPLY_MSG_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_REPLY_MSG_ERR.getErrMsg() + EventMeshUtil.stackTrace(ex, 2)));
+ replyMessageResponseHeader,
+ SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_REPLY_MSG_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_REPLY_MSG_ERR.getErrMsg() + EventMeshUtil.stackTrace(ex, 2)));
asyncContext.onComplete(err);
long endTime = System.currentTimeMillis();
messageLogger.error("message|eventMesh2mq|RSP|SYNC|reply2MQCost={}|topic={}|origTopic={}|bizSeqNo={}|uniqueId={}",
- endTime - startTime, replyTopic, origTopic, bizNo, uniqueId, ex);
- eventMeshHTTPServer.metrics.summaryMetrics.recordReplyMsgFailed();
- eventMeshHTTPServer.metrics.summaryMetrics.recordReplyMsgCost(endTime - startTime);
+ endTime - startTime, replyTopic, origTopic, bizNo, uniqueId, ex);
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordReplyMsgFailed();
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordReplyMsgCost(endTime - startTime);
}
-
- return;
}
@Override
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java
index fbb8082..8264385 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java
@@ -173,7 +173,7 @@ public class SendAsyncMessageProcessor implements HttpRequestProcessor {
sendMessageResponseHeader,
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getRetCode(),
EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getErrMsg()));
- eventMeshHTTPServer.metrics.summaryMetrics.recordHTTPDiscard();
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordHTTPDiscard();
asyncContext.onComplete(responseEventMeshCommand);
return;
}
@@ -237,7 +237,7 @@ public class SendAsyncMessageProcessor implements HttpRequestProcessor {
final SendMessageContext sendMessageContext = new SendMessageContext(bizNo, event, eventMeshProducer,
eventMeshHTTPServer);
- eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsg();
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsg();
long startTime = System.currentTimeMillis();
@@ -249,7 +249,7 @@ public class SendAsyncMessageProcessor implements HttpRequestProcessor {
httpLogger.debug("{}", httpCommand);
}
eventMeshHTTPServer.sendResponse(ctx, httpCommand.httpResponse());
- eventMeshHTTPServer.metrics.summaryMetrics.recordHTTPReqResTimeCost(
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordHTTPReqResTimeCost(
System.currentTimeMillis() - asyncContext.getRequest().getReqTime());
} catch (Exception ex) {
//ignore
@@ -273,7 +273,7 @@ public class SendAsyncMessageProcessor implements HttpRequestProcessor {
EventMeshRetCode.SUCCESS.getErrMsg() + sendResult.toString()));
asyncContext.onComplete(succ, handler);
long endTime = System.currentTimeMillis();
- eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsgCost(endTime - startTime);
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgCost(endTime - startTime);
messageLogger.info("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
endTime - startTime, topic, bizNo, uniqueId);
}
@@ -289,8 +289,8 @@ public class SendAsyncMessageProcessor implements HttpRequestProcessor {
eventMeshHTTPServer.getHttpRetryer().pushRetry(sendMessageContext.delay(10000));
long endTime = System.currentTimeMillis();
- eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsgFailed();
- eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsgCost(endTime - startTime);
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgFailed();
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgCost(endTime - startTime);
messageLogger.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
endTime - startTime, topic, bizNo, uniqueId, context.getException());
}
@@ -307,8 +307,8 @@ public class SendAsyncMessageProcessor implements HttpRequestProcessor {
long endTime = System.currentTimeMillis();
messageLogger.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
endTime - startTime, topic, bizNo, uniqueId, ex);
- eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsgFailed();
- eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsgCost(endTime - startTime);
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgFailed();
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgCost(endTime - startTime);
}
return;
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendSyncMessageProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendSyncMessageProcessor.java
index dc43e53..b80fec4 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendSyncMessageProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendSyncMessageProcessor.java
@@ -182,11 +182,11 @@ public class SendSyncMessageProcessor implements HttpRequestProcessor {
.tryAcquire(EventMeshConstants.DEFAULT_FASTFAIL_TIMEOUT_IN_MILLISECONDS,
TimeUnit.MILLISECONDS)) {
responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- sendMessageResponseHeader,
- SendMessageResponseBody
- .buildBody(EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getErrMsg()));
- eventMeshHTTPServer.metrics.summaryMetrics.recordHTTPDiscard();
+ sendMessageResponseHeader,
+ SendMessageResponseBody
+ .buildBody(EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR.getErrMsg()));
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordHTTPDiscard();
asyncContext.onComplete(responseEventMeshCommand);
return;
}
@@ -230,7 +230,7 @@ public class SendSyncMessageProcessor implements HttpRequestProcessor {
final SendMessageContext sendMessageContext =
new SendMessageContext(bizNo, event, eventMeshProducer,
eventMeshHTTPServer);
- eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsg();
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsg();
long startTime = System.currentTimeMillis();
@@ -242,8 +242,8 @@ public class SendSyncMessageProcessor implements HttpRequestProcessor {
httpLogger.debug("{}", httpCommand);
}
eventMeshHTTPServer.sendResponse(ctx, httpCommand.httpResponse());
- eventMeshHTTPServer.metrics.summaryMetrics.recordHTTPReqResTimeCost(
- System.currentTimeMillis() - asyncContext.getRequest().getReqTime());
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordHTTPReqResTimeCost(
+ System.currentTimeMillis() - asyncContext.getRequest().getReqTime());
} catch (Exception ex) {
// ignore
}
@@ -309,20 +309,20 @@ public class SendSyncMessageProcessor implements HttpRequestProcessor {
}, Integer.parseInt(ttl));
} catch (Exception ex) {
HttpCommand err = asyncContext.getRequest().createHttpCommandResponse(
- sendMessageResponseHeader,
- SendMessageResponseBody
- .buildBody(EventMeshRetCode.EVENTMESH_SEND_SYNC_MSG_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_SEND_SYNC_MSG_ERR.getErrMsg()
- + EventMeshUtil.stackTrace(ex, 2)));
+ sendMessageResponseHeader,
+ SendMessageResponseBody
+ .buildBody(EventMeshRetCode.EVENTMESH_SEND_SYNC_MSG_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_SEND_SYNC_MSG_ERR.getErrMsg()
+ + EventMeshUtil.stackTrace(ex, 2)));
asyncContext.onComplete(err);
eventMeshHTTPServer.getHttpRetryer().pushRetry(sendMessageContext.delay(10000));
long endTime = System.currentTimeMillis();
- eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsgFailed();
- eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsgCost(endTime - startTime);
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgFailed();
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgCost(endTime - startTime);
messageLogger.error(
- "message|eventMesh2mq|REQ|SYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
- endTime - startTime, topic, bizNo, uniqueId, ex);
+ "message|eventMesh2mq|REQ|SYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
+ endTime - startTime, topic, bizNo, uniqueId, ex);
}
return;
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
index 98aed2c..7e28f51 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
@@ -233,8 +233,8 @@ public class SubscribeProcessor implements HttpRequestProcessor {
httpLogger.debug("{}", httpCommand);
}
eventMeshHTTPServer.sendResponse(ctx, httpCommand.httpResponse());
- eventMeshHTTPServer.metrics.summaryMetrics.recordHTTPReqResTimeCost(
- System.currentTimeMillis() - request.getReqTime());
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordHTTPReqResTimeCost(
+ System.currentTimeMillis() - request.getReqTime());
} catch (Exception ex) {
// ignore
}
@@ -245,20 +245,20 @@ public class SubscribeProcessor implements HttpRequestProcessor {
asyncContext.onComplete(responseEventMeshCommand, handler);
} catch (Exception e) {
HttpCommand err = asyncContext.getRequest().createHttpCommandResponse(
- subscribeResponseHeader,
- SubscribeResponseBody
- .buildBody(EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR.getErrMsg()
- + EventMeshUtil.stackTrace(e, 2)));
+ subscribeResponseHeader,
+ SubscribeResponseBody
+ .buildBody(EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_SUBSCRIBE_ERR.getErrMsg()
+ + EventMeshUtil.stackTrace(e, 2)));
asyncContext.onComplete(err);
long endTime = System.currentTimeMillis();
httpLogger.error(
- "message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}"
- + "|bizSeqNo={}|uniqueId={}", endTime - startTime,
- JsonUtils.serialize(subscribeRequestBody.getTopics()),
- subscribeRequestBody.getUrl(), e);
- eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsgFailed();
- eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsgCost(endTime - startTime);
+ "message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}"
+ + "|bizSeqNo={}|uniqueId={}", endTime - startTime,
+ JsonUtils.serialize(subscribeRequestBody.getTopics()),
+ subscribeRequestBody.getUrl(), e);
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgFailed();
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgCost(endTime - startTime);
}
}
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java
index 3e1c2a1..43eeb38 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java
@@ -129,7 +129,7 @@ public class UnSubscribeProcessor implements HttpRequestProcessor {
httpLogger.debug("{}", httpCommand);
}
eventMeshHTTPServer.sendResponse(ctx, httpCommand.httpResponse());
- eventMeshHTTPServer.metrics.summaryMetrics.recordHTTPReqResTimeCost(
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordHTTPReqResTimeCost(
System.currentTimeMillis() - asyncContext.getRequest().getReqTime());
} catch (Exception ex) {
// ignore
@@ -226,8 +226,8 @@ public class UnSubscribeProcessor implements HttpRequestProcessor {
+ "|topic={}|bizSeqNo={}|uniqueId={}", endTime - startTime,
JsonUtils.serialize(unSubscribeRequestBody.getTopics()),
unSubscribeRequestBody.getUrl(), e);
- eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsgFailed();
- eventMeshHTTPServer.metrics.summaryMetrics
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgFailed();
+ eventMeshHTTPServer.metrics.getSummaryMetrics()
.recordSendMsgCost(endTime - startTime);
}
} else {
@@ -258,9 +258,8 @@ public class UnSubscribeProcessor implements HttpRequestProcessor {
+ "|topic={}|bizSeqNo={}|uniqueId={}", endTime - startTime,
JsonUtils.serialize(unSubscribeRequestBody.getTopics()),
unSubscribeRequestBody.getUrl(), e);
- eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsgFailed();
- eventMeshHTTPServer.metrics.summaryMetrics
- .recordSendMsgCost(endTime - startTime);
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgFailed();
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendMsgCost(endTime - startTime);
}
}
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/producer/SendMessageContext.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/producer/SendMessageContext.java
index 3153b55..054e283 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/producer/SendMessageContext.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/producer/SendMessageContext.java
@@ -143,7 +143,7 @@ public class SendMessageContext extends RetryContext {
@Override
public void onException(OnExceptionContext context) {
logger.warn("", context.getException());
- eventMeshHTTPServer.metrics.summaryMetrics.recordSendBatchMsgFailed(1);
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordSendBatchMsgFailed(1);
}
});
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java
index 8d687a2..89c59c9 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java
@@ -159,7 +159,7 @@ public class AsyncHTTPPushRequest extends AbstractHTTPPushRequest {
builder.setEntity(new UrlEncodedFormEntity(body, StandardCharsets.UTF_8));
- eventMeshHTTPServer.metrics.summaryMetrics.recordPushMsg();
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordPushMsg();
this.lastPushTime = System.currentTimeMillis();
@@ -174,9 +174,9 @@ public class AsyncHTTPPushRequest extends AbstractHTTPPushRequest {
public Object handleResponse(HttpResponse response) {
removeWaitingMap(AsyncHTTPPushRequest.this);
long cost = System.currentTimeMillis() - lastPushTime;
- eventMeshHTTPServer.metrics.summaryMetrics.recordHTTPPushTimeCost(cost);
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordHTTPPushTimeCost(cost);
if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
- eventMeshHTTPServer.metrics.summaryMetrics.recordHttpPushMsgFailed();
+ eventMeshHTTPServer.metrics.getSummaryMetrics().recordHttpPushMsgFailed();
messageLogger.info(
"message|eventMesh2client|exception|url={}|topic={}|bizSeqNo={}"
+ "|uniqueId={}|cost={}", currPushUrl, handleMsgContext.getTopic(),
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/retry/HttpRetryer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/retry/HttpRetryer.java
index f7d2a01..6448db3 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/retry/HttpRetryer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/retry/HttpRetryer.java
@@ -41,7 +41,7 @@ public class HttpRetryer {
this.eventMeshHTTPServer = eventMeshHTTPServer;
}
- private DelayQueue<DelayRetryable> failed = new DelayQueue<DelayRetryable>();
+ private DelayQueue<DelayRetryable> failed = new DelayQueue<>();
private ThreadPoolExecutor pool;
@@ -102,6 +102,13 @@ public class HttpRetryer {
return failed.size();
}
+ /**
+ * Get failed queue, this method is just used for metrics.
+ */
+ public DelayQueue<DelayRetryable> getFailedQueue() {
+ return failed;
+ }
+
public void shutdown() {
dispatcher.interrupt();
pool.shutdown();
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/EventMeshTcp2Client.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/EventMeshTcp2Client.java
index 0d300f7..35f17aa 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/EventMeshTcp2Client.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/EventMeshTcp2Client.java
@@ -105,7 +105,7 @@ public class EventMeshTcp2Client {
EventMeshTcpMonitor eventMeshTcpMonitor) {
long startTime = System.currentTimeMillis();
Package pkg = new Package(new Header(SERVER_GOODBYE_REQUEST, OPStatus.FAIL.getCode(), errMsg, null));
- eventMeshTcpMonitor.getEventMesh2clientMsgNum().incrementAndGet();
+ eventMeshTcpMonitor.getTcpSummaryMetrics().getEventMesh2clientMsgNum().incrementAndGet();
logger.info("goodBye2Client client[{}]", RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
ctx.writeAndFlush(pkg).addListener(
new ChannelFutureListener() {
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/EventMeshTcpMessageDispatcher.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/EventMeshTcpMessageDispatcher.java
index 8b6c6e7..9b39b65 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/EventMeshTcpMessageDispatcher.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/EventMeshTcpMessageDispatcher.java
@@ -55,7 +55,7 @@ public class EventMeshTcpMessageDispatcher extends SimpleChannelInboundHandler<P
protected void channelRead0(ChannelHandlerContext ctx, Package pkg) throws Exception {
long startTime = System.currentTimeMillis();
validateMsg(pkg);
- eventMeshTCPServer.getEventMeshTcpMonitor().getClient2EventMeshMsgNum().incrementAndGet();
+ eventMeshTCPServer.getEventMeshTcpMonitor().getTcpSummaryMetrics().getClient2eventMeshMsgNum().incrementAndGet();
Command cmd = null;
try {
Runnable task;
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java
index 809b1ee..0b3e4db 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java
@@ -63,6 +63,7 @@ import org.slf4j.LoggerFactory;
import io.cloudevents.CloudEvent;
import io.cloudevents.core.builder.CloudEventBuilder;
+import com.google.common.base.Preconditions;
public class ClientGroupWrapper {
@@ -119,7 +120,7 @@ public class ClientGroupWrapper {
this.eventMeshTCPServer = eventMeshTCPServer;
this.eventMeshTCPConfiguration = eventMeshTCPServer.getEventMeshTCPConfiguration();
this.eventMeshTcpRetryer = eventMeshTCPServer.getEventMeshTcpRetryer();
- this.eventMeshTcpMonitor = eventMeshTCPServer.getEventMeshTcpMonitor();
+ this.eventMeshTcpMonitor = Preconditions.checkNotNull(eventMeshTCPServer.getEventMeshTcpMonitor());
this.downstreamDispatchStrategy = downstreamDispatchStrategy;
this.persistentMsgConsumer = new MQConsumerWrapper(
eventMeshTCPServer.getEventMeshTCPConfiguration().eventMeshConnectorPluginType);
@@ -455,12 +456,12 @@ public class ClientGroupWrapper {
@Override
public void consume(CloudEvent event, AsyncConsumeContext context) {
- eventMeshTcpMonitor.getMq2EventMeshMsgNum().incrementAndGet();
+ eventMeshTcpMonitor.getTcpSummaryMetrics().getMq2eventMeshMsgNum().incrementAndGet();
event = CloudEventBuilder.from(event)
- .withExtension(EventMeshConstants.REQ_MQ2EVENTMESH_TIMESTAMP,
- String.valueOf(System.currentTimeMillis()))
- .withExtension(EventMeshConstants.REQ_RECEIVE_EVENTMESH_IP,
- eventMeshTCPConfiguration.eventMeshServerIp).build();
+ .withExtension(EventMeshConstants.REQ_MQ2EVENTMESH_TIMESTAMP,
+ String.valueOf(System.currentTimeMillis()))
+ .withExtension(EventMeshConstants.REQ_RECEIVE_EVENTMESH_IP,
+ eventMeshTCPConfiguration.eventMeshServerIp).build();
String topic = event.getSubject();
// message.getSystemProperties(Constants.PROPERTY_MESSAGE_DESTINATION);
//message.getSystemProperties().put(EventMeshConstants.REQ_MQ2EVENTMESH_TIMESTAMP,
@@ -513,75 +514,72 @@ public class ClientGroupWrapper {
};
broadCastMsgConsumer.subscribe(subscriptionItem.getTopic(), listener);
} else {
- listener = new EventListener() {
- @Override
- public void consume(CloudEvent event, AsyncConsumeContext context) {
- eventMeshTcpMonitor.getMq2EventMeshMsgNum().incrementAndGet();
- event = CloudEventBuilder.from(event)
- .withExtension(EventMeshConstants.REQ_MQ2EVENTMESH_TIMESTAMP,
- String.valueOf(System.currentTimeMillis()))
- .withExtension(EventMeshConstants.REQ_RECEIVE_EVENTMESH_IP,
- eventMeshTCPConfiguration.eventMeshServerIp).build();
- String topic = event.getSubject();
+ listener = (event, context) -> {
+ eventMeshTcpMonitor.getTcpSummaryMetrics().getMq2eventMeshMsgNum().incrementAndGet();
+ event = CloudEventBuilder.from(event)
+ .withExtension(EventMeshConstants.REQ_MQ2EVENTMESH_TIMESTAMP,
+ String.valueOf(System.currentTimeMillis()))
+ .withExtension(EventMeshConstants.REQ_RECEIVE_EVENTMESH_IP,
+ eventMeshTCPConfiguration.eventMeshServerIp).build();
+ String topic = event.getSubject();
+
+ EventMeshAsyncConsumeContext eventMeshAsyncConsumeContext =
+ (EventMeshAsyncConsumeContext) context;
+ Session session = downstreamDispatchStrategy
+ .select(consumerGroup, topic, groupConsumerSessions);
+ String bizSeqNo = EventMeshUtil.getMessageBizSeq(event);
+ if (session == null) {
+ try {
+ Integer sendBackTimes = 0;
+ String sendBackFromEventMeshIp = "";
+ if (StringUtils.isNotBlank(Objects.requireNonNull(event.getExtension(
+ EventMeshConstants.EVENTMESH_SEND_BACK_TIMES)).toString())) {
+ sendBackTimes = (Integer) event.getExtension(EventMeshConstants.EVENTMESH_SEND_BACK_TIMES);
+ }
+ if (StringUtils.isNotBlank(Objects.requireNonNull(event.getExtension(
+ EventMeshConstants.EVENTMESH_SEND_BACK_IP)).toString())) {
+ sendBackFromEventMeshIp = (String) event.getExtension(EventMeshConstants.EVENTMESH_SEND_BACK_IP);
+ }
- EventMeshAsyncConsumeContext eventMeshAsyncConsumeContext =
- (EventMeshAsyncConsumeContext) context;
- Session session = downstreamDispatchStrategy
- .select(consumerGroup, topic, groupConsumerSessions);
- String bizSeqNo = EventMeshUtil.getMessageBizSeq(event);
- if (session == null) {
- try {
- Integer sendBackTimes = new Integer(0);
- String sendBackFromEventMeshIp = "";
- if (StringUtils.isNotBlank(Objects.requireNonNull(event.getExtension(
- EventMeshConstants.EVENTMESH_SEND_BACK_TIMES)).toString())) {
- sendBackTimes = (Integer) event.getExtension(EventMeshConstants.EVENTMESH_SEND_BACK_TIMES);
- }
- if (StringUtils.isNotBlank(Objects.requireNonNull(event.getExtension(
- EventMeshConstants.EVENTMESH_SEND_BACK_IP)).toString())) {
- sendBackFromEventMeshIp = (String) event.getExtension(EventMeshConstants.EVENTMESH_SEND_BACK_IP);
- }
+ logger.error(
+ "found no session to downstream msg,groupName:{}, topic:{}, "
+ + "bizSeqNo:{}, sendBackTimes:{}, sendBackFromEventMeshIp:{}",
+ consumerGroup, topic, bizSeqNo, sendBackTimes,
+ sendBackFromEventMeshIp);
+ if (sendBackTimes >= eventMeshTCPServer
+ .getEventMeshTCPConfiguration().eventMeshTcpSendBackMaxTimes) {
logger.error(
- "found no session to downstream msg,groupName:{}, topic:{}, "
- + "bizSeqNo:{}, sendBackTimes:{}, sendBackFromEventMeshIp:{}",
- consumerGroup, topic, bizSeqNo, sendBackTimes,
- sendBackFromEventMeshIp);
-
- if (sendBackTimes >= eventMeshTCPServer
- .getEventMeshTCPConfiguration().eventMeshTcpSendBackMaxTimes) {
- logger.error(
- "sendBack to broker over max times:{}, groupName:{}, topic:{}, "
- + "bizSeqNo:{}", eventMeshTCPServer
- .getEventMeshTCPConfiguration()
- .eventMeshTcpSendBackMaxTimes,
- consumerGroup, topic, bizSeqNo);
- } else {
- sendBackTimes++;
- event = CloudEventBuilder.from(event)
- .withExtension(EventMeshConstants.EVENTMESH_SEND_BACK_TIMES,
- sendBackTimes.toString())
- .withExtension(EventMeshConstants.EVENTMESH_SEND_BACK_IP,
- eventMeshTCPConfiguration.eventMeshServerIp).build();
- sendMsgBackToBroker(event, bizSeqNo);
- }
- } catch (Exception e) {
- logger.warn("handle msg exception when no session found", e);
+ "sendBack to broker over max times:{}, groupName:{}, topic:{}, "
+ + "bizSeqNo:{}", eventMeshTCPServer
+ .getEventMeshTCPConfiguration()
+ .eventMeshTcpSendBackMaxTimes,
+ consumerGroup, topic, bizSeqNo);
+ } else {
+ sendBackTimes++;
+ event = CloudEventBuilder.from(event)
+ .withExtension(EventMeshConstants.EVENTMESH_SEND_BACK_TIMES,
+ sendBackTimes.toString())
+ .withExtension(EventMeshConstants.EVENTMESH_SEND_BACK_IP,
+ eventMeshTCPConfiguration.eventMeshServerIp).build();
+ sendMsgBackToBroker(event, bizSeqNo);
}
-
- eventMeshAsyncConsumeContext.commit(EventMeshAction.CommitMessage);
- return;
+ } catch (Exception e) {
+ logger.warn("handle msg exception when no session found", e);
}
- DownStreamMsgContext downStreamMsgContext =
- new DownStreamMsgContext(event, session, persistentMsgConsumer,
- eventMeshAsyncConsumeContext.getAbstractContext(), false,
- subscriptionItem);
- //msg put in eventmesh,waiting client ack
- session.getPusher().unAckMsg(downStreamMsgContext.seq, downStreamMsgContext);
- session.downstreamMsg(downStreamMsgContext);
- eventMeshAsyncConsumeContext.commit(EventMeshAction.ManualAck);
+ eventMeshAsyncConsumeContext.commit(EventMeshAction.CommitMessage);
+ return;
}
+
+ DownStreamMsgContext downStreamMsgContext =
+ new DownStreamMsgContext(event, session, persistentMsgConsumer,
+ eventMeshAsyncConsumeContext.getAbstractContext(), false,
+ subscriptionItem);
+ //msg put in eventmesh,waiting client ack
+ session.getPusher().unAckMsg(downStreamMsgContext.seq, downStreamMsgContext);
+ session.downstreamMsg(downStreamMsgContext);
+ eventMeshAsyncConsumeContext.commit(EventMeshAction.ManualAck);
};
persistentMsgConsumer.subscribe(subscriptionItem.getTopic(), listener);
}
@@ -725,7 +723,7 @@ public class ClientGroupWrapper {
}
});
- eventMeshTcpMonitor.getEventMesh2mqMsgNum().incrementAndGet();
+ eventMeshTcpMonitor.getTcpSummaryMetrics().getEventMesh2mqMsgNum().incrementAndGet();
} catch (Exception e) {
logger.warn("try send msg back to broker failed");
throw e;
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/Session.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/Session.java
index c1a4dc9..37e7b5e 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/Session.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/Session.java
@@ -218,7 +218,8 @@ public class Session {
if (!future.isSuccess()) {
messageLogger.error("write2Client fail, pkg[{}] session[{}]", pkg, this);
} else {
- clientGroupWrapper.get().getEventMeshTcpMonitor().getEventMesh2clientMsgNum().incrementAndGet();
+ clientGroupWrapper.get().getEventMeshTcpMonitor().getTcpSummaryMetrics().getEventMesh2clientMsgNum()
+ .incrementAndGet();
}
}
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/push/SessionPusher.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/push/SessionPusher.java
index 55ac95f..ad75a24 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/push/SessionPusher.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/push/SessionPusher.java
@@ -107,7 +107,7 @@ public class SessionPusher {
retCode = -1;
retMsg = e.toString();
} finally {
- session.getClientGroupWrapper().get().getEventMeshTcpMonitor().getEventMesh2clientMsgNum().incrementAndGet();
+ session.getClientGroupWrapper().get().getEventMeshTcpMonitor().getTcpSummaryMetrics().getEventMesh2clientMsgNum().incrementAndGet();
session.getContext().writeAndFlush(pkg).addListener(
new ChannelFutureListener() {
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/send/SessionSender.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/send/SessionSender.java
index f4615f2..44fccad 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/send/SessionSender.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/send/SessionSender.java
@@ -126,7 +126,7 @@ public class SessionSender {
session.getClientGroupWrapper().get().send(upStreamMsgContext, sendCallback);
}
- session.getClientGroupWrapper().get().getEventMeshTcpMonitor().getEventMesh2mqMsgNum().incrementAndGet();
+ session.getClientGroupWrapper().get().getEventMeshTcpMonitor().getTcpSummaryMetrics().getEventMesh2mqMsgNum().incrementAndGet();
} else {
logger.warn("send too fast,session flow control,session:{}", session.getClient());
return new EventMeshTcpSendResult(header.getSeq(), EventMeshTcpSendStatus.SEND_TOO_FAST, EventMeshTcpSendStatus.SEND_TOO_FAST.name());
@@ -158,7 +158,7 @@ public class SessionSender {
.withExtension(EventMeshConstants.RSP_MQ2EVENTMESH_TIMESTAMP, String.valueOf(System.currentTimeMillis()))
.withExtension(EventMeshConstants.RSP_RECEIVE_EVENTMESH_IP, session.getEventMeshTCPConfiguration().eventMeshServerIp)
.build();
- session.getClientGroupWrapper().get().getEventMeshTcpMonitor().getMq2EventMeshMsgNum().incrementAndGet();
+ session.getClientGroupWrapper().get().getEventMeshTcpMonitor().getTcpSummaryMetrics().getMq2eventMeshMsgNum().incrementAndGet();
Command cmd;
if (header.getCmd().equals(Command.REQUEST_TO_SERVER)) {
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HTTPMetricsServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HTTPMetricsServer.java
index eecfdf7..80fdc4d 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HTTPMetricsServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HTTPMetricsServer.java
@@ -17,9 +17,11 @@
package org.apache.eventmesh.runtime.metrics.http;
+import org.apache.eventmesh.metrics.api.MetricsRegistry;
+import org.apache.eventmesh.metrics.api.model.HttpSummaryMetrics;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
-import org.apache.eventmesh.runtime.metrics.opentelemetry.OpenTelemetryHTTPMetricsExporter;
+import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
@@ -29,45 +31,38 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.codahale.metrics.MetricRegistry;
-
public class HTTPMetricsServer {
private EventMeshHTTPServer eventMeshHTTPServer;
- private MetricRegistry metricRegistry = new MetricRegistry();
-
- public SummaryMetrics summaryMetrics;
-
- public HealthMetrics healthMetrics;
-
- public TopicMetrics topicMetrics;
-
- public GroupMetrics groupMetrics;
-
- public OpenTelemetryHTTPMetricsExporter openTelemetryHTTPMetricsExporter;
-
private Logger httpLogger = LoggerFactory.getLogger("httpMonitor");
private Logger logger = LoggerFactory.getLogger(this.getClass());
- public HTTPMetricsServer(EventMeshHTTPServer eventMeshHTTPServer) {
+ private List<MetricsRegistry> metricsRegistries;
+
+ private final HttpSummaryMetrics summaryMetrics;
+
+ public HTTPMetricsServer(EventMeshHTTPServer eventMeshHTTPServer, List<MetricsRegistry> metricsRegistries) {
this.eventMeshHTTPServer = eventMeshHTTPServer;
+ this.metricsRegistries = metricsRegistries;
+ this.summaryMetrics = new HttpSummaryMetrics(
+ eventMeshHTTPServer.batchMsgExecutor,
+ eventMeshHTTPServer.sendMsgExecutor,
+ eventMeshHTTPServer.pushMsgExecutor,
+ eventMeshHTTPServer.getHttpRetryer().getFailedQueue());
}
public void init() throws Exception {
- summaryMetrics = new SummaryMetrics(this.eventMeshHTTPServer, this.metricRegistry);
- topicMetrics = new TopicMetrics(this.eventMeshHTTPServer, this.metricRegistry);
- groupMetrics = new GroupMetrics(this.eventMeshHTTPServer, this.metricRegistry);
- healthMetrics = new HealthMetrics(this.eventMeshHTTPServer, this.metricRegistry);
-
- openTelemetryHTTPMetricsExporter = new OpenTelemetryHTTPMetricsExporter(this, this.eventMeshHTTPServer.getEventMeshHttpConfiguration());
-
- logger.info("HTTPMetricsServer inited......");
+ metricsRegistries.forEach(MetricsRegistry::start);
+ logger.info("HTTPMetricsServer initialized......");
}
public void start() throws Exception {
- openTelemetryHTTPMetricsExporter.start();
+ metricsRegistries.forEach(metricsRegistry -> {
+ metricsRegistry.register(summaryMetrics);
+ logger.info("Register httpMetrics to " + metricsRegistry.getClass().getName());
+ });
metricsSchedule.scheduleAtFixedRate(() -> {
try {
summaryMetrics.snapshotHTTPTPS();
@@ -78,21 +73,21 @@ public class HTTPMetricsServer {
logger.warn("eventMesh snapshot tps metrics err", ex);
}
}, 0, 1000, TimeUnit.MILLISECONDS);
-
+
metricsSchedule.scheduleAtFixedRate(() -> {
try {
logPrintServerMetrics();
} catch (Exception ex) {
logger.warn("eventMesh print metrics err", ex);
}
- }, 1000, SummaryMetrics.STATIC_PERIOD, TimeUnit.MILLISECONDS);
+ }, 1000, 30 * 1000, TimeUnit.MILLISECONDS);
logger.info("HTTPMetricsServer started......");
}
public void shutdown() throws Exception {
metricsSchedule.shutdown();
- openTelemetryHTTPMetricsExporter.shutdown();
+ metricsRegistries.forEach(MetricsRegistry::showdown);
logger.info("HTTPMetricsServer shutdown......");
}
@@ -108,88 +103,65 @@ public class HTTPMetricsServer {
}
});
+ // todo: move this into standalone metrics plugin
private void logPrintServerMetrics() {
httpLogger.info("===========================================SERVER METRICS==================================================");
- httpLogger.info(String.format(SummaryMetrics.EVENTMESH_MONITOR_FORMAT_HTTP,
- summaryMetrics.maxHTTPTPS(),
- summaryMetrics.avgHTTPTPS(),
- summaryMetrics.maxHTTPCost(),
- summaryMetrics.avgHTTPCost(),
- summaryMetrics.avgHTTPBodyDecodeCost(),
- summaryMetrics.getHttpDiscard()));
+ httpLogger.info(String.format(HttpSummaryMetrics.EVENTMESH_MONITOR_FORMAT_HTTP,
+ summaryMetrics.maxHTTPTPS(),
+ summaryMetrics.avgHTTPTPS(),
+ summaryMetrics.maxHTTPCost(),
+ summaryMetrics.avgHTTPCost(),
+ summaryMetrics.avgHTTPBodyDecodeCost(),
+ summaryMetrics.getHttpDiscard()));
summaryMetrics.httpStatInfoClear();
- httpLogger.info(String.format(SummaryMetrics.EVENTMESH_MONITOR_FORMAT_BATCHSENDMSG,
- summaryMetrics.maxSendBatchMsgTPS(),
- summaryMetrics.avgSendBatchMsgTPS(),
- summaryMetrics.getSendBatchMsgNumSum(),
- summaryMetrics.getSendBatchMsgFailNumSum(),
- summaryMetrics.getSendBatchMsgFailRate(),
- summaryMetrics.getSendBatchMsgDiscardNumSum()
+ httpLogger.info(String.format(HttpSummaryMetrics.EVENTMESH_MONITOR_FORMAT_BATCHSENDMSG,
+ summaryMetrics.maxSendBatchMsgTPS(),
+ summaryMetrics.avgSendBatchMsgTPS(),
+ summaryMetrics.getSendBatchMsgNumSum(),
+ summaryMetrics.getSendBatchMsgFailNumSum(),
+ summaryMetrics.getSendBatchMsgFailRate(),
+ summaryMetrics.getSendBatchMsgDiscardNumSum()
));
summaryMetrics.cleanSendBatchStat();
- httpLogger.info(String.format(SummaryMetrics.EVENTMESH_MONITOR_FORMAT_SENDMSG,
- summaryMetrics.maxSendMsgTPS(),
- summaryMetrics.avgSendMsgTPS(),
- summaryMetrics.getSendMsgNumSum(),
- summaryMetrics.getSendMsgFailNumSum(),
- summaryMetrics.getSendMsgFailRate(),
- summaryMetrics.getReplyMsgNumSum(),
- summaryMetrics.getReplyMsgFailNumSum()
+ httpLogger.info(String.format(HttpSummaryMetrics.EVENTMESH_MONITOR_FORMAT_SENDMSG,
+ summaryMetrics.maxSendMsgTPS(),
+ summaryMetrics.avgSendMsgTPS(),
+ summaryMetrics.getSendMsgNumSum(),
+ summaryMetrics.getSendMsgFailNumSum(),
+ summaryMetrics.getSendMsgFailRate(),
+ summaryMetrics.getReplyMsgNumSum(),
+ summaryMetrics.getReplyMsgFailNumSum()
));
summaryMetrics.cleanSendMsgStat();
- httpLogger.info(String.format(SummaryMetrics.EVENTMESH_MONITOR_FORMAT_PUSHMSG,
- summaryMetrics.maxPushMsgTPS(),
- summaryMetrics.avgPushMsgTPS(),
- summaryMetrics.getHttpPushMsgNumSum(),
- summaryMetrics.getHttpPushFailNumSum(),
- summaryMetrics.getHttpPushMsgFailRate(),
- summaryMetrics.maxHTTPPushLatency(),
- summaryMetrics.avgHTTPPushLatency()
+ httpLogger.info(String.format(HttpSummaryMetrics.EVENTMESH_MONITOR_FORMAT_PUSHMSG,
+ summaryMetrics.maxPushMsgTPS(),
+ summaryMetrics.avgPushMsgTPS(),
+ summaryMetrics.getHttpPushMsgNumSum(),
+ summaryMetrics.getHttpPushFailNumSum(),
+ summaryMetrics.getHttpPushMsgFailRate(),
+ summaryMetrics.maxHTTPPushLatency(),
+ summaryMetrics.avgHTTPPushLatency()
));
summaryMetrics.cleanHttpPushMsgStat();
- httpLogger.info(String.format(SummaryMetrics.EVENTMESH_MONITOR_FORMAT_BLOCKQ,
- eventMeshHTTPServer.getBatchMsgExecutor().getQueue().size(),
- eventMeshHTTPServer.getSendMsgExecutor().getQueue().size(),
- eventMeshHTTPServer.getPushMsgExecutor().getQueue().size(),
- eventMeshHTTPServer.getHttpRetryer().size()));
+ httpLogger.info(String.format(HttpSummaryMetrics.EVENTMESH_MONITOR_FORMAT_BLOCKQ,
+ eventMeshHTTPServer.getBatchMsgExecutor().getQueue().size(),
+ eventMeshHTTPServer.getSendMsgExecutor().getQueue().size(),
+ eventMeshHTTPServer.getPushMsgExecutor().getQueue().size(),
+ eventMeshHTTPServer.getHttpRetryer().size()));
- httpLogger.info(String.format(SummaryMetrics.EVENTMESH_MONITOR_FORMAT_MQ_CLIENT,
- summaryMetrics.avgBatchSendMsgCost(),
- summaryMetrics.avgSendMsgCost(),
- summaryMetrics.avgReplyMsgCost()));
+ httpLogger.info(String.format(HttpSummaryMetrics.EVENTMESH_MONITOR_FORMAT_MQ_CLIENT,
+ summaryMetrics.avgBatchSendMsgCost(),
+ summaryMetrics.avgSendMsgCost(),
+ summaryMetrics.avgReplyMsgCost()));
summaryMetrics.send2MQStatInfoClear();
}
- public int getBatchMsgQ() {
- return eventMeshHTTPServer.getBatchMsgExecutor().getQueue().size();
- }
-
- public int getSendMsgQ() {
- return eventMeshHTTPServer.getSendMsgExecutor().getQueue().size();
- }
-
- public int getPushMsgQ() {
- return eventMeshHTTPServer.getPushMsgExecutor().getQueue().size();
- }
-
- public int getHttpRetryQ() {
- return eventMeshHTTPServer.getHttpRetryer().size();
- }
-
- public HealthMetrics getHealthMetrics() {
- return healthMetrics;
- }
-
- public TopicMetrics getTopicMetrics() {
- return topicMetrics;
- }
-
- public GroupMetrics getGroupMetrics() {
- return groupMetrics;
+ public HttpSummaryMetrics getSummaryMetrics() {
+ return summaryMetrics;
}
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/TcpSummaryMetrics.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/TcpSummaryMetrics.java
deleted file mode 100644
index 579c71b..0000000
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/TcpSummaryMetrics.java
+++ /dev/null
@@ -1,403 +0,0 @@
-/*
- * 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.runtime.metrics.http;
-
-import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
-
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.codahale.metrics.MetricRegistry;
-
-public class TcpSummaryMetrics {
-
- public Logger logger = LoggerFactory.getLogger("httpMonitor");
-
- private EventMeshHTTPServer eventMeshHTTPServer;
- private MetricRegistry metricRegistry;
-
- public TcpSummaryMetrics(EventMeshHTTPServer eventMeshHTTPServer, MetricRegistry metricRegistry) {
- this.eventMeshHTTPServer = eventMeshHTTPServer;
- this.metricRegistry = metricRegistry;
- }
-
- public static final int STATIC_PERIOD = 30 * 1000;
-
- private float avg(LinkedList<Integer> linkedList) {
- float sum = 0.0f;
- if (linkedList.isEmpty()) {
- return sum;
- }
-
- Iterator<Integer> it = linkedList.iterator();
- while (it.hasNext()) {
- float tps = (float) it.next();
- sum += tps;
- }
-
- return sum / linkedList.size();
- }
-
- ////////////////////////////////////////////////////////////////////////////////
- public static final String EVENTMESH_MONITOR_FORMAT_HTTP = "%15s : {\"maxHTTPTPS\":\"%.1f\",\"avgHTTPTPS\":\"%.1f\","
- + //EVENTMESH tps related to accepting external http requests
- "\"maxHTTPCOST\":\"%s\",\"avgHTTPCOST\":\"%.1f\",\"avgHTTPBodyDecodeCost\":\"%.1f\"}";
-
- private float wholeCost = 0f;
-
- private AtomicLong wholeRequestNum = new AtomicLong(0);
-
- private AtomicLong maxCost = new AtomicLong(0);
-
- private AtomicLong httpRequestPerSecond = new AtomicLong(0);
-
- private LinkedList<Integer> httpRequestTPSSnapshots = new LinkedList<Integer>();
-
- public float avgHTTPCost() {
- float cost = (wholeRequestNum.longValue() == 0L) ? 0f : wholeCost / wholeRequestNum.longValue();
- return cost;
- }
-
- public long maxHTTPCost() {
- long cost = maxCost.longValue();
- return cost;
- }
-
- public void recordHTTPRequest() {
- httpRequestPerSecond.incrementAndGet();
- }
-
- public void snapshotHTTPTPS() {
- Integer tps = httpRequestPerSecond.intValue();
- httpRequestTPSSnapshots.add(tps);
- httpRequestPerSecond.set(0);
- if (httpRequestTPSSnapshots.size() > STATIC_PERIOD / 1000) {
- httpRequestTPSSnapshots.removeFirst();
- }
- }
-
- public float maxHTTPTPS() {
- float tps = Collections.max(httpRequestTPSSnapshots);
- return tps;
- }
-
- public float avgHTTPTPS() {
- float tps = avg(httpRequestTPSSnapshots);
- return tps;
- }
-
- public void recordHTTPReqResTimeCost(long cost) {
- wholeRequestNum.incrementAndGet();
- wholeCost = wholeCost + cost;
- if (cost > maxCost.longValue()) {
- maxCost.set(cost);
- }
- }
-
- public void httpStatInfoClear() {
- wholeRequestNum.set(0L);
- wholeCost = 0f;
- maxCost.set(0L);
- httpDecodeNum.set(0L);
- httpDecodeTimeCost = 0f;
- }
-
- private float httpDecodeTimeCost = 0f;
-
- private AtomicLong httpDecodeNum = new AtomicLong(0);
-
- public void recordDecodeTimeCost(long cost) {
- httpDecodeNum.incrementAndGet();
- httpDecodeTimeCost = httpDecodeTimeCost + cost;
- }
-
- public float avgHTTPBodyDecodeCost() {
- float cost = (httpDecodeNum.longValue() == 0L) ? 0f : httpDecodeTimeCost / httpDecodeNum.longValue();
- return cost;
- }
-
-
- //////////////////////////////////////////////////////////////////////////
- public static final String EVENTMESH_MONITOR_FORMAT_BATCHSENDMSG = "%15s : {\"maxBatchSendMsgTPS\":\"%.1f\",\"avgBatchSendMsgTPS\":\"%.1f\","
- +
- " \"sum\":\"%s\", \"sumFail\":\"%s\", \"sumFailRate\":\"%.2f\", \"discard\":\"%s\"}";
-
- private AtomicLong sendBatchMsgNumPerSecond = new AtomicLong(0);
-
- private AtomicLong sendBatchMsgNumSum = new AtomicLong(0);
-
- private AtomicLong sendBatchMsgFailNumSum = new AtomicLong(0);
-
- private AtomicLong sendBatchMsgDiscardNumSum = new AtomicLong(0);
-
- public void recordSendBatchMsgDiscard(long delta) {
- sendBatchMsgDiscardNumSum.addAndGet(delta);
- }
-
- private LinkedList<Integer> sendBatchMsgTPSSnapshots = new LinkedList<Integer>();
-
- public void snapshotSendBatchMsgTPS() {
- Integer tps = sendBatchMsgNumPerSecond.intValue();
- sendBatchMsgTPSSnapshots.add(tps);
- sendBatchMsgNumPerSecond.set(0);
- if (sendBatchMsgTPSSnapshots.size() > STATIC_PERIOD / 1000) {
- sendBatchMsgTPSSnapshots.removeFirst();
- }
- }
-
- public float maxSendBatchMsgTPS() {
- float tps = Collections.max(sendBatchMsgTPSSnapshots);
- return tps;
- }
-
- public float avgSendBatchMsgTPS() {
- float tps = avg(sendBatchMsgTPSSnapshots);
- return tps;
- }
-
- public void recordSendBatchMsg(long delta) {
- sendBatchMsgNumPerSecond.addAndGet(delta);
- sendBatchMsgNumSum.addAndGet(delta);
- }
-
- public void recordSendBatchMsgFailed(long delta) {
- sendBatchMsgFailNumSum.getAndAdd(delta);
- }
-
- public long getSendBatchMsgNumSum() {
- return sendBatchMsgNumSum.longValue();
- }
-
- public long getSendBatchMsgFailNumSum() {
- return sendBatchMsgFailNumSum.longValue();
- }
-
- public float getSendBatchMsgFailRate() {
- return (sendBatchMsgNumSum.longValue() == 0L) ? 0f : sendBatchMsgFailNumSum.floatValue() / sendBatchMsgNumSum.longValue();
- }
-
- public void cleanSendBatchStat() {
- sendBatchMsgNumSum.set(0L);
- sendBatchMsgFailNumSum.set(0L);
- }
-
- public long getSendBatchMsgDiscardNumSum() {
- return sendBatchMsgDiscardNumSum.longValue();
- }
-
- //////////////////////////////////////////////////////////////////////////
- public static final String EVENTMESH_MONITOR_FORMAT_SENDMSG = "%15s : {\"maxSendMsgTPS\":\"%.1f\",\"avgSendMsgTPS\":\"%.1f\","
- +
- " \"sum\":\"%s\", \"sumFail\":\"%s\", \"sumFailRate\":\"%.2f\", \"discard\":\"%s\"}";
-
- private AtomicLong sendMsgNumSum = new AtomicLong(0);
-
- private AtomicLong sendMsgFailNumSum = new AtomicLong(0);
-
- private AtomicLong sendMsgDiscardNumSum = new AtomicLong(0);
-
- private AtomicLong sendMsgNumPerSecond = new AtomicLong(0);
-
- private LinkedList<Integer> sendMsgTPSSnapshots = new LinkedList<Integer>();
-
- public void snapshotSendMsgTPS() {
- Integer tps = sendMsgNumPerSecond.intValue();
- sendMsgTPSSnapshots.add(tps);
- sendMsgNumPerSecond.set(0);
- if (sendMsgTPSSnapshots.size() > STATIC_PERIOD / 1000) {
- sendMsgTPSSnapshots.removeFirst();
- }
- }
-
- public float maxSendMsgTPS() {
- float tps = Collections.max(sendMsgTPSSnapshots);
- return tps;
- }
-
- public float avgSendMsgTPS() {
- float tps = avg(sendMsgTPSSnapshots);
- return tps;
- }
-
- public void recordSendMsg() {
- sendMsgNumPerSecond.incrementAndGet();
- sendMsgNumSum.incrementAndGet();
- }
-
- public long getSendMsgNumSum() {
- return sendMsgNumSum.longValue();
- }
-
- public long getSendMsgFailNumSum() {
- return sendMsgFailNumSum.longValue();
- }
-
- public float getSendMsgFailRate() {
- return (sendMsgNumSum.longValue() == 0L) ? 0f : sendMsgFailNumSum.floatValue() / sendMsgNumSum.longValue();
- }
-
- public void recordSendMsgFailed() {
- sendMsgFailNumSum.incrementAndGet();
- }
-
- public void recordSendMsgDiscard() {
- sendMsgDiscardNumSum.incrementAndGet();
- }
-
- public void cleanSendMsgStat() {
- sendMsgNumSum.set(0L);
- sendMsgFailNumSum.set(0L);
- }
-
- public long getSendMsgDiscardNumSum() {
- return sendMsgDiscardNumSum.longValue();
- }
-
- ////////////////////////////////////////////////////////////////////////////
- public static final String EVENTMESH_MONITOR_FORMAT_PUSHMSG = "%15s : {\"maxPushMsgTPS\":\"%.1f\",\"avgPushMsgTPS\":\"%.1f\","
- +
- " \"sum\":\"%s\", \"sumFail\":\"%s\", \"sumFailRate\":\"%.1f\", "
- + "\"maxClientLatency\":\"%.1f\", \"avgClientLatency\":\"%.1f\"}";
-
- private float wholePushCost = 0f;
-
- private AtomicLong wholePushRequestNum = new AtomicLong(0);
-
- private AtomicLong maxHttpPushLatency = new AtomicLong(0);
-
- private AtomicLong pushMsgNumPerSecond = new AtomicLong(0);
-
- private LinkedList<Integer> pushMsgTPSSnapshots = new LinkedList<Integer>();
-
- private AtomicLong httpPushMsgNumSum = new AtomicLong(0);
-
- private AtomicLong httpPushFailNumSum = new AtomicLong(0);
-
- public void snapshotPushMsgTPS() {
- Integer tps = pushMsgNumPerSecond.intValue();
- pushMsgTPSSnapshots.add(tps);
- pushMsgNumPerSecond.set(0);
- if (pushMsgTPSSnapshots.size() > STATIC_PERIOD / 1000) {
- pushMsgTPSSnapshots.removeFirst();
- }
- }
-
- public void recordHTTPPushTimeCost(long cost) {
- wholePushRequestNum.incrementAndGet();
- wholePushCost = wholePushCost + cost;
- if (cost > maxHttpPushLatency.longValue()) {
- maxHttpPushLatency.set(cost);
- }
- }
-
- public float avgHTTPPushLatency() {
- return (wholePushRequestNum.longValue() == 0L) ? 0f : wholePushCost / wholePushRequestNum.longValue();
- }
-
- public float maxHTTPPushLatency() {
- return maxHttpPushLatency.floatValue();
- }
-
- public float maxPushMsgTPS() {
- float tps = Collections.max(pushMsgTPSSnapshots);
- return tps;
- }
-
- public float avgPushMsgTPS() {
- float tps = avg(pushMsgTPSSnapshots);
- return tps;
- }
-
- public void recordPushMsg() {
- pushMsgNumPerSecond.incrementAndGet();
- httpPushMsgNumSum.incrementAndGet();
- }
-
- public long getHttpPushMsgNumSum() {
- return httpPushMsgNumSum.longValue();
- }
-
- public long getHttpPushFailNumSum() {
- return httpPushFailNumSum.longValue();
- }
-
- public float getHttpPushMsgFailRate() {
- return (httpPushMsgNumSum.longValue() == 0L) ? 0f : httpPushFailNumSum.floatValue() / httpPushMsgNumSum.longValue();
- }
-
- public void recordHttpPushMsgFailed() {
- sendMsgFailNumSum.incrementAndGet();
- }
-
- public void cleanHttpPushMsgStat() {
- httpPushFailNumSum.set(0L);
- httpPushMsgNumSum.set(0L);
- wholeRequestNum.set(0L);
- wholeCost = 0f;
- maxCost.set(0L);
- }
-
- ///////////////////////////////////////////////////////////////////////////////////////////////////////////////
- public static final String EVENTMESH_MONITOR_FORMAT_BLOCKQ = "%15s : {\"batchMsgQ\":\"%s\",\"sendMsgQ\":\"%s\","
- +
- "\"pushMsgQ\":\"%s\",\"consumeRetryQ\":\"%s\"}";
-
- ///////////////////////////////////////////////////////////////////////////
- public static final String EVENTMESH_MONITOR_FORMAT_MQ_CLIENT = "%15s : {\"batchAvgSend2MQCost\":\"%.1f\", "
- + "\"avgSend2MQCost\":\"%.1f\"}";
-
- private float batchSend2MQWholeCost = 0f;
-
- private AtomicLong batchSend2MQNum = new AtomicLong(0);
-
- private float send2MQWholeCost = 0f;
-
- private AtomicLong send2MQNum = new AtomicLong(0);
-
- public void recordBatchSendMsgCost(long cost) {
- batchSend2MQNum.incrementAndGet();
- batchSend2MQWholeCost = batchSend2MQWholeCost + cost;
- }
-
- public float avgBatchSendMsgCost() {
- float cost = (batchSend2MQNum.intValue() == 0) ? 0f : batchSend2MQWholeCost / batchSend2MQNum.intValue();
- return cost;
- }
-
- public void recordSendMsgCost(long cost) {
- send2MQNum.incrementAndGet();
- send2MQWholeCost = send2MQWholeCost + cost;
- }
-
- public float avgSendMsgCost() {
- float cost = (send2MQNum.intValue() == 0) ? 0f : send2MQWholeCost / send2MQNum.intValue();
- return cost;
- }
-
- public void send2MQStatInfoClear() {
- batchSend2MQWholeCost = 0f;
- batchSend2MQNum.set(0L);
- send2MQWholeCost = 0f;
- send2MQNum.set(0L);
- }
-}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/opentelemetry/OpenTelemetryHTTPMetricsExporter.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/opentelemetry/OpenTelemetryHTTPMetricsExporter.java
deleted file mode 100644
index 65d1357..0000000
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/opentelemetry/OpenTelemetryHTTPMetricsExporter.java
+++ /dev/null
@@ -1,310 +0,0 @@
-/*
- * 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.runtime.metrics.opentelemetry;
-
-import org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
-import org.apache.eventmesh.runtime.metrics.http.HTTPMetricsServer;
-import org.apache.eventmesh.runtime.metrics.http.SummaryMetrics;
-
-import io.opentelemetry.api.metrics.GlobalMeterProvider;
-import io.opentelemetry.api.metrics.Meter;
-import io.opentelemetry.api.metrics.common.Labels;
-
-public class OpenTelemetryHTTPMetricsExporter {
-
- private final SummaryMetrics summaryMetrics;
-
- private final HTTPMetricsServer httpMetricsServer;
-
- public OpenTelemetryHTTPMetricsExporter(HTTPMetricsServer httpMetricsServer, EventMeshHTTPConfiguration eventMeshHTTPConfiguration) {
- OpenTelemetryPrometheusExporter.initialize(eventMeshHTTPConfiguration);
- this.httpMetricsServer = httpMetricsServer;
- this.summaryMetrics = httpMetricsServer.summaryMetrics;
- }
-
- public void start() {
- Meter meter = GlobalMeterProvider.getMeter("apache-eventmesh");
- //maxHTTPTPS
- meter
- .doubleValueObserverBuilder("eventmesh.http.request.tps.max")
- .setDescription("max TPS of HTTP.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.maxHTTPTPS(), Labels.empty()))
- .build();
-
- //avgHTTPTPS
- meter
- .doubleValueObserverBuilder("eventmesh.http.request.tps.avg")
- .setDescription("avg TPS of HTTP.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.avgHTTPTPS(), Labels.empty()))
- .build();
-
- //maxHTTPCost
- meter
- .longValueObserverBuilder("eventmesh.http.request.cost.max")
- .setDescription("max cost of HTTP.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.maxHTTPCost(), Labels.empty()))
- .build();
-
- //avgHTTPCost
- meter
- .doubleValueObserverBuilder("eventmesh.http.request.cost.avg")
- .setDescription("avg cost of HTTP.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.avgHTTPCost(), Labels.empty()))
- .build();
-
- //avgHTTPBodyDecodeCost
- meter
- .doubleValueObserverBuilder("eventmesh.http.body.decode.cost.avg")
- .setDescription("avg body decode cost of HTTP.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.avgHTTPBodyDecodeCost(), Labels.empty()))
- .build();
-
- //httpDiscard
- meter
- .longValueObserverBuilder("eventmesh.http.request.discard.num")
- .setDescription("http request discard num.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.getHttpDiscard(), Labels.empty()))
- .build();
-
- //maxBatchSendMsgTPS
- meter
- .doubleValueObserverBuilder("eventmesh.batch.send.message.tps.max")
- .setDescription("max of batch send message tps.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.maxSendBatchMsgTPS(), Labels.empty()))
- .build();
-
- //avgBatchSendMsgTPS
- meter
- .doubleValueObserverBuilder("eventmesh.batch.send.message.tps.avg")
- .setDescription("avg of batch send message tps.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.avgSendBatchMsgTPS(), Labels.empty()))
- .build();
-
- //sum
- meter
- .doubleValueObserverBuilder("eventmesh.batch.send.message.num")
- .setDescription("sum of batch send message number.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.getSendBatchMsgNumSum(), Labels.empty()))
- .build();
-
- //sumFail
- meter
- .doubleValueObserverBuilder("eventmesh.batch.send.message.fail.num")
- .setDescription("sum of batch send message fail message number.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.getSendBatchMsgFailNumSum(), Labels.empty()))
- .build();
-
- //sumFailRate
- meter
- .doubleValueObserverBuilder("eventmesh.batch.send.message.fail.rate")
- .setDescription("send batch message fail rate.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.getSendBatchMsgFailRate(), Labels.empty()))
- .build();
-
- //discard
- meter
- .doubleValueObserverBuilder("eventmesh.batch.send.message.discard.num")
- .setDescription("sum of send batch message discard number.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.getSendBatchMsgDiscardNumSum(), Labels.empty()))
- .build();
-
- //maxSendMsgTPS
- meter
- .doubleValueObserverBuilder("eventmesh.send.message.tps.max")
- .setDescription("max of send message tps.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.maxSendMsgTPS(), Labels.empty()))
- .build();
-
- //avgSendMsgTPS
- meter
- .doubleValueObserverBuilder("eventmesh.send.message.tps.avg")
- .setDescription("avg of send message tps.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.avgSendMsgTPS(), Labels.empty()))
- .build();
-
- //sum
- meter
- .doubleValueObserverBuilder("eventmesh.send.message.num")
- .setDescription("sum of send message number.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.getSendMsgNumSum(), Labels.empty()))
- .build();
-
- //sumFail
- meter
- .doubleValueObserverBuilder("eventmesh.send.message.fail.num")
- .setDescription("sum of send message fail number.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.getSendMsgFailNumSum(), Labels.empty()))
- .build();
-
- //sumFailRate
- meter
- .doubleValueObserverBuilder("eventmesh.send.message.fail.rate")
- .setDescription("send message fail rate.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.getSendMsgFailRate(), Labels.empty()))
- .build();
-
- //replyMsg
- meter
- .doubleValueObserverBuilder("eventmesh.reply.message.num")
- .setDescription("sum of reply message number.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.getReplyMsgNumSum(), Labels.empty()))
- .build();
-
- //replyFail
- meter
- .doubleValueObserverBuilder("eventmesh.reply.message.fail.num")
- .setDescription("sum of reply message fail number.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.getReplyMsgFailNumSum(), Labels.empty()))
- .build();
-
- //maxPushMsgTPS
- meter
- .doubleValueObserverBuilder("eventmesh.push.message.tps.max")
- .setDescription("max of push message tps.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.maxPushMsgTPS(), Labels.empty()))
- .build();
-
- //avgPushMsgTPS
- meter
- .doubleValueObserverBuilder("eventmesh.push.message.tps.avg")
- .setDescription("avg of push message tps.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.avgPushMsgTPS(), Labels.empty()))
- .build();
-
- //sum
- meter
- .doubleValueObserverBuilder("eventmesh.http.push.message.num")
- .setDescription("sum of http push message number.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.getHttpPushMsgNumSum(), Labels.empty()))
- .build();
-
- //sumFail
- meter
- .doubleValueObserverBuilder("eventmesh.http.push.message.fail.num")
- .setDescription("sum of http push message fail number.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.getHttpPushFailNumSum(), Labels.empty()))
- .build();
-
- //sumFailRate
- meter
- .doubleValueObserverBuilder("eventmesh.http.push.message.fail.rate")
- .setDescription("http push message fail rate.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.getHttpPushMsgFailRate(), Labels.empty()))
- .build();
-
- //maxClientLatency
- meter
- .doubleValueObserverBuilder("eventmesh.http.push.latency.max")
- .setDescription("max of http push latency.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.maxHTTPPushLatency(), Labels.empty()))
- .build();
-
- //avgClientLatency
- meter
- .doubleValueObserverBuilder("eventmesh.http.push.latency.avg")
- .setDescription("avg of http push latency.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.avgHTTPPushLatency(), Labels.empty()))
- .build();
-
- //batchMsgQ
- meter
- .longValueObserverBuilder("eventmesh.batch.message.queue.size")
- .setDescription("size of batch message queue.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(httpMetricsServer.getBatchMsgQ(), Labels.empty()))
- .build();
-
- //sendMsgQ
- meter
- .longValueObserverBuilder("eventmesh.send.message.queue.size")
- .setDescription("size of send message queue.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(httpMetricsServer.getSendMsgQ(), Labels.empty()))
- .build();
-
- //pushMsgQ
- meter
- .longValueObserverBuilder("eventmesh.push.message.queue.size")
- .setDescription("size of push message queue.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(httpMetricsServer.getPushMsgQ(), Labels.empty()))
- .build();
-
- //httpRetryQ
- meter
- .longValueObserverBuilder("eventmesh.http.retry.queue.size")
- .setDescription("size of http retry queue.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(httpMetricsServer.getHttpRetryQ(), Labels.empty()))
- .build();
-
- //batchAvgSend2MQCost
- meter
- .doubleValueObserverBuilder("eventmesh.batch.send.message.cost.avg")
- .setDescription("avg of batch send message cost.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.avgBatchSendMsgCost(), Labels.empty()))
- .build();
-
- //avgSend2MQCost
- meter
- .doubleValueObserverBuilder("eventmesh.send.message.cost.avg")
- .setDescription("avg of send message cost.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.avgSendMsgCost(), Labels.empty()))
- .build();
-
- //avgReply2MQCost
- meter
- .doubleValueObserverBuilder("eventmesh.reply.message.cost.avg")
- .setDescription("avg of reply message cost.")
- .setUnit("HTTP")
- .setUpdater(result -> result.observe(summaryMetrics.avgReplyMsgCost(), Labels.empty()))
- .build();
- }
-
- public void shutdown() {
- OpenTelemetryPrometheusExporter.shutdown();
- }
-}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/opentelemetry/OpenTelemetryPrometheusExporter.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/opentelemetry/OpenTelemetryPrometheusExporter.java
deleted file mode 100644
index f3ca627..0000000
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/opentelemetry/OpenTelemetryPrometheusExporter.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * 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.runtime.metrics.opentelemetry;
-
-import org.apache.eventmesh.common.config.CommonConfiguration;
-
-import java.io.IOException;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import io.opentelemetry.exporter.prometheus.PrometheusCollector;
-import io.opentelemetry.sdk.metrics.SdkMeterProvider;
-import io.prometheus.client.exporter.HTTPServer;
-
-public class OpenTelemetryPrometheusExporter {
-
- private static final Logger logger = LoggerFactory.getLogger(OpenTelemetryPrometheusExporter.class);
-
- private static HTTPServer server;
-
- /**
- * Initializes the Meter SDK and configures the prometheus collector with all default settings.
- *
- * @param configuration configuration.
- */
- public static synchronized void initialize(CommonConfiguration configuration) {
- if (server != null) {
- return;
- }
- PrometheusCollector.builder().setMetricProducer(
- SdkMeterProvider.builder().buildAndRegisterGlobal()).buildAndRegister();
- int port = configuration.eventMeshPrometheusPort;
- try {
- //Use the daemon thread to start an HTTP server to serve the default Prometheus registry.
- server = new HTTPServer(port, true);
- } catch (IOException e) {
- logger.error("failed to start prometheus server, port: {} due to {}", port, e.getMessage());
- }
- }
-
- public static void shutdown() {
- if (server != null) {
- server.stop();
- server = null;
- }
- }
-}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/opentelemetry/OpenTelemetryTCPMetricsExporter.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/opentelemetry/OpenTelemetryTCPMetricsExporter.java
deleted file mode 100644
index 5b4f6ce..0000000
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/opentelemetry/OpenTelemetryTCPMetricsExporter.java
+++ /dev/null
@@ -1,102 +0,0 @@
-/*
- * 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.runtime.metrics.opentelemetry;
-
-import org.apache.eventmesh.runtime.configuration.EventMeshTCPConfiguration;
-import org.apache.eventmesh.runtime.core.protocol.tcp.client.EventMeshTcpConnectionHandler;
-import org.apache.eventmesh.runtime.metrics.tcp.EventMeshTcpMonitor;
-
-import io.opentelemetry.api.metrics.GlobalMeterProvider;
-import io.opentelemetry.api.metrics.Meter;
-import io.opentelemetry.api.metrics.common.Labels;
-
-public class OpenTelemetryTCPMetricsExporter {
-
- private final EventMeshTcpMonitor eventMeshTcpMonitor;
-
- public OpenTelemetryTCPMetricsExporter(EventMeshTcpMonitor eventMeshTcpMonitor,
- EventMeshTCPConfiguration eventMeshTCPConfiguration) {
- OpenTelemetryPrometheusExporter.initialize(eventMeshTCPConfiguration);
- this.eventMeshTcpMonitor = eventMeshTcpMonitor;
- }
-
- public void start() {
- Meter meter = GlobalMeterProvider.getMeter("apache-eventmesh");
- //retryQueueSize
- meter.doubleValueObserverBuilder("eventmesh.tcp.retry.queue.size")
- .setDescription("get size of retry queue.")
- .setUnit("TCP")
- .setUpdater(result -> result.observe(
- eventMeshTcpMonitor.getEventMeshTCPServer()
- .getEventMeshTcpRetryer().getRetrySize(), Labels.empty()))
- .build();
-
- //client2eventMeshTPS
- meter.doubleValueObserverBuilder("eventmesh.tcp.server.tps")
- .setDescription("get tps of client to eventMesh.")
- .setUnit("TCP")
- .setUpdater(result -> result.observe(eventMeshTcpMonitor.getClient2eventMeshTPS(), Labels.empty()))
- .build();
-
- //eventMesh2mqTPS
- meter.doubleValueObserverBuilder("eventmesh.tcp.mq.provider.tps")
- .setDescription("get tps of eventMesh to mq.")
- .setUnit("TCP")
- .setUpdater(result -> result.observe(eventMeshTcpMonitor.getEventMesh2mqTPS(), Labels.empty()))
- .build();
-
- //mq2eventMeshTPS
- meter.doubleValueObserverBuilder("eventmesh.tcp.mq.consumer.tps")
- .setDescription("get tps of mq to eventMesh.")
- .setUnit("TCP")
- .setUpdater(result -> result.observe(eventMeshTcpMonitor.getMq2eventMeshTPS(), Labels.empty()))
- .build();
-
- //eventMesh2clientTPS
- meter.doubleValueObserverBuilder("eventmesh.tcp.client.tps")
- .setDescription("get tps of eventMesh to client.")
- .setUnit("TCP")
- .setUpdater(result -> result.observe(eventMeshTcpMonitor.getEventMesh2clientTPS(), Labels.empty()))
- .build();
-
- //allTPS
- meter.doubleValueObserverBuilder("eventmesh.tcp.all.tps")
- .setDescription("get all TPS.")
- .setUnit("TCP")
- .setUpdater(result -> result.observe(eventMeshTcpMonitor.getAllTPS(), Labels.empty()))
- .build();
-
- //EventMeshTcpConnectionHandler.connections
- meter.doubleValueObserverBuilder("eventmesh.tcp.connection.num")
- .setDescription("EventMeshTcpConnectionHandler.connections.")
- .setUnit("TCP")
- .setUpdater(result -> result.observe(EventMeshTcpConnectionHandler.connections.doubleValue(), Labels.empty()))
- .build();
-
- //subTopicNum
- meter.doubleValueObserverBuilder("eventmesh.tcp.sub.topic.num")
- .setDescription("get sub topic num.")
- .setUnit("TCP")
- .setUpdater(result -> result.observe(eventMeshTcpMonitor.getSubTopicNum(), Labels.empty()))
- .build();
- }
-
- public void shutdown() {
- OpenTelemetryPrometheusExporter.shutdown();
- }
-}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/tcp/EventMeshTcpMonitor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/tcp/EventMeshTcpMonitor.java
index 9bf4d39..dea7233 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/tcp/EventMeshTcpMonitor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/tcp/EventMeshTcpMonitor.java
@@ -17,29 +17,32 @@
package org.apache.eventmesh.runtime.metrics.tcp;
+import org.apache.eventmesh.metrics.api.MetricsRegistry;
+import org.apache.eventmesh.metrics.api.model.TcpSummaryMetrics;
import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
import org.apache.eventmesh.runtime.core.protocol.tcp.client.EventMeshTcpConnectionHandler;
import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
import org.apache.eventmesh.runtime.metrics.MonitorMetricConstants;
-import org.apache.eventmesh.runtime.metrics.opentelemetry.OpenTelemetryTCPMetricsExporter;
import java.net.InetSocketAddress;
import java.util.HashSet;
import java.util.Iterator;
+import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.base.Preconditions;
+
public class EventMeshTcpMonitor {
- private EventMeshTCPServer eventMeshTCPServer;
+ private final EventMeshTCPServer eventMeshTCPServer;
public EventMeshTCPServer getEventMeshTCPServer() {
return eventMeshTCPServer;
@@ -57,63 +60,47 @@ public class EventMeshTcpMonitor {
private static int PRINT_THREADPOOLSTATE_INTERVAL = 1;
- private AtomicInteger client2eventMeshMsgNum;
- private AtomicInteger eventMesh2mqMsgNum;
- private AtomicInteger mq2eventMeshMsgNum;
- private AtomicInteger eventMesh2clientMsgNum;
-
- private int client2eventMeshTPS;
- private int eventMesh2clientTPS;
- private int eventMesh2mqTPS;
- private int mq2eventMeshTPS;
- private int allTPS;
- private int subTopicNum;
-
- private OpenTelemetryTCPMetricsExporter metricsExporter;
-
public ScheduledFuture<?> monitorTpsTask;
public ScheduledFuture<?> monitorThreadPoolTask;
- public EventMeshTcpMonitor(EventMeshTCPServer eventMeshTCPServer) {
+ private final TcpSummaryMetrics tcpSummaryMetrics;
+
+ private final List<MetricsRegistry> metricsRegistries;
+
+ public EventMeshTcpMonitor(EventMeshTCPServer eventMeshTCPServer, List<MetricsRegistry> metricsRegistries) {
this.eventMeshTCPServer = eventMeshTCPServer;
+ this.tcpSummaryMetrics = new TcpSummaryMetrics();
+ this.metricsRegistries = Preconditions.checkNotNull(metricsRegistries);
}
public void init() throws Exception {
- this.client2eventMeshMsgNum = new AtomicInteger(0);
- this.eventMesh2mqMsgNum = new AtomicInteger(0);
- this.mq2eventMeshMsgNum = new AtomicInteger(0);
- this.eventMesh2clientMsgNum = new AtomicInteger(0);
- this.metricsExporter = new OpenTelemetryTCPMetricsExporter(
- this, eventMeshTCPServer.getEventMeshTCPConfiguration());
- logger.info("EventMeshTcpMonitor inited......");
+ metricsRegistries.forEach(MetricsRegistry::start);
+ logger.info("EventMeshTcpMonitor initialized......");
}
public void start() throws Exception {
- metricsExporter.start();
+ metricsRegistries.forEach(metricsRegistry -> {
+ metricsRegistry.register(tcpSummaryMetrics);
+ logger.info("Register tcpMetrics to " + metricsRegistry.getClass().getName());
+ });
monitorTpsTask = eventMeshTCPServer.getScheduler().scheduleAtFixedRate((() -> {
- int msgNum = client2eventMeshMsgNum.intValue();
- client2eventMeshMsgNum = new AtomicInteger(0);
- client2eventMeshTPS = 1000 * msgNum / period;
+ int msgNum = tcpSummaryMetrics.client2eventMeshMsgNum();
+ tcpSummaryMetrics.setClient2eventMeshTPS(1000 * msgNum / period);
- msgNum = eventMesh2clientMsgNum.intValue();
- eventMesh2clientMsgNum = new AtomicInteger(0);
- eventMesh2clientTPS = 1000 * msgNum / period;
+ msgNum = tcpSummaryMetrics.eventMesh2clientMsgNum();
+ tcpSummaryMetrics.setEventMesh2clientTPS(1000 * msgNum / period);
- msgNum = eventMesh2mqMsgNum.intValue();
- eventMesh2mqMsgNum = new AtomicInteger(0);
- eventMesh2mqTPS = 1000 * msgNum / period;
+ msgNum = tcpSummaryMetrics.eventMesh2mqMsgNum();
+ tcpSummaryMetrics.setEventMesh2mqTPS(1000 * msgNum / period);
- msgNum = mq2eventMeshMsgNum.intValue();
- mq2eventMeshMsgNum = new AtomicInteger(0);
- mq2eventMeshTPS = 1000 * msgNum / period;
-
- allTPS = client2eventMeshTPS + eventMesh2clientTPS;
+ msgNum = tcpSummaryMetrics.mq2eventMeshMsgNum();
+ tcpSummaryMetrics.setMq2eventMeshTPS(1000 * msgNum / period);
//count topics subscribed by client in this eventMesh
- ConcurrentHashMap<InetSocketAddress, Session> sessionMap =
- eventMeshTCPServer.getClientSessionGroupMapping().getSessionMap();
+ ConcurrentHashMap<InetSocketAddress, Session> sessionMap =
+ eventMeshTCPServer.getClientSessionGroupMapping().getSessionMap();
Iterator<Session> sessionIterator = sessionMap.values().iterator();
Set<String> topicSet = new HashSet<>();
while (sessionIterator.hasNext()) {
@@ -125,28 +112,55 @@ public class EventMeshTcpMonitor {
int subscribeTopics = session.getSessionContext().subscribeTopics.size();
tcpLogger.info("session|deliveredFailCount={}|deliveredMsgsCount={}|unAckMsgsCount={}|sendTopics={}|subscribeTopics={}|user={}",
- deliveredFailCount.longValue(), deliveredMsgsCount.longValue(),
- unAckMsgsCount, sendTopics, subscribeTopics, session.getClient());
+ deliveredFailCount.longValue(), deliveredMsgsCount.longValue(),
+ unAckMsgsCount, sendTopics, subscribeTopics, session.getClient());
topicSet.addAll(session.getSessionContext().subscribeTopics.keySet());
}
- subTopicNum = topicSet.size();
-
- appLogger.info(String.format(MonitorMetricConstants.EVENTMESH_MONITOR_FORMAT_COMMON,
- EventMeshConstants.PROTOCOL_TCP, MonitorMetricConstants.CLIENT_2_EVENTMESH_TPS, client2eventMeshTPS));
- appLogger.info(String.format(MonitorMetricConstants.EVENTMESH_MONITOR_FORMAT_COMMON,
- EventMeshConstants.PROTOCOL_TCP, MonitorMetricConstants.EVENTMESH_2_MQ_TPS, eventMesh2mqTPS));
- appLogger.info(String.format(MonitorMetricConstants.EVENTMESH_MONITOR_FORMAT_COMMON,
- EventMeshConstants.PROTOCOL_TCP, MonitorMetricConstants.MQ_2_EVENTMESH_TPS, mq2eventMeshTPS));
- appLogger.info(String.format(MonitorMetricConstants.EVENTMESH_MONITOR_FORMAT_COMMON,
- EventMeshConstants.PROTOCOL_TCP, MonitorMetricConstants.EVENTMESH_2_CLIENT_TPS, eventMesh2clientTPS));
- appLogger.info(String.format(MonitorMetricConstants.EVENTMESH_MONITOR_FORMAT_COMMON,
- EventMeshConstants.PROTOCOL_TCP, MonitorMetricConstants.ALL_TPS, allTPS));
- appLogger.info(String.format(MonitorMetricConstants.EVENTMESH_MONITOR_FORMAT_COMMON,
- EventMeshConstants.PROTOCOL_TCP, MonitorMetricConstants.CONNECTION,
- EventMeshTcpConnectionHandler.connections));
- appLogger.info(String.format(MonitorMetricConstants.EVENTMESH_MONITOR_FORMAT_COMMON,
- EventMeshConstants.PROTOCOL_TCP, MonitorMetricConstants.SUB_TOPIC_NUM, subTopicNum));
+ tcpSummaryMetrics.setSubTopicNum(topicSet.size());
+ tcpSummaryMetrics.setAllConnections(EventMeshTcpConnectionHandler.connections.get());
+
+ appLogger.info(String.format(
+ MonitorMetricConstants.EVENTMESH_MONITOR_FORMAT_COMMON,
+ EventMeshConstants.PROTOCOL_TCP,
+ MonitorMetricConstants.CLIENT_2_EVENTMESH_TPS,
+ tcpSummaryMetrics.getClient2eventMeshTPS()));
+
+ appLogger.info(String.format(
+ MonitorMetricConstants.EVENTMESH_MONITOR_FORMAT_COMMON,
+ EventMeshConstants.PROTOCOL_TCP,
+ MonitorMetricConstants.EVENTMESH_2_MQ_TPS,
+ tcpSummaryMetrics.getEventMesh2mqTPS()));
+
+ appLogger.info(String.format(
+ MonitorMetricConstants.EVENTMESH_MONITOR_FORMAT_COMMON,
+ EventMeshConstants.PROTOCOL_TCP,
+ MonitorMetricConstants.MQ_2_EVENTMESH_TPS,
+ tcpSummaryMetrics.getMq2eventMeshTPS()));
+
+ appLogger.info(String.format(
+ MonitorMetricConstants.EVENTMESH_MONITOR_FORMAT_COMMON,
+ EventMeshConstants.PROTOCOL_TCP,
+ MonitorMetricConstants.EVENTMESH_2_CLIENT_TPS,
+ tcpSummaryMetrics.getEventMesh2clientTPS()));
+
+ appLogger.info(String.format(
+ MonitorMetricConstants.EVENTMESH_MONITOR_FORMAT_COMMON,
+ EventMeshConstants.PROTOCOL_TCP,
+ MonitorMetricConstants.ALL_TPS,
+ tcpSummaryMetrics.getAllTPS()));
+
+ appLogger.info(String.format(
+ MonitorMetricConstants.EVENTMESH_MONITOR_FORMAT_COMMON,
+ EventMeshConstants.PROTOCOL_TCP,
+ MonitorMetricConstants.CONNECTION,
+ tcpSummaryMetrics.getAllConnections()));
+
+ appLogger.info(String.format(
+ MonitorMetricConstants.EVENTMESH_MONITOR_FORMAT_COMMON,
+ EventMeshConstants.PROTOCOL_TCP,
+ MonitorMetricConstants.SUB_TOPIC_NUM,
+ tcpSummaryMetrics.getSubTopicNum()));
}), delay, period, TimeUnit.MILLISECONDS);
monitorThreadPoolTask = eventMeshTCPServer.getScheduler().scheduleAtFixedRate(() -> {
@@ -155,57 +169,25 @@ public class EventMeshTcpMonitor {
eventMeshTCPServer.getEventMeshTcpRetryer().printRetryThreadPoolState();
//monitor retry queue size
- int retrySize = eventMeshTCPServer.getEventMeshTcpRetryer().getRetrySize();
- appLogger.info(String.format(MonitorMetricConstants.EVENTMESH_MONITOR_FORMAT_COMMON,
- EventMeshConstants.PROTOCOL_TCP, MonitorMetricConstants.RETRY_QUEUE_SIZE, retrySize));
+ tcpSummaryMetrics.setRetrySize(eventMeshTCPServer.getEventMeshTcpRetryer().getRetrySize());
+ appLogger.info(String.format(
+ MonitorMetricConstants.EVENTMESH_MONITOR_FORMAT_COMMON,
+ EventMeshConstants.PROTOCOL_TCP,
+ MonitorMetricConstants.RETRY_QUEUE_SIZE,
+ tcpSummaryMetrics.getRetrySize()));
+
}, 10, PRINT_THREADPOOLSTATE_INTERVAL, TimeUnit.SECONDS);
logger.info("EventMeshTcpMonitor started......");
}
+ public TcpSummaryMetrics getTcpSummaryMetrics() {
+ return tcpSummaryMetrics;
+ }
+
public void shutdown() throws Exception {
monitorTpsTask.cancel(true);
monitorThreadPoolTask.cancel(true);
- metricsExporter.shutdown();
+ metricsRegistries.forEach(MetricsRegistry::showdown);
logger.info("EventMeshTcpMonitor shutdown......");
}
-
- public AtomicInteger getClient2EventMeshMsgNum() {
- return client2eventMeshMsgNum;
- }
-
- public AtomicInteger getEventMesh2mqMsgNum() {
- return eventMesh2mqMsgNum;
- }
-
- public AtomicInteger getMq2EventMeshMsgNum() {
- return mq2eventMeshMsgNum;
- }
-
- public AtomicInteger getEventMesh2clientMsgNum() {
- return eventMesh2clientMsgNum;
- }
-
- public int getClient2eventMeshTPS() {
- return client2eventMeshTPS;
- }
-
- public int getEventMesh2clientTPS() {
- return eventMesh2clientTPS;
- }
-
- public int getEventMesh2mqTPS() {
- return eventMesh2mqTPS;
- }
-
- public int getMq2eventMeshTPS() {
- return mq2eventMeshTPS;
- }
-
- public int getAllTPS() {
- return allTPS;
- }
-
- public int getSubTopicNum() {
- return subTopicNum;
- }
}
\ No newline at end of file
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/Utils.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/Utils.java
index e6be33a..d2b357c 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/Utils.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/Utils.java
@@ -67,8 +67,8 @@ public class Utils {
logSucceedMessageFlow(pkg, user, startTime, taskExecuteTime);
if (session != null) {
- session.getClientGroupWrapper().get().getEventMeshTcpMonitor()
- .getEventMesh2clientMsgNum().incrementAndGet();
+ session.getClientGroupWrapper().get()
+ .getEventMeshTcpMonitor().getTcpSummaryMetrics().getEventMesh2clientMsgNum().incrementAndGet();
}
}
}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
index cbe9f3e..e126bed 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
@@ -26,6 +26,7 @@ public enum EventMeshExtensionType {
REGISTRY("registry"),
SECURITY("security"),
PROTOCOL("protocol"),
+ METRICS("metrics"),
;
private final String extensionTypeName;
diff --git a/settings.gradle b/settings.gradle
index 1d94585..e4ab6f3 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -36,4 +36,7 @@ include 'eventmesh-protocol-plugin:eventmesh-protocol-openmessage'
include 'eventmesh-protocol-plugin:eventmesh-protocol-cloudevents'
include 'eventmesh-protocol-plugin:eventmesh-protocol-meshmessage'
include 'eventmesh-admin:eventmesh-admin-rocketmq'
+include 'eventmesh-metrics-plugin'
+include 'eventmesh-metrics-plugin:eventmesh-metrics-api'
+include 'eventmesh-metrics-plugin:eventmesh-metrics-opentelemetry'
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org