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 b9b3f7688b..04eb6798a2 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 @@ -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(); 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 2ce63a1f49..e9fad05e51 100644 --- a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java +++ b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java @@ -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 = ""; diff --git a/common/src/main/java/org/apache/rocketmq/common/ControllerConfig.java b/common/src/main/java/org/apache/rocketmq/common/ControllerConfig.java index b35198fc63..1e9c80b222 100644 --- a/common/src/main/java/org/apache/rocketmq/common/ControllerConfig.java +++ b/common/src/main/java/org/apache/rocketmq/common/ControllerConfig.java @@ -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; + } } diff --git a/common/src/main/java/org/apache/rocketmq/common/UtilAll.java b/common/src/main/java/org/apache/rocketmq/common/UtilAll.java index 03d1624509..d2b7c374b7 100644 --- a/common/src/main/java/org/apache/rocketmq/common/UtilAll.java +++ b/common/src/main/java/org/apache/rocketmq/common/UtilAll.java @@ -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; + } } diff --git a/common/src/main/java/org/apache/rocketmq/common/metrics/MetricsExporterType.java b/common/src/main/java/org/apache/rocketmq/common/metrics/MetricsExporterType.java new file mode 100644 index 0000000000..5f065b4534 --- /dev/null +++ b/common/src/main/java/org/apache/rocketmq/common/metrics/MetricsExporterType.java @@ -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; + } +} diff --git a/common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java b/common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java index 6306d56847..f568a65f4d 100644 --- a/common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java +++ b/common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java @@ -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(); + } } diff --git a/controller/pom.xml b/controller/pom.xml index 9e9ad287e2..f82ad2a72a 100644 --- a/controller/pom.xml +++ b/controller/pom.xml @@ -58,5 +58,9 @@ ${project.groupId} rocketmq-srvutil + + org.slf4j + jul-to-slf4j + \ No newline at end of file diff --git a/controller/src/main/java/org/apache/rocketmq/controller/BrokerHeartbeatManager.java b/controller/src/main/java/org/apache/rocketmq/controller/BrokerHeartbeatManager.java index 71b274c09b..f38a03a7b4 100644 --- a/controller/src/main/java/org/apache/rocketmq/controller/BrokerHeartbeatManager.java +++ b/controller/src/main/java/org/apache/rocketmq/controller/BrokerHeartbeatManager.java @@ -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> getActiveBrokersNum(); } diff --git a/controller/src/main/java/org/apache/rocketmq/controller/ControllerManager.java b/controller/src/main/java/org/apache/rocketmq/controller/ControllerManager.java index 46826517cd..7c91e70da5 100644 --- a/controller/src/main/java/org/apache/rocketmq/controller/ControllerManager.java +++ b/controller/src/main/java/org/apache/rocketmq/controller/ControllerManager.java @@ -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; } diff --git a/controller/src/main/java/org/apache/rocketmq/controller/impl/DLedgerController.java b/controller/src/main/java/org/apache/rocketmq/controller/impl/DLedgerController.java index b6007fe09b..fa91f288e2 100644 --- a/controller/src/main/java/org/apache/rocketmq/controller/impl/DLedgerController.java +++ b/controller/src/main/java/org/apache/rocketmq/controller/impl/DLedgerController.java @@ -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 electMaster(final ElectMasterRequestHeader request) { return this.scheduler.appendEvent("electMaster", - () -> this.replicasInfoManager.electMaster(request, this.electPolicy), true); + () -> { + ControllerResult 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 dLedgerFuture = (AppendFuture) 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) { diff --git a/controller/src/main/java/org/apache/rocketmq/controller/impl/heartbeat/DefaultBrokerHeartbeatManager.java b/controller/src/main/java/org/apache/rocketmq/controller/impl/heartbeat/DefaultBrokerHeartbeatManager.java index dc824281b6..2fbddb9cdf 100644 --- a/controller/src/main/java/org/apache/rocketmq/controller/impl/heartbeat/DefaultBrokerHeartbeatManager.java +++ b/controller/src/main/java/org/apache/rocketmq/controller/impl/heartbeat/DefaultBrokerHeartbeatManager.java @@ -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> getActiveBrokersNum() { + Map> 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; + } } diff --git a/controller/src/main/java/org/apache/rocketmq/controller/metrics/ControllerMetricsConstant.java b/controller/src/main/java/org/apache/rocketmq/controller/metrics/ControllerMetricsConstant.java new file mode 100644 index 0000000000..1efae43fe9 --- /dev/null +++ b/controller/src/main/java/org/apache/rocketmq/controller/metrics/ControllerMetricsConstant.java @@ -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(); + } + } + +} diff --git a/controller/src/main/java/org/apache/rocketmq/controller/metrics/ControllerMetricsManager.java b/controller/src/main/java/org/apache/rocketmq/controller/metrics/ControllerMetricsManager.java new file mode 100644 index 0000000000..9b30a3b43b --- /dev/null +++ b/controller/src/main/java/org/apache/rocketmq/controller/metrics/ControllerMetricsManager.java @@ -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 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 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> 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 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 headerMap = new HashMap<>(); + List 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); + } + +} diff --git a/controller/src/main/java/org/apache/rocketmq/controller/processor/ControllerRequestProcessor.java b/controller/src/main/java/org/apache/rocketmq/controller/processor/ControllerRequestProcessor.java index 6010f81eeb..93ecbbd9dd 100644 --- a/controller/src/main/java/org/apache/rocketmq/controller/processor/ControllerRequestProcessor.java +++ b/controller/src/main/java/org/apache/rocketmq/controller/processor/ControllerRequestProcessor.java @@ -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); 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 560cd89f5a..4f57a7052a 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 @@ -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() { diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsManager.java b/proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsManager.java index 8474a687bd..f5050858f6 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsManager.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsManager.java @@ -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();