You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by li...@apache.org on 2022/10/27 09:23:39 UTC
[rocketmq] branch develop updated: [ISSUE #5354] [RIP-46] Implement proxy metrics framework (#5407)
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new 4ba8aa238 [ISSUE #5354] [RIP-46] Implement proxy metrics framework (#5407)
4ba8aa238 is described below
commit 4ba8aa238d8cd7104cdc1119d0f35514fb47cfc2
Author: SSpirits <ad...@lv5.moe>
AuthorDate: Thu Oct 27 17:23:31 2022 +0800
[ISSUE #5354] [RIP-46] Implement proxy metrics framework (#5407)
* implement proxy metrics framework
* fix bazel dependency
* fix proxy startup test
* fix warning in bazel test
* minor improve
---
.../apache/rocketmq/broker/BrokerController.java | 4 +
.../broker/metrics/BrokerMetricsConstant.java | 6 +-
.../broker/metrics/BrokerMetricsManager.java | 69 ++++-----
.../org/apache/rocketmq/common/BrokerConfig.java | 76 +++-------
example/pom.xml | 4 +
pom.xml | 4 +
proxy/BUILD.bazel | 10 ++
.../org/apache/rocketmq/proxy/ProxyStartup.java | 6 +-
.../apache/rocketmq/proxy/config/ProxyConfig.java | 128 ++++++++++++++++
.../proxy/metrics/ProxyMetricsConstant.java | 24 +++
.../proxy/metrics/ProxyMetricsManager.java | 164 ++++++++++++---------
.../apache/rocketmq/proxy/ProxyStartupTest.java | 14 +-
remoting/pom.xml | 4 +
13 files changed, 343 insertions(+), 170 deletions(-)
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
index 6a450b7aa..d5041c2a7 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
@@ -418,6 +418,10 @@ public class BrokerController {
return queryThreadPoolQueue;
}
+ public BrokerMetricsManager getBrokerMetricsManager() {
+ return brokerMetricsManager;
+ }
+
protected void initializeRemotingServer() throws CloneNotSupportedException {
this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.clientHousekeepingService);
NettyServerConfig fastConfig = (NettyServerConfig) this.nettyServerConfig.clone();
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/metrics/BrokerMetricsConstant.java b/broker/src/main/java/org/apache/rocketmq/broker/metrics/BrokerMetricsConstant.java
index 14ace640b..48c73b1b8 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/metrics/BrokerMetricsConstant.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/metrics/BrokerMetricsConstant.java
@@ -17,16 +17,12 @@
package org.apache.rocketmq.broker.metrics;
public class BrokerMetricsConstant {
- public static final String SLS_OTEL_PROJECT_HEADER_KEY = "x-sls-otel-project";
- public static final String SLS_OTEL_INSTANCE_ID_KEY = "x-sls-otel-instance-id";
- public static final String SLS_OTEL_AK_ID_KEY = "x-sls-otel-ak-id";
- public static final String SLS_OTEL_AK_SECRET_KEY = "x-sls-otel-ak-secret";
public static final String OPEN_TELEMETRY_METER_NAME = "broker-meter";
public static final String GAUGE_BROKER_PERMISSION = "rocketmq_broker_permission";
public static final String LABEL_CLUSTER_NAME = "cluster";
public static final String LABEL_NODE_TYPE = "node_type";
- public static final String BROKER_NODE_TYPE = "broker";
+ public static final String NODE_TYPE_BROKER = "broker";
public static final String LABEL_NODE_ID = "node_id";
public static final String LABEL_AGGREGATION = "aggregation";
public static final String AGGREGATION_DELTA = "delta";
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/metrics/BrokerMetricsManager.java b/broker/src/main/java/org/apache/rocketmq/broker/metrics/BrokerMetricsManager.java
index 2e2ff94b2..e6fab145f 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/metrics/BrokerMetricsManager.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/metrics/BrokerMetricsManager.java
@@ -44,17 +44,13 @@ import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.store.MessageStore;
import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.AGGREGATION_DELTA;
-import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.BROKER_NODE_TYPE;
+import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.NODE_TYPE_BROKER;
import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.GAUGE_BROKER_PERMISSION;
import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_AGGREGATION;
import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_CLUSTER_NAME;
import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_NODE_ID;
import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_NODE_TYPE;
import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.OPEN_TELEMETRY_METER_NAME;
-import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.SLS_OTEL_AK_ID_KEY;
-import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.SLS_OTEL_AK_SECRET_KEY;
-import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.SLS_OTEL_INSTANCE_ID_KEY;
-import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.SLS_OTEL_PROJECT_HEADER_KEY;
public class BrokerMetricsManager {
private static final InternalLogger LOGGER = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
@@ -62,10 +58,11 @@ public class BrokerMetricsManager {
private final BrokerConfig brokerConfig;
private final MessageStore messageStore;
private final BrokerController brokerController;
- public final static Map<String, String> LABEL_MAP = new HashMap<>();
+ private final static Map<String, String> LABEL_MAP = new HashMap<>();
private OtlpGrpcMetricExporter metricExporter;
private PeriodicMetricReader periodicMetricReader;
private PrometheusHttpServer prometheusHttpServer;
+ private Meter brokerMeter;
public static ObservableLongGauge brokerPermission = null;
@@ -82,6 +79,10 @@ public class BrokerMetricsManager {
return attributesBuilder;
}
+ public Meter getBrokerMeter() {
+ return brokerMeter;
+ }
+
private boolean checkConfig() {
if (brokerConfig == null) {
return false;
@@ -93,13 +94,7 @@ public class BrokerMetricsManager {
switch (exporterType) {
case OTLP_GRPC:
- return StringUtils.isNotBlank(brokerConfig.getMetricsGrpcCollectorEndpoint());
- case OTLP_GRPC_SLS:
- return StringUtils.isNotBlank(brokerConfig.getMetricsGrpcCollectorEndpoint()) &&
- StringUtils.isNotBlank(brokerConfig.getMetricsSlsProjectName()) &&
- StringUtils.isNotBlank(brokerConfig.getMetricsSlsInstanceName()) &&
- StringUtils.isNotBlank(brokerConfig.getMetricsSlsAccessKey()) &&
- StringUtils.isNotBlank(brokerConfig.getMetricsSlsSecretKey());
+ return StringUtils.isNotBlank(brokerConfig.getMetricsGrpcExporterTarget());
case PROM:
return true;
}
@@ -108,54 +103,61 @@ public class BrokerMetricsManager {
private void init() {
if (!checkConfig()) {
- LOGGER.error("check broker metrics config failed, will not export metrics");
+ LOGGER.error("check metrics config failed, will not export metrics");
return;
}
- String labels = brokerConfig.getBrokerMetricsLabel();
+ String labels = brokerConfig.getMetricsLabel();
if (StringUtils.isNotBlank(labels)) {
List<String> kvPairs = Splitter.on(',').omitEmptyStrings().splitToList(labels);
for (String item : kvPairs) {
String[] split = item.split(":");
if (split.length != 2) {
- LOGGER.warn("brokerMetricsLabel is not valid: {}", brokerConfig.getBrokerMetricsLabel());
+ LOGGER.warn("metricsLabel is not valid: {}", labels);
continue;
}
LABEL_MAP.put(split[0], split[1]);
}
}
- if (brokerConfig.isBrokerMetricsPreferDelta()) {
+ if (brokerConfig.isMetricsInDelta()) {
LABEL_MAP.put(LABEL_AGGREGATION, AGGREGATION_DELTA);
}
- LABEL_MAP.put(LABEL_NODE_TYPE, BROKER_NODE_TYPE);
+ LABEL_MAP.put(LABEL_NODE_TYPE, NODE_TYPE_BROKER);
LABEL_MAP.put(LABEL_CLUSTER_NAME, brokerConfig.getBrokerClusterName());
LABEL_MAP.put(LABEL_NODE_ID, brokerConfig.getBrokerName());
SdkMeterProviderBuilder providerBuilder = SdkMeterProvider.builder()
.setResource(Resource.empty());
- if (brokerConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.OTLP_GRPC ||
- brokerConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.OTLP_GRPC_SLS) {
- String endpoint = brokerConfig.getMetricsGrpcCollectorEndpoint();
- if (!endpoint.startsWith("https://")) {
+ if (brokerConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.OTLP_GRPC) {
+ String endpoint = brokerConfig.getMetricsGrpcExporterTarget();
+ if (!endpoint.startsWith("http")) {
endpoint = "https://" + endpoint;
}
OtlpGrpcMetricExporterBuilder metricExporterBuilder = OtlpGrpcMetricExporter.builder()
.setEndpoint(endpoint)
.setTimeout(brokerConfig.getMetricGrpcExporterTimeOutInMills(), TimeUnit.MILLISECONDS)
.setAggregationTemporalitySelector(type -> {
- if (brokerConfig.isBrokerMetricsPreferDelta() &&
+ if (brokerConfig.isMetricsInDelta() &&
(type == InstrumentType.COUNTER || type == InstrumentType.OBSERVABLE_COUNTER || type == InstrumentType.HISTOGRAM)) {
return AggregationTemporality.DELTA;
}
return AggregationTemporality.CUMULATIVE;
});
- if (brokerConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.OTLP_GRPC_SLS) {
- metricExporterBuilder.addHeader(SLS_OTEL_PROJECT_HEADER_KEY, brokerConfig.getMetricsSlsProjectName())
- .addHeader(SLS_OTEL_INSTANCE_ID_KEY, brokerConfig.getMetricsSlsInstanceName())
- .addHeader(SLS_OTEL_AK_ID_KEY, brokerConfig.getMetricsSlsAccessKey())
- .addHeader(SLS_OTEL_AK_SECRET_KEY, brokerConfig.getMetricsSlsSecretKey());
+ String headers = brokerConfig.getMetricsGrpcExporterHeader();
+ if (StringUtils.isNotBlank(headers)) {
+ Map<String, String> headerMap = new HashMap<>();
+ List<String> kvPairs = Splitter.on(',').omitEmptyStrings().splitToList(headers);
+ for (String item : kvPairs) {
+ String[] split = item.split(":");
+ if (split.length != 2) {
+ LOGGER.warn("metricsGrpcExporterHeader is not valid: {}", headers);
+ continue;
+ }
+ headerMap.put(split[0], split[1]);
+ }
+ headerMap.forEach(metricExporterBuilder::addHeader);
}
metricExporter = metricExporterBuilder.build();
@@ -179,24 +181,23 @@ public class BrokerMetricsManager {
providerBuilder.registerMetricReader(prometheusHttpServer);
}
- Meter brokerMeter = OpenTelemetrySdk.builder()
+ brokerMeter = OpenTelemetrySdk.builder()
.setMeterProvider(providerBuilder.build())
.build()
.getMeter(OPEN_TELEMETRY_METER_NAME);
- initStatsMetrics(brokerMeter);
+ initStatsMetrics();
}
- private void initStatsMetrics(Meter meter) {
- brokerPermission = meter.gaugeBuilder(GAUGE_BROKER_PERMISSION)
+ private void initStatsMetrics() {
+ brokerPermission = brokerMeter.gaugeBuilder(GAUGE_BROKER_PERMISSION)
.setDescription("Broker permission")
.ofLongs()
.buildWithCallback(measurement -> measurement.record(brokerConfig.getBrokerPermission(), newAttributesBuilder().build()));
}
public void shutdown() {
- if (brokerConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.OTLP_GRPC ||
- brokerConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.OTLP_GRPC_SLS) {
+ if (brokerConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.OTLP_GRPC) {
periodicMetricReader.forceFlush();
periodicMetricReader.shutdown();
metricExporter.shutdown();
diff --git a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
index 1a61f11c0..dbb320afc 100644
--- a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
+++ b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
@@ -338,8 +338,7 @@ public class BrokerConfig extends BrokerIdentity {
public enum MetricsExporterType {
DISABLE(0),
OTLP_GRPC(1),
- OTLP_GRPC_SLS(2),
- PROM(3);
+ PROM(2);
private final int value;
@@ -356,8 +355,6 @@ public class BrokerConfig extends BrokerIdentity {
case 1:
return OTLP_GRPC;
case 2:
- return OTLP_GRPC_SLS;
- case 3:
return PROM;
default:
return DISABLE;
@@ -371,23 +368,18 @@ public class BrokerConfig extends BrokerIdentity {
private MetricsExporterType metricsExporterType = MetricsExporterType.DISABLE;
- private String metricsGrpcCollectorEndpoint = "";
-
- private String metricsSlsProjectName = "";
- private String metricsSlsInstanceName = "";
- private String metricsSlsAccessKey = "";
- private String metricsSlsSecretKey = "";
-
+ private String metricsGrpcExporterTarget = "";
+ private String metricsGrpcExporterHeader = "";
private long metricGrpcExporterTimeOutInMills = 3 * 1000;
private long metricGrpcExporterIntervalInMills = 60 * 1000;
- private int metricsPromExporterPort = 8080;
+ private int metricsPromExporterPort = 5557;
private String metricsPromExporterHost = "";
// Label pairs in CSV. Each label follows pattern of Key:Value. eg: instance_id:xxx,uid:xxx
- private String brokerMetricsLabel = "";
+ private String metricsLabel = "";
- private boolean brokerMetricsPreferDelta = true;
+ private boolean metricsInDelta = false;
public long getMaxPopPollingSize() {
return maxPopPollingSize;
@@ -1493,44 +1485,20 @@ public class BrokerConfig extends BrokerIdentity {
this.metricsExporterType = MetricsExporterType.valueOf(metricsExporterType);
}
- public String getMetricsGrpcCollectorEndpoint() {
- return metricsGrpcCollectorEndpoint;
- }
-
- public void setMetricsGrpcCollectorEndpoint(String metricsGrpcCollectorEndpoint) {
- this.metricsGrpcCollectorEndpoint = metricsGrpcCollectorEndpoint;
- }
-
- public String getMetricsSlsProjectName() {
- return metricsSlsProjectName;
- }
-
- public void setMetricsSlsProjectName(String metricsSlsProjectName) {
- this.metricsSlsProjectName = metricsSlsProjectName;
- }
-
- public String getMetricsSlsInstanceName() {
- return metricsSlsInstanceName;
- }
-
- public void setMetricsSlsInstanceName(String metricsSlsInstanceName) {
- this.metricsSlsInstanceName = metricsSlsInstanceName;
- }
-
- public String getMetricsSlsAccessKey() {
- return metricsSlsAccessKey;
+ public String getMetricsGrpcExporterTarget() {
+ return metricsGrpcExporterTarget;
}
- public void setMetricsSlsAccessKey(String metricsSlsAccessKey) {
- this.metricsSlsAccessKey = metricsSlsAccessKey;
+ public void setMetricsGrpcExporterTarget(String metricsGrpcExporterTarget) {
+ this.metricsGrpcExporterTarget = metricsGrpcExporterTarget;
}
- public String getMetricsSlsSecretKey() {
- return metricsSlsSecretKey;
+ public String getMetricsGrpcExporterHeader() {
+ return metricsGrpcExporterHeader;
}
- public void setMetricsSlsSecretKey(String metricsSlsSecretKey) {
- this.metricsSlsSecretKey = metricsSlsSecretKey;
+ public void setMetricsGrpcExporterHeader(String metricsGrpcExporterHeader) {
+ this.metricsGrpcExporterHeader = metricsGrpcExporterHeader;
}
public long getMetricGrpcExporterTimeOutInMills() {
@@ -1549,20 +1517,20 @@ public class BrokerConfig extends BrokerIdentity {
this.metricGrpcExporterIntervalInMills = metricGrpcExporterIntervalInMills;
}
- public String getBrokerMetricsLabel() {
- return brokerMetricsLabel;
+ public String getMetricsLabel() {
+ return metricsLabel;
}
- public void setBrokerMetricsLabel(String brokerMetricsLabel) {
- this.brokerMetricsLabel = brokerMetricsLabel;
+ public void setMetricsLabel(String metricsLabel) {
+ this.metricsLabel = metricsLabel;
}
- public boolean isBrokerMetricsPreferDelta() {
- return brokerMetricsPreferDelta;
+ public boolean isMetricsInDelta() {
+ return metricsInDelta;
}
- public void setBrokerMetricsPreferDelta(boolean brokerMetricsPreferDelta) {
- this.brokerMetricsPreferDelta = brokerMetricsPreferDelta;
+ public void setMetricsInDelta(boolean metricsInDelta) {
+ this.metricsInDelta = metricsInDelta;
}
public int getMetricsPromExporterPort() {
diff --git a/example/pom.xml b/example/pom.xml
index 42134709b..22a01c6c8 100644
--- a/example/pom.xml
+++ b/example/pom.xml
@@ -68,6 +68,10 @@
<groupId>io.jaegertracing</groupId>
<artifactId>jaeger-client</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.jaegertracing</groupId>
+ <artifactId>jaeger-thrift</artifactId>
+ </dependency>
<dependency>
<groupId>commons-cli</groupId>
<artifactId>commons-cli</artifactId>
diff --git a/pom.xml b/pom.xml
index b2176e6ce..615a53028 100644
--- a/pom.xml
+++ b/pom.xml
@@ -907,6 +907,10 @@
<groupId>org.jetbrains.kotlin</groupId>
<artifactId>kotlin-stdlib</artifactId>
</exclusion>
+ <exclusion>
+ <groupId>org.jetbrains.kotlin</groupId>
+ <artifactId>kotlin-stdlib-common</artifactId>
+ </exclusion>
</exclusions>
</dependency>
<dependency>
diff --git a/proxy/BUILD.bazel b/proxy/BUILD.bazel
index fa67fc018..929b8c01d 100644
--- a/proxy/BUILD.bazel
+++ b/proxy/BUILD.bazel
@@ -47,6 +47,13 @@ java_library(
"@maven//:io_grpc_grpc_services",
"@maven//:io_grpc_grpc_stub",
"@maven//:io_netty_netty_all",
+ "@maven//:io_openmessaging_storage_dledger",
+ "@maven//:io_opentelemetry_opentelemetry_api",
+ "@maven//:io_opentelemetry_opentelemetry_exporter_otlp",
+ "@maven//:io_opentelemetry_opentelemetry_exporter_prometheus",
+ "@maven//:io_opentelemetry_opentelemetry_sdk",
+ "@maven//:io_opentelemetry_opentelemetry_sdk_common",
+ "@maven//:io_opentelemetry_opentelemetry_sdk_metrics",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:org_apache_rocketmq_rocketmq_proto",
"@maven//:org_checkerframework_checker_qual",
@@ -84,6 +91,9 @@ java_library(
"@maven//:io_grpc_grpc_stub",
"@maven//:io_netty_netty_all",
"@maven//:org_apache_commons_commons_lang3",
+ "@maven//:io_opentelemetry_opentelemetry_exporter_otlp",
+ "@maven//:io_opentelemetry_opentelemetry_exporter_prometheus",
+ "@maven//:io_opentelemetry_opentelemetry_sdk",
"@maven//:org_apache_rocketmq_rocketmq_proto",
"@maven//:org_checkerframework_checker_qual",
"@maven//:org_slf4j_slf4j_api",
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java b/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java
index c051854a8..c5222a004 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java
@@ -48,6 +48,7 @@ import org.apache.rocketmq.proxy.config.ProxyConfig;
import org.apache.rocketmq.proxy.grpc.GrpcServer;
import org.apache.rocketmq.proxy.grpc.GrpcServerBuilder;
import org.apache.rocketmq.proxy.grpc.v2.GrpcMessagingApplication;
+import org.apache.rocketmq.proxy.metrics.ProxyMetricsManager;
import org.apache.rocketmq.proxy.processor.DefaultMessagingProcessor;
import org.apache.rocketmq.proxy.processor.MessagingProcessor;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
@@ -168,8 +169,11 @@ public class ProxyStartup {
if (ProxyMode.isClusterMode(proxyModeStr)) {
messagingProcessor = DefaultMessagingProcessor.createForClusterMode();
+ ProxyMetricsManager proxyMetricsManager = ProxyMetricsManager.initClusterMode(ConfigurationManager.getProxyConfig());
+ PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(proxyMetricsManager);
} else if (ProxyMode.isLocalMode(proxyModeStr)) {
BrokerController brokerController = createBrokerController();
+ ProxyMetricsManager.initLocalMode(brokerController.getBrokerMetricsManager(), ConfigurationManager.getProxyConfig());
StartAndShutdown brokerControllerWrapper = new StartAndShutdown() {
@Override
public void start() throws Exception {
@@ -256,4 +260,4 @@ public class ProxyStartup {
}
configurator.doConfigure(home + "/conf/logback_proxy.xml");
}
-}
\ No newline at end of file
+}
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java b/proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java
index 00a6cc35e..383f819fc 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java
@@ -17,6 +17,8 @@
package org.apache.rocketmq.proxy.config;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
import java.time.Duration;
import java.util.Comparator;
import java.util.HashMap;
@@ -25,6 +27,8 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.proxy.ProxyMode;
@@ -35,8 +39,21 @@ public class ProxyConfig implements ConfigFile {
private final static Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
public final static String DEFAULT_CONFIG_FILE_NAME = "rmq-proxy.json";
private static final int PROCESSOR_NUMBER = Runtime.getRuntime().availableProcessors();
+ private static final String DEFAULT_CLUSTER_NAME = "DefaultCluster";
+
+ private static String localHostName;
+
+ static {
+ try {
+ localHostName = InetAddress.getLocalHost().getHostName();
+ } catch (UnknownHostException e) {
+ log.error("Failed to obtain the host name", e);
+ }
+ }
private String rocketMQClusterName = "";
+ private String proxyClusterName = DEFAULT_CLUSTER_NAME;
+ private String proxyName = StringUtils.isEmpty(localHostName) ? "DEFAULT_PROXY" : localHostName;
/**
* configuration for ThreadPoolMonitor
@@ -161,6 +178,21 @@ public class ProxyConfig implements ConfigFile {
// Example address: 127.0.0.1:1234
private String metricCollectorAddress = "";
+ private BrokerConfig.MetricsExporterType metricsExporterType = BrokerConfig.MetricsExporterType.DISABLE;
+
+ private String metricsGrpcExporterTarget = "";
+ private String metricsGrpcExporterHeader = "";
+ private long metricGrpcExporterTimeOutInMills = 3 * 1000;
+ private long metricGrpcExporterIntervalInMills = 60 * 1000;
+
+ private int metricsPromExporterPort = 5557;
+ private String metricsPromExporterHost = "";
+
+ // Label pairs in CSV. Each label follows pattern of Key:Value. eg: instance_id:xxx,uid:xxx
+ private String metricsLabel = "";
+
+ private boolean metricsInDelta = false;
+
@Override
public void initData() {
parseDelayLevel();
@@ -211,6 +243,22 @@ public class ProxyConfig implements ConfigFile {
this.rocketMQClusterName = rocketMQClusterName;
}
+ public String getProxyClusterName() {
+ return proxyClusterName;
+ }
+
+ public void setProxyClusterName(String proxyClusterName) {
+ this.proxyClusterName = proxyClusterName;
+ }
+
+ public String getProxyName() {
+ return proxyName;
+ }
+
+ public void setProxyName(String proxyName) {
+ this.proxyName = proxyName;
+ }
+
public boolean isEnablePrintJstack() {
return enablePrintJstack;
}
@@ -910,4 +958,84 @@ public class ProxyConfig implements ConfigFile {
public void setGrpcClientIdleTimeMills(final long grpcClientIdleTimeMills) {
this.grpcClientIdleTimeMills = grpcClientIdleTimeMills;
}
+
+ public BrokerConfig.MetricsExporterType getMetricsExporterType() {
+ return metricsExporterType;
+ }
+
+ public void setMetricsExporterType(BrokerConfig.MetricsExporterType metricsExporterType) {
+ this.metricsExporterType = metricsExporterType;
+ }
+
+ public void setMetricsExporterType(int metricsExporterType) {
+ this.metricsExporterType = BrokerConfig.MetricsExporterType.valueOf(metricsExporterType);
+ }
+
+ public void setMetricsExporterType(String metricsExporterType) {
+ this.metricsExporterType = BrokerConfig.MetricsExporterType.valueOf(metricsExporterType);
+ }
+
+ public String getMetricsGrpcExporterTarget() {
+ return metricsGrpcExporterTarget;
+ }
+
+ public void setMetricsGrpcExporterTarget(String metricsGrpcExporterTarget) {
+ this.metricsGrpcExporterTarget = metricsGrpcExporterTarget;
+ }
+
+ public String getMetricsGrpcExporterHeader() {
+ return metricsGrpcExporterHeader;
+ }
+
+ public void setMetricsGrpcExporterHeader(String metricsGrpcExporterHeader) {
+ this.metricsGrpcExporterHeader = metricsGrpcExporterHeader;
+ }
+
+ public long getMetricGrpcExporterTimeOutInMills() {
+ return metricGrpcExporterTimeOutInMills;
+ }
+
+ public void setMetricGrpcExporterTimeOutInMills(long metricGrpcExporterTimeOutInMills) {
+ this.metricGrpcExporterTimeOutInMills = metricGrpcExporterTimeOutInMills;
+ }
+
+ public long getMetricGrpcExporterIntervalInMills() {
+ return metricGrpcExporterIntervalInMills;
+ }
+
+ public void setMetricGrpcExporterIntervalInMills(long metricGrpcExporterIntervalInMills) {
+ this.metricGrpcExporterIntervalInMills = metricGrpcExporterIntervalInMills;
+ }
+
+ public int getMetricsPromExporterPort() {
+ return metricsPromExporterPort;
+ }
+
+ public void setMetricsPromExporterPort(int metricsPromExporterPort) {
+ this.metricsPromExporterPort = metricsPromExporterPort;
+ }
+
+ public String getMetricsPromExporterHost() {
+ return metricsPromExporterHost;
+ }
+
+ public void setMetricsPromExporterHost(String metricsPromExporterHost) {
+ this.metricsPromExporterHost = metricsPromExporterHost;
+ }
+
+ public String getMetricsLabel() {
+ return metricsLabel;
+ }
+
+ public void setMetricsLabel(String metricsLabel) {
+ this.metricsLabel = metricsLabel;
+ }
+
+ public boolean isMetricsInDelta() {
+ return metricsInDelta;
+ }
+
+ public void setMetricsInDelta(boolean metricsInDelta) {
+ this.metricsInDelta = metricsInDelta;
+ }
}
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsConstant.java b/proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsConstant.java
new file mode 100644
index 000000000..a7682e6cf
--- /dev/null
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsConstant.java
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.proxy.metrics;
+
+public class ProxyMetricsConstant {
+ public static final String GAUGE_PROXY_UP = "rocketmq_proxy_up";
+
+ public static final String LABEL_PROXY_MODE = "proxy_mode";
+ public static final String NODE_TYPE_PROXY = "proxy";
+}
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/metrics/BrokerMetricsManager.java b/proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsManager.java
similarity index 51%
copy from broker/src/main/java/org/apache/rocketmq/broker/metrics/BrokerMetricsManager.java
copy to proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsManager.java
index 2e2ff94b2..3d27220a2 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/metrics/BrokerMetricsManager.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsManager.java
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.rocketmq.broker.metrics;
+package org.apache.rocketmq.proxy.metrics;
import com.google.common.base.Splitter;
import io.opentelemetry.api.common.Attributes;
@@ -35,173 +35,193 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
import org.apache.commons.lang3.StringUtils;
-import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.broker.metrics.BrokerMetricsManager;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.constant.LoggerName;
-import org.apache.rocketmq.logging.InternalLogger;
-import org.apache.rocketmq.logging.InternalLoggerFactory;
-import org.apache.rocketmq.store.MessageStore;
+import org.apache.rocketmq.proxy.common.StartAndShutdown;
+import org.apache.rocketmq.proxy.config.ProxyConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.AGGREGATION_DELTA;
-import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.BROKER_NODE_TYPE;
-import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.GAUGE_BROKER_PERMISSION;
import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_AGGREGATION;
import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_CLUSTER_NAME;
import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_NODE_ID;
import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.LABEL_NODE_TYPE;
import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.OPEN_TELEMETRY_METER_NAME;
-import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.SLS_OTEL_AK_ID_KEY;
-import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.SLS_OTEL_AK_SECRET_KEY;
-import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.SLS_OTEL_INSTANCE_ID_KEY;
-import static org.apache.rocketmq.broker.metrics.BrokerMetricsConstant.SLS_OTEL_PROJECT_HEADER_KEY;
-
-public class BrokerMetricsManager {
- private static final InternalLogger LOGGER = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
-
- private final BrokerConfig brokerConfig;
- private final MessageStore messageStore;
- private final BrokerController brokerController;
- public final static Map<String, String> LABEL_MAP = new HashMap<>();
+import static org.apache.rocketmq.proxy.metrics.ProxyMetricsConstant.GAUGE_PROXY_UP;
+import static org.apache.rocketmq.proxy.metrics.ProxyMetricsConstant.LABEL_PROXY_MODE;
+import static org.apache.rocketmq.proxy.metrics.ProxyMetricsConstant.NODE_TYPE_PROXY;
+
+public class ProxyMetricsManager implements StartAndShutdown {
+ private final static Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
+
+ private static ProxyConfig proxyConfig;
+ private final static Map<String, String> LABEL_MAP = new HashMap<>();
+ public static Supplier<AttributesBuilder> attributesBuilderSupplier;
+
private OtlpGrpcMetricExporter metricExporter;
private PeriodicMetricReader periodicMetricReader;
private PrometheusHttpServer prometheusHttpServer;
- public static ObservableLongGauge brokerPermission = null;
+ public static ObservableLongGauge proxyUp = null;
+
+ public static void initLocalMode(BrokerMetricsManager brokerMetricsManager, ProxyConfig proxyConfig) {
+ ProxyMetricsManager.proxyConfig = proxyConfig;
+ LABEL_MAP.put(LABEL_NODE_TYPE, NODE_TYPE_PROXY);
+ LABEL_MAP.put(LABEL_CLUSTER_NAME, proxyConfig.getProxyClusterName());
+ LABEL_MAP.put(LABEL_NODE_ID, proxyConfig.getProxyName());
+ LABEL_MAP.put(LABEL_PROXY_MODE, proxyConfig.getProxyMode().toLowerCase());
+ initMetrics(brokerMetricsManager.getBrokerMeter(), BrokerMetricsManager::newAttributesBuilder);
+ }
- public BrokerMetricsManager(BrokerController brokerController) {
- this.brokerController = brokerController;
- brokerConfig = brokerController.getBrokerConfig();
- this.messageStore = brokerController.getMessageStore();
- init();
+ public static ProxyMetricsManager initClusterMode(ProxyConfig proxyConfig) {
+ ProxyMetricsManager.proxyConfig = proxyConfig;
+ return new ProxyMetricsManager();
}
public static AttributesBuilder newAttributesBuilder() {
- AttributesBuilder attributesBuilder = Attributes.builder();
+ AttributesBuilder attributesBuilder;
+ if (attributesBuilderSupplier == null) {
+ attributesBuilder = Attributes.builder();
+ LABEL_MAP.forEach(attributesBuilder::put);
+ return attributesBuilder;
+ }
+ attributesBuilder = attributesBuilderSupplier.get();
LABEL_MAP.forEach(attributesBuilder::put);
return attributesBuilder;
}
+ private static void initMetrics(Meter meter, Supplier<AttributesBuilder> attributesBuilderSupplier) {
+ ProxyMetricsManager.attributesBuilderSupplier = attributesBuilderSupplier;
+
+ proxyUp = meter.gaugeBuilder(GAUGE_PROXY_UP)
+ .setDescription("proxy status")
+ .ofLongs()
+ .buildWithCallback(measurement -> measurement.record(1, newAttributesBuilder().build()));
+ }
+
+ public ProxyMetricsManager() {
+ }
+
private boolean checkConfig() {
- if (brokerConfig == null) {
+ if (proxyConfig == null) {
return false;
}
- BrokerConfig.MetricsExporterType exporterType = brokerConfig.getMetricsExporterType();
+ BrokerConfig.MetricsExporterType exporterType = proxyConfig.getMetricsExporterType();
if (!exporterType.isEnable()) {
return false;
}
switch (exporterType) {
case OTLP_GRPC:
- return StringUtils.isNotBlank(brokerConfig.getMetricsGrpcCollectorEndpoint());
- case OTLP_GRPC_SLS:
- return StringUtils.isNotBlank(brokerConfig.getMetricsGrpcCollectorEndpoint()) &&
- StringUtils.isNotBlank(brokerConfig.getMetricsSlsProjectName()) &&
- StringUtils.isNotBlank(brokerConfig.getMetricsSlsInstanceName()) &&
- StringUtils.isNotBlank(brokerConfig.getMetricsSlsAccessKey()) &&
- StringUtils.isNotBlank(brokerConfig.getMetricsSlsSecretKey());
+ return StringUtils.isNotBlank(proxyConfig.getMetricsGrpcExporterTarget());
case PROM:
return true;
}
return false;
}
- private void init() {
+ @Override
+ public void start() throws Exception {
if (!checkConfig()) {
- LOGGER.error("check broker metrics config failed, will not export metrics");
+ log.error("check metrics config failed, will not export metrics");
return;
}
- String labels = brokerConfig.getBrokerMetricsLabel();
+ String labels = proxyConfig.getMetricsLabel();
if (StringUtils.isNotBlank(labels)) {
List<String> kvPairs = Splitter.on(',').omitEmptyStrings().splitToList(labels);
for (String item : kvPairs) {
String[] split = item.split(":");
if (split.length != 2) {
- LOGGER.warn("brokerMetricsLabel is not valid: {}", brokerConfig.getBrokerMetricsLabel());
+ log.warn("metricsLabel is not valid: {}", labels);
continue;
}
LABEL_MAP.put(split[0], split[1]);
}
}
- if (brokerConfig.isBrokerMetricsPreferDelta()) {
+ if (proxyConfig.isMetricsInDelta()) {
LABEL_MAP.put(LABEL_AGGREGATION, AGGREGATION_DELTA);
}
- LABEL_MAP.put(LABEL_NODE_TYPE, BROKER_NODE_TYPE);
- LABEL_MAP.put(LABEL_CLUSTER_NAME, brokerConfig.getBrokerClusterName());
- LABEL_MAP.put(LABEL_NODE_ID, brokerConfig.getBrokerName());
+ LABEL_MAP.put(LABEL_NODE_TYPE, NODE_TYPE_PROXY);
+ LABEL_MAP.put(LABEL_CLUSTER_NAME, proxyConfig.getProxyClusterName());
+ LABEL_MAP.put(LABEL_NODE_ID, proxyConfig.getProxyName());
+ LABEL_MAP.put(LABEL_PROXY_MODE, proxyConfig.getProxyMode().toLowerCase());
SdkMeterProviderBuilder providerBuilder = SdkMeterProvider.builder()
.setResource(Resource.empty());
- if (brokerConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.OTLP_GRPC ||
- brokerConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.OTLP_GRPC_SLS) {
- String endpoint = brokerConfig.getMetricsGrpcCollectorEndpoint();
- if (!endpoint.startsWith("https://")) {
+ if (proxyConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.OTLP_GRPC) {
+ String endpoint = proxyConfig.getMetricsGrpcExporterTarget();
+ if (!endpoint.startsWith("http")) {
endpoint = "https://" + endpoint;
}
OtlpGrpcMetricExporterBuilder metricExporterBuilder = OtlpGrpcMetricExporter.builder()
.setEndpoint(endpoint)
- .setTimeout(brokerConfig.getMetricGrpcExporterTimeOutInMills(), TimeUnit.MILLISECONDS)
+ .setTimeout(proxyConfig.getMetricGrpcExporterTimeOutInMills(), TimeUnit.MILLISECONDS)
.setAggregationTemporalitySelector(type -> {
- if (brokerConfig.isBrokerMetricsPreferDelta() &&
+ if (proxyConfig.isMetricsInDelta() &&
(type == InstrumentType.COUNTER || type == InstrumentType.OBSERVABLE_COUNTER || type == InstrumentType.HISTOGRAM)) {
return AggregationTemporality.DELTA;
}
return AggregationTemporality.CUMULATIVE;
});
- if (brokerConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.OTLP_GRPC_SLS) {
- metricExporterBuilder.addHeader(SLS_OTEL_PROJECT_HEADER_KEY, brokerConfig.getMetricsSlsProjectName())
- .addHeader(SLS_OTEL_INSTANCE_ID_KEY, brokerConfig.getMetricsSlsInstanceName())
- .addHeader(SLS_OTEL_AK_ID_KEY, brokerConfig.getMetricsSlsAccessKey())
- .addHeader(SLS_OTEL_AK_SECRET_KEY, brokerConfig.getMetricsSlsSecretKey());
+ String headers = proxyConfig.getMetricsGrpcExporterHeader();
+ if (StringUtils.isNotBlank(headers)) {
+ Map<String, String> headerMap = new HashMap<>();
+ List<String> kvPairs = Splitter.on(',').omitEmptyStrings().splitToList(headers);
+ for (String item : kvPairs) {
+ String[] split = item.split(":");
+ if (split.length != 2) {
+ log.warn("metricsGrpcExporterHeader is not valid: {}", headers);
+ continue;
+ }
+ headerMap.put(split[0], split[1]);
+ }
+ headerMap.forEach(metricExporterBuilder::addHeader);
}
metricExporter = metricExporterBuilder.build();
periodicMetricReader = PeriodicMetricReader.builder(metricExporter)
- .setInterval(brokerConfig.getMetricGrpcExporterIntervalInMills(), TimeUnit.MILLISECONDS)
+ .setInterval(proxyConfig.getMetricGrpcExporterIntervalInMills(), TimeUnit.MILLISECONDS)
.build();
providerBuilder.registerMetricReader(periodicMetricReader);
}
- if (brokerConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.PROM) {
- String promExporterHost = brokerConfig.getMetricsPromExporterHost();
+ if (proxyConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.PROM) {
+ String promExporterHost = proxyConfig.getMetricsPromExporterHost();
if (StringUtils.isBlank(promExporterHost)) {
- promExporterHost = brokerConfig.getBrokerIP1();
+ promExporterHost = "0.0.0.0";
}
prometheusHttpServer = PrometheusHttpServer.builder()
.setHost(promExporterHost)
- .setPort(brokerConfig.getMetricsPromExporterPort())
+ .setPort(proxyConfig.getMetricsPromExporterPort())
.build();
providerBuilder.registerMetricReader(prometheusHttpServer);
}
- Meter brokerMeter = OpenTelemetrySdk.builder()
+ Meter proxyMeter = OpenTelemetrySdk.builder()
.setMeterProvider(providerBuilder.build())
.build()
.getMeter(OPEN_TELEMETRY_METER_NAME);
- initStatsMetrics(brokerMeter);
- }
-
- private void initStatsMetrics(Meter meter) {
- brokerPermission = meter.gaugeBuilder(GAUGE_BROKER_PERMISSION)
- .setDescription("Broker permission")
- .ofLongs()
- .buildWithCallback(measurement -> measurement.record(brokerConfig.getBrokerPermission(), newAttributesBuilder().build()));
+ initMetrics(proxyMeter, null);
}
- public void shutdown() {
- if (brokerConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.OTLP_GRPC ||
- brokerConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.OTLP_GRPC_SLS) {
+ @Override
+ public void shutdown() throws Exception {
+ if (proxyConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.OTLP_GRPC) {
periodicMetricReader.forceFlush();
periodicMetricReader.shutdown();
metricExporter.shutdown();
}
- if (brokerConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.PROM) {
+ if (proxyConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.PROM) {
prometheusHttpServer.forceFlush();
prometheusHttpServer.shutdown();
}
diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/ProxyStartupTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/ProxyStartupTest.java
index 0fe987bb9..f857ef6da 100644
--- a/proxy/src/test/java/org/apache/rocketmq/proxy/ProxyStartupTest.java
+++ b/proxy/src/test/java/org/apache/rocketmq/proxy/ProxyStartupTest.java
@@ -19,7 +19,7 @@ package org.apache.rocketmq.proxy;
import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
-
+import io.opentelemetry.sdk.OpenTelemetrySdk;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
@@ -28,10 +28,11 @@ import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
-import java.util.UUID;
import java.util.Iterator;
+import java.util.UUID;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.BrokerStartup;
+import org.apache.rocketmq.broker.metrics.BrokerMetricsManager;
import org.apache.rocketmq.client.log.ClientLogger;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.proxy.config.Configuration;
@@ -44,6 +45,7 @@ import org.junit.Before;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.MockedStatic;
+import org.mockito.Mockito;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
@@ -187,8 +189,12 @@ public class ProxyStartupTest {
try (MockedStatic<BrokerStartup> brokerStartupMocked = mockStatic(BrokerStartup.class);
MockedStatic<DefaultMessagingProcessor> messagingProcessorMocked = mockStatic(DefaultMessagingProcessor.class)) {
ArgumentCaptor<Object> args = ArgumentCaptor.forClass(Object.class);
+ BrokerController brokerControllerMocked = mock(BrokerController.class);
+ BrokerMetricsManager brokerMetricsManagerMocked = mock(BrokerMetricsManager.class);
+ Mockito.when(brokerMetricsManagerMocked.getBrokerMeter()).thenReturn(OpenTelemetrySdk.builder().build().getMeter("test"));
+ Mockito.when(brokerControllerMocked.getBrokerMetricsManager()).thenReturn(brokerMetricsManagerMocked);
brokerStartupMocked.when(() -> BrokerStartup.createBrokerController((String[]) args.capture()))
- .thenReturn(mock(BrokerController.class));
+ .thenReturn(brokerControllerMocked);
messagingProcessorMocked.when(() -> DefaultMessagingProcessor.createForLocalMode(any(), any()))
.thenReturn(mock(DefaultMessagingProcessor.class));
@@ -288,4 +294,4 @@ public class ProxyStartupTest {
assertSame(processor, ProxyStartup.createMessagingProcessor());
}
}
-}
\ No newline at end of file
+}
diff --git a/remoting/pom.xml b/remoting/pom.xml
index 403e527e3..b471f67c2 100644
--- a/remoting/pom.xml
+++ b/remoting/pom.xml
@@ -74,5 +74,9 @@
<groupId>io.grpc</groupId>
<artifactId>grpc-netty-shaded</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.squareup.okio</groupId>
+ <artifactId>okio-jvm</artifactId>
+ </dependency>
</dependencies>
</project>