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/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new e748f2472 [ISSUE #3912]There are boilerplate codes in
PrometheusTcpExporter, PrometheusHttpExporter (#3916)
e748f2472 is described below
commit e748f24722708f80a45f2aaaaf79fc3e54ccf83a
Author: pandaapo <[email protected]>
AuthorDate: Tue Jun 6 15:05:59 2023 +0800
[ISSUE #3912]There are boilerplate codes in PrometheusTcpExporter,
PrometheusHttpExporter (#3916)
* Replace boilerplate codes with shared method.
* Add comment for new field.
* Fix CI error.
---
.../prometheus/metrics/PrometheusGrpcExporter.java | 45 ++-
.../prometheus/metrics/PrometheusHttpExporter.java | 415 ++++++++-------------
.../prometheus/metrics/PrometheusTcpExporter.java | 110 +++---
.../utils/PrometheusExporterConstants.java | 35 ++
.../prometheus/utils/PrometheusExporterUtils.java | 67 ++++
5 files changed, 332 insertions(+), 340 deletions(-)
diff --git
a/eventmesh-metrics-plugin/eventmesh-metrics-prometheus/src/main/java/org/apache/eventmesh/metrics/prometheus/metrics/PrometheusGrpcExporter.java
b/eventmesh-metrics-plugin/eventmesh-metrics-prometheus/src/main/java/org/apache/eventmesh/metrics/prometheus/metrics/PrometheusGrpcExporter.java
index a99da83b6..d3846bba1 100644
---
a/eventmesh-metrics-plugin/eventmesh-metrics-prometheus/src/main/java/org/apache/eventmesh/metrics/prometheus/metrics/PrometheusGrpcExporter.java
+++
b/eventmesh-metrics-plugin/eventmesh-metrics-prometheus/src/main/java/org/apache/eventmesh/metrics/prometheus/metrics/PrometheusGrpcExporter.java
@@ -17,40 +17,49 @@
package org.apache.eventmesh.metrics.prometheus.metrics;
+import static
org.apache.eventmesh.metrics.prometheus.utils.PrometheusExporterConstants.GRPC;
+import static
org.apache.eventmesh.metrics.prometheus.utils.PrometheusExporterConstants.METRICS_GRPC_PREFIX;
+import static
org.apache.eventmesh.metrics.prometheus.utils.PrometheusExporterUtils.join;
+import static
org.apache.eventmesh.metrics.prometheus.utils.PrometheusExporterUtils.observeOfValue;
+
import org.apache.eventmesh.metrics.api.model.GrpcSummaryMetrics;
-import java.util.function.Supplier;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Function;
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 PrometheusGrpcExporter {
- private static final String UNIT = "GRPC";
- private static final String METRICS_NAME_PREFIX = "eventmesh.grpc.";
+ /**
+ * Map structure : [metric name, description of name] -> the method of get
corresponding metric.
+ */
+ private Map<String[], Function<GrpcSummaryMetrics, Number>> paramPairs;
+
+ static {
+ paramPairs = new HashMap<String[], Function<GrpcSummaryMetrics,
Number>>() {
+ {
+ put(join("sub.topic.num", "get sub topic num."),
GrpcSummaryMetrics::getSubscribeTopicNum);
+ put(join("retry.queue.size", "get size of retry queue."),
GrpcSummaryMetrics::getRetrySize);
+
+ put(join("server.tps", "get size of retry queue."),
GrpcSummaryMetrics::getClient2EventMeshTPS);
+ put(join("client.tps", "get tps of eventMesh to mq."),
GrpcSummaryMetrics::getEventMesh2ClientTPS);
- private void observeOfValue(Meter meter, String name, String desc,
Supplier<Long> supplier) {
- meter.doubleValueObserverBuilder(METRICS_NAME_PREFIX + name)
- .setDescription(desc)
- .setUnit(UNIT)
- .setUpdater(result -> result.observe(supplier.get(),
Labels.empty()))
- .build();
+ put(join("mq.provider.tps", "get tps of eventMesh to mq."),
GrpcSummaryMetrics::getEventMesh2MqTPS);
+ put(join("mq.consumer.tps", "get tps of eventMesh to mq."),
GrpcSummaryMetrics::getMq2EventMeshTPS);
+ }
+ };
}
public static void export(final String meterName, final GrpcSummaryMetrics
summaryMetrics) {
final Meter meter = GlobalMeterProvider.getMeter(meterName);
- observeOfValue(meter, "sub.topic.num", "get sub topic num.",
summaryMetrics::getSubscribeTopicNum);
- observeOfValue(meter, "retry.queue.size", "get size of retry queue.",
summaryMetrics::getRetrySize);
-
- observeOfValue(meter, "server.tps", "get size of retry queue.",
summaryMetrics::getClient2EventMeshTPS);
- observeOfValue(meter, "client.tps", "get tps of eventMesh to mq.",
summaryMetrics::getEventMesh2ClientTPS);
-
- observeOfValue(meter, "mq.provider.tps", "get tps of eventMesh to
mq.", summaryMetrics::getEventMesh2MqTPS);
- observeOfValue(meter, "mq.consumer.tps", "get tps of eventMesh to
mq.", summaryMetrics::getMq2EventMeshTPS);
+ paramPairs.forEach((k, v) ->
+ observeOfValue(meter, METRICS_GRPC_PREFIX + k[0], k[1], GRPC,
v.apply(summaryMetrics)));
}
}
diff --git
a/eventmesh-metrics-plugin/eventmesh-metrics-prometheus/src/main/java/org/apache/eventmesh/metrics/prometheus/metrics/PrometheusHttpExporter.java
b/eventmesh-metrics-plugin/eventmesh-metrics-prometheus/src/main/java/org/apache/eventmesh/metrics/prometheus/metrics/PrometheusHttpExporter.java
index 0f87aa8ea..bba9a89a0 100644
---
a/eventmesh-metrics-plugin/eventmesh-metrics-prometheus/src/main/java/org/apache/eventmesh/metrics/prometheus/metrics/PrometheusHttpExporter.java
+++
b/eventmesh-metrics-plugin/eventmesh-metrics-prometheus/src/main/java/org/apache/eventmesh/metrics/prometheus/metrics/PrometheusHttpExporter.java
@@ -17,282 +17,167 @@
package org.apache.eventmesh.metrics.prometheus.metrics;
+import static
org.apache.eventmesh.metrics.prometheus.utils.PrometheusExporterConstants.HTTP;
+import static
org.apache.eventmesh.metrics.prometheus.utils.PrometheusExporterUtils.join;
+import static
org.apache.eventmesh.metrics.prometheus.utils.PrometheusExporterUtils.observeOfValue;
+
import org.apache.eventmesh.metrics.api.model.HttpSummaryMetrics;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Function;
+
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 PrometheusHttpExporter {
- public static void export(String name, HttpSummaryMetrics summaryMetrics) {
+ /**
+ * Map structure : [metric name, description of name] -> the method of get
corresponding metric.
+ */
+ private Map<String[], Function<HttpSummaryMetrics, Number>> paramPairs;
+
+ static {
+ paramPairs = new HashMap<String[], Function<HttpSummaryMetrics,
Number>>() {
+ {
+ //maxHTTPTPS
+ put(join("eventmesh.http.request.tps.max", "max TPS of HTTP."),
+ HttpSummaryMetrics::maxHTTPTPS);
+
+ //avgHTTPTPS
+ put(join("eventmesh.http.request.tps.avg", "avg TPS of HTTP."),
+ HttpSummaryMetrics::avgHTTPTPS);
+
+ //maxHTTPCost
+ put(join("eventmesh.http.request.cost.max", "max cost of
HTTP."),
+ HttpSummaryMetrics::maxHTTPCost);
+
+ //avgHTTPCost
+ put(join("eventmesh.http.request.cost.avg", "avg cost of
HTTP."),
+ HttpSummaryMetrics::avgHTTPCost);
+
+ //avgHTTPBodyDecodeCost
+ put(join("eventmesh.http.body.decode.cost.avg", "avg body
decode cost of HTTP."),
+ HttpSummaryMetrics::avgHTTPBodyDecodeCost);
+
+ //httpDiscard
+ put(join("eventmesh.http.request.discard.num", "http request
discard num."),
+ HttpSummaryMetrics::getHttpDiscard);
+
+ //maxBatchSendMsgTPS
+ put(join("eventmesh.batch.send.message.tps.max", "max of batch
send message tps."),
+ HttpSummaryMetrics::maxSendBatchMsgTPS);
+
+ //avgBatchSendMsgTPS
+ put(join("eventmesh.batch.send.message.tps.avg", "avg of batch
send message tps."),
+ HttpSummaryMetrics::avgSendBatchMsgTPS);
+
+ //sum
+ put(join("eventmesh.batch.send.message.num", "sum of batch
send message number."),
+ HttpSummaryMetrics::getSendBatchMsgNumSum);
+
+ //sumFail
+ put(join("eventmesh.batch.send.message.fail.num", "sum of
batch send message fail message number."),
+ HttpSummaryMetrics::getSendBatchMsgFailNumSum);
+
+ //sumFailRate
+ put(join("eventmesh.batch.send.message.fail.rate", "send batch
message fail rate."),
+ HttpSummaryMetrics::getSendBatchMsgFailRate);
+ //discard
+ put(join("eventmesh.batch.send.message.discard.num", "sum of
send batch message discard number."),
+ HttpSummaryMetrics::getSendBatchMsgDiscardNumSum);
+
+ //maxSendMsgTPS
+ put(join("eventmesh.send.message.tps.max", "max of send
message tps."),
+ HttpSummaryMetrics::maxSendMsgTPS);
+
+ //avgSendMsgTPS
+ put(join("eventmesh.send.message.tps.avg", "avg of send
message tps."),
+ HttpSummaryMetrics::avgSendMsgTPS);
+
+ //sum
+ put(join("eventmesh.send.message.num", "sum of send message
number."),
+ HttpSummaryMetrics::getSendMsgNumSum);
+
+ //sumFail
+ put(join("eventmesh.send.message.fail.num", "sum of send
message fail number."),
+ HttpSummaryMetrics::getSendMsgFailNumSum);
+
+ //sumFailRate
+ put(join("eventmesh.send.message.fail.rate", "send message
fail rate."),
+ HttpSummaryMetrics::getSendMsgFailRate);
+
+ //replyMsg
+ put(join("eventmesh.reply.message.num", "sum of reply message
number."),
+ HttpSummaryMetrics::getReplyMsgNumSum);
+
+ //replyFail
+ put(join("eventmesh.reply.message.fail.num", "sum of reply
message fail number."),
+ HttpSummaryMetrics::getReplyMsgFailNumSum);
+
+ //maxPushMsgTPS
+ put(join("eventmesh.push.message.tps.max", "max of push
message tps."),
+ HttpSummaryMetrics::maxPushMsgTPS);
+
+ //avgPushMsgTPS
+ put(join("eventmesh.push.message.tps.avg", "avg of push
message tps."),
+ HttpSummaryMetrics::avgPushMsgTPS);
+
+ //sum
+ put(join("eventmesh.http.push.message.num", "sum of http push
message number."),
+ HttpSummaryMetrics::getHttpPushMsgNumSum);
+ //sumFail
+ put(join("eventmesh.http.push.message.fail.num", "sum of http
push message fail number."),
+ HttpSummaryMetrics::getHttpPushFailNumSum);
+
+ //sumFailRate
+ put(join("eventmesh.http.push.message.fail.rate", "http push
message fail rate."),
+ HttpSummaryMetrics::getHttpPushMsgFailRate);
+
+ //maxClientLatency
+ put(join("eventmesh.http.push.latency.max", "max of http push
latency."),
+ HttpSummaryMetrics::maxHTTPPushLatency);
+
+ //avgClientLatency
+ put(join("eventmesh.http.push.latency.avg", "avg of http push
latency."),
+ HttpSummaryMetrics::avgHTTPPushLatency);
+ //batchMsgQ
+ put(join("eventmesh.batch.message.queue.size", "size of batch
message queue."),
+ HttpSummaryMetrics::getBatchMsgQueueSize);
+
+ //sendMsgQ
+ put(join("eventmesh.send.message.queue.size", "size of send
message queue."),
+ HttpSummaryMetrics::getSendMsgQueueSize);
+
+ //pushMsgQ
+ put(join("eventmesh.push.message.queue.size", "size of push
message queue."),
+ HttpSummaryMetrics::getPushMsgQueueSize);
+
+ //httpRetryQ
+ put(join("eventmesh.http.retry.queue.size", "size of http
retry queue."),
+ HttpSummaryMetrics::getHttpRetryQueueSize);
+
+ //batchAvgSend2MQCost
+ put(join("eventmesh.batch.send.message.cost.avg", "avg of
batch send message cost."),
+ HttpSummaryMetrics::avgBatchSendMsgCost);
+
+ //avgSend2MQCost
+ put(join("eventmesh.send.message.cost.avg", "avg of send
message cost."),
+ HttpSummaryMetrics::avgSendMsgCost);
+
+ //avgReply2MQCost
+ put(join("eventmesh.reply.message.cost.avg", "avg of reply
message cost."),
+ HttpSummaryMetrics::avgReplyMsgCost);
+ }
+ };
+ }
+
+ public 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();
+ paramPairs.forEach((k, v) -> observeOfValue(meter, k[0], k[1], HTTP,
v.apply(summaryMetrics)));
}
}
diff --git
a/eventmesh-metrics-plugin/eventmesh-metrics-prometheus/src/main/java/org/apache/eventmesh/metrics/prometheus/metrics/PrometheusTcpExporter.java
b/eventmesh-metrics-plugin/eventmesh-metrics-prometheus/src/main/java/org/apache/eventmesh/metrics/prometheus/metrics/PrometheusTcpExporter.java
index 01edb5549..da39828a9 100644
---
a/eventmesh-metrics-plugin/eventmesh-metrics-prometheus/src/main/java/org/apache/eventmesh/metrics/prometheus/metrics/PrometheusTcpExporter.java
+++
b/eventmesh-metrics-plugin/eventmesh-metrics-prometheus/src/main/java/org/apache/eventmesh/metrics/prometheus/metrics/PrometheusTcpExporter.java
@@ -17,73 +17,69 @@
package org.apache.eventmesh.metrics.prometheus.metrics;
+import static
org.apache.eventmesh.metrics.prometheus.utils.PrometheusExporterConstants.METRICS_TCP_PREFIX;
+import static
org.apache.eventmesh.metrics.prometheus.utils.PrometheusExporterConstants.TCP;
+import static
org.apache.eventmesh.metrics.prometheus.utils.PrometheusExporterUtils.join;
+import static
org.apache.eventmesh.metrics.prometheus.utils.PrometheusExporterUtils.observeOfValue;
+
import org.apache.eventmesh.metrics.api.model.TcpSummaryMetrics;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Function;
+
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 PrometheusTcpExporter {
- public static void export(final String meterName, final TcpSummaryMetrics
summaryMetrics) {
+ /**
+ * Map structure : [metric name, description of name] -> the method of get
corresponding metric.
+ */
+ private Map<String[], Function<TcpSummaryMetrics, Number>> paramPairs;
+
+ static {
+ paramPairs = new HashMap<String[], Function<TcpSummaryMetrics,
Number>>() {
+ {
+ //retryQueueSize
+ put(join("retry.queue.size", "get size of retry queue."),
+ TcpSummaryMetrics::getRetrySize);
+
+ //client2eventMeshTPS
+ put(join("server.tps", "get tps of client to eventMesh."),
+ TcpSummaryMetrics::getClient2eventMeshTPS);
+
+ //eventMesh2mqTPS
+ put(join("mq.provider.tps", "get tps of eventMesh to mq."),
+ TcpSummaryMetrics::getEventMesh2mqTPS);
+
+ //mq2eventMeshTPS
+ put(join("mq.consumer.tps", "get tps of mq to eventMesh."),
+ TcpSummaryMetrics::getMq2eventMeshTPS);
+
+ //eventMesh2clientTPS
+ put(join("client.tps", "get tps of eventMesh to client."),
+ TcpSummaryMetrics::getEventMesh2clientTPS);
+
+ //allTPS
+ put(join("all.tps", "get all TPS."),
TcpSummaryMetrics::getAllTPS);
+
+ //EventMeshTcpConnectionHandler.connections
+ put(join("connection.num",
"EventMeshTcpConnectionHandler.connections."),
+ TcpSummaryMetrics::getAllConnections);
+
+ //subTopicNum
+ put(join("sub.topic.num", "get sub topic num."),
TcpSummaryMetrics::getSubTopicNum);
+ }
+ };
+ }
+
+ public 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();
+ paramPairs.forEach((k, v) ->
+ observeOfValue(meter, METRICS_TCP_PREFIX + k[0], k[1], TCP,
v.apply(summaryMetrics)));
}
}
diff --git
a/eventmesh-metrics-plugin/eventmesh-metrics-prometheus/src/main/java/org/apache/eventmesh/metrics/prometheus/utils/PrometheusExporterConstants.java
b/eventmesh-metrics-plugin/eventmesh-metrics-prometheus/src/main/java/org/apache/eventmesh/metrics/prometheus/utils/PrometheusExporterConstants.java
new file mode 100644
index 000000000..d1b970569
--- /dev/null
+++
b/eventmesh-metrics-plugin/eventmesh-metrics-prometheus/src/main/java/org/apache/eventmesh/metrics/prometheus/utils/PrometheusExporterConstants.java
@@ -0,0 +1,35 @@
+/*
+ * 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.prometheus.utils;
+
+/**
+ * Constants in metrics-prometheus module
+ */
+public class PrometheusExporterConstants {
+
+ public static final String HTTP = "HTTP";
+
+ public static final String GRPC = "GRPC";
+
+ public static final String TCP = "TCP";
+
+ public static final String METRICS_GRPC_PREFIX = "eventmesh.grpc.";
+
+ public static final String METRICS_TCP_PREFIX = "eventmesh.tcp.";
+
+}
diff --git
a/eventmesh-metrics-plugin/eventmesh-metrics-prometheus/src/main/java/org/apache/eventmesh/metrics/prometheus/utils/PrometheusExporterUtils.java
b/eventmesh-metrics-plugin/eventmesh-metrics-prometheus/src/main/java/org/apache/eventmesh/metrics/prometheus/utils/PrometheusExporterUtils.java
new file mode 100644
index 000000000..f898df057
--- /dev/null
+++
b/eventmesh-metrics-plugin/eventmesh-metrics-prometheus/src/main/java/org/apache/eventmesh/metrics/prometheus/utils/PrometheusExporterUtils.java
@@ -0,0 +1,67 @@
+/*
+ * 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.prometheus.utils;
+
+import io.opentelemetry.api.metrics.Meter;
+import io.opentelemetry.api.metrics.common.Labels;
+
+/**
+ * Utils for metrics-prometheus module
+ */
+public class PrometheusExporterUtils {
+
+ /**
+ * Build the OpenTelemetry's Meter
+ *
+ * @param meter
+ * @param name
+ * @param desc
+ * @param protocol
+ * @param number
+ */
+ public static void observeOfValue(Meter meter, String name, String desc,
String protocol, Number number) {
+ if (number instanceof Long) {
+ meter.longValueObserverBuilder(name)
+ .setDescription(desc)
+ .setUnit(protocol)
+ .setUpdater(result -> result.observe((long) number,
Labels.empty()))
+ .build();
+ } else if (number instanceof Double) {
+ meter.doubleValueObserverBuilder(name)
+ .setDescription(desc)
+ .setUnit(protocol)
+ .setUpdater(result -> result.observe((double) number,
Labels.empty()))
+ .build();
+ }
+ }
+
+
+
+ /**
+ * create and init an array contains 2 String.
+ *
+ * @param metricName the metric name
+ * @param desc the description of metric
+ * @return
+ */
+ public static String[] join(String metricName, String desc) {
+ String[] array = {metricName, desc};
+ return array;
+ }
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
