opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-30 18:10:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #6777] Support metric in controller (#6778)

Browse Source 
* feat(controller): build metrics framework in controller module

1. build metrics framework in controller module

* feat(controller): record metrics about controller

1. record metrics about controller

* fix(controller): fix after review

1. fix after review

Closes https://github.com/apache/rocketmq/issues/6777

* fix(controller): fix after review

1. fix lack of dependency

Closes https://github.com/apache/rocketmq/issues/6777
This commit is contained in:
TheR1sing3un
2023-05-30 14:17:59 +08:00
committed by GitHub
parent 40ada807b3
commit 529976f8fb
16 changed files with 930 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/metrics/BrokerMetricsManager.java
+10 -9
View File
@@ -54,6 +54,7 @@ import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.metrics.MetricsExporterType;
import org.apache.rocketmq.common.metrics.NopLongCounter;
import org.apache.rocketmq.common.metrics.NopLongHistogram;
import org.apache.rocketmq.common.metrics.NopObservableLongGauge;
@@ -214,7 +215,7 @@ public class BrokerMetricsManager {
if (brokerConfig == null) {
return false;
}
BrokerConfig.MetricsExporterType exporterType = brokerConfig.getMetricsExporterType();
MetricsExporterType exporterType = brokerConfig.getMetricsExporterType();
if (!exporterType.isEnable()) {
return false;
}
@@ -231,8 +232,8 @@ public class BrokerMetricsManager {
}
private void init() {
BrokerConfig.MetricsExporterType metricsExporterType = brokerConfig.getMetricsExporterType();
if (metricsExporterType == BrokerConfig.MetricsExporterType.DISABLE) {
MetricsExporterType metricsExporterType = brokerConfig.getMetricsExporterType();
if (metricsExporterType == MetricsExporterType.DISABLE) {
return;
}
@@ -263,7 +264,7 @@ public class BrokerMetricsManager {
SdkMeterProviderBuilder providerBuilder = SdkMeterProvider.builder()
.setResource(Resource.empty());
if (metricsExporterType == BrokerConfig.MetricsExporterType.OTLP_GRPC) {
if (metricsExporterType == MetricsExporterType.OTLP_GRPC) {
String endpoint = brokerConfig.getMetricsGrpcExporterTarget();
if (!endpoint.startsWith("http")) {
endpoint = "https://" + endpoint;
@@ -303,7 +304,7 @@ public class BrokerMetricsManager {
providerBuilder.registerMetricReader(periodicMetricReader);
}
if (metricsExporterType == BrokerConfig.MetricsExporterType.PROM) {
if (metricsExporterType == MetricsExporterType.PROM) {
String promExporterHost = brokerConfig.getMetricsPromExporterHost();
if (StringUtils.isBlank(promExporterHost)) {
promExporterHost = brokerConfig.getBrokerIP1();
@@ -315,7 +316,7 @@ public class BrokerMetricsManager {
providerBuilder.registerMetricReader(prometheusHttpServer);
}
if (metricsExporterType == BrokerConfig.MetricsExporterType.LOG) {
if (metricsExporterType == MetricsExporterType.LOG) {
SLF4JBridgeHandler.removeHandlersForRootLogger();
SLF4JBridgeHandler.install();
loggingMetricExporter = LoggingMetricExporter.create(brokerConfig.isMetricsInDelta() ? AggregationTemporality.DELTA : AggregationTemporality.CUMULATIVE);
@@ -533,16 +534,16 @@ public class BrokerMetricsManager {
}
public void shutdown() {
if (brokerConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.OTLP_GRPC) {
if (brokerConfig.getMetricsExporterType() == MetricsExporterType.OTLP_GRPC) {
periodicMetricReader.forceFlush();
periodicMetricReader.shutdown();
metricExporter.shutdown();
}
if (brokerConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.PROM) {
if (brokerConfig.getMetricsExporterType() == MetricsExporterType.PROM) {
prometheusHttpServer.forceFlush();
prometheusHttpServer.shutdown();
}
if (brokerConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.LOG) {
if (brokerConfig.getMetricsExporterType() == MetricsExporterType.LOG) {
periodicMetricReader.forceFlush();
periodicMetricReader.shutdown();
loggingMetricExporter.shutdown();
common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
+1 -34
View File
@@ -20,6 +20,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.common.annotation.ImportantField;
import org.apache.rocketmq.common.constant.PermName;
import org.apache.rocketmq.common.message.MessageRequestMode;
import org.apache.rocketmq.common.metrics.MetricsExporterType;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.common.utils.NetworkUtil;
@@ -345,40 +346,6 @@ public class BrokerConfig extends BrokerIdentity {
private boolean useStaticSubscription = false;
public enum MetricsExporterType {
DISABLE(0),
OTLP_GRPC(1),
PROM(2),
LOG(3);
private final int value;
MetricsExporterType(int value) {
this.value = value;
}
public int getValue() {
return value;
}
public static MetricsExporterType valueOf(int value) {
switch (value) {
case 1:
return OTLP_GRPC;
case 2:
return PROM;
case 3:
return LOG;
default:
return DISABLE;
}
}
public boolean isEnable() {
return this.value > 0;
}
}
private MetricsExporterType metricsExporterType = MetricsExporterType.DISABLE;
private String metricsGrpcExporterTarget = "";
common/src/main/java/org/apache/rocketmq/common/ControllerConfig.java
+104
View File
@@ -17,6 +17,8 @@
package org.apache.rocketmq.common;
import java.io.File;
import java.util.Arrays;
import org.apache.rocketmq.common.metrics.MetricsExporterType;
public class ControllerConfig {
@@ -65,6 +67,22 @@ public class ControllerConfig {
*/
private long scanInactiveMasterInterval = 5 * 1000;
private MetricsExporterType metricsExporterType = MetricsExporterType.DISABLE;
private String metricsGrpcExporterTarget = "";
private String metricsGrpcExporterHeader = "";
private long metricGrpcExporterTimeOutInMills = 3 * 1000;
private long metricGrpcExporterIntervalInMills = 60 * 1000;
private long metricLoggingExporterIntervalInMills = 10 * 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;
public String getRocketmqHome() {
return rocketmqHome;
}
@@ -176,4 +194,90 @@ public class ControllerConfig {
public void setScanInactiveMasterInterval(long scanInactiveMasterInterval) {
this.scanInactiveMasterInterval = scanInactiveMasterInterval;
}
public String getDLedgerAddress() {
return Arrays.stream(this.controllerDLegerPeers.split(";"))
.filter(x -> this.controllerDLegerSelfId.equals(x.split("-")[0]))
.map(x -> x.split("-")[1]).findFirst().get();
}
public MetricsExporterType getMetricsExporterType() {
return metricsExporterType;
}
public void setMetricsExporterType(MetricsExporterType metricsExporterType) {
this.metricsExporterType = 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 long getMetricLoggingExporterIntervalInMills() {
return metricLoggingExporterIntervalInMills;
}
public void setMetricLoggingExporterIntervalInMills(long metricLoggingExporterIntervalInMills) {
this.metricLoggingExporterIntervalInMills = metricLoggingExporterIntervalInMills;
}
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;
}
}
common/src/main/java/org/apache/rocketmq/common/UtilAll.java
+27
View File
@@ -28,6 +28,7 @@ import java.net.Inet6Address;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.text.NumberFormat;
@@ -765,4 +766,30 @@ public class UtilAll {
STORE_LOG.info(dirName + " mkdir " + (result ? "OK" : "Failed"));
}
}
public static long calculateFileSizeInPath(File path) {
long size = 0;
try {
if (!path.exists() || Files.isSymbolicLink(path.toPath())) {
return 0;
}
if (path.isFile()) {
return path.length();
}
if (path.isDirectory()) {
File[] files = path.listFiles();
if (files != null && files.length > 0) {
for (File file : files) {
long fileSize = calculateFileSizeInPath(file);
if (fileSize == -1) return -1;
size += fileSize;
}
}
}
} catch (Exception e) {
log.error("calculate all file size in: {} error", path.getAbsolutePath(), e);
return -1;
}
return size;
}
}
common/src/main/java/org/apache/rocketmq/common/metrics/MetricsExporterType.java
+53
View File
@@ -0,0 +1,53 @@
/*
* 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.common.metrics;
public enum MetricsExporterType {
DISABLE(0),
OTLP_GRPC(1),
PROM(2),
LOG(3);
private final int value;
MetricsExporterType(int value) {
this.value = value;
}
public int getValue() {
return value;
}
public static MetricsExporterType valueOf(int value) {
switch (value) {
case 1:
return OTLP_GRPC;
case 2:
return PROM;
case 3:
return LOG;
default:
return DISABLE;
}
}
public boolean isEnable() {
return this.value > 0;
}
}
common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java
+61
View File
@@ -17,6 +17,8 @@
package org.apache.rocketmq.common;
import java.io.File;
import java.io.FileOutputStream;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
@@ -29,6 +31,7 @@ import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.within;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class UtilAllTest {
@@ -220,4 +223,62 @@ public class UtilAllTest {
public void testInvoke() throws Exception {
UtilAll.invoke(new Object(), "noMethod");
}
@Test
public void testCalculateFileSizeInPath() throws Exception {
/**
* testCalculateFileSizeInPath
* - file_0
* - dir_1
* - file_1_0
* - file_1_1
* - dir_1_2
* - file_1_2_0
* - dir_2
*/
String basePath = System.getProperty("java.io.tmpdir") + File.separator + "testCalculateFileSizeInPath";
File baseFile = new File(basePath);
// test empty path
assertEquals(0, UtilAll.calculateFileSizeInPath(baseFile));
// create baseDir
assertTrue(baseFile.mkdirs());
File file0 = new File(baseFile, "file_0");
assertTrue(file0.createNewFile());
writeFixedBytesToFile(file0, 1313);
assertEquals(1313, UtilAll.calculateFileSizeInPath(baseFile));
// build a file tree like above
File dir1 = new File(baseFile, "dir_1");
dir1.mkdirs();
File file10 = new File(dir1, "file_1_0");
File file11 = new File(dir1, "file_1_1");
File dir12 = new File(dir1, "dir_1_2");
dir12.mkdirs();
File file120 = new File(dir12, "file_1_2_0");
File dir2 = new File(baseFile, "dir_2");
dir2.mkdirs();
// write all file with 1313 bytes data
assertTrue(file10.createNewFile());
writeFixedBytesToFile(file10, 1313);
assertTrue(file11.createNewFile());
writeFixedBytesToFile(file11, 1313);
assertTrue(file120.createNewFile());
writeFixedBytesToFile(file120, 1313);
assertEquals(1313 * 4, UtilAll.calculateFileSizeInPath(baseFile));
// clear all file
baseFile.deleteOnExit();
}
private void writeFixedBytesToFile(File file, int size) throws Exception {
FileOutputStream outputStream = new FileOutputStream(file);
byte[] bytes = new byte[size];
outputStream.write(bytes, 0, size);
outputStream.close();
}
}
controller/pom.xml
+4
View File
@@ -58,5 +58,9 @@
<groupId>${project.groupId}</groupId>
<artifactId>rocketmq-srvutil</artifactId>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>jul-to-slf4j</artifactId>
</dependency>
</dependencies>
</project>
controller/src/main/java/org/apache/rocketmq/controller/BrokerHeartbeatManager.java
+7
View File
@@ -17,6 +17,7 @@
package org.apache.rocketmq.controller;
import io.netty.channel.Channel;
import java.util.Map;
import org.apache.rocketmq.controller.helper.BrokerLifecycleListener;
import org.apache.rocketmq.controller.impl.heartbeat.BrokerLiveInfo;
@@ -63,4 +64,10 @@ public interface BrokerHeartbeatManager {
* Check whether broker active
*/
boolean isBrokerActive(final String clusterName, final String brokerName, final Long brokerId);
/**
* Count the number of active brokers in each broker-set of each cluster
* @return active brokers count
*/
Map<String/*cluster*/, Map<String/*broker-set*/, Integer/*active broker num*/>> getActiveBrokersNum();
}
controller/src/main/java/org/apache/rocketmq/controller/ControllerManager.java
+4
View File
@@ -39,6 +39,7 @@ import org.apache.rocketmq.common.future.FutureTaskExt;
import org.apache.rocketmq.controller.elect.impl.DefaultElectPolicy;
import org.apache.rocketmq.controller.impl.DLedgerController;
import org.apache.rocketmq.controller.impl.heartbeat.DefaultBrokerHeartbeatManager;
import org.apache.rocketmq.controller.metrics.ControllerMetricsManager;
import org.apache.rocketmq.controller.processor.ControllerRequestProcessor;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
@@ -75,6 +76,8 @@ public class ControllerManager {
private NotifyService notifyService;
private ControllerMetricsManager controllerMetricsManager;
public ControllerManager(ControllerConfig controllerConfig, NettyServerConfig nettyServerConfig,
NettyClientConfig nettyClientConfig) {
this.controllerConfig = controllerConfig;
@@ -120,6 +123,7 @@ public class ControllerManager {
this.heartbeatManager.registerBrokerLifecycleListener(this::onBrokerInactive);
this.controller.registerBrokerLifecycleListener(this::onBrokerInactive);
registerProcessor();
this.controllerMetricsManager = ControllerMetricsManager.getInstance(this);
return true;
}
controller/src/main/java/org/apache/rocketmq/controller/impl/DLedgerController.java
+55 -1
View File
@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.controller.impl;
import com.google.common.base.Stopwatch;
import io.openmessaging.storage.dledger.AppendFuture;
import io.openmessaging.storage.dledger.DLedgerConfig;
import io.openmessaging.storage.dledger.DLedgerLeaderElector;
@@ -24,6 +25,7 @@ import io.openmessaging.storage.dledger.MemberState;
import io.openmessaging.storage.dledger.protocol.AppendEntryRequest;
import io.openmessaging.storage.dledger.protocol.AppendEntryResponse;
import io.openmessaging.storage.dledger.protocol.BatchAppendEntryRequest;
import io.opentelemetry.api.common.AttributesBuilder;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -35,6 +37,7 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;
import org.apache.rocketmq.common.ControllerConfig;
@@ -50,6 +53,8 @@ import org.apache.rocketmq.controller.impl.event.ControllerResult;
import org.apache.rocketmq.controller.impl.event.EventMessage;
import org.apache.rocketmq.controller.impl.event.EventSerializer;
import org.apache.rocketmq.controller.impl.manager.ReplicasInfoManager;
import org.apache.rocketmq.controller.metrics.ControllerMetricsConstant;
import org.apache.rocketmq.controller.metrics.ControllerMetricsManager;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.ChannelEventListener;
@@ -61,6 +66,7 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.SyncStateSet;
import org.apache.rocketmq.remoting.protocol.header.controller.AlterSyncStateSetRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.admin.CleanControllerBrokerDataRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.GetMetaDataResponseHeader;
@@ -69,6 +75,12 @@ import org.apache.rocketmq.remoting.protocol.header.controller.register.ApplyBro
import org.apache.rocketmq.remoting.protocol.header.controller.register.GetNextBrokerIdRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.RegisterBrokerToControllerRequestHeader;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.LABEL_BROKER_SET;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.LABEL_CLUSTER_NAME;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.LABEL_DLEDGER_OPERATION;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.LABEL_DLEDGER_OPERATION_STATUS;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.LABEL_ELECTION_RESULT;
/**
* The implementation of controller, based on DLedger (raft).
*/
@@ -174,7 +186,30 @@ public class DLedgerController implements Controller {
@Override
public CompletableFuture<RemotingCommand> electMaster(final ElectMasterRequestHeader request) {
return this.scheduler.appendEvent("electMaster",
() -> this.replicasInfoManager.electMaster(request, this.electPolicy), true);
() -> {
ControllerResult<ElectMasterResponseHeader> electResult = this.replicasInfoManager.electMaster(request, this.electPolicy);
AttributesBuilder attributesBuilder = ControllerMetricsManager.newAttributesBuilder()
.put(LABEL_CLUSTER_NAME, request.getClusterName())
.put(LABEL_BROKER_SET, request.getBrokerName());
switch (electResult.getResponseCode()) {
case ResponseCode.SUCCESS:
ControllerMetricsManager.electionTotal.add(1,
attributesBuilder.put(LABEL_ELECTION_RESULT, ControllerMetricsConstant.ElectionResult.NEW_MASTER_ELECTED.getLowerCaseName()).build());
break;
case ResponseCode.CONTROLLER_MASTER_STILL_EXIST:
ControllerMetricsManager.electionTotal.add(1,
attributesBuilder.put(LABEL_ELECTION_RESULT, ControllerMetricsConstant.ElectionResult.KEEP_CURRENT_MASTER.getLowerCaseName()).build());
break;
case ResponseCode.CONTROLLER_MASTER_NOT_AVAILABLE:
case ResponseCode.CONTROLLER_ELECT_MASTER_FAILED:
ControllerMetricsManager.electionTotal.add(1,
attributesBuilder.put(LABEL_ELECTION_RESULT, ControllerMetricsConstant.ElectionResult.NO_MASTER_ELECTED.getLowerCaseName()).build());
break;
default:
break;
}
return electResult;
}, true);
}
@Override
@@ -258,14 +293,30 @@ public class DLedgerController implements Controller {
if (request != null) {
request.setGroup(this.dLedgerConfig.getGroup());
request.setRemoteId(this.dLedgerConfig.getSelfId());
Stopwatch stopwatch = Stopwatch.createStarted();
AttributesBuilder attributesBuilder = ControllerMetricsManager.newAttributesBuilder()
.put(LABEL_DLEDGER_OPERATION, ControllerMetricsConstant.DLedgerOperation.APPEND.getLowerCaseName());
try {
final AppendFuture<AppendEntryResponse> dLedgerFuture = (AppendFuture<AppendEntryResponse>) dLedgerServer.handleAppend(request);
if (dLedgerFuture.getPos() == -1) {
ControllerMetricsManager.dLedgerOpTotal.add(1,
attributesBuilder.put(LABEL_DLEDGER_OPERATION_STATUS, ControllerMetricsConstant.DLedgerOperationStatus.FAILED.getLowerCaseName()).build());
return false;
}
dLedgerFuture.get(5, TimeUnit.SECONDS);
ControllerMetricsManager.dLedgerOpTotal.add(1,
attributesBuilder.put(LABEL_DLEDGER_OPERATION_STATUS, ControllerMetricsConstant.DLedgerOperationStatus.SUCCESS.getLowerCaseName()).build());
ControllerMetricsManager.dLedgerOpLatency.record(stopwatch.elapsed(TimeUnit.MICROSECONDS),
attributesBuilder.build());
} catch (Exception e) {
log.error("Failed to append entry to DLedger", e);
if (e instanceof TimeoutException) {
ControllerMetricsManager.dLedgerOpTotal.add(1,
attributesBuilder.put(LABEL_DLEDGER_OPERATION_STATUS, ControllerMetricsConstant.DLedgerOperationStatus.TIMEOUT.getLowerCaseName()).build());
} else {
ControllerMetricsManager.dLedgerOpTotal.add(1,
attributesBuilder.put(LABEL_DLEDGER_OPERATION_STATUS, ControllerMetricsConstant.DLedgerOperationStatus.FAILED.getLowerCaseName()).build());
}
return false;
}
return true;
@@ -474,12 +525,14 @@ public class DLedgerController implements Controller {
Runnable runnable = () -> {
switch (role) {
case CANDIDATE:
ControllerMetricsManager.recordRole(role, this.currentRole);
this.currentRole = MemberState.Role.CANDIDATE;
log.info("Controller {} change role to candidate", this.selfId);
DLedgerController.this.stopScheduling();
DLedgerController.this.cancelScanInactiveFuture();
break;
case FOLLOWER:
ControllerMetricsManager.recordRole(role, this.currentRole);
this.currentRole = MemberState.Role.FOLLOWER;
log.info("Controller {} change role to Follower, leaderId:{}", this.selfId, getMemberState().getLeaderId());
DLedgerController.this.stopScheduling();
@@ -496,6 +549,7 @@ public class DLedgerController implements Controller {
request.setBody(new byte[0]);
try {
if (appendToDLedgerAndWait(request)) {
ControllerMetricsManager.recordRole(role, this.currentRole);
this.currentRole = MemberState.Role.LEADER;
DLedgerController.this.startScheduling();
if (DLedgerController.this.scanInactiveMasterFuture == null) {
controller/src/main/java/org/apache/rocketmq/controller/impl/heartbeat/DefaultBrokerHeartbeatManager.java
+16 -1
View File
@@ -18,6 +18,7 @@ package org.apache.rocketmq.controller.impl.heartbeat;
import io.netty.channel.Channel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -106,7 +107,8 @@ public class DefaultBrokerHeartbeatManager implements BrokerHeartbeatManager {
@Override
public void onBrokerHeartbeat(String clusterName, String brokerName, String brokerAddr, Long brokerId,
Long timeoutMillis, Channel channel, Integer epoch, Long maxOffset, Long confirmOffset, Integer electionPriority) {
Long timeoutMillis, Channel channel, Integer epoch, Long maxOffset, Long confirmOffset,
Integer electionPriority) {
BrokerIdentityInfo brokerIdentityInfo = new BrokerIdentityInfo(clusterName, brokerName, brokerId);
BrokerLiveInfo prev = this.brokerLiveTable.get(brokerIdentityInfo);
int realEpoch = Optional.ofNullable(epoch).orElse(-1);
@@ -173,4 +175,17 @@ public class DefaultBrokerHeartbeatManager implements BrokerHeartbeatManager {
return false;
}
@Override
public Map<String, Map<String, Integer>> getActiveBrokersNum() {
Map<String, Map<String, Integer>> map = new HashMap<>();
this.brokerLiveTable.keySet().stream()
.filter(brokerIdentity -> this.isBrokerActive(brokerIdentity.getClusterName(), brokerIdentity.getBrokerName(), brokerIdentity.getBrokerId()))
.forEach(id -> {
map.computeIfAbsent(id.getClusterName(), k -> new HashMap<>());
map.get(id.getClusterName()).compute(id.getBrokerName(), (broker, num) ->
num == null ? 0 : num + 1
);
});
return map;
}
}
controller/src/main/java/org/apache/rocketmq/controller/metrics/ControllerMetricsConstant.java
+148
View File
@@ -0,0 +1,148 @@
/*
* 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.controller.metrics;
import org.apache.rocketmq.remoting.protocol.RequestCode;
public class ControllerMetricsConstant {
public static final String LABEL_ADDRESS = "address";
public static final String LABEL_GROUP = "group";
public static final String LABEL_PEER_ID = "peer_id";
public static final String LABEL_AGGREGATION = "aggregation";
public static final String AGGREGATION_DELTA = "delta";
public static final String OPEN_TELEMETRY_METER_NAME = "controller";
public static final String GAUGE_ROLE = "role";
// unit: B
public static final String GAUGE_DLEDGER_DISK_USAGE = "dledger_disk_usage";
public static final String GAUGE_ACTIVE_BROKER_NUM = "active_broker_num";
public static final String COUNTER_REQUEST_TOTAL = "request_total";
public static final String COUNTER_DLEDGER_OP_TOTAL = "dledger_op_total";
public static final String COUNTER_ELECTION_TOTAL = "election_total";
// unit: us
public static final String HISTOGRAM_REQUEST_LATENCY = "request_latency";
// unit: us
public static final String HISTOGRAM_DLEDGER_OP_LATENCY = "dledger_op_latency";
public static final String LABEL_CLUSTER_NAME = "cluster";
public static final String LABEL_BROKER_SET = "broker_set";
public static final String LABEL_REQUEST_TYPE = "request_type";
public static final String LABEL_REQUEST_HANDLE_STATUS = "request_handle_status";
public static final String LABEL_DLEDGER_OPERATION = "dledger_operation";
public static final String LABEL_DLEDGER_OPERATION_STATUS = "dLedger_operation_status";
public static final String LABEL_ELECTION_RESULT = "election_result";
public enum RequestType {
CONTROLLER_ALTER_SYNC_STATE_SET(RequestCode.CONTROLLER_ALTER_SYNC_STATE_SET),
CONTROLLER_ELECT_MASTER(RequestCode.CONTROLLER_ELECT_MASTER),
CONTROLLER_REGISTER_BROKER(RequestCode.CONTROLLER_REGISTER_BROKER),
CONTROLLER_GET_REPLICA_INFO(RequestCode.CONTROLLER_GET_REPLICA_INFO),
CONTROLLER_GET_METADATA_INFO(RequestCode.CONTROLLER_GET_METADATA_INFO),
CONTROLLER_GET_SYNC_STATE_DATA(RequestCode.CONTROLLER_GET_SYNC_STATE_DATA),
CONTROLLER_GET_BROKER_EPOCH_CACHE(RequestCode.GET_BROKER_EPOCH_CACHE),
CONTROLLER_NOTIFY_BROKER_ROLE_CHANGED(RequestCode.NOTIFY_BROKER_ROLE_CHANGED),
CONTROLLER_BROKER_HEARTBEAT(RequestCode.BROKER_HEARTBEAT),
CONTROLLER_UPDATE_CONTROLLER_CONFIG(RequestCode.UPDATE_CONTROLLER_CONFIG),
CONTROLLER_GET_CONTROLLER_CONFIG(RequestCode.GET_CONTROLLER_CONFIG),
CONTROLLER_CLEAN_BROKER_DATA(RequestCode.CLEAN_BROKER_DATA),
CONTROLLER_GET_NEXT_BROKER_ID(RequestCode.CONTROLLER_GET_NEXT_BROKER_ID),
CONTROLLER_APPLY_BROKER_ID(RequestCode.CONTROLLER_APPLY_BROKER_ID);
private final int code;
RequestType(int code) {
this.code = code;
}
public static String getLowerCaseNameByCode(int code) {
for (RequestType requestType : RequestType.values()) {
if (requestType.code == code) {
return requestType.name();
}
}
return null;
}
}
public enum RequestHandleStatus {
SUCCESS,
FAILED,
TIMEOUT;
public String getLowerCaseName() {
return this.name().toLowerCase();
}
}
public enum DLedgerOperation {
APPEND;
public String getLowerCaseName() {
return this.name().toLowerCase();
}
}
public enum DLedgerOperationStatus {
SUCCESS,
FAILED,
TIMEOUT;
public String getLowerCaseName() {
return this.name().toLowerCase();
}
}
public enum ElectionResult {
NEW_MASTER_ELECTED,
KEEP_CURRENT_MASTER,
NO_MASTER_ELECTED;
public String getLowerCaseName() {
return this.name().toLowerCase();
}
}
}
controller/src/main/java/org/apache/rocketmq/controller/metrics/ControllerMetricsManager.java
+383
View File
@@ -0,0 +1,383 @@
/*
* 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.controller.metrics;
import com.google.common.base.Splitter;
import io.openmessaging.storage.dledger.MemberState;
import io.opentelemetry.api.common.Attributes;
import io.opentelemetry.api.common.AttributesBuilder;
import io.opentelemetry.api.metrics.LongCounter;
import io.opentelemetry.api.metrics.LongHistogram;
import io.opentelemetry.api.metrics.LongUpDownCounter;
import io.opentelemetry.api.metrics.Meter;
import io.opentelemetry.api.metrics.ObservableLongGauge;
import io.opentelemetry.exporter.logging.LoggingMetricExporter;
import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporter;
import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporterBuilder;
import io.opentelemetry.exporter.prometheus.PrometheusHttpServer;
import io.opentelemetry.sdk.OpenTelemetrySdk;
import io.opentelemetry.sdk.metrics.Aggregation;
import io.opentelemetry.sdk.metrics.InstrumentSelector;
import io.opentelemetry.sdk.metrics.InstrumentType;
import io.opentelemetry.sdk.metrics.SdkMeterProvider;
import io.opentelemetry.sdk.metrics.SdkMeterProviderBuilder;
import io.opentelemetry.sdk.metrics.View;
import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader;
import io.opentelemetry.sdk.resources.Resource;
import java.io.File;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.ControllerConfig;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.metrics.MetricsExporterType;
import org.apache.rocketmq.common.metrics.NopLongCounter;
import org.apache.rocketmq.common.metrics.NopLongHistogram;
import org.apache.rocketmq.common.metrics.NopLongUpDownCounter;
import org.apache.rocketmq.common.metrics.NopObservableLongGauge;
import org.apache.rocketmq.controller.ControllerManager;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.slf4j.bridge.SLF4JBridgeHandler;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.AGGREGATION_DELTA;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.COUNTER_DLEDGER_OP_TOTAL;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.COUNTER_ELECTION_TOTAL;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.COUNTER_REQUEST_TOTAL;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.GAUGE_ACTIVE_BROKER_NUM;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.GAUGE_DLEDGER_DISK_USAGE;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.GAUGE_ROLE;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.HISTOGRAM_DLEDGER_OP_LATENCY;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.HISTOGRAM_REQUEST_LATENCY;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.LABEL_ADDRESS;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.LABEL_AGGREGATION;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.LABEL_BROKER_SET;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.LABEL_CLUSTER_NAME;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.LABEL_GROUP;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.LABEL_PEER_ID;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.OPEN_TELEMETRY_METER_NAME;
public class ControllerMetricsManager {
private static final Logger logger = LoggerFactory.getLogger(LoggerName.CONTROLLER_LOGGER_NAME);
private static volatile ControllerMetricsManager instance;
private static final Map<String, String> LABEL_MAP = new HashMap<>();
// metrics about node status
public static LongUpDownCounter role = new NopLongUpDownCounter();
public static ObservableLongGauge dLedgerDiskUsage = new NopObservableLongGauge();
public static ObservableLongGauge activeBrokerNum = new NopObservableLongGauge();
public static LongCounter requestTotal = new NopLongCounter();
public static LongCounter dLedgerOpTotal = new NopLongCounter();
public static LongCounter electionTotal = new NopLongCounter();
// metrics about latency
public static LongHistogram requestLatency = new NopLongHistogram();
public static LongHistogram dLedgerOpLatency = new NopLongHistogram();
private static double us = 1d;
private static double ms = 1000 * us;
private static double s = 1000 * ms;
private final ControllerManager controllerManager;
private final ControllerConfig config;
private Meter controllerMeter;
private OtlpGrpcMetricExporter metricExporter;
private PeriodicMetricReader periodicMetricReader;
private PrometheusHttpServer prometheusHttpServer;
private LoggingMetricExporter loggingMetricExporter;
public static ControllerMetricsManager getInstance(ControllerManager controllerManager) {
if (instance == null) {
synchronized (ControllerMetricsManager.class) {
if (instance == null) {
instance = new ControllerMetricsManager(controllerManager);
}
}
}
return instance;
}
public static AttributesBuilder newAttributesBuilder() {
AttributesBuilder builder = Attributes.builder();
LABEL_MAP.forEach(builder::put);
return builder;
}
public static void recordRole(MemberState.Role newRole, MemberState.Role oldRole) {
role.add(getRoleValue(newRole) - getRoleValue(oldRole),
newAttributesBuilder().build());
}
private static int getRoleValue(MemberState.Role role) {
switch (role) {
case UNKNOWN:
return 0;
case CANDIDATE:
return 1;
case FOLLOWER:
return 2;
case LEADER:
return 3;
default:
logger.error("Unknown role {}", role);
return 0;
}
}
private ControllerMetricsManager(ControllerManager controllerManager) {
this.controllerManager = controllerManager;
this.config = this.controllerManager.getControllerConfig();
this.LABEL_MAP.put(LABEL_ADDRESS, this.config.getDLedgerAddress());
this.LABEL_MAP.put(LABEL_GROUP, this.config.getControllerDLegerGroup());
this.LABEL_MAP.put(LABEL_PEER_ID, this.config.getControllerDLegerSelfId());
this.init();
}
private boolean checkConfig() {
if (config == null) {
return false;
}
MetricsExporterType exporterType = config.getMetricsExporterType();
if (!exporterType.isEnable()) {
return false;
}
switch (exporterType) {
case OTLP_GRPC:
return StringUtils.isNotBlank(config.getMetricsGrpcExporterTarget());
case PROM:
return true;
case LOG:
return true;
}
return false;
}
private void registerMetricsView(SdkMeterProviderBuilder providerBuilder) {
// define latency bucket
List<Double> latencyBuckets = Arrays.asList(
1 * us, 3 * us, 5 * us,
10 * us, 30 * us, 50 * us,
100 * us, 300 * us, 500 * us,
1 * ms, 3 * ms, 5 * ms,
10 * ms, 30 * ms, 50 * ms,
100 * ms, 300 * ms, 500 * ms,
1 * s, 3 * s, 5 * s,
10 * s
);
View latecyView = View.builder()
.setAggregation(Aggregation.explicitBucketHistogram(latencyBuckets))
.build();
InstrumentSelector requestLatencySelector = InstrumentSelector.builder()
.setType(InstrumentType.HISTOGRAM)
.setName(HISTOGRAM_REQUEST_LATENCY)
.build();
InstrumentSelector dLedgerOpLatencySelector = InstrumentSelector.builder()
.setType(InstrumentType.HISTOGRAM)
.setName(HISTOGRAM_DLEDGER_OP_LATENCY)
.build();
providerBuilder.registerView(requestLatencySelector, latecyView);
providerBuilder.registerView(dLedgerOpLatencySelector, latecyView);
}
private void initMetric(Meter meter) {
role = meter.upDownCounterBuilder(GAUGE_ROLE)
.setDescription("role of current node")
.build();
dLedgerDiskUsage = meter.gaugeBuilder(GAUGE_DLEDGER_DISK_USAGE)
.setDescription("disk usage of dledger")
.setUnit("bytes")
.ofLongs()
.buildWithCallback(measurement -> {
String path = config.getControllerStorePath();
if (!UtilAll.isPathExists(path)) {
return;
}
File file = new File(path);
Long diskUsage = UtilAll.calculateFileSizeInPath(file);
if (diskUsage == -1) {
logger.error("calculateFileSizeInPath error, path: {}", path);
return;
}
measurement.record(diskUsage, newAttributesBuilder().build());
});
activeBrokerNum = meter.gaugeBuilder(GAUGE_ACTIVE_BROKER_NUM)
.setDescription("now active brokers num")
.ofLongs()
.buildWithCallback(measurement -> {
Map<String, Map<String, Integer>> activeBrokersNum = controllerManager.getHeartbeatManager().getActiveBrokersNum();
activeBrokersNum.forEach((cluster, brokerSetAndNum) -> {
brokerSetAndNum.forEach((brokerSet, num) -> measurement.record(num,
newAttributesBuilder().put(LABEL_CLUSTER_NAME, cluster).put(LABEL_BROKER_SET, brokerSet).build()));
});
});
requestTotal = meter.counterBuilder(COUNTER_REQUEST_TOTAL)
.setDescription("total request num")
.build();
dLedgerOpTotal = meter.counterBuilder(COUNTER_DLEDGER_OP_TOTAL)
.setDescription("total dledger operation num")
.build();
electionTotal = meter.counterBuilder(COUNTER_ELECTION_TOTAL)
.setDescription("total elect num")
.build();
requestLatency = meter.histogramBuilder(HISTOGRAM_REQUEST_LATENCY)
.setDescription("request latency")
.setUnit("us")
.ofLongs()
.build();
dLedgerOpLatency = meter.histogramBuilder(HISTOGRAM_DLEDGER_OP_LATENCY)
.setDescription("dledger operation latency")
.setUnit("us")
.ofLongs()
.build();
}
public void init() {
MetricsExporterType type = this.config.getMetricsExporterType();
if (type == MetricsExporterType.DISABLE) {
return;
}
if (!checkConfig()) {
logger.error("check metric config failed, will not export metrics");
return;
}
String labels = config.getMetricsLabel();
if (StringUtils.isNotBlank(labels)) {
List<String> labelList = Splitter.on(',').omitEmptyStrings().splitToList(labels);
for (String label : labelList) {
String[] pair = label.split(":");
if (pair.length != 2) {
logger.warn("metrics label is not valid: {}", label);
continue;
}
LABEL_MAP.put(pair[0], pair[1]);
}
}
if (config.isMetricsInDelta()) {
LABEL_MAP.put(LABEL_AGGREGATION, AGGREGATION_DELTA);
}
SdkMeterProviderBuilder providerBuilder = SdkMeterProvider.builder().setResource(Resource.empty());
if (type == MetricsExporterType.OTLP_GRPC) {
String endpoint = config.getMetricsGrpcExporterTarget();
if (!endpoint.startsWith("http")) {
endpoint = "https://" + endpoint;
}
OtlpGrpcMetricExporterBuilder metricExporterBuilder = OtlpGrpcMetricExporter.builder()
.setEndpoint(endpoint)
.setTimeout(config.getMetricGrpcExporterTimeOutInMills(), TimeUnit.MILLISECONDS)
.setAggregationTemporalitySelector(x -> {
if (config.isMetricsInDelta() &&
(x == InstrumentType.COUNTER || x == InstrumentType.OBSERVABLE_COUNTER || x == InstrumentType.HISTOGRAM)) {
return AggregationTemporality.DELTA;
}
return AggregationTemporality.CUMULATIVE;
});
String headers = config.getMetricsGrpcExporterHeader();
if (StringUtils.isNotBlank(headers)) {
Map<String, String> headerMap = new HashMap<>();
List<String> headerList = Splitter.on(',').omitEmptyStrings().splitToList(headers);
for (String header : headerList) {
String[] pair = header.split(":");
if (pair.length != 2) {
logger.warn("metricsGrpcExporterHeader is not valid: {}", headers);
continue;
}
headerMap.put(pair[0], pair[1]);
}
headerMap.forEach(metricExporterBuilder::addHeader);
}
metricExporter = metricExporterBuilder.build();
periodicMetricReader = PeriodicMetricReader.builder(metricExporter)
.setInterval(config.getMetricGrpcExporterIntervalInMills(), TimeUnit.MILLISECONDS)
.build();
providerBuilder.registerMetricReader(periodicMetricReader);
}
if (type == MetricsExporterType.PROM) {
String promExporterHost = config.getMetricsPromExporterHost();
if (StringUtils.isBlank(promExporterHost)) {
promExporterHost = "0.0.0.0";
}
prometheusHttpServer = PrometheusHttpServer.builder()
.setHost(promExporterHost)
.setPort(config.getMetricsPromExporterPort())
.build();
providerBuilder.registerMetricReader(prometheusHttpServer);
}
if (type == MetricsExporterType.LOG) {
SLF4JBridgeHandler.removeHandlersForRootLogger();
SLF4JBridgeHandler.install();
loggingMetricExporter = LoggingMetricExporter.create(config.isMetricsInDelta() ? AggregationTemporality.DELTA : AggregationTemporality.CUMULATIVE);
java.util.logging.Logger.getLogger(LoggingMetricExporter.class.getName()).setLevel(java.util.logging.Level.FINEST);
periodicMetricReader = PeriodicMetricReader.builder(loggingMetricExporter)
.setInterval(config.getMetricLoggingExporterIntervalInMills(), TimeUnit.MILLISECONDS)
.build();
providerBuilder.registerMetricReader(periodicMetricReader);
}
registerMetricsView(providerBuilder);
controllerMeter = OpenTelemetrySdk.builder().setMeterProvider(providerBuilder.build())
.build().getMeter(OPEN_TELEMETRY_METER_NAME);
initMetric(controllerMeter);
}
}
controller/src/main/java/org/apache/rocketmq/controller/processor/ControllerRequestProcessor.java
+40
View File
@@ -16,17 +16,22 @@
*/
package org.apache.rocketmq.controller.processor;
import com.google.common.base.Stopwatch;
import io.netty.channel.ChannelHandlerContext;
import io.opentelemetry.api.common.Attributes;
import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.controller.BrokerHeartbeatManager;
import org.apache.rocketmq.controller.ControllerManager;
import org.apache.rocketmq.controller.metrics.ControllerMetricsConstant;
import org.apache.rocketmq.controller.metrics.ControllerMetricsManager;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.common.RemotingHelper;
@@ -45,6 +50,8 @@ import org.apache.rocketmq.remoting.protocol.header.controller.admin.CleanContro
import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.GetReplicaInfoRequestHeader;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.LABEL_REQUEST_HANDLE_STATUS;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.LABEL_REQUEST_TYPE;
import static org.apache.rocketmq.remoting.protocol.RequestCode.CONTROLLER_APPLY_BROKER_ID;
import static org.apache.rocketmq.remoting.protocol.RequestCode.BROKER_HEARTBEAT;
import static org.apache.rocketmq.remoting.protocol.RequestCode.CLEAN_BROKER_DATA;
@@ -80,6 +87,39 @@ public class ControllerRequestProcessor implements NettyRequestProcessor {
RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
request);
}
Stopwatch stopwatch = Stopwatch.createStarted();
try {
RemotingCommand resp = handleRequest(ctx, request);
Attributes attributes = ControllerMetricsManager.newAttributesBuilder()
.put(LABEL_REQUEST_TYPE, ControllerMetricsConstant.RequestType.getLowerCaseNameByCode(request.getCode()))
.put(LABEL_REQUEST_HANDLE_STATUS, ControllerMetricsConstant.RequestHandleStatus.SUCCESS.getLowerCaseName())
.build();
ControllerMetricsManager.requestTotal.add(1, attributes);
attributes = ControllerMetricsManager.newAttributesBuilder()
.put(LABEL_REQUEST_TYPE, ControllerMetricsConstant.RequestType.getLowerCaseNameByCode(request.getCode()))
.build();
ControllerMetricsManager.requestLatency.record(stopwatch.elapsed(TimeUnit.MICROSECONDS), attributes);
return resp;
} catch (Exception e) {
log.error("process request: {} error, ", request, e);
Attributes attributes;
if (e instanceof TimeoutException) {
attributes = ControllerMetricsManager.newAttributesBuilder()
.put(LABEL_REQUEST_TYPE, ControllerMetricsConstant.RequestType.getLowerCaseNameByCode(request.getCode()))
.put(LABEL_REQUEST_HANDLE_STATUS, ControllerMetricsConstant.RequestHandleStatus.TIMEOUT.getLowerCaseName())
.build();
} else {
attributes = ControllerMetricsManager.newAttributesBuilder()
.put(LABEL_REQUEST_TYPE, ControllerMetricsConstant.RequestType.getLowerCaseNameByCode(request.getCode()))
.put(LABEL_REQUEST_HANDLE_STATUS, ControllerMetricsConstant.RequestHandleStatus.FAILED.getLowerCaseName())
.build();
}
ControllerMetricsManager.requestTotal.add(1, attributes);
throw e;
}
}
private RemotingCommand handleRequest(ChannelHandlerContext ctx, RemotingCommand request) throws Exception {
switch (request.getCode()) {
case CONTROLLER_ALTER_SYNC_STATE_SET:
return this.handleAlterSyncStateSet(ctx, request);
proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java
+6 -6
View File
@@ -28,9 +28,9 @@ 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.common.metrics.MetricsExporterType;
import org.apache.rocketmq.common.utils.NetworkUtil;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
@@ -205,7 +205,7 @@ public class ProxyConfig implements ConfigFile {
private boolean traceOn = false;
private BrokerConfig.MetricsExporterType metricsExporterType = BrokerConfig.MetricsExporterType.DISABLE;
private MetricsExporterType metricsExporterType = MetricsExporterType.DISABLE;
private String metricsGrpcExporterTarget = "";
private String metricsGrpcExporterHeader = "";
@@ -1116,20 +1116,20 @@ public class ProxyConfig implements ConfigFile {
this.remotingAccessAddr = remotingAccessAddr;
}
public BrokerConfig.MetricsExporterType getMetricsExporterType() {
public MetricsExporterType getMetricsExporterType() {
return metricsExporterType;
}
public void setMetricsExporterType(BrokerConfig.MetricsExporterType metricsExporterType) {
public void setMetricsExporterType(MetricsExporterType metricsExporterType) {
this.metricsExporterType = metricsExporterType;
}
public void setMetricsExporterType(int metricsExporterType) {
this.metricsExporterType = BrokerConfig.MetricsExporterType.valueOf(metricsExporterType);
this.metricsExporterType = MetricsExporterType.valueOf(metricsExporterType);
}
public void setMetricsExporterType(String metricsExporterType) {
this.metricsExporterType = BrokerConfig.MetricsExporterType.valueOf(metricsExporterType);
this.metricsExporterType = MetricsExporterType.valueOf(metricsExporterType);
}
public String getMetricsGrpcExporterTarget() {
proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsManager.java
+11 -11
View File
@@ -39,8 +39,8 @@ import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.broker.metrics.BrokerMetricsManager;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.metrics.MetricsExporterType;
import org.apache.rocketmq.common.utils.StartAndShutdown;
import org.apache.rocketmq.proxy.config.ProxyConfig;
import org.apache.rocketmq.logging.org.slf4j.Logger;
@@ -72,7 +72,7 @@ public class ProxyMetricsManager implements StartAndShutdown {
public static ObservableLongGauge proxyUp = null;
public static void initLocalMode(BrokerMetricsManager brokerMetricsManager, ProxyConfig proxyConfig) {
if (proxyConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.DISABLE) {
if (proxyConfig.getMetricsExporterType() == MetricsExporterType.DISABLE) {
return;
}
ProxyMetricsManager.proxyConfig = proxyConfig;
@@ -116,7 +116,7 @@ public class ProxyMetricsManager implements StartAndShutdown {
if (proxyConfig == null) {
return false;
}
BrokerConfig.MetricsExporterType exporterType = proxyConfig.getMetricsExporterType();
MetricsExporterType exporterType = proxyConfig.getMetricsExporterType();
if (!exporterType.isEnable()) {
return false;
}
@@ -134,8 +134,8 @@ public class ProxyMetricsManager implements StartAndShutdown {
@Override
public void start() throws Exception {
BrokerConfig.MetricsExporterType metricsExporterType = proxyConfig.getMetricsExporterType();
if (metricsExporterType == BrokerConfig.MetricsExporterType.DISABLE) {
MetricsExporterType metricsExporterType = proxyConfig.getMetricsExporterType();
if (metricsExporterType == MetricsExporterType.DISABLE) {
return;
}
if (!checkConfig()) {
@@ -166,7 +166,7 @@ public class ProxyMetricsManager implements StartAndShutdown {
SdkMeterProviderBuilder providerBuilder = SdkMeterProvider.builder()
.setResource(Resource.empty());
if (metricsExporterType == BrokerConfig.MetricsExporterType.OTLP_GRPC) {
if (metricsExporterType == MetricsExporterType.OTLP_GRPC) {
String endpoint = proxyConfig.getMetricsGrpcExporterTarget();
if (!endpoint.startsWith("http")) {
endpoint = "https://" + endpoint;
@@ -206,7 +206,7 @@ public class ProxyMetricsManager implements StartAndShutdown {
providerBuilder.registerMetricReader(periodicMetricReader);
}
if (metricsExporterType == BrokerConfig.MetricsExporterType.PROM) {
if (metricsExporterType == MetricsExporterType.PROM) {
String promExporterHost = proxyConfig.getMetricsPromExporterHost();
if (StringUtils.isBlank(promExporterHost)) {
promExporterHost = "0.0.0.0";
@@ -218,7 +218,7 @@ public class ProxyMetricsManager implements StartAndShutdown {
providerBuilder.registerMetricReader(prometheusHttpServer);
}
if (metricsExporterType == BrokerConfig.MetricsExporterType.LOG) {
if (metricsExporterType == MetricsExporterType.LOG) {
SLF4JBridgeHandler.removeHandlersForRootLogger();
SLF4JBridgeHandler.install();
loggingMetricExporter = LoggingMetricExporter.create(proxyConfig.isMetricsInDelta() ? AggregationTemporality.DELTA : AggregationTemporality.CUMULATIVE);
@@ -239,16 +239,16 @@ public class ProxyMetricsManager implements StartAndShutdown {
@Override
public void shutdown() throws Exception {
if (proxyConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.OTLP_GRPC) {
if (proxyConfig.getMetricsExporterType() == MetricsExporterType.OTLP_GRPC) {
periodicMetricReader.forceFlush();
periodicMetricReader.shutdown();
metricExporter.shutdown();
}
if (proxyConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.PROM) {
if (proxyConfig.getMetricsExporterType() == MetricsExporterType.PROM) {
prometheusHttpServer.forceFlush();
prometheusHttpServer.shutdown();
}
if (proxyConfig.getMetricsExporterType() == BrokerConfig.MetricsExporterType.LOG) {
if (proxyConfig.getMetricsExporterType() == MetricsExporterType.LOG) {
periodicMetricReader.forceFlush();
periodicMetricReader.shutdown();
loggingMetricExporter.shutdown();