| /* |
| * 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); |
| } |
| |
| } |
