You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2026/09/29 05:59:51 UTC
(iotdb) branch master updated: Fix metrics lost after metric service restart and conflicting metric series (#18748)
This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new b2cc8b52126 Fix metrics lost after metric service restart and conflicting metric series (#18748)
b2cc8b52126 is described below
commit b2cc8b521262aec0993d3d223b59502904031c82
Author: Jackie Tien <ja...@gmail.com>
AuthorDate: Tue Sep 29 13:59:44 2026 +0800
Fix metrics lost after metric service restart and conflicting metric series (#18748)
---
.../iotdb/confignode/i18n/ManagerMessages.java | 8 -
.../iotdb/confignode/i18n/ManagerMessages.java | 8 -
.../PipeConfigNodeRemainingTimeMetrics.java | 20 +-
.../PipeTemporaryMetaInCoordinatorMetrics.java | 18 +-
.../metric/sink/PipeConfigRegionSinkMetrics.java | 17 +-
.../source/PipeConfigRegionSourceMetrics.java | 17 +-
.../metric/PipeConfigNodeMetricsRestartTest.java | 184 +++++++++
iotdb-core/consensus/pom.xml | 5 +
.../logdispatcher/LogDispatcherThreadMetrics.java | 104 +++--
.../consensus/pipe/IoTConsensusV2ServerImpl.java | 5 +
.../pipe/metric/IoTConsensusV2ServerMetrics.java | 3 -
.../consensus/ratis/metrics/CounterProxy.java | 8 +-
.../ratis/metrics/IoTDBMetricRegistry.java | 34 +-
.../ratis/metrics/MetricRegistryManager.java | 14 +
.../consensus/ratis/metrics/RatisMetricSet.java | 8 +-
.../iotdb/consensus/ratis/metrics/TimerProxy.java | 7 +-
.../LogDispatcherThreadMetricsTest.java | 146 +++++++
.../metric/IoTConsensusV2ServerMetricsTest.java | 107 ++++++
.../ratis/metrics/RatisMetricSetTest.java | 134 +++++++
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 24 --
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 23 --
.../overview/PipeDataNodeSinglePipeMetrics.java | 20 +-
.../overview/PipeTsFileToTabletsMetrics.java | 16 +-
.../metric/processor/PipeProcessorMetrics.java | 19 +-
.../schema/PipeSchemaRegionListenerMetrics.java | 16 +-
.../metric/schema/PipeSchemaRegionSinkMetrics.java | 16 +-
.../schema/PipeSchemaRegionSourceMetrics.java | 16 +-
.../metric/sink/PipeDataRegionSinkMetrics.java | 11 +-
.../db/pipe/metric/source/PipeAssignerMetrics.java | 13 +-
.../metric/source/PipeDataRegionSourceMetrics.java | 20 +-
.../protocol/airgap/IoTDBDataRegionAirGapSink.java | 3 +-
.../thrift/async/IoTDBDataRegionAsyncSink.java | 3 +-
.../thrift/sync/IoTDBDataRegionSyncSink.java | 3 +-
.../handler/RPCServiceThriftHandlerMetrics.java | 37 +-
.../analyze/cache/partition/PartitionCache.java | 2 +-
.../schemaengine/metric/SchemaEngineMemMetric.java | 7 +-
.../schemaengine/metric/SchemaRegionMemMetric.java | 7 +-
.../rescon/MemSchemaEngineStatistics.java | 4 +
.../iotdb/db/service/metrics/CacheMetrics.java | 17 +
.../db/service/metrics/DataNodeMetricsHelper.java | 1 +
.../iotdb/db/service/metrics/WritingMetrics.java | 80 +++-
.../db/service/metrics/file/TsFileMetrics.java | 22 ++
.../dataregion/DataRegionMetrics.java | 49 ++-
.../memtable/TsFileProcessorInfoMetrics.java | 55 ++-
.../load/memory/LoadTsFileMemoryBlock.java | 20 +-
.../load/metrics/ActiveLoadingFilesMetricsSet.java | 6 +
.../ActiveLoadingFilesNumberMetricsSet.java | 16 +-
.../load/metrics/LoadTsFileMemMetricSet.java | 19 +-
...nsensusSubscriptionPrefetchingQueueMetrics.java | 12 +-
.../SubscriptionPrefetchingQueueMetrics.java | 55 ++-
.../db/pipe/metric/PipeMetricsRestartTest.java | 426 +++++++++++++++++++++
.../PipeSchemaRegionListenerMetricsTest.java | 116 ++++++
.../metric/SchemaMemMetricTableTest.java | 133 +++++++
.../metrics/DataNodeMetricsRestartTest.java | 290 ++++++++++++++
.../db/service/metrics/file/TsFileMetricsTest.java | 148 +++++++
.../dataregion/DatabaseMemMetricsTest.java | 144 +++++++
.../metric/SubscriptionMetricsRestartTest.java | 188 +++++++++
.../iotdb/metrics/metricsets/jvm/JvmGcMetrics.java | 86 ++---
.../metrics/metricsets/system/SystemMetrics.java | 13 +-
.../apache/iotdb/commons/i18n/PipeMessages.java | 2 -
.../apache/iotdb/commons/i18n/ServiceMessages.java | 2 +
.../apache/iotdb/commons/i18n/PipeMessages.java | 2 -
.../apache/iotdb/commons/i18n/ServiceMessages.java | 2 +
.../commons/concurrent/ThreadPoolMetrics.java | 74 ++--
.../WrappedScheduledExecutorService.java | 4 +-
.../WrappedSingleThreadExecutorService.java | 4 +-
.../WrappedSingleThreadScheduledExecutor.java | 4 +-
.../threadpool/WrappedThreadPoolExecutor.java | 4 +-
.../pipe/metric/PipeEventCommitMetrics.java | 20 +-
.../service/metric/JvmGcMonitorMetrics.java | 38 +-
.../commons/service/metric/MetricService.java | 13 +-
.../iotdb/commons/service/metric/enums/Metric.java | 3 +
.../commons/concurrent/ThreadPoolMetricsTest.java | 157 ++++++++
.../pipe/metric/PipeEventCommitMetricsTest.java | 99 +++++
.../service/metric/JvmGcMonitorMetricsTest.java | 72 ++++
.../service/metric/MetricServiceRestartTest.java | 218 +++++++++++
76 files changed, 3235 insertions(+), 486 deletions(-)
diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
index 6a36ac71302..48e1a390d18 100644
--- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
+++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
@@ -204,14 +204,6 @@ public final class ManagerMessages {
"Failed to sync template {} extension info to DataNode {}";
public static final String FAILED_TO_SYNC_TOPIC_META_RESULT_STATUS =
"Failed to sync topic meta. Result status: {}.";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_CONFIG_REGION_CONNECTOR_METRICS_CONNECTOR =
- "Failed to unbind from pipe config region connector metrics, connector map not empty";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_CONFIG_REGION_EXTRACTOR_METRICS_EXTRACTOR =
- "Failed to unbind from pipe config region extractor metrics, extractor map not empty";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_REMAINING_TIME_METRICS_REMAININGTIMEOPERATOR_MAP =
- "Failed to unbind from pipe remaining time metrics, RemainingTimeOperator map not empty";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_TEMPORARY_META_METRICS_PIPETEMPORARYMETA_MAP =
- "Failed to unbind from pipe temporary meta metrics, PipeTemporaryMeta map not empty";
public static final String FAILED_TO_UPDATE_PIPE_PROCEDURE_TIMER_PIPEPROCEDURE_DOES_NOT_EXIST =
"Failed to update pipe procedure timer, PipeProcedure({}) does not exist";
public static final String FAILED_TO_UPDATE_THE_LAST_EXECUTION_TIME_OF_CQ_BECAUSE =
diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
index 3c35171dd06..a06ae420eac 100644
--- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
+++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
@@ -202,14 +202,6 @@ public final class ManagerMessages {
"将模板 {} 的扩展信息同步到 DataNode {} 失败";
public static final String FAILED_TO_SYNC_TOPIC_META_RESULT_STATUS =
"同步 topic 元数据失败。结果状态：{}。";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_CONFIG_REGION_CONNECTOR_METRICS_CONNECTOR =
- "从 pipe config region connector 指标解绑失败，connector map 不为空";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_CONFIG_REGION_EXTRACTOR_METRICS_EXTRACTOR =
- "从 pipe config region extractor 指标解绑失败，extractor map 不为空";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_REMAINING_TIME_METRICS_REMAININGTIMEOPERATOR_MAP =
- "从 pipe remaining time 指标解绑失败，RemainingTimeOperator map 不为空";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_TEMPORARY_META_METRICS_PIPETEMPORARYMETA_MAP =
- "从 pipe temporary meta 指标解绑失败，PipeTemporaryMeta map 不为空";
public static final String FAILED_TO_UPDATE_PIPE_PROCEDURE_TIMER_PIPEPROCEDURE_DOES_NOT_EXIST =
"更新 pipe procedure timer 失败，PipeProcedure({}) 不存在";
public static final String FAILED_TO_UPDATE_THE_LAST_EXECUTION_TIME_OF_CQ_BECAUSE =
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/overview/PipeConfigNodeRemainingTimeMetrics.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/overview/PipeConfigNodeRemainingTimeMetrics.java
index 0e9c44f05d2..bd585a98a8c 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/overview/PipeConfigNodeRemainingTimeMetrics.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/overview/PipeConfigNodeRemainingTimeMetrics.java
@@ -51,7 +51,7 @@ public class PipeConfigNodeRemainingTimeMetrics implements IMetricSet {
//////////////////////////// bindTo & unbindFrom (metric framework) ////////////////////////////
@Override
- public void bindTo(final AbstractMetricService metricService) {
+ public synchronized void bindTo(final AbstractMetricService metricService) {
this.metricService = metricService;
ImmutableSet.copyOf(remainingTimeOperatorMap.keySet()).forEach(this::createMetrics);
}
@@ -74,13 +74,11 @@ public class PipeConfigNodeRemainingTimeMetrics implements IMetricSet {
}
@Override
- public void unbindFrom(final AbstractMetricService metricService) {
- ImmutableSet.copyOf(remainingTimeOperatorMap.keySet()).forEach(this::deregister);
- if (!remainingTimeOperatorMap.isEmpty()) {
- LOGGER.warn(
- ManagerMessages
- .FAILED_TO_UNBIND_FROM_PIPE_REMAINING_TIME_METRICS_REMAININGTIMEOPERATOR_MAP);
- }
+ public synchronized void unbindFrom(final AbstractMetricService metricService) {
+ // Keep the operators registered: they hold the states of the pipes and register only once, so
+ // a metric service restart must be able to bind them again.
+ // Synchronized with the (de)registrations, which may remove a registration being unbound.
+ ImmutableSet.copyOf(remainingTimeOperatorMap.keySet()).forEach(this::removeMetrics);
}
private void removeMetrics(final String pipeID) {
@@ -96,12 +94,11 @@ public class PipeConfigNodeRemainingTimeMetrics implements IMetricSet {
operator.getPipeName(),
Tag.CREATION_TIME.toString(),
String.valueOf(operator.getCreationTime()));
- remainingTimeOperatorMap.remove(pipeID);
}
//////////////////////////// register & deregister (pipe integration) ////////////////////////////
- public void register(final IoTDBConfigRegionSource extractor) {
+ public synchronized void register(final IoTDBConfigRegionSource extractor) {
// The metric is global thus the regionId is omitted
final String pipeID = extractor.getPipeName() + "_" + extractor.getCreationTime();
remainingTimeOperatorMap
@@ -132,7 +129,7 @@ public class PipeConfigNodeRemainingTimeMetrics implements IMetricSet {
remainingTimeOperatorMap.get(pipeID).freezeRate(true);
}
- public void deregister(final String pipeID) {
+ public synchronized void deregister(final String pipeID) {
if (!remainingTimeOperatorMap.containsKey(pipeID)) {
LOGGER.warn(
ManagerMessages
@@ -143,6 +140,7 @@ public class PipeConfigNodeRemainingTimeMetrics implements IMetricSet {
if (Objects.nonNull(metricService)) {
removeMetrics(pipeID);
}
+ remainingTimeOperatorMap.remove(pipeID);
}
public void markRegionCommit(final String pipeID, final boolean isDataRegion) {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/overview/PipeTemporaryMetaInCoordinatorMetrics.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/overview/PipeTemporaryMetaInCoordinatorMetrics.java
index 19604e0d6c1..c4e3ca285e2 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/overview/PipeTemporaryMetaInCoordinatorMetrics.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/overview/PipeTemporaryMetaInCoordinatorMetrics.java
@@ -58,7 +58,7 @@ public class PipeTemporaryMetaInCoordinatorMetrics implements IMetricSet {
//////////////////////////// bindTo & unbindFrom (metric framework) ////////////////////////////
@Override
- public void bindTo(final AbstractMetricService metricService) {
+ public synchronized void bindTo(final AbstractMetricService metricService) {
this.metricService = metricService;
ImmutableSet.copyOf(pipeTemporaryMetaMap.keySet()).forEach(this::createMetrics);
}
@@ -91,12 +91,10 @@ public class PipeTemporaryMetaInCoordinatorMetrics implements IMetricSet {
}
@Override
- public void unbindFrom(final AbstractMetricService metricService) {
- ImmutableSet.copyOf(pipeTemporaryMetaMap.keySet()).forEach(this::deregister);
- if (!pipeTemporaryMetaMap.isEmpty()) {
- LOGGER.warn(
- ManagerMessages.FAILED_TO_UNBIND_FROM_PIPE_TEMPORARY_META_METRICS_PIPETEMPORARYMETA_MAP);
- }
+ public synchronized void unbindFrom(final AbstractMetricService metricService) {
+ // Keep the pipes registered so that a metric service restart can bind them again.
+ // Synchronized with the (de)registrations, which may remove a registration being unbound.
+ ImmutableSet.copyOf(pipeTemporaryMetaMap.keySet()).forEach(this::removeMetrics);
}
private void removeMetrics(final String pipeID) {
@@ -119,12 +117,11 @@ public class PipeTemporaryMetaInCoordinatorMetrics implements IMetricSet {
pipeNameAndCreationTime[0],
Tag.CREATION_TIME.toString(),
pipeNameAndCreationTime[1]);
- pipeTemporaryMetaMap.remove(pipeID);
}
//////////////////////////// register & deregister (pipe integration) ////////////////////////////
- public void register(final PipeMeta pipeMeta) {
+ public synchronized void register(final PipeMeta pipeMeta) {
final String taskID =
pipeMeta.getStaticMeta().getPipeName() + "_" + pipeMeta.getStaticMeta().getCreationTime();
pipeTemporaryMetaMap.putIfAbsent(
@@ -134,7 +131,7 @@ public class PipeTemporaryMetaInCoordinatorMetrics implements IMetricSet {
}
}
- public void deregister(final String pipeID) {
+ public synchronized void deregister(final String pipeID) {
if (!pipeTemporaryMetaMap.containsKey(pipeID)) {
LOGGER.warn(
ManagerMessages
@@ -145,6 +142,7 @@ public class PipeTemporaryMetaInCoordinatorMetrics implements IMetricSet {
if (Objects.nonNull(metricService)) {
removeMetrics(pipeID);
}
+ pipeTemporaryMetaMap.remove(pipeID);
}
public void handleTemporaryMetaChanges(final Iterable<PipeMeta> pipeMetaList) {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/sink/PipeConfigRegionSinkMetrics.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/sink/PipeConfigRegionSinkMetrics.java
index 0ca45d4c364..3b36730cafc 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/sink/PipeConfigRegionSinkMetrics.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/sink/PipeConfigRegionSinkMetrics.java
@@ -50,7 +50,7 @@ public class PipeConfigRegionSinkMetrics implements IMetricSet {
//////////////////////////// bindTo & unbindFrom (metric framework) ////////////////////////////
@Override
- public void bindTo(final AbstractMetricService metricService) {
+ public synchronized void bindTo(final AbstractMetricService metricService) {
this.metricService = metricService;
ImmutableSet.copyOf(subtaskMap.keySet()).forEach(this::createMetrics);
}
@@ -74,12 +74,11 @@ public class PipeConfigRegionSinkMetrics implements IMetricSet {
}
@Override
- public void unbindFrom(final AbstractMetricService metricService) {
- ImmutableSet.copyOf(subtaskMap.keySet()).forEach(this::deregister);
- if (!subtaskMap.isEmpty()) {
- LOGGER.warn(
- ManagerMessages.FAILED_TO_UNBIND_FROM_PIPE_CONFIG_REGION_CONNECTOR_METRICS_CONNECTOR);
- }
+ public synchronized void unbindFrom(final AbstractMetricService metricService) {
+ // Keep the subtasks registered: they register only once, so a metric service restart
+ // must be able to bind them again.
+ // Synchronized with the (de)registrations, which may remove a registration being unbound.
+ ImmutableSet.copyOf(subtaskMap.keySet()).forEach(this::removeMetrics);
}
private void removeMetrics(final String taskID) {
@@ -101,7 +100,7 @@ public class PipeConfigRegionSinkMetrics implements IMetricSet {
//////////////////////////// register & deregister (pipe integration) ////////////////////////////
- public void register(final PipeConfigNodeSubtask pipeConfigNodeSubtask) {
+ public synchronized void register(final PipeConfigNodeSubtask pipeConfigNodeSubtask) {
final String taskID = pipeConfigNodeSubtask.getTaskID();
subtaskMap.putIfAbsent(taskID, pipeConfigNodeSubtask);
if (Objects.nonNull(metricService)) {
@@ -109,7 +108,7 @@ public class PipeConfigRegionSinkMetrics implements IMetricSet {
}
}
- public void deregister(final String taskID) {
+ public synchronized void deregister(final String taskID) {
if (!subtaskMap.containsKey(taskID)) {
LOGGER.warn(ManagerMessages.FAILED_TO_DEREGISTER_PIPE_CONFIG_REGION_CONNECTOR, taskID);
return;
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/source/PipeConfigRegionSourceMetrics.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/source/PipeConfigRegionSourceMetrics.java
index 428b4b40263..44dcdf2c522 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/source/PipeConfigRegionSourceMetrics.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/source/PipeConfigRegionSourceMetrics.java
@@ -47,7 +47,7 @@ public class PipeConfigRegionSourceMetrics implements IMetricSet {
//////////////////////////// bindTo & unbindFrom (metric framework) ////////////////////////////
@Override
- public void bindTo(final AbstractMetricService metricService) {
+ public synchronized void bindTo(final AbstractMetricService metricService) {
this.metricService = metricService;
ImmutableSet.copyOf(extractorMap.keySet()).forEach(this::createMetrics);
}
@@ -70,12 +70,11 @@ public class PipeConfigRegionSourceMetrics implements IMetricSet {
}
@Override
- public void unbindFrom(final AbstractMetricService metricService) {
- ImmutableSet.copyOf(extractorMap.keySet()).forEach(this::deregister);
- if (!extractorMap.isEmpty()) {
- LOGGER.warn(
- ManagerMessages.FAILED_TO_UNBIND_FROM_PIPE_CONFIG_REGION_EXTRACTOR_METRICS_EXTRACTOR);
- }
+ public synchronized void unbindFrom(final AbstractMetricService metricService) {
+ // Keep the extractors registered: they register only once, so a metric service restart
+ // must be able to bind them again.
+ // Synchronized with the (de)registrations, which may remove a registration being unbound.
+ ImmutableSet.copyOf(extractorMap.keySet()).forEach(this::removeMetrics);
}
private void removeMetrics(final String taskID) {
@@ -96,7 +95,7 @@ public class PipeConfigRegionSourceMetrics implements IMetricSet {
//////////////////////////// pipe integration ////////////////////////////
- public void register(final IoTDBConfigRegionSource extractor) {
+ public synchronized void register(final IoTDBConfigRegionSource extractor) {
final String taskID = extractor.getTaskID();
extractorMap.putIfAbsent(taskID, extractor);
if (Objects.nonNull(metricService)) {
@@ -104,7 +103,7 @@ public class PipeConfigRegionSourceMetrics implements IMetricSet {
}
}
- public void deregister(final String taskID) {
+ public synchronized void deregister(final String taskID) {
if (!extractorMap.containsKey(taskID)) {
LOGGER.warn(ManagerMessages.FAILED_TO_DEREGISTER_PIPE_CONFIG_REGION_EXTRACTOR, taskID);
return;
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/metric/PipeConfigNodeMetricsRestartTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/metric/PipeConfigNodeMetricsRestartTest.java
new file mode 100644
index 00000000000..0bd5e52aff1
--- /dev/null
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/metric/PipeConfigNodeMetricsRestartTest.java
@@ -0,0 +1,184 @@
+/*
+ * 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.iotdb.confignode.manager.pipe.metric;
+
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTemporaryMetaInCoordinator;
+import org.apache.iotdb.commons.service.metric.MetricService;
+import org.apache.iotdb.commons.service.metric.enums.Metric;
+import org.apache.iotdb.confignode.manager.pipe.agent.task.PipeConfigNodeSubtask;
+import org.apache.iotdb.confignode.manager.pipe.metric.overview.PipeConfigNodeRemainingTimeMetrics;
+import org.apache.iotdb.confignode.manager.pipe.metric.overview.PipeTemporaryMetaInCoordinatorMetrics;
+import org.apache.iotdb.confignode.manager.pipe.metric.sink.PipeConfigRegionSinkMetrics;
+import org.apache.iotdb.confignode.manager.pipe.metric.source.PipeConfigRegionSourceMetrics;
+import org.apache.iotdb.confignode.manager.pipe.source.IoTDBConfigRegionSource;
+import org.apache.iotdb.metrics.config.MetricConfig;
+import org.apache.iotdb.metrics.config.MetricConfigDescriptor;
+import org.apache.iotdb.metrics.metricsets.IMetricSet;
+import org.apache.iotdb.metrics.type.IMetric;
+import org.apache.iotdb.metrics.type.Rate;
+import org.apache.iotdb.metrics.utils.MetricInfo;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.when;
+
+/** Pipe metrics must survive a metric service restart, which drops and rebinds all metrics. */
+public class PipeConfigNodeMetricsRestartTest {
+
+ private static final String PIPE = "pipe";
+ private static final long CREATION_TIME = 1L;
+ private static final String PIPE_ID = PIPE + "_" + CREATION_TIME;
+ private static final String TASK_ID = "task";
+
+ private final MetricService service = MetricService.getInstance();
+ private final MetricConfig config = MetricConfigDescriptor.getInstance().getMetricConfig();
+ private final List<IMetricSet> boundMetricSets = new ArrayList<>();
+ private MetricLevel originalLevel;
+ private String originalReporters;
+
+ @Before
+ public void setUp() {
+ originalLevel = config.getMetricLevel();
+ originalReporters =
+ config.getMetricReporterList().stream().map(Enum::name).collect(Collectors.joining(","));
+ config.setMetricLevel(MetricLevel.IMPORTANT);
+ config.setMetricReporterList("");
+ service.startService();
+ }
+
+ @After
+ public void tearDown() {
+ boundMetricSets.forEach(service::removeMetricSet);
+ service.stopService();
+ config.setMetricLevel(originalLevel);
+ config.setMetricReporterList(originalReporters);
+ }
+
+ @Test
+ public void testConfigRegionSourceMetrics() {
+ final PipeConfigRegionSourceMetrics metrics = PipeConfigRegionSourceMetrics.getInstance();
+ bind(metrics);
+ metrics.register(mockConfigRegionSource());
+ try {
+ service.restartService();
+ assertEquals(1, count(Metric.UNTRANSFERRED_CONFIG_COUNT));
+ } finally {
+ metrics.deregister(TASK_ID);
+ }
+ assertEquals(0, count(Metric.UNTRANSFERRED_CONFIG_COUNT));
+ }
+
+ @Test
+ public void testConfigRegionSinkMetrics() {
+ final PipeConfigRegionSinkMetrics metrics = PipeConfigRegionSinkMetrics.getInstance();
+ bind(metrics);
+ final PipeConfigNodeSubtask subtask = Mockito.mock(PipeConfigNodeSubtask.class);
+ when(subtask.getTaskID()).thenReturn(TASK_ID);
+ when(subtask.getPipeName()).thenReturn(PIPE);
+ when(subtask.getCreationTime()).thenReturn(CREATION_TIME);
+ metrics.register(subtask);
+ try {
+ service.restartService();
+ metrics.markConfigEvent(TASK_ID);
+ assertEquals(1, ((Rate) get(Metric.PIPE_CONNECTOR_CONFIG_TRANSFER)).getCount());
+ } finally {
+ metrics.deregister(TASK_ID);
+ }
+ assertEquals(0, count(Metric.PIPE_CONNECTOR_CONFIG_TRANSFER));
+ }
+
+ @Test
+ public void testRemainingTimeMetrics() {
+ final PipeConfigNodeRemainingTimeMetrics metrics =
+ PipeConfigNodeRemainingTimeMetrics.getInstance();
+ bind(metrics);
+ metrics.register(mockConfigRegionSource());
+ try {
+ service.restartService();
+ assertEquals(1, count(Metric.PIPE_CONFIGNODE_REMAINING_TIME));
+ } finally {
+ metrics.deregister(PIPE_ID);
+ }
+ assertEquals(0, count(Metric.PIPE_CONFIGNODE_REMAINING_TIME));
+ }
+
+ @Test
+ public void testTemporaryMetaInCoordinatorMetrics() {
+ final PipeTemporaryMetaInCoordinatorMetrics metrics =
+ PipeTemporaryMetaInCoordinatorMetrics.getInstance();
+ bind(metrics);
+ final PipeStaticMeta staticMeta = Mockito.mock(PipeStaticMeta.class);
+ when(staticMeta.getPipeName()).thenReturn(PIPE);
+ when(staticMeta.getCreationTime()).thenReturn(CREATION_TIME);
+ final PipeMeta pipeMeta = Mockito.mock(PipeMeta.class);
+ when(pipeMeta.getStaticMeta()).thenReturn(staticMeta);
+ final PipeTemporaryMetaInCoordinator temporaryMeta =
+ Mockito.mock(PipeTemporaryMetaInCoordinator.class);
+ when(pipeMeta.getTemporaryMeta()).thenReturn(temporaryMeta);
+ metrics.register(pipeMeta);
+ try {
+ service.restartService();
+ assertEquals(1, count(Metric.PIPE_GLOBAL_REMAINING_EVENT_COUNT));
+ } finally {
+ metrics.deregister(PIPE_ID);
+ }
+ assertEquals(0, count(Metric.PIPE_GLOBAL_REMAINING_EVENT_COUNT));
+ }
+
+ private void bind(final IMetricSet metricSet) {
+ service.addMetricSet(metricSet);
+ boundMetricSets.add(metricSet);
+ }
+
+ private static IoTDBConfigRegionSource mockConfigRegionSource() {
+ final IoTDBConfigRegionSource source = Mockito.mock(IoTDBConfigRegionSource.class);
+ when(source.getTaskID()).thenReturn(TASK_ID);
+ when(source.getPipeName()).thenReturn(PIPE);
+ when(source.getCreationTime()).thenReturn(CREATION_TIME);
+ return source;
+ }
+
+ private long count(final Metric metric) {
+ return service.getAllMetrics().keySet().stream()
+ .filter(info -> metric.toString().equals(info.getName()))
+ .count();
+ }
+
+ private IMetric get(final Metric metric) {
+ for (final Map.Entry<MetricInfo, IMetric> entry : service.getAllMetrics().entrySet()) {
+ if (metric.toString().equals(entry.getKey().getName())) {
+ return entry.getValue();
+ }
+ }
+ throw new AssertionError(metric + " is not registered");
+ }
+}
diff --git a/iotdb-core/consensus/pom.xml b/iotdb-core/consensus/pom.xml
index 5d4b785ef73..7130af74828 100644
--- a/iotdb-core/consensus/pom.xml
+++ b/iotdb-core/consensus/pom.xml
@@ -124,6 +124,11 @@
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcherThreadMetrics.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcherThreadMetrics.java
index f30f8c23094..9a5dd6c4b28 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcherThreadMetrics.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcherThreadMetrics.java
@@ -28,7 +28,20 @@ import org.apache.iotdb.metrics.type.Timer;
import org.apache.iotdb.metrics.utils.MetricLevel;
import org.apache.iotdb.metrics.utils.MetricType;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
public class LogDispatcherThreadMetrics implements IMetricSet {
+ // The stage timers are shared by the dispatcher threads of all peers of a region, so only the
+ // last of them to unbind may remove the timers, otherwise the others keep recording into timers
+ // that are no longer exported, e.g. after the metric service restarts. A metric set may be bound
+ // twice without being unbound in between, e.g. when it is added before the metric service
+ // starts, so track the owners rather than count the bindings.
+ private static final Map<String, Set<LogDispatcherThreadMetrics>> REGION_TO_STAGE_TIMER_OWNERS =
+ new HashMap<>();
+
private final LogDispatcher.LogDispatcherThread logDispatcherThread;
private final String peerGroupId;
@@ -91,47 +104,62 @@ public class LogDispatcherThreadMetrics implements IMetricSet {
}
private void bindStageTimer(AbstractMetricService metricService) {
- constructBatchTimer =
- metricService.getOrCreateTimer(
- Metric.IOT_SEND_LOG.toString(),
- MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- Metric.IOT_CONSENSUS.toString(),
- Tag.STAGE.toString(),
- "constructBatch",
- Tag.REGION.toString(),
- peerGroupId);
- syncLogTimePerRequestTimer =
- metricService.getOrCreateTimer(
- Metric.IOT_SEND_LOG.toString(),
- MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- Metric.IOT_CONSENSUS.toString(),
- Tag.STAGE.toString(),
- "syncLogTimePerRequest",
- Tag.REGION.toString(),
- peerGroupId);
+ synchronized (REGION_TO_STAGE_TIMER_OWNERS) {
+ REGION_TO_STAGE_TIMER_OWNERS
+ .computeIfAbsent(peerGroupId, region -> new HashSet<>())
+ .add(this);
+ constructBatchTimer =
+ metricService.getOrCreateTimer(
+ Metric.IOT_SEND_LOG.toString(),
+ MetricLevel.IMPORTANT,
+ Tag.NAME.toString(),
+ Metric.IOT_CONSENSUS.toString(),
+ Tag.STAGE.toString(),
+ "constructBatch",
+ Tag.REGION.toString(),
+ peerGroupId);
+ syncLogTimePerRequestTimer =
+ metricService.getOrCreateTimer(
+ Metric.IOT_SEND_LOG.toString(),
+ MetricLevel.IMPORTANT,
+ Tag.NAME.toString(),
+ Metric.IOT_CONSENSUS.toString(),
+ Tag.STAGE.toString(),
+ "syncLogTimePerRequest",
+ Tag.REGION.toString(),
+ peerGroupId);
+ }
}
private void unbindStageTimer(AbstractMetricService metricService) {
- metricService.remove(
- MetricType.TIMER,
- Metric.IOT_SEND_LOG.toString(),
- Tag.NAME.toString(),
- Metric.IOT_CONSENSUS.toString(),
- Tag.STAGE.toString(),
- "constructBatch",
- Tag.REGION.toString(),
- peerGroupId);
- metricService.remove(
- MetricType.TIMER,
- Metric.IOT_SEND_LOG.toString(),
- Tag.NAME.toString(),
- Metric.IOT_CONSENSUS.toString(),
- Tag.STAGE.toString(),
- "syncLogTimePerRequest",
- Tag.REGION.toString(),
- peerGroupId);
+ synchronized (REGION_TO_STAGE_TIMER_OWNERS) {
+ final Set<LogDispatcherThreadMetrics> owners = REGION_TO_STAGE_TIMER_OWNERS.get(peerGroupId);
+ if (owners != null) {
+ owners.remove(this);
+ if (!owners.isEmpty()) {
+ return;
+ }
+ REGION_TO_STAGE_TIMER_OWNERS.remove(peerGroupId);
+ }
+ metricService.remove(
+ MetricType.TIMER,
+ Metric.IOT_SEND_LOG.toString(),
+ Tag.NAME.toString(),
+ Metric.IOT_CONSENSUS.toString(),
+ Tag.STAGE.toString(),
+ "constructBatch",
+ Tag.REGION.toString(),
+ peerGroupId);
+ metricService.remove(
+ MetricType.TIMER,
+ Metric.IOT_SEND_LOG.toString(),
+ Tag.NAME.toString(),
+ Metric.IOT_CONSENSUS.toString(),
+ Tag.STAGE.toString(),
+ "syncLogTimePerRequest",
+ Tag.REGION.toString(),
+ peerGroupId);
+ }
}
@Override
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2ServerImpl.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2ServerImpl.java
index 415326cbf52..52df1138161 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2ServerImpl.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2ServerImpl.java
@@ -54,6 +54,7 @@ import org.apache.iotdb.consensus.iotconsensusv2.thrift.TWaitReleaseAllRegionRel
import org.apache.iotdb.consensus.pipe.consensuspipe.ConsensusPipeName;
import org.apache.iotdb.consensus.pipe.consensuspipe.ReplicateProgressManager;
import org.apache.iotdb.consensus.pipe.metric.IoTConsensusV2ServerMetrics;
+import org.apache.iotdb.consensus.pipe.metric.IoTConsensusV2SyncLagManager;
import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.rpc.RpcUtils;
@@ -122,12 +123,16 @@ public class IoTConsensusV2ServerImpl {
public synchronized void stop() {
MetricService.getInstance().removeMetricSet(this.iotConsensusV2ServerMetrics);
+ // Release it here rather than when unbinding the metrics, which also happens when the metric
+ // service restarts
+ IoTConsensusV2SyncLagManager.release(consensusGroupId);
stateMachine.stop();
isStarted.set(false);
}
public synchronized void clear() {
MetricService.getInstance().removeMetricSet(this.iotConsensusV2ServerMetrics);
+ IoTConsensusV2SyncLagManager.release(consensusGroupId);
peerManager.clear();
stateMachine.stop();
isStarted.set(false);
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/metric/IoTConsensusV2ServerMetrics.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/metric/IoTConsensusV2ServerMetrics.java
index 2c10d434b51..0f39e14206d 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/metric/IoTConsensusV2ServerMetrics.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/metric/IoTConsensusV2ServerMetrics.java
@@ -68,9 +68,6 @@ public class IoTConsensusV2ServerMetrics implements IMetricSet {
unbindAutoGauge(metricService);
unbindGauge(metricService);
unbindStageTimer(metricService);
-
- // release corresponding resource
- IoTConsensusV2SyncLagManager.release(impl.getConsensusGroupId());
}
public void bindGauge(AbstractMetricService metricService) {
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/CounterProxy.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/CounterProxy.java
index 33d2470af20..04c8441eba3 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/CounterProxy.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/CounterProxy.java
@@ -26,13 +26,17 @@ import org.apache.ratis.metrics.LongCounter;
/** CounterProxy will route Ratis' internal counter metrics to our IoTDB {@link Counter} */
public class CounterProxy implements LongCounter {
- /** IoTDB Counter. */
- private final Counter counter;
+ /** IoTDB Counter, replaced when the metric service restarts. */
+ private volatile Counter counter;
CounterProxy(Counter counter) {
this.counter = counter;
}
+ void setCounter(Counter counter) {
+ this.counter = counter;
+ }
+
@Override
public void inc() {
inc(1L);
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/IoTDBMetricRegistry.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/IoTDBMetricRegistry.java
index 41619b28e63..2288e37ef6f 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/IoTDBMetricRegistry.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/IoTDBMetricRegistry.java
@@ -43,7 +43,7 @@ public class IoTDBMetricRegistry implements RatisMetricRegistry {
private final Map<String, String> metricNameCache = new ConcurrentHashMap<>();
private final Map<String, CounterProxy> counterCache = new ConcurrentHashMap<>();
private final Map<String, TimerProxy> timerCache = new ConcurrentHashMap<>();
- private final Map<String, Boolean> gaugeCache = new ConcurrentHashMap<>();
+ private final Map<String, GaugeProxy<?>> gaugeCache = new ConcurrentHashMap<>();
/** Time taken to flush log. */
public static final String RAFT_LOG_FLUSH_TIME = "flushTime";
@@ -149,7 +149,11 @@ public class IoTDBMetricRegistry implements RatisMetricRegistry {
metricService.remove(MetricType.AUTO_GAUGE, getMetricName(name));
} catch (IllegalArgumentException ignored) {
}
-
+ // Forget the metric so that rebinding does not restore it and Ratis can create it again.
+ final String fullName = getMetricName(name);
+ counterCache.remove(fullName);
+ timerCache.remove(fullName);
+ gaugeCache.remove(fullName);
return true;
}
@@ -161,7 +165,7 @@ public class IoTDBMetricRegistry implements RatisMetricRegistry {
fn -> {
final GaugeProxy<T> gauge = new GaugeProxy<>(supplier);
metricService.createAutoGauge(fn, getMetricLevel(fn), gauge, GaugeProxy::getDoubleValue);
- return true;
+ return gauge;
});
}
@@ -170,10 +174,32 @@ public class IoTDBMetricRegistry implements RatisMetricRegistry {
return info;
}
- void removeAll() {
+ /**
+ * Registers all metrics again. Ratis creates its metrics once and keeps using them, so they must
+ * be restored after the metric service drops them, e.g. when it restarts.
+ */
+ void bindMetrics() {
+ counterCache.forEach(
+ (name, counter) ->
+ counter.setCounter(metricService.getOrCreateCounter(name, getMetricLevel(name))));
+ timerCache.forEach(
+ (name, timer) ->
+ timer.setTimer(metricService.getOrCreateTimer(name, getMetricLevel(name))));
+ gaugeCache.forEach(
+ (name, gauge) ->
+ metricService.createAutoGauge(
+ name, getMetricLevel(name), gauge, GaugeProxy::getDoubleValue));
+ }
+
+ /** Removes all metrics from the metric service but keeps them for {@link #bindMetrics()}. */
+ void unbindMetrics() {
counterCache.forEach((name, counter) -> metricService.remove(MetricType.COUNTER, name));
gaugeCache.forEach((name, gauge) -> metricService.remove(MetricType.AUTO_GAUGE, name));
timerCache.forEach((name, timer) -> metricService.remove(MetricType.TIMER, name));
+ }
+
+ void removeAll() {
+ unbindMetrics();
metricNameCache.clear();
counterCache.clear();
gaugeCache.clear();
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/MetricRegistryManager.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/MetricRegistryManager.java
index 2a8ca243295..104b8faaf7e 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/MetricRegistryManager.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/MetricRegistryManager.java
@@ -52,6 +52,20 @@ public class MetricRegistryManager extends MetricRegistries {
this.registries.clear();
}
+ /** Registers the metrics of all registries again, e.g. after the metric service restarts. */
+ void bindMetrics() {
+ registries.values().stream()
+ .map(IoTDBMetricRegistry.class::cast)
+ .forEach(IoTDBMetricRegistry::bindMetrics);
+ }
+
+ /** Removes the metrics of all registries from the metric service without dropping them. */
+ void unbindMetrics() {
+ registries.values().stream()
+ .map(IoTDBMetricRegistry.class::cast)
+ .forEach(IoTDBMetricRegistry::unbindMetrics);
+ }
+
@Override
public RatisMetricRegistry create(MetricRegistryInfo metricRegistryInfo) {
return registries.put(
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/RatisMetricSet.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/RatisMetricSet.java
index c13efd591dc..9903cff2813 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/RatisMetricSet.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/RatisMetricSet.java
@@ -100,6 +100,9 @@ public class RatisMetricSet implements IMetricSet {
@Override
public void bindTo(AbstractMetricService metricService) {
manager = MetricRegistries.global();
+ if (manager instanceof MetricRegistryManager) {
+ ((MetricRegistryManager) manager).bindMetrics();
+ }
for (MetricInfo metricInfo : metricInfoMap.values()) {
metricService.getOrCreateTimer(
metricInfo.getName(), MetricLevel.CORE, metricInfo.getTagsInArray());
@@ -108,7 +111,10 @@ public class RatisMetricSet implements IMetricSet {
@Override
public void unbindFrom(AbstractMetricService metricService) {
- manager.clear();
+ // Keep the registries: Ratis still holds their metrics and never creates them again.
+ if (manager instanceof MetricRegistryManager) {
+ ((MetricRegistryManager) manager).unbindMetrics();
+ }
for (MetricInfo metricInfo : metricInfoMap.values()) {
metricService.remove(MetricType.TIMER, metricInfo.getName(), metricInfo.getTagsInArray());
}
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/TimerProxy.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/TimerProxy.java
index 07a01abe04a..8aff57a9bc7 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/TimerProxy.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/metrics/TimerProxy.java
@@ -46,12 +46,17 @@ public class TimerProxy implements Timekeeper {
}
}
- private final Timer timer;
+ /** IoTDB Timer, replaced when the metric service restarts. */
+ private volatile Timer timer;
TimerProxy(Timer timer) {
this.timer = timer;
}
+ void setTimer(Timer timer) {
+ this.timer = timer;
+ }
+
@Override
public Context time() {
return new TimerContext(timer);
diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcherThreadMetricsTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcherThreadMetricsTest.java
new file mode 100644
index 00000000000..4da003936a9
--- /dev/null
+++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcherThreadMetricsTest.java
@@ -0,0 +1,146 @@
+/*
+ * 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.iotdb.consensus.iot.logdispatcher;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.consensus.DataRegionId;
+import org.apache.iotdb.commons.service.metric.MetricService;
+import org.apache.iotdb.commons.service.metric.enums.Metric;
+import org.apache.iotdb.commons.service.metric.enums.Tag;
+import org.apache.iotdb.consensus.common.Peer;
+import org.apache.iotdb.metrics.config.MetricConfig;
+import org.apache.iotdb.metrics.config.MetricConfigDescriptor;
+import org.apache.iotdb.metrics.type.IMetric;
+import org.apache.iotdb.metrics.type.Timer;
+import org.apache.iotdb.metrics.utils.MetricInfo;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+public class LogDispatcherThreadMetricsTest {
+
+ private static final DataRegionId REGION = new DataRegionId(1);
+
+ private final MetricService service = MetricService.getInstance();
+ private final MetricConfig config = MetricConfigDescriptor.getInstance().getMetricConfig();
+ private MetricLevel originalLevel;
+ private String originalReporters;
+
+ @Before
+ public void setUp() {
+ originalLevel = config.getMetricLevel();
+ originalReporters =
+ config.getMetricReporterList().stream().map(Enum::name).collect(Collectors.joining(","));
+ config.setMetricLevel(MetricLevel.IMPORTANT);
+ config.setMetricReporterList("");
+ service.startService();
+ }
+
+ @After
+ public void tearDown() {
+ service.stopService();
+ config.setMetricLevel(originalLevel);
+ config.setMetricReporterList(originalReporters);
+ }
+
+ /** The dispatcher threads of all peers of a region record into the same stage timers. */
+ @Test
+ public void testStageTimersAreSharedByThePeersOfARegion() {
+ final LogDispatcherThreadMetrics first = new LogDispatcherThreadMetrics(mockThread(2));
+ final LogDispatcherThreadMetrics second = new LogDispatcherThreadMetrics(mockThread(3));
+ service.addMetricSet(first);
+ service.addMetricSet(second);
+ try {
+ service.restartService();
+ first.recordConstructBatchTime(1);
+ second.recordConstructBatchTime(1);
+ assertEquals(2, getConstructBatchTimer().getCount());
+
+ // A peer leaving the region must not remove the timers still used by the other peers
+ service.removeMetricSet(second);
+ first.recordConstructBatchTime(1);
+ assertEquals(3, getConstructBatchTimer().getCount());
+ } finally {
+ service.removeMetricSet(first);
+ service.removeMetricSet(second);
+ }
+ assertNull(findConstructBatchTimer());
+ }
+
+ /**
+ * The dispatchers recovered at startup register their metrics before the metric service starts,
+ * which binds them again, so a metric set may be bound twice without being unbound in between.
+ */
+ @Test
+ public void testStageTimersAreRemovedAfterBindingTwice() {
+ service.stopService();
+ final LogDispatcherThreadMetrics first = new LogDispatcherThreadMetrics(mockThread(2));
+ final LogDispatcherThreadMetrics second = new LogDispatcherThreadMetrics(mockThread(3));
+ service.addMetricSet(first);
+ service.addMetricSet(second);
+ service.startService();
+ try {
+ first.recordConstructBatchTime(1);
+ second.recordConstructBatchTime(1);
+ assertEquals(2, getConstructBatchTimer().getCount());
+ } finally {
+ service.removeMetricSet(first);
+ service.removeMetricSet(second);
+ }
+ assertNull(findConstructBatchTimer());
+ }
+
+ private static LogDispatcher.LogDispatcherThread mockThread(final int peerNodeId) {
+ final LogDispatcher.LogDispatcherThread thread =
+ Mockito.mock(LogDispatcher.LogDispatcherThread.class);
+ Mockito.when(thread.getPeer())
+ .thenReturn(new Peer(REGION, peerNodeId, new TEndPoint("127.0.0.1", 10000 + peerNodeId)));
+ return thread;
+ }
+
+ private Timer getConstructBatchTimer() {
+ final IMetric timer = findConstructBatchTimer();
+ if (timer == null) {
+ throw new AssertionError("The constructBatch timer of " + REGION + " is not registered");
+ }
+ return (Timer) timer;
+ }
+
+ private IMetric findConstructBatchTimer() {
+ for (final Map.Entry<MetricInfo, IMetric> entry : service.getAllMetrics().entrySet()) {
+ final Map<String, String> tags = entry.getKey().getTags();
+ if (Metric.IOT_SEND_LOG.toString().equals(entry.getKey().getName())
+ && "constructBatch".equals(tags.get(Tag.STAGE.toString()))
+ && REGION.toString().equals(tags.get(Tag.REGION.toString()))) {
+ return entry.getValue();
+ }
+ }
+ return null;
+ }
+}
diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/pipe/metric/IoTConsensusV2ServerMetricsTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/pipe/metric/IoTConsensusV2ServerMetricsTest.java
new file mode 100644
index 00000000000..86ef55e58a8
--- /dev/null
+++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/pipe/metric/IoTConsensusV2ServerMetricsTest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.iotdb.consensus.pipe.metric;
+
+import org.apache.iotdb.commons.consensus.DataRegionId;
+import org.apache.iotdb.commons.service.metric.MetricService;
+import org.apache.iotdb.commons.service.metric.enums.Metric;
+import org.apache.iotdb.commons.service.metric.enums.Tag;
+import org.apache.iotdb.consensus.pipe.IoTConsensusV2ServerImpl;
+import org.apache.iotdb.consensus.pipe.consensuspipe.ConsensusPipeName;
+import org.apache.iotdb.consensus.pipe.consensuspipe.ConsensusPipeSink;
+import org.apache.iotdb.metrics.config.MetricConfig;
+import org.apache.iotdb.metrics.config.MetricConfigDescriptor;
+import org.apache.iotdb.metrics.type.AutoGauge;
+import org.apache.iotdb.metrics.type.IMetric;
+import org.apache.iotdb.metrics.utils.MetricInfo;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+public class IoTConsensusV2ServerMetricsTest {
+
+ private static final DataRegionId GROUP_ID = new DataRegionId(1);
+
+ private final MetricService service = MetricService.getInstance();
+ private final MetricConfig config = MetricConfigDescriptor.getInstance().getMetricConfig();
+ private MetricLevel originalLevel;
+ private String originalReporters;
+
+ @Before
+ public void setUp() {
+ originalLevel = config.getMetricLevel();
+ originalReporters =
+ config.getMetricReporterList().stream().map(Enum::name).collect(Collectors.joining(","));
+ config.setMetricLevel(MetricLevel.IMPORTANT);
+ config.setMetricReporterList("");
+ service.startService();
+ IoTConsensusV2SyncLagManager.build();
+ }
+
+ @After
+ public void tearDown() {
+ IoTConsensusV2SyncLagManager.release(GROUP_ID.toString());
+ service.stopService();
+ config.setMetricLevel(originalLevel);
+ config.setMetricReporterList(originalReporters);
+ }
+
+ @Test
+ public void testSyncLagSurvivesMetricServiceRestart() {
+ final IoTConsensusV2ServerImpl impl = Mockito.mock(IoTConsensusV2ServerImpl.class);
+ Mockito.when(impl.getConsensusGroupId()).thenReturn(GROUP_ID.toString());
+ final ConsensusPipeSink sink = Mockito.mock(ConsensusPipeSink.class);
+ Mockito.when(sink.getLeaderReplicateProgress()).thenReturn(10L);
+ Mockito.when(sink.getFollowerApplyProgress()).thenReturn(3L);
+ IoTConsensusV2SyncLagManager.getInstance(GROUP_ID.toString())
+ .addConsensusPipeConnector(new ConsensusPipeName(GROUP_ID, 1, 2), sink);
+
+ final IoTConsensusV2ServerMetrics metrics = new IoTConsensusV2ServerMetrics(impl);
+ service.addMetricSet(metrics);
+ try {
+ assertEquals(7, getSyncLag(), 0);
+ service.restartService();
+ assertEquals(7, getSyncLag(), 0);
+ } finally {
+ service.removeMetricSet(metrics);
+ }
+ }
+
+ private double getSyncLag() {
+ AutoGauge gauge = null;
+ for (final Map.Entry<MetricInfo, IMetric> entry : service.getAllMetrics().entrySet()) {
+ if (Metric.IOT_CONSENSUS_V2.toString().equals(entry.getKey().getName())
+ && "syncLag".equals(entry.getKey().getTags().get(Tag.TYPE.toString()))) {
+ gauge = (AutoGauge) entry.getValue();
+ }
+ }
+ assertNotNull(gauge);
+ return gauge.getValue();
+ }
+}
diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/metrics/RatisMetricSetTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/metrics/RatisMetricSetTest.java
new file mode 100644
index 00000000000..4fcf57f0ca6
--- /dev/null
+++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/metrics/RatisMetricSetTest.java
@@ -0,0 +1,134 @@
+/*
+ * 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.iotdb.consensus.ratis.metrics;
+
+import org.apache.iotdb.commons.service.metric.MetricService;
+import org.apache.iotdb.metrics.config.MetricConfig;
+import org.apache.iotdb.metrics.config.MetricConfigDescriptor;
+import org.apache.iotdb.metrics.type.AutoGauge;
+import org.apache.iotdb.metrics.type.Counter;
+import org.apache.iotdb.metrics.type.IMetric;
+import org.apache.iotdb.metrics.type.Timer;
+import org.apache.iotdb.metrics.utils.MetricInfo;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+
+import org.apache.ratis.metrics.LongCounter;
+import org.apache.ratis.metrics.MetricRegistries;
+import org.apache.ratis.metrics.MetricRegistryInfo;
+import org.apache.ratis.metrics.RatisMetricRegistry;
+import org.apache.ratis.metrics.Timekeeper;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+public class RatisMetricSetTest {
+
+ private static final String COMPONENT = "ratis_metric_set_test";
+
+ private final MetricRegistryInfo info =
+ new MetricRegistryInfo("test", "ratis", COMPONENT, "RatisMetricSetTest");
+ private final RatisMetricSet metricSet = new RatisMetricSet();
+ private final MetricService service = MetricService.getInstance();
+ private final MetricConfig config = MetricConfigDescriptor.getInstance().getMetricConfig();
+ private MetricLevel originalLevel;
+ private String originalReporters;
+ private RatisMetricRegistry registry;
+
+ @Before
+ public void setUp() {
+ originalLevel = config.getMetricLevel();
+ originalReporters =
+ config.getMetricReporterList().stream().map(Enum::name).collect(Collectors.joining(","));
+ config.setMetricLevel(MetricLevel.IMPORTANT);
+ config.setMetricReporterList("");
+ service.startService();
+ service.addMetricSet(metricSet);
+ registry = MetricRegistries.global().create(info);
+ }
+
+ @After
+ public void tearDown() {
+ MetricRegistries.global().remove(info);
+ service.removeMetricSet(metricSet);
+ service.stopService();
+ config.setMetricLevel(originalLevel);
+ config.setMetricReporterList(originalReporters);
+ }
+
+ @Test
+ public void testRatisMetricsSurviveMetricServiceRestart() {
+ // Ratis creates these metrics once and keeps using the returned instances.
+ final LongCounter counter = registry.counter("counter");
+ final Timekeeper timer = registry.timer("timer");
+ registry.gauge("gauge", () -> () -> 42);
+ counter.inc();
+
+ for (final MetricLevel level :
+ new MetricLevel[] {MetricLevel.ALL, MetricLevel.OFF, MetricLevel.IMPORTANT}) {
+ config.setMetricLevel(level);
+ service.restartService();
+ if (level == MetricLevel.OFF) {
+ assertNull(getMetric("counter"));
+ assertNull(getMetric("timer"));
+ assertNull(getMetric("gauge"));
+ } else {
+ assertNotNull(getMetric("counter"));
+ assertNotNull(getMetric("timer"));
+ assertEquals(42, ((AutoGauge) getMetric("gauge")).getValue(), 0);
+ }
+ }
+
+ counter.inc();
+ timer.time().stop();
+ assertEquals(1, ((Counter) getMetric("counter")).getCount());
+ assertEquals(1, ((Timer) getMetric("timer")).getCount());
+ }
+
+ @Test
+ public void testRemovedGaugeIsNotRestoredAndCanBeAddedAgain() {
+ registry.gauge("gauge", () -> () -> 1);
+ registry.remove("gauge");
+ assertNull(getMetric("gauge"));
+ service.restartService();
+ assertNull(getMetric("gauge"));
+
+ // Ratis adds a gauge again with a new supplier, e.g. when a server becomes leader again.
+ registry.gauge("gauge", () -> () -> 2);
+ assertEquals(2, ((AutoGauge) getMetric("gauge")).getValue(), 0);
+ service.restartService();
+ assertEquals(2, ((AutoGauge) getMetric("gauge")).getValue(), 0);
+ }
+
+ private IMetric getMetric(final String name) {
+ for (final Map.Entry<MetricInfo, IMetric> entry : service.getAllMetrics().entrySet()) {
+ if (entry.getKey().getName().endsWith(COMPONENT + name)) {
+ return entry.getValue();
+ }
+ }
+ return null;
+ }
+}
diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 59f58407dc4..8fd63338892 100644
--- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -1524,25 +1524,6 @@ public final class DataNodePipeMessages {
public static final String FAILED_TO_SET_RECENT_PROCESSED_TSFILE_EPOCH =
"Failed to set recent processed tsfile epoch state, PipeRealtimeDataRegionExtractor({}) "
+ "does not exist";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_ASSIGNER_METRICS =
- "Failed to unbind from pipe assigner metrics, assigner map not empty";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_DATA_REGION =
- "Failed to unbind from pipe data region sink metrics, sink map not empty";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_EXTRACTOR_METRICS =
- "Failed to unbind from pipe extractor metrics, extractor map not empty";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_PROCESSOR_METRICS =
- "Failed to unbind from pipe processor metrics, processor map not empty";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_REMAINING_EVENT =
- "Failed to unbind from pipe remaining event and time metrics, "
- + "RemainingEventAndTimeOperator map not empty";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_SCHEMA_REGION =
- "Failed to unbind from pipe schema region connector metrics, connector map not empty";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_SCHEMA_REGION_1 =
- "Failed to unbind from pipe schema region extractor metrics, extractor map not empty";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_SCHEMA_REGION_2 =
- "Failed to unbind from pipe schema region listener metrics, listening queue map not empty";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_TSFILE_TO =
- "Failed to unbind from pipe tsfile to tablets metrics, pipe map is not empty, pipe: {}";
// ---------------------------------------------------------------------------
// pipe – PipeRow
@@ -1872,16 +1853,11 @@ public final class DataNodePipeMessages {
"Subscription: consumer {} poll excessive payload {} for event {} (outdated: {}) with "
+ "request: {}, something unexpected happened with parameter configuration or payload "
+ "control...";
- public static final String PIPE_LOG_FAILED_TO_UNBIND_FROM_SUBSCRIPTION_PREFETCHING_QUEUE_METRICS_6614388C =
- "Failed to unbind from subscription prefetching queue metrics, prefetching queue map not "
- + "empty";
public static final String PIPE_LOG_FAILED_TO_DEREGISTER_SUBSCRIPTION_PREFETCHING_QUEUE_METRICS_F08479A7 =
"Failed to deregister subscription prefetching queue metrics, "
+ "SubscriptionPrefetchingQueue({}) does not exist";
public static final String PIPE_LOG_FAILED_TO_MARK_TRANSFER_EVENT_RATE_SUBSCRIPTIONPREFETCHINGQUEUE_7DEF95B5 =
"Failed to mark transfer event rate, SubscriptionPrefetchingQueue({}) does not exist";
- public static final String PIPE_LOG_FAILED_TO_UNBIND_FROM_CONSENSUS_SUBSCRIPTION_PREFETCHING_A8F920D9 =
- "Failed to unbind from consensus subscription prefetching queue metrics, queue map not empty";
public static final String PIPE_LOG_FAILED_TO_DEREGISTER_CONSENSUS_SUBSCRIPTION_PREFETCHING_8B180091 =
"Failed to deregister consensus subscription prefetching queue metrics, "
+ "ConsensusPrefetchingQueue({}) does not exist";
diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 48ec49342f3..62918788834 100644
--- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -1423,25 +1423,6 @@ public final class DataNodePipeMessages {
public static final String FAILED_TO_SET_RECENT_PROCESSED_TSFILE_EPOCH =
"设置 recent processed tsfile epoch state, PipeRealtimeDataRegionExtractor({}) does not "
+ "exist 失败";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_ASSIGNER_METRICS =
- "解绑 from pipe assigner metrics, assigner map not empty 失败";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_DATA_REGION =
- "解绑 from pipe data region sink metrics, sink map not empty 失败";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_EXTRACTOR_METRICS =
- "解绑 from pipe extractor metrics, extractor map not empty 失败";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_PROCESSOR_METRICS =
- "解绑 from pipe processor metrics, processor map not empty 失败";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_REMAINING_EVENT =
- "解绑 from pipe remaining event and time metrics, RemainingEventAndTimeOperator map not "
- + "empty 失败";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_SCHEMA_REGION =
- "解绑 from pipe schema region connector metrics, connector map not empty 失败";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_SCHEMA_REGION_1 =
- "解绑 from pipe schema region extractor metrics, extractor map not empty 失败";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_SCHEMA_REGION_2 =
- "解绑 from pipe schema region listener metrics, listening queue map not empty 失败";
- public static final String FAILED_TO_UNBIND_FROM_PIPE_TSFILE_TO =
- "解绑 from pipe tsfile to tablets metrics, pipe map is not empty, pipe: {} 失败";
// ---------------------------------------------------------------------------
// pipe – PipeRow
@@ -1740,14 +1721,10 @@ public final class DataNodePipeMessages {
public static final String PIPE_LOG_SUBSCRIPTION_CONSUMER_POLL_EXCESSIVE_PAYLOAD_FOR_EVENT_OUTDATED_2BFF690B =
"Subscription：consumer {} poll 到过大的 payload {}，event={}（outdated：{}），请求：{}，参数配置或 payload "
+ "控制可能出现意外情况……";
- public static final String PIPE_LOG_FAILED_TO_UNBIND_FROM_SUBSCRIPTION_PREFETCHING_QUEUE_METRICS_6614388C =
- "解绑 subscription prefetching queue metrics 失败，prefetching queue map 非空";
public static final String PIPE_LOG_FAILED_TO_DEREGISTER_SUBSCRIPTION_PREFETCHING_QUEUE_METRICS_F08479A7 =
"注销 subscription prefetching queue metrics 失败，SubscriptionPrefetchingQueue({}) 不存在";
public static final String PIPE_LOG_FAILED_TO_MARK_TRANSFER_EVENT_RATE_SUBSCRIPTIONPREFETCHINGQUEUE_7DEF95B5 =
"标记传输事件速率失败，SubscriptionPrefetchingQueue({}) 不存在";
- public static final String PIPE_LOG_FAILED_TO_UNBIND_FROM_CONSENSUS_SUBSCRIPTION_PREFETCHING_A8F920D9 =
- "解绑 consensus subscription prefetching queue metrics 失败，queue map 非空";
public static final String PIPE_LOG_FAILED_TO_DEREGISTER_CONSENSUS_SUBSCRIPTION_PREFETCHING_8B180091 =
"注销 consensus subscription prefetching queue metrics 失败，ConsensusPrefetchingQueue({}) 不存在";
public static final String PIPE_LOG_FAILED_TO_MARK_TRANSFER_EVENT_RATE_CONSENSUSPREFETCHINGQUEUE_FE9B91C3 =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeSinglePipeMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeSinglePipeMetrics.java
index b6efd64ed2c..0f537fd5800 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeSinglePipeMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeSinglePipeMetrics.java
@@ -56,7 +56,7 @@ public class PipeDataNodeSinglePipeMetrics implements IMetricSet {
//////////////////////////// bindTo & unbindFrom (metric framework) ////////////////////////////
@Override
- public void bindTo(final AbstractMetricService metricService) {
+ public synchronized void bindTo(final AbstractMetricService metricService) {
this.metricService = metricService;
ImmutableSet.copyOf(remainingEventAndTimeOperatorMap.keySet()).forEach(this::createMetrics);
}
@@ -135,11 +135,11 @@ public class PipeDataNodeSinglePipeMetrics implements IMetricSet {
}
@Override
- public void unbindFrom(final AbstractMetricService metricService) {
- ImmutableSet.copyOf(remainingEventAndTimeOperatorMap.keySet()).forEach(this::deregister);
- if (!remainingEventAndTimeOperatorMap.isEmpty()) {
- LOGGER.warn(DataNodePipeMessages.FAILED_TO_UNBIND_FROM_PIPE_REMAINING_EVENT);
- }
+ public synchronized void unbindFrom(final AbstractMetricService metricService) {
+ // Keep the operators registered: they hold the states of the pipes and register only once, so
+ // a metric service restart must be able to bind them again.
+ // Synchronized with the (de)registrations, which may remove a registration being unbound.
+ ImmutableSet.copyOf(remainingEventAndTimeOperatorMap.keySet()).forEach(this::removeMetrics);
}
private void removeMetrics(final String pipeID) {
@@ -194,12 +194,11 @@ public class PipeDataNodeSinglePipeMetrics implements IMetricSet {
Metric.PIPE_TSFILE_EVENT_TRANSFER_TIME.toString(),
Tag.NAME.toString(),
operator.getPipeName());
- remainingEventAndTimeOperatorMap.remove(pipeID);
}
//////////////////////////// register & deregister (pipe integration) ////////////////////////////
- public void register(final IoTDBDataRegionSource source) {
+ public synchronized void register(final IoTDBDataRegionSource source) {
// The metric is global thus the regionId is omitted
final String pipeID = source.getPipeName() + "_" + source.getCreationTime();
remainingEventAndTimeOperatorMap
@@ -214,7 +213,7 @@ public class PipeDataNodeSinglePipeMetrics implements IMetricSet {
}
}
- public void register(final IoTDBSchemaRegionSource source) {
+ public synchronized void register(final IoTDBSchemaRegionSource source) {
// The metric is global thus the regionId is omitted
final String pipeID = source.getPipeName() + "_" + source.getCreationTime();
remainingEventAndTimeOperatorMap
@@ -349,7 +348,7 @@ public class PipeDataNodeSinglePipeMetrics implements IMetricSet {
remainingEventAndTimeOperatorMap.get(pipeID).freezeRate(true);
}
- public void deregister(final String pipeID) {
+ public synchronized void deregister(final String pipeID) {
if (!remainingEventAndTimeOperatorMap.containsKey(pipeID)) {
LOGGER.warn(DataNodePipeMessages.FAILED_TO_DEREGISTER_PIPE_REMAINING_EVENT_AND, pipeID);
return;
@@ -357,6 +356,7 @@ public class PipeDataNodeSinglePipeMetrics implements IMetricSet {
if (Objects.nonNull(metricService)) {
removeMetrics(pipeID);
}
+ remainingEventAndTimeOperatorMap.remove(pipeID);
}
public void markRegionCommit(final String pipeID, final boolean isDataRegion) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeTsFileToTabletsMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeTsFileToTabletsMetrics.java
index 1a958c7d3a6..39aef7896eb 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeTsFileToTabletsMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeTsFileToTabletsMetrics.java
@@ -57,7 +57,7 @@ public class PipeTsFileToTabletsMetrics implements IMetricSet {
//////////////////////////// bindTo & unbindFrom (metric framework) ////////////////////////////
@Override
- public void bindTo(final AbstractMetricService metricService) {
+ public synchronized void bindTo(final AbstractMetricService metricService) {
this.metricService = metricService;
ImmutableSet.copyOf(pipe).forEach(this::createMetrics);
}
@@ -101,11 +101,11 @@ public class PipeTsFileToTabletsMetrics implements IMetricSet {
}
@Override
- public void unbindFrom(final AbstractMetricService metricService) {
- ImmutableSet.copyOf(pipe).forEach(this::deregister);
- if (!pipe.isEmpty()) {
- LOGGER.warn(DataNodePipeMessages.FAILED_TO_UNBIND_FROM_PIPE_TSFILE_TO, pipe);
- }
+ public synchronized void unbindFrom(final AbstractMetricService metricService) {
+ // Keep the pipes registered: they register only once, so a metric service restart
+ // must be able to bind them again.
+ // Synchronized with the (de)registrations, which may remove a registration being unbound.
+ ImmutableSet.copyOf(pipe).forEach(this::removeMetrics);
}
private void removeMetrics(final String pipeID) {
@@ -147,7 +147,7 @@ public class PipeTsFileToTabletsMetrics implements IMetricSet {
//////////////////////////// register & deregister ////////////////////////////
- public void register(final IoTDBDataRegionSource extractor) {
+ public synchronized void register(final IoTDBDataRegionSource extractor) {
final String pipeID = extractor.getPipeName() + "_" + extractor.getCreationTime();
pipe.add(pipeID);
if (Objects.nonNull(metricService)) {
@@ -155,7 +155,7 @@ public class PipeTsFileToTabletsMetrics implements IMetricSet {
}
}
- public void deregister(final String pipeID) {
+ public synchronized void deregister(final String pipeID) {
if (!pipe.contains(pipeID)) {
LOGGER.info(DataNodePipeMessages.SKIP_DEREGISTER_PIPE_TSFILE_TO_TABLETS, pipeID);
return;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/processor/PipeProcessorMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/processor/PipeProcessorMetrics.java
index 97bb7f46341..97fb46bdb23 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/processor/PipeProcessorMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/processor/PipeProcessorMetrics.java
@@ -55,7 +55,7 @@ public class PipeProcessorMetrics implements IMetricSet {
//////////////////////////// bindTo & unbindFrom (metric framework) ////////////////////////////
@Override
- public void bindTo(final AbstractMetricService metricService) {
+ public synchronized void bindTo(final AbstractMetricService metricService) {
this.metricService = metricService;
final ImmutableSet<String> taskIDs = ImmutableSet.copyOf(processorMap.keySet());
for (final String taskID : taskIDs) {
@@ -106,14 +106,11 @@ public class PipeProcessorMetrics implements IMetricSet {
}
@Override
- public void unbindFrom(final AbstractMetricService metricService) {
- final ImmutableSet<String> taskIDs = ImmutableSet.copyOf(processorMap.keySet());
- for (final String taskID : taskIDs) {
- deregister(taskID);
- }
- if (!processorMap.isEmpty()) {
- LOGGER.warn(DataNodePipeMessages.FAILED_TO_UNBIND_FROM_PIPE_PROCESSOR_METRICS);
- }
+ public synchronized void unbindFrom(final AbstractMetricService metricService) {
+ // Keep the subtasks registered: they register only once, so a metric service restart
+ // must be able to bind them again.
+ // Synchronized with the (de)registrations, which may remove a registration being unbound.
+ ImmutableSet.copyOf(processorMap.keySet()).forEach(this::removeMetrics);
}
private void removeMetrics(final String taskID) {
@@ -157,7 +154,7 @@ public class PipeProcessorMetrics implements IMetricSet {
//////////////////////////// register & deregister (pipe integration) ////////////////////////////
- public void register(final PipeProcessorSubtask pipeProcessorSubtask) {
+ public synchronized void register(final PipeProcessorSubtask pipeProcessorSubtask) {
final String taskID = pipeProcessorSubtask.getTaskID();
processorMap.putIfAbsent(taskID, pipeProcessorSubtask);
if (Objects.nonNull(metricService)) {
@@ -165,7 +162,7 @@ public class PipeProcessorMetrics implements IMetricSet {
}
}
- public void deregister(final String taskID) {
+ public synchronized void deregister(final String taskID) {
if (!processorMap.containsKey(taskID)) {
// Allow calls from schema region tasks
return;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionListenerMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionListenerMetrics.java
index 526ec7df281..8e89d156829 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionListenerMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionListenerMetrics.java
@@ -50,7 +50,7 @@ public class PipeSchemaRegionListenerMetrics implements IMetricSet {
//////////////////////////// bindTo & unbindFrom (metric framework) ////////////////////////////
@Override
- public void bindTo(final AbstractMetricService metricService) {
+ public synchronized void bindTo(final AbstractMetricService metricService) {
this.metricService = metricService;
ImmutableSet.copyOf(listeningQueueMap.keySet()).forEach(this::createMetrics);
}
@@ -70,11 +70,11 @@ public class PipeSchemaRegionListenerMetrics implements IMetricSet {
}
@Override
- public void unbindFrom(final AbstractMetricService metricService) {
- ImmutableSet.copyOf(listeningQueueMap.keySet()).forEach(this::deregister);
- if (!listeningQueueMap.isEmpty()) {
- LOGGER.warn(DataNodePipeMessages.FAILED_TO_UNBIND_FROM_PIPE_SCHEMA_REGION_2);
- }
+ public synchronized void unbindFrom(final AbstractMetricService metricService) {
+ // Keep the queues registered: they register only once, so a metric service restart
+ // must be able to bind them again.
+ // Synchronized with the (de)registrations, which may remove a registration being unbound.
+ ImmutableSet.copyOf(listeningQueueMap.keySet()).forEach(this::removeMetrics);
}
private void removeMetrics(final Integer schemaRegionId) {
@@ -91,7 +91,7 @@ public class PipeSchemaRegionListenerMetrics implements IMetricSet {
//////////////////////////// register & deregister (pipe integration) ////////////////////////////
- public void register(
+ public synchronized void register(
final SchemaRegionListeningQueue schemaRegionListeningQueue, final Integer schemaRegionId) {
listeningQueueMap.putIfAbsent(schemaRegionId, schemaRegionListeningQueue);
if (Objects.nonNull(metricService)) {
@@ -99,7 +99,7 @@ public class PipeSchemaRegionListenerMetrics implements IMetricSet {
}
}
- public void deregister(final Integer schemaRegionId) {
+ public synchronized void deregister(final Integer schemaRegionId) {
if (!listeningQueueMap.containsKey(schemaRegionId)) {
LOGGER.warn(
DataNodePipeMessages.FAILED_TO_DEREGISTER_SCHEMA_REGION_LISTENER_METRICS, schemaRegionId);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionSinkMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionSinkMetrics.java
index cc8c74e2984..d611592a3ee 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionSinkMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionSinkMetrics.java
@@ -51,7 +51,7 @@ public class PipeSchemaRegionSinkMetrics implements IMetricSet {
//////////////////////////// bindTo & unbindFrom (metric framework) ////////////////////////////
@Override
- public void bindTo(final AbstractMetricService metricService) {
+ public synchronized void bindTo(final AbstractMetricService metricService) {
this.metricService = metricService;
ImmutableSet.copyOf(connectorMap.keySet()).forEach(this::createMetrics);
}
@@ -129,11 +129,11 @@ public class PipeSchemaRegionSinkMetrics implements IMetricSet {
}
@Override
- public void unbindFrom(final AbstractMetricService metricService) {
- ImmutableSet.copyOf(connectorMap.keySet()).forEach(this::deregister);
- if (!connectorMap.isEmpty()) {
- LOGGER.warn(DataNodePipeMessages.FAILED_TO_UNBIND_FROM_PIPE_SCHEMA_REGION);
- }
+ public synchronized void unbindFrom(final AbstractMetricService metricService) {
+ // Keep the subtasks registered: they register only once, so a metric service restart
+ // must be able to bind them again.
+ // Synchronized with the (de)registrations, which may remove a registration being unbound.
+ ImmutableSet.copyOf(connectorMap.keySet()).forEach(this::removeMetrics);
}
private void removeMetrics(final String taskID) {
@@ -167,7 +167,7 @@ public class PipeSchemaRegionSinkMetrics implements IMetricSet {
//////////////////////////// Register & deregister (pipe integration) ////////////////////////////
- public void register(final PipeSinkSubtask pipeSinkSubtask) {
+ public synchronized void register(final PipeSinkSubtask pipeSinkSubtask) {
final String taskID = pipeSinkSubtask.getTaskID();
connectorMap.putIfAbsent(taskID, pipeSinkSubtask);
if (Objects.nonNull(metricService)) {
@@ -175,7 +175,7 @@ public class PipeSchemaRegionSinkMetrics implements IMetricSet {
}
}
- public void deregister(final String taskID) {
+ public synchronized void deregister(final String taskID) {
if (!connectorMap.containsKey(taskID)) {
LOGGER.warn(
DataNodePipeMessages.FAILED_TO_DEREGISTER_PIPE_SCHEMA_REGION_CONNECTOR,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionSourceMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionSourceMetrics.java
index 459fa9fcaa3..9e9aeaebe0e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionSourceMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionSourceMetrics.java
@@ -48,7 +48,7 @@ public class PipeSchemaRegionSourceMetrics implements IMetricSet {
//////////////////////////// bindTo & unbindFrom (metric framework) ////////////////////////////
@Override
- public void bindTo(final AbstractMetricService metricService) {
+ public synchronized void bindTo(final AbstractMetricService metricService) {
this.metricService = metricService;
ImmutableSet.copyOf(sourceMap.keySet()).forEach(this::createMetrics);
}
@@ -73,11 +73,11 @@ public class PipeSchemaRegionSourceMetrics implements IMetricSet {
}
@Override
- public void unbindFrom(final AbstractMetricService metricService) {
- ImmutableSet.copyOf(sourceMap.keySet()).forEach(this::deregister);
- if (!sourceMap.isEmpty()) {
- LOGGER.warn(DataNodePipeMessages.FAILED_TO_UNBIND_FROM_PIPE_SCHEMA_REGION_1);
- }
+ public synchronized void unbindFrom(final AbstractMetricService metricService) {
+ // Keep the sources registered: they register only once, so a metric service restart
+ // must be able to bind them again.
+ // Synchronized with the (de)registrations, which may remove a registration being unbound.
+ ImmutableSet.copyOf(sourceMap.keySet()).forEach(this::removeMetrics);
}
private void removeMetrics(final String taskID) {
@@ -100,7 +100,7 @@ public class PipeSchemaRegionSourceMetrics implements IMetricSet {
//////////////////////////// register & deregister (pipe integration) ////////////////////////////
- public void register(final IoTDBSchemaRegionSource source) {
+ public synchronized void register(final IoTDBSchemaRegionSource source) {
final String taskID = source.getTaskID();
sourceMap.putIfAbsent(taskID, source);
if (Objects.nonNull(metricService)) {
@@ -108,7 +108,7 @@ public class PipeSchemaRegionSourceMetrics implements IMetricSet {
}
}
- public void deregister(final String taskID) {
+ public synchronized void deregister(final String taskID) {
if (!sourceMap.containsKey(taskID)) {
LOGGER.warn(DataNodePipeMessages.FAILED_TO_DEREGISTER_PIPE_SCHEMA_REGION_SOURCE, taskID);
return;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetrics.java
index 9990625dd79..63189a8e14a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetrics.java
@@ -248,13 +248,10 @@ public class PipeDataRegionSinkMetrics implements IMetricSet {
@Override
public synchronized void unbindFrom(final AbstractMetricService metricService) {
- final ImmutableSet<String> taskIDs = ImmutableSet.copyOf(sinkMap.keySet());
- for (final String taskID : taskIDs) {
- deregister(taskID);
- }
- if (!sinkMap.isEmpty()) {
- LOGGER.warn(DataNodePipeMessages.FAILED_TO_UNBIND_FROM_PIPE_DATA_REGION);
- }
+ // Keep the subtasks registered: they register only once, so a metric service restart
+ // must be able to bind them again.
+ // Synchronized with the (de)registrations, which may remove a registration being unbound.
+ ImmutableSet.copyOf(sinkMap.keySet()).forEach(this::removeMetrics);
}
private void removeMetrics(final String taskID) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/source/PipeAssignerMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/source/PipeAssignerMetrics.java
index 3665ec24294..b2a44eef42a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/source/PipeAssignerMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/source/PipeAssignerMetrics.java
@@ -28,7 +28,6 @@ import org.apache.iotdb.metrics.metricsets.IMetricSet;
import org.apache.iotdb.metrics.utils.MetricLevel;
import org.apache.iotdb.metrics.utils.MetricType;
-import com.google.common.collect.ImmutableSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -86,12 +85,12 @@ public class PipeAssignerMetrics implements IMetricSet {
@Override
public void unbindFrom(AbstractMetricService metricService) {
- ImmutableSet<Integer> dataRegionIds = ImmutableSet.copyOf(assignerMap.keySet());
- for (int dataRegionId : dataRegionIds) {
- deregister(dataRegionId);
- }
- if (!assignerMap.isEmpty()) {
- LOGGER.warn(DataNodePipeMessages.FAILED_TO_UNBIND_FROM_PIPE_ASSIGNER_METRICS);
+ // Keep the assigners registered: they register only once, so a metric service restart
+ // must be able to bind them again.
+ synchronized (this) {
+ for (int dataRegionId : assignerMap.keySet()) {
+ removeMetrics(dataRegionId);
+ }
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/source/PipeDataRegionSourceMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/source/PipeDataRegionSourceMetrics.java
index f44799e0550..11622f4bae1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/source/PipeDataRegionSourceMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/source/PipeDataRegionSourceMetrics.java
@@ -63,7 +63,7 @@ public class PipeDataRegionSourceMetrics implements IMetricSet {
//////////////////////////// bindTo & unbindFrom (metric framework) ////////////////////////////
@Override
- public void bindTo(final AbstractMetricService metricService) {
+ public synchronized void bindTo(final AbstractMetricService metricService) {
this.metricService = metricService;
final ImmutableSet<String> taskIDs = ImmutableSet.copyOf(extractorMap.keySet());
for (final String taskID : taskIDs) {
@@ -181,14 +181,11 @@ public class PipeDataRegionSourceMetrics implements IMetricSet {
}
@Override
- public void unbindFrom(final AbstractMetricService metricService) {
- final ImmutableSet<String> taskIDs = ImmutableSet.copyOf(extractorMap.keySet());
- for (final String taskID : taskIDs) {
- deregister(taskID);
- }
- if (!extractorMap.isEmpty()) {
- LOGGER.warn(DataNodePipeMessages.FAILED_TO_UNBIND_FROM_PIPE_EXTRACTOR_METRICS);
- }
+ public synchronized void unbindFrom(final AbstractMetricService metricService) {
+ // Keep the extractors registered: they register only once, so a metric service restart
+ // must be able to bind them again.
+ // Synchronized with the (de)registrations, which may remove a registration being unbound.
+ ImmutableSet.copyOf(extractorMap.keySet()).forEach(this::removeMetrics);
}
private void removeMetrics(final String taskID) {
@@ -285,11 +282,12 @@ public class PipeDataRegionSourceMetrics implements IMetricSet {
String.valueOf(extractor.getRegionId()),
Tag.CREATION_TIME.toString(),
String.valueOf(extractor.getCreationTime()));
+ recentProcessedTsFileEpochStateMap.remove(taskID);
}
//////////////////////////// register & deregister (pipe integration) ////////////////////////////
- public void register(final IoTDBDataRegionSource extractor) {
+ public synchronized void register(final IoTDBDataRegionSource extractor) {
final String taskID = extractor.getTaskID();
extractorMap.putIfAbsent(taskID, extractor);
if (Objects.nonNull(metricService)) {
@@ -297,7 +295,7 @@ public class PipeDataRegionSourceMetrics implements IMetricSet {
}
}
- public void deregister(final String taskID) {
+ public synchronized void deregister(final String taskID) {
if (!extractorMap.containsKey(taskID)) {
LOGGER.warn(DataNodePipeMessages.FAILED_TO_DEREGISTER_PIPE_DATA_REGION_EXTRACTOR, taskID);
return;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java
index 6ef1c47ad2e..7261648ab5f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java
@@ -665,7 +665,8 @@ public class IoTDBDataRegionAirGapSink extends IoTDBDataNodeAirGapSink {
@Override
protected byte[] compressIfNeeded(final byte[] reqInBytes) throws IOException {
- if (Objects.isNull(compressionTimer) && Objects.nonNull(sinkTaskId)) {
+ if (Objects.nonNull(sinkTaskId)) {
+ // Do not cache the timer, the metric service replaces it when it restarts
compressionTimer = PipeDataRegionSinkMetrics.getInstance().getCompressionTimer(sinkTaskId);
}
return super.compressIfNeeded(reqInBytes);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
index c0da24658fd..4b7f332dbe3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
@@ -707,7 +707,8 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink implements PipeSinkWithS
@Override
public TPipeTransferReq compressIfNeeded(final TPipeTransferReq req) throws IOException {
- if (Objects.isNull(compressionTimer) && Objects.nonNull(sinkTaskId)) {
+ if (Objects.nonNull(sinkTaskId)) {
+ // Do not cache the timer, the metric service replaces it when it restarts
compressionTimer = PipeDataRegionSinkMetrics.getInstance().getCompressionTimer(sinkTaskId);
}
return super.compressIfNeeded(req);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java
index d7538907dd6..3b858b9d712 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java
@@ -791,7 +791,8 @@ public class IoTDBDataRegionSyncSink extends IoTDBDataNodeSyncSink {
@Override
public TPipeTransferReq compressIfNeeded(final TPipeTransferReq req) throws IOException {
- if (Objects.isNull(compressionTimer) && Objects.nonNull(sinkTaskId)) {
+ if (Objects.nonNull(sinkTaskId)) {
+ // Do not cache the timer, the metric service replaces it when it restarts
compressionTimer = PipeDataRegionSinkMetrics.getInstance().getCompressionTimer(sinkTaskId);
}
return super.compressIfNeeded(req);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/handler/RPCServiceThriftHandlerMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/handler/RPCServiceThriftHandlerMetrics.java
index 4189d5f4b21..12695ee933b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/handler/RPCServiceThriftHandlerMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/handler/RPCServiceThriftHandlerMetrics.java
@@ -31,9 +31,12 @@ import java.util.Objects;
import java.util.concurrent.atomic.AtomicLong;
public class RPCServiceThriftHandlerMetrics implements IMetricSet {
+ // The singleton owns the decoding metrics, and each handler registers its own instance owning the
+ // connection metric. A metric must have a single owner, otherwise rebinding one of the instances,
+ // e.g. when the metric service restarts, drops the metrics bound by the other.
private static final RPCServiceThriftHandlerMetrics INSTANCE =
- new RPCServiceThriftHandlerMetrics(new AtomicLong(0));
- private AtomicLong thriftConnectionNumber;
+ new RPCServiceThriftHandlerMetrics(null);
+ private final AtomicLong thriftConnectionNumber;
// region begin
private Gauge unCompressionSizeTimer = DoNothingMetricManager.DO_NOTHING_GAUGE;
@@ -70,13 +73,16 @@ public class RPCServiceThriftHandlerMetrics implements IMetricSet {
@Override
public void bindTo(AbstractMetricService metricService) {
- metricService.createAutoGauge(
- Metric.THRIFT_CONNECTIONS.toString(),
- MetricLevel.CORE,
- thriftConnectionNumber,
- AtomicLong::get,
- Tag.NAME.toString(),
- "ClientRPC");
+ if (Objects.nonNull(thriftConnectionNumber)) {
+ metricService.createAutoGauge(
+ Metric.THRIFT_CONNECTIONS.toString(),
+ MetricLevel.CORE,
+ thriftConnectionNumber,
+ AtomicLong::get,
+ Tag.NAME.toString(),
+ "ClientRPC");
+ return;
+ }
unCompressionSizeTimer =
metricService.getOrCreateGauge(
@@ -116,11 +122,14 @@ public class RPCServiceThriftHandlerMetrics implements IMetricSet {
@Override
public void unbindFrom(AbstractMetricService metricService) {
- metricService.remove(
- MetricType.AUTO_GAUGE,
- Metric.THRIFT_CONNECTIONS.toString(),
- Tag.NAME.toString(),
- "ClientRPC");
+ if (Objects.nonNull(thriftConnectionNumber)) {
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.THRIFT_CONNECTIONS.toString(),
+ Tag.NAME.toString(),
+ "ClientRPC");
+ return;
+ }
metricService.remove(
MetricType.GAUGE,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java
index db58481285e..0b8dfbed002 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java
@@ -149,7 +149,7 @@ public class PartitionCache {
this.partitionExecutor =
SeriesPartitionExecutor.getSeriesPartitionExecutor(
this.seriesSlotExecutorName, this.seriesPartitionSlotNum);
- this.cacheMetrics = new CacheMetrics();
+ this.cacheMetrics = CacheMetrics.getInstance();
}
protected void failIfMetadataLeaseFenced() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaEngineMemMetric.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaEngineMemMetric.java
index e91811c50ec..e87c8cfba17 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaEngineMemMetric.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaEngineMemMetric.java
@@ -101,12 +101,14 @@ public class SchemaEngineMemMetric implements ISchemaEngineMetric {
MetricLevel.IMPORTANT,
Tag.NAME.toString(),
SCHEMA_CONSENSUS);
+ // A table binds its metrics only when its first device is added, so bind the existing ones here
+ engineStatistics.getTable2DevicesNumMap().keySet().forEach(this::bindTableMetrics);
}
@Override
public void bindTableMetrics(final String tableName) {
metricService.createAutoGauge(
- Metric.SCHEMA_ENGINE.toString(),
+ Metric.SCHEMA_ENGINE_TABLE.toString(),
MetricLevel.IMPORTANT,
engineStatistics,
statistics -> statistics.getTableDeviceNumber(tableName),
@@ -138,13 +140,14 @@ public class SchemaEngineMemMetric implements ISchemaEngineMetric {
MetricType.GAUGE, Metric.SCHEMA_ENGINE.toString(), Tag.NAME.toString(), SCHEMA_ENGINE_MODE);
metricService.remove(
MetricType.GAUGE, Metric.SCHEMA_ENGINE.toString(), Tag.NAME.toString(), SCHEMA_CONSENSUS);
+ engineStatistics.getTable2DevicesNumMap().keySet().forEach(this::unbindTableMetrics);
}
@Override
public void unbindTableMetrics(final String tableName) {
metricService.remove(
MetricType.AUTO_GAUGE,
- Metric.SCHEMA_ENGINE.toString(),
+ Metric.SCHEMA_ENGINE_TABLE.toString(),
Tag.NAME.toString(),
TABLE_DEVICE_NUMBER,
TABLE,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaRegionMemMetric.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaRegionMemMetric.java
index 31f02e9c353..8969feb3f74 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaRegionMemMetric.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaRegionMemMetric.java
@@ -133,12 +133,14 @@ public class SchemaRegionMemMetric implements ISchemaRegionMetric {
regionTagValue,
Tag.DATABASE.toString(),
database);
+ // A table binds its metrics only when its first device is added, so bind the existing ones here
+ regionStatistics.getTable2DevicesNumMap().keySet().forEach(this::bindTableMetrics);
}
@Override
public void bindTableMetrics(final String tableName) {
metricService.createAutoGauge(
- Metric.SCHEMA_REGION.toString(),
+ Metric.SCHEMA_REGION_TABLE.toString(),
MetricLevel.IMPORTANT,
regionStatistics,
statistics -> statistics.getTableDevicesNumber(tableName),
@@ -218,13 +220,14 @@ public class SchemaRegionMemMetric implements ISchemaRegionMetric {
regionTagValue,
Tag.DATABASE.toString(),
database);
+ regionStatistics.getTable2DevicesNumMap().keySet().forEach(this::unbindTableMetrics);
}
@Override
public void unbindTableMetrics(final String tableName) {
metricService.remove(
MetricType.AUTO_GAUGE,
- Metric.SCHEMA_REGION.toString(),
+ Metric.SCHEMA_REGION_TABLE.toString(),
Tag.NAME.toString(),
TABLE_DEVICE_NUMBER,
Tag.REGION.toString(),
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/rescon/MemSchemaEngineStatistics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/rescon/MemSchemaEngineStatistics.java
index 7c8c2677841..56fb941b577 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/rescon/MemSchemaEngineStatistics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/rescon/MemSchemaEngineStatistics.java
@@ -144,6 +144,10 @@ public class MemSchemaEngineStatistics implements ISchemaEngineStatistics {
return tableDeviceNumber.getOrDefault(tableName, 0L);
}
+ public Map<String, Long> getTable2DevicesNumMap() {
+ return tableDeviceNumber;
+ }
+
@Override
public int getSchemaRegionNumber() {
return SchemaEngine.getInstance().getSchemaRegionNumber();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/CacheMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/CacheMetrics.java
index 2225a377f9f..2eb1b3ed9dd 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/CacheMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/CacheMetrics.java
@@ -147,4 +147,21 @@ public class CacheMetrics implements IMetricSet {
break;
}
}
+
+ private static class CacheMetricsHolder {
+
+ private static final CacheMetrics INSTANCE = new CacheMetrics();
+
+ private CacheMetricsHolder() {
+ // empty constructor
+ }
+ }
+
+ public static CacheMetrics getInstance() {
+ return CacheMetricsHolder.INSTANCE;
+ }
+
+ private CacheMetrics() {
+ // empty constructor
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsHelper.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsHelper.java
index d857f546e2d..55e71ca9ecf 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsHelper.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsHelper.java
@@ -87,6 +87,7 @@ public class DataNodeMetricsHelper {
metricService.addMetricSet(DataExchangeCountMetricSet.getInstance());
metricService.addMetricSet(DriverSchedulerMetricSet.getInstance());
metricService.addMetricSet(QueryRelatedResourceMetricSet.getInstance());
+ metricService.addMetricSet(CacheMetrics.getInstance());
// bind performance overview related metrics
metricService.addMetricSet(PerformanceOverviewMetrics.getInstance());
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/WritingMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/WritingMetrics.java
index bd2f124e060..9333a0e6082 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/WritingMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/WritingMetrics.java
@@ -43,10 +43,14 @@ import org.apache.iotdb.metrics.utils.MetricType;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.concurrent.atomic.DoubleAdder;
import java.util.concurrent.atomic.LongAdder;
+import java.util.stream.Collectors;
public class WritingMetrics implements IMetricSet {
private static final WritingMetrics INSTANCE = new WritingMetrics();
@@ -204,6 +208,8 @@ public class WritingMetrics implements IMetricSet {
private Histogram readWALBufferCostHistogram = DoNothingMetricManager.DO_NOTHING_HISTOGRAM;
private Histogram writeWALBufferCostHistogram = DoNothingMetricManager.DO_NOTHING_HISTOGRAM;
private Gauge walQueueMaxMemSizeGauge = DoNothingMetricManager.DO_NOTHING_GAUGE;
+ // Recorded rarely, so keep the value to restore it when the metric service recreates the gauge
+ private volatile long walQueueMaxMemSize;
private void bindWALMetrics(AbstractMetricService metricService) {
metricService.createAutoGauge(
@@ -277,6 +283,7 @@ public class WritingMetrics implements IMetricSet {
MetricLevel.IMPORTANT,
Tag.NAME.toString(),
WAL_QUEUE_MAX_MEM_COST);
+ walQueueMaxMemSizeGauge.set(walQueueMaxMemSize);
SystemInfo systemInfo = SystemInfo.getInstance();
metricService.createAutoGauge(
Metric.WAL_QUEUE_MEM_COST.toString(),
@@ -455,6 +462,16 @@ public class WritingMetrics implements IMetricSet {
private Gauge flushThreholdGauge = DoNothingMetricManager.DO_NOTHING_GAUGE;
private Gauge rejectThreholdGauge = DoNothingMetricManager.DO_NOTHING_GAUGE;
+ // Recorded rarely, so keep the values to restore them when the metric service recreates the
+ // gauges
+ private volatile long flushThreshold;
+ private volatile long rejectThreshold;
+
+ // The active memtable and time partition counters are levels maintained by deltas, so keep the
+ // levels to restore them when the metric service recreates the counters
+ private final Map<String, Long> activeMemTableCounts = new HashMap<>();
+ private long activeTimePartitionCount;
+ private final Object activeCountLock = new Object();
private Timer memtableLiveTimer = DoNothingMetricManager.DO_NOTHING_TIMER;
@@ -482,6 +499,8 @@ public class WritingMetrics implements IMetricSet {
manualFlushMemtableCounter = createManualFlushMemTableCounterMetrics();
memControlFlushMemtableCounter = createMemControlFlushMemTableCounterMetrics();
+ restoreActiveCounts(allDataRegionIds);
+
flushThreholdGauge =
MetricService.getInstance()
.getOrCreateGauge(
@@ -489,6 +508,7 @@ public class WritingMetrics implements IMetricSet {
MetricLevel.IMPORTANT,
Tag.TYPE.toString(),
FLUSH_THRESHOLD);
+ flushThreholdGauge.set(flushThreshold);
rejectThreholdGauge =
MetricService.getInstance()
.getOrCreateGauge(
@@ -496,6 +516,7 @@ public class WritingMetrics implements IMetricSet {
MetricLevel.IMPORTANT,
Tag.TYPE.toString(),
REJECT_THRESHOLD);
+ rejectThreholdGauge.set(rejectThreshold);
memtableLiveTimer =
MetricService.getInstance()
@@ -981,15 +1002,18 @@ public class WritingMetrics implements IMetricSet {
}
public void recordWALQueueMaxMemorySize(long size) {
+ walQueueMaxMemSize = size;
walQueueMaxMemSizeGauge.set(size);
}
public void recordFlushThreshold(double flushThreshold) {
- flushThreholdGauge.set((long) flushThreshold);
+ this.flushThreshold = (long) flushThreshold;
+ flushThreholdGauge.set(this.flushThreshold);
}
public void recordRejectThreshold(double rejectThreshold) {
- rejectThreholdGauge.set((long) rejectThreshold);
+ this.rejectThreshold = (long) rejectThreshold;
+ rejectThreholdGauge.set(this.rejectThreshold);
}
public void recordMemTableLiveDuration(long durationMillis) {
@@ -1013,18 +1037,52 @@ public class WritingMetrics implements IMetricSet {
}
public void recordActiveMemTableCount(String dataRegionId, int number) {
- MetricService.getInstance()
- .count(
- number,
- Metric.ACTIVE_MEMTABLE_COUNT.toString(),
- MetricLevel.IMPORTANT,
- Tag.REGION.toString(),
- dataRegionId);
+ // Use the region tag of the counter created and removed with the region
+ String region = new DataRegionId(Integer.parseInt(dataRegionId)).toString();
+ synchronized (activeCountLock) {
+ activeMemTableCounts.merge(region, (long) number, Long::sum);
+ MetricService.getInstance()
+ .count(
+ number,
+ Metric.ACTIVE_MEMTABLE_COUNT.toString(),
+ MetricLevel.IMPORTANT,
+ Tag.REGION.toString(),
+ region);
+ }
}
public void recordActiveTimePartitionCount(int number) {
- MetricService.getInstance()
- .count(number, Metric.ACTIVE_TIME_PARTITION_COUNT.toString(), MetricLevel.IMPORTANT);
+ synchronized (activeCountLock) {
+ activeTimePartitionCount += number;
+ MetricService.getInstance()
+ .count(number, Metric.ACTIVE_TIME_PARTITION_COUNT.toString(), MetricLevel.IMPORTANT);
+ }
+ }
+
+ private void restoreActiveCounts(List<DataRegionId> dataRegionIds) {
+ Set<String> existingDataRegionIds =
+ dataRegionIds.stream().map(DataRegionId::toString).collect(Collectors.toSet());
+ synchronized (activeCountLock) {
+ // Do not bring back the counters of the removed regions
+ activeMemTableCounts
+ .entrySet()
+ .removeIf(
+ entry -> entry.getValue() == 0 && !existingDataRegionIds.contains(entry.getKey()));
+ activeMemTableCounts.forEach(
+ (dataRegionId, count) ->
+ restoreCounter(
+ count,
+ Metric.ACTIVE_MEMTABLE_COUNT.toString(),
+ Tag.REGION.toString(),
+ dataRegionId));
+ restoreCounter(activeTimePartitionCount, Metric.ACTIVE_TIME_PARTITION_COUNT.toString());
+ }
+ }
+
+ private static void restoreCounter(long count, String metric, String... tags) {
+ Counter counter =
+ MetricService.getInstance().getOrCreateCounter(metric, MetricLevel.IMPORTANT, tags);
+ counter.inc(count - counter.getCount());
}
// endregion
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/TsFileMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/TsFileMetrics.java
index de0c0fd23d6..f9c5ff089f0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/TsFileMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/TsFileMetrics.java
@@ -83,6 +83,17 @@ public class TsFileMetrics implements IMetricSet {
@Override
public void bindTo(AbstractMetricService metricService) {
this.metricService.set(metricService);
+ // The cached gauges may have been dropped by the metric service, e.g. when it restarts, so
+ // recreate all of them from the recorded values.
+ Arrays.asList(seqFileCountMap, unseqFileCountMap)
+ .forEach(map -> map.values().forEach(TsFileMetrics::clearGauges));
+ Arrays.asList(seqFileSizeMap, unseqFileSizeMap)
+ .forEach(map -> map.values().forEach(TsFileMetrics::clearGauges));
+ Arrays.asList(seqLevelTsFileCountMap, unseqLevelTsFileCountMap)
+ .forEach(TsFileMetrics::clearGauges);
+ Arrays.asList(seqLevelTsFileSizeMap, unseqLevelTsFileSizeMap)
+ .forEach(TsFileMetrics::clearGauges);
+ hasRemainData.set(true);
checkIfThereRemainingData();
}
@@ -91,6 +102,17 @@ public class TsFileMetrics implements IMetricSet {
// do nothing here
}
+ private static <K, V> void clearGauges(Map<K, Pair<V, Gauge>> map) {
+ for (K key : map.keySet()) {
+ map.computeIfPresent(
+ key,
+ (k, v) -> {
+ v.setRight(null);
+ return v;
+ });
+ }
+ }
+
// region external update tsfile related metrics
public void addTsFile(TsFileResource tsFileResource) {
if (!tsFileResource.markAsRecordedByMetric()) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionMetrics.java
index 7fc0f6b8c8b..dcdc7f533eb 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionMetrics.java
@@ -26,9 +26,18 @@ import org.apache.iotdb.metrics.metricsets.IMetricSet;
import org.apache.iotdb.metrics.utils.MetricLevel;
import org.apache.iotdb.metrics.utils.MetricType;
+import java.util.HashMap;
+import java.util.Map;
import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
public class DataRegionMetrics implements IMetricSet {
+ // A database has several data regions but a single memory gauge, which sums up the memory of the
+ // regions bound to it. Each region adds itself when binding, and the last region to unbind
+ // removes the gauge.
+ private static final Map<String, Set<DataRegion>> DATABASE_TO_DATA_REGIONS = new HashMap<>();
+
private final DataRegion dataRegion;
private final String databaseName;
@@ -39,22 +48,38 @@ public class DataRegionMetrics implements IMetricSet {
@Override
public void bindTo(AbstractMetricService metricService) {
- metricService.createAutoGauge(
- Metric.MEM.toString(),
- MetricLevel.IMPORTANT,
- dataRegion,
- DataRegion::getMemCost,
- Tag.NAME.toString(),
- "database_" + databaseName);
+ synchronized (DATABASE_TO_DATA_REGIONS) {
+ Set<DataRegion> dataRegions =
+ DATABASE_TO_DATA_REGIONS.computeIfAbsent(
+ databaseName, database -> ConcurrentHashMap.newKeySet());
+ dataRegions.add(dataRegion);
+ metricService.createAutoGauge(
+ Metric.MEM.toString(),
+ MetricLevel.IMPORTANT,
+ dataRegions,
+ regions -> regions.stream().mapToLong(DataRegion::getMemCost).sum(),
+ Tag.NAME.toString(),
+ "database_" + databaseName);
+ }
}
@Override
public void unbindFrom(AbstractMetricService metricService) {
- metricService.remove(
- MetricType.AUTO_GAUGE,
- Metric.MEM.toString(),
- Tag.NAME.toString(),
- "database_" + databaseName);
+ synchronized (DATABASE_TO_DATA_REGIONS) {
+ Set<DataRegion> dataRegions = DATABASE_TO_DATA_REGIONS.get(databaseName);
+ if (Objects.nonNull(dataRegions)) {
+ dataRegions.remove(dataRegion);
+ if (!dataRegions.isEmpty()) {
+ return;
+ }
+ DATABASE_TO_DATA_REGIONS.remove(databaseName);
+ }
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.MEM.toString(),
+ Tag.NAME.toString(),
+ "database_" + databaseName);
+ }
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorInfoMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorInfoMetrics.java
index fb106a9bac7..dc19f3131ca 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorInfoMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorInfoMetrics.java
@@ -18,7 +18,6 @@
*/
package org.apache.iotdb.db.storageengine.dataregion.memtable;
-import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.commons.service.metric.enums.Metric;
import org.apache.iotdb.commons.service.metric.enums.Tag;
import org.apache.iotdb.metrics.AbstractMetricService;
@@ -26,7 +25,19 @@ import org.apache.iotdb.metrics.metricsets.IMetricSet;
import org.apache.iotdb.metrics.utils.MetricLevel;
import org.apache.iotdb.metrics.utils.MetricType;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
public class TsFileProcessorInfoMetrics implements IMetricSet {
+ // A database has many TsFileProcessors but a single chunk metadata memory gauge, which sums up
+ // the memory of the processors bound to it. Each processor adds itself when binding, and the last
+ // processor to unbind removes the gauge.
+ private static final Map<String, Set<TsFileProcessorInfo>> DATABASE_TO_PROCESSOR_INFOS =
+ new HashMap<>();
+
private final String databaseName;
private final TsFileProcessorInfo tsFileProcessorInfo;
@@ -38,23 +49,37 @@ public class TsFileProcessorInfoMetrics implements IMetricSet {
@Override
public void bindTo(AbstractMetricService metricService) {
- MetricService.getInstance()
- .createAutoGauge(
- Metric.MEM.toString(),
- MetricLevel.IMPORTANT,
- tsFileProcessorInfo,
- TsFileProcessorInfo::getMemCost,
- Tag.NAME.toString(),
- "chunkMetaData_" + databaseName);
+ synchronized (DATABASE_TO_PROCESSOR_INFOS) {
+ Set<TsFileProcessorInfo> processorInfos =
+ DATABASE_TO_PROCESSOR_INFOS.computeIfAbsent(
+ databaseName, database -> ConcurrentHashMap.newKeySet());
+ processorInfos.add(tsFileProcessorInfo);
+ metricService.createAutoGauge(
+ Metric.MEM.toString(),
+ MetricLevel.IMPORTANT,
+ processorInfos,
+ infos -> infos.stream().mapToLong(TsFileProcessorInfo::getMemCost).sum(),
+ Tag.NAME.toString(),
+ "chunkMetaData_" + databaseName);
+ }
}
@Override
public void unbindFrom(AbstractMetricService metricService) {
- MetricService.getInstance()
- .remove(
- MetricType.AUTO_GAUGE,
- Metric.MEM.toString(),
- Tag.NAME.toString(),
- "chunkMetaData_" + databaseName);
+ synchronized (DATABASE_TO_PROCESSOR_INFOS) {
+ Set<TsFileProcessorInfo> processorInfos = DATABASE_TO_PROCESSOR_INFOS.get(databaseName);
+ if (Objects.nonNull(processorInfos)) {
+ processorInfos.remove(tsFileProcessorInfo);
+ if (!processorInfos.isEmpty()) {
+ return;
+ }
+ DATABASE_TO_PROCESSOR_INFOS.remove(databaseName);
+ }
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.MEM.toString(),
+ Tag.NAME.toString(),
+ "chunkMetaData_" + databaseName);
+ }
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryBlock.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryBlock.java
index 43c239642d0..fd3a76b0122 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryBlock.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryBlock.java
@@ -19,12 +19,8 @@
package org.apache.iotdb.db.storageengine.load.memory;
-import org.apache.iotdb.commons.service.metric.MetricService;
-import org.apache.iotdb.commons.service.metric.enums.Metric;
-import org.apache.iotdb.commons.service.metric.enums.Tag;
import org.apache.iotdb.db.i18n.StorageEngineMessages;
import org.apache.iotdb.db.storageengine.load.metrics.LoadTsFileMemMetricSet;
-import org.apache.iotdb.metrics.utils.MetricLevel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -56,13 +52,7 @@ public class LoadTsFileMemoryBlock extends LoadTsFileAbstractMemoryBlock {
LOGGER.debug(StorageEngineMessages.EXCEED_TOTAL_MEMORY_SIZE, this);
}
- MetricService.getInstance()
- .getOrCreateGauge(
- Metric.LOAD_MEM.toString(),
- MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- LoadTsFileMemMetricSet.LOAD_TSFILE_OTHER_MEMORY)
- .incr(memoryInBytes);
+ LoadTsFileMemMetricSet.getInstance().updateOtherMemory(memoryInBytes);
}
@Override
@@ -71,13 +61,7 @@ public class LoadTsFileMemoryBlock extends LoadTsFileAbstractMemoryBlock {
LOGGER.warn(StorageEngineMessages.REDUCE_MEMORY_USAGE_TO_NEGATIVE, this);
}
- MetricService.getInstance()
- .getOrCreateGauge(
- Metric.LOAD_MEM.toString(),
- MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- LoadTsFileMemMetricSet.LOAD_TSFILE_OTHER_MEMORY)
- .decr(memoryInBytes);
+ LoadTsFileMemMetricSet.getInstance().updateOtherMemory(-memoryInBytes);
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/metrics/ActiveLoadingFilesMetricsSet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/metrics/ActiveLoadingFilesMetricsSet.java
index cf15065d5b1..c3934e15676 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/metrics/ActiveLoadingFilesMetricsSet.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/metrics/ActiveLoadingFilesMetricsSet.java
@@ -147,6 +147,12 @@ public abstract class ActiveLoadingFilesMetricsSet implements IMetricSet {
// Dir2PendingFileCounters' binding is triggered by updatePendingDirList
// FailedDirCounter's binding is triggered by updateFailedDir
+ // Both are only triggered by changes, so bind them again for the existing dirs, e.g. when the
+ // metric service restarts
+ rebindDir2PendingFileCounters();
+ if (Objects.nonNull(failedDir.get())) {
+ rebindFailedDirCounter();
+ }
bindOtherCounters(metricService);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/metrics/ActiveLoadingFilesNumberMetricsSet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/metrics/ActiveLoadingFilesNumberMetricsSet.java
index 11105ed14c1..69765820a30 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/metrics/ActiveLoadingFilesNumberMetricsSet.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/metrics/ActiveLoadingFilesNumberMetricsSet.java
@@ -35,17 +35,23 @@ public class ActiveLoadingFilesNumberMetricsSet extends ActiveLoadingFilesMetric
private Counter queuingFileCounter = DoNothingMetricManager.DO_NOTHING_COUNTER;
private Counter loadingFileCounter = DoNothingMetricManager.DO_NOTHING_COUNTER;
+ // The queuing and loading counters are levels maintained by deltas, so keep the levels to restore
+ // them when the metric service recreates the counters
+ private long queuingFileNumber = 0;
+ private long loadingFileNumber = 0;
- public void increaseQueuingFileCounter(final long number) {
+ public synchronized void increaseQueuingFileCounter(final long number) {
+ queuingFileNumber += number;
queuingFileCounter.inc(number);
}
- public void increaseLoadingFileCounter(final long number) {
+ public synchronized void increaseLoadingFileCounter(final long number) {
+ loadingFileNumber += number;
loadingFileCounter.inc(number);
}
@Override
- protected void bindOtherCounters(final AbstractMetricService metricService) {
+ protected synchronized void bindOtherCounters(final AbstractMetricService metricService) {
totalPendingFileCounter =
metricService.getOrCreateCounter(
Metric.ACTIVE_LOADING_FILES_NUMBER.toString(),
@@ -58,16 +64,18 @@ public class ActiveLoadingFilesNumberMetricsSet extends ActiveLoadingFilesMetric
MetricLevel.IMPORTANT,
Tag.TYPE.toString(),
QUEUING);
+ queuingFileCounter.inc(queuingFileNumber - queuingFileCounter.getCount());
loadingFileCounter =
metricService.getOrCreateCounter(
Metric.ACTIVE_LOADING_FILES_NUMBER.toString(),
MetricLevel.IMPORTANT,
Tag.TYPE.toString(),
LOADING);
+ loadingFileCounter.inc(loadingFileNumber - loadingFileCounter.getCount());
}
@Override
- protected void unbindOtherCounters(final AbstractMetricService metricService) {
+ protected synchronized void unbindOtherCounters(final AbstractMetricService metricService) {
totalPendingFileCounter = DoNothingMetricManager.DO_NOTHING_COUNTER;
queuingFileCounter = DoNothingMetricManager.DO_NOTHING_COUNTER;
loadingFileCounter = DoNothingMetricManager.DO_NOTHING_COUNTER;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/metrics/LoadTsFileMemMetricSet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/metrics/LoadTsFileMemMetricSet.java
index 63608ae12f4..200ba19e509 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/metrics/LoadTsFileMemMetricSet.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/metrics/LoadTsFileMemMetricSet.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.storageengine.load.metrics;
+import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.commons.service.metric.enums.Metric;
import org.apache.iotdb.commons.service.metric.enums.Tag;
import org.apache.iotdb.db.storageengine.load.memory.LoadTsFileMemoryManager;
@@ -34,8 +35,11 @@ public class LoadTsFileMemMetricSet implements IMetricSet {
private static final String LOAD_TSFILE_DATA_CACHE_MEMORY = "LoadTsFileDataCacheMemory";
+ // Keep the size to restore it when the metric service recreates the gauge, e.g. on restart
+ private long otherMemorySizeInBytes = 0;
+
@Override
- public void bindTo(AbstractMetricService metricService) {
+ public synchronized void bindTo(AbstractMetricService metricService) {
metricService.createAutoGauge(
Metric.LOAD_MEM.toString(),
MetricLevel.IMPORTANT,
@@ -58,7 +62,18 @@ public class LoadTsFileMemMetricSet implements IMetricSet {
MetricLevel.IMPORTANT,
Tag.NAME.toString(),
LOAD_TSFILE_OTHER_MEMORY)
- .set(0L);
+ .set(otherMemorySizeInBytes);
+ }
+
+ public synchronized void updateOtherMemory(long deltaInBytes) {
+ otherMemorySizeInBytes += deltaInBytes;
+ MetricService.getInstance()
+ .getOrCreateGauge(
+ Metric.LOAD_MEM.toString(),
+ MetricLevel.IMPORTANT,
+ Tag.NAME.toString(),
+ LOAD_TSFILE_OTHER_MEMORY)
+ .set(otherMemorySizeInBytes);
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java
index b7331749b8c..19076e9abae 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java
@@ -60,15 +60,9 @@ public class ConsensusSubscriptionPrefetchingQueueMetrics implements IMetricSet
@Override
public synchronized void unbindFrom(final AbstractMetricService metricService) {
- final ImmutableSet<QueueMetricsKey> keys = ImmutableSet.copyOf(queueMap.keySet());
- for (final QueueMetricsKey key : keys) {
- deregister(key);
- }
- if (!queueMap.isEmpty()) {
- LOGGER.warn(
- DataNodePipeMessages
- .PIPE_LOG_FAILED_TO_UNBIND_FROM_CONSENSUS_SUBSCRIPTION_PREFETCHING_A8F920D9);
- }
+ // Keep the queues registered: they register only once, so a metric service restart
+ // must be able to bind them again.
+ ImmutableSet.copyOf(queueMap.keySet()).forEach(this::removeMetrics);
}
//////////////////////////// register & deregister ////////////////////////////
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/SubscriptionPrefetchingQueueMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/SubscriptionPrefetchingQueueMetrics.java
index 250191fcdf1..0bfbf893200 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/SubscriptionPrefetchingQueueMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/SubscriptionPrefetchingQueueMetrics.java
@@ -50,7 +50,7 @@ public class SubscriptionPrefetchingQueueMetrics implements IMetricSet {
private final Map<String, Rate> rateMap = new ConcurrentHashMap<>();
@Override
- public void bindTo(final AbstractMetricService metricService) {
+ public synchronized void bindTo(final AbstractMetricService metricService) {
this.metricService = metricService;
final ImmutableSet<String> ids = ImmutableSet.copyOf(prefetchingQueueMap.keySet());
for (final String id : ids) {
@@ -59,21 +59,16 @@ public class SubscriptionPrefetchingQueueMetrics implements IMetricSet {
}
@Override
- public void unbindFrom(final AbstractMetricService metricService) {
- final ImmutableSet<String> ids = ImmutableSet.copyOf(prefetchingQueueMap.keySet());
- for (final String id : ids) {
- deregister(id);
- }
- if (!prefetchingQueueMap.isEmpty()) {
- LOGGER.warn(
- DataNodePipeMessages
- .PIPE_LOG_FAILED_TO_UNBIND_FROM_SUBSCRIPTION_PREFETCHING_QUEUE_METRICS_6614388C);
- }
+ public synchronized void unbindFrom(final AbstractMetricService metricService) {
+ // Keep the queues registered: they register only once, so a metric service restart
+ // must be able to bind them again.
+ // Synchronized with the (de)registrations, which may remove a registration being unbound.
+ ImmutableSet.copyOf(prefetchingQueueMap.keySet()).forEach(this::removeMetrics);
}
//////////////////////////// register & deregister ////////////////////////////
- public void register(final SubscriptionPrefetchingQueue prefetchingQueue) {
+ public synchronized void register(final SubscriptionPrefetchingQueue prefetchingQueue) {
final String id = prefetchingQueue.getPrefetchingQueueId();
prefetchingQueueMap.putIfAbsent(id, prefetchingQueue);
if (Objects.nonNull(metricService)) {
@@ -94,16 +89,14 @@ public class SubscriptionPrefetchingQueueMetrics implements IMetricSet {
MetricLevel.IMPORTANT,
queue,
SubscriptionPrefetchingQueue::getSubscriptionUncommittedEventCount,
- Tag.NAME.toString(),
- queue.getPrefetchingQueueId());
+ getTags(queue));
// current commit id
metricService.createAutoGauge(
Metric.SUBSCRIPTION_CURRENT_COMMIT_ID.toString(),
MetricLevel.IMPORTANT,
queue,
SubscriptionPrefetchingQueue::getCurrentCommitId,
- Tag.NAME.toString(),
- queue.getPrefetchingQueueId());
+ getTags(queue));
}
private void createRate(final String id) {
@@ -112,13 +105,10 @@ public class SubscriptionPrefetchingQueueMetrics implements IMetricSet {
rateMap.put(
id,
metricService.getOrCreateRate(
- Metric.SUBSCRIPTION_EVENT_TRANSFER.toString(),
- MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- queue.getPrefetchingQueueId()));
+ Metric.SUBSCRIPTION_EVENT_TRANSFER.toString(), MetricLevel.IMPORTANT, getTags(queue)));
}
- public void deregister(final String id) {
+ public synchronized void deregister(final String id) {
if (!prefetchingQueueMap.containsKey(id)) {
LOGGER.warn(
DataNodePipeMessages
@@ -143,24 +133,27 @@ public class SubscriptionPrefetchingQueueMetrics implements IMetricSet {
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.SUBSCRIPTION_UNCOMMITTED_EVENT_COUNT.toString(),
- Tag.NAME.toString(),
- queue.getPrefetchingQueueId());
+ getTags(queue));
// current commit id
metricService.remove(
- MetricType.AUTO_GAUGE,
- Metric.SUBSCRIPTION_CURRENT_COMMIT_ID.toString(),
- Tag.NAME.toString(),
- queue.getPrefetchingQueueId());
+ MetricType.AUTO_GAUGE, Metric.SUBSCRIPTION_CURRENT_COMMIT_ID.toString(), getTags(queue));
}
private void removeRate(final String id) {
final SubscriptionPrefetchingQueue queue = prefetchingQueueMap.get(id);
// transfer event rate
metricService.remove(
- MetricType.RATE,
- Metric.SUBSCRIPTION_EVENT_TRANSFER.toString(),
- Tag.NAME.toString(),
- queue.getPrefetchingQueueId());
+ MetricType.RATE, Metric.SUBSCRIPTION_EVENT_TRANSFER.toString(), getTags(queue));
+ rateMap.remove(id);
+ }
+
+ // The consensus-based queues export these metrics with a region tag. A metric name must have the
+ // same tag keys everywhere, otherwise the queues registered later get no metrics at all, so give
+ // them the region tag too. These queues are not bound to a region, hence the empty value.
+ private static String[] getTags(final SubscriptionPrefetchingQueue queue) {
+ return new String[] {
+ Tag.NAME.toString(), queue.getPrefetchingQueueId(), Tag.REGION.toString(), "",
+ };
}
public void mark(final String id, final long size) {
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/PipeMetricsRestartTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/PipeMetricsRestartTest.java
new file mode 100644
index 00000000000..ab982c78af0
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/PipeMetricsRestartTest.java
@@ -0,0 +1,426 @@
+/*
+ * 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.iotdb.db.pipe.metric;
+
+import org.apache.iotdb.commons.pipe.sink.protocol.IoTDBSink;
+import org.apache.iotdb.commons.service.metric.MetricService;
+import org.apache.iotdb.commons.service.metric.enums.Metric;
+import org.apache.iotdb.db.pipe.agent.task.subtask.processor.PipeProcessorSubtask;
+import org.apache.iotdb.db.pipe.agent.task.subtask.sink.PipeSinkSubtask;
+import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics;
+import org.apache.iotdb.db.pipe.metric.overview.PipeTsFileToTabletsMetrics;
+import org.apache.iotdb.db.pipe.metric.processor.PipeProcessorMetrics;
+import org.apache.iotdb.db.pipe.metric.schema.PipeSchemaRegionSinkMetrics;
+import org.apache.iotdb.db.pipe.metric.schema.PipeSchemaRegionSourceMetrics;
+import org.apache.iotdb.db.pipe.metric.sink.PipeDataRegionSinkMetrics;
+import org.apache.iotdb.db.pipe.metric.source.PipeAssignerMetrics;
+import org.apache.iotdb.db.pipe.metric.source.PipeDataRegionSourceMetrics;
+import org.apache.iotdb.db.pipe.sink.protocol.airgap.IoTDBDataRegionAirGapSink;
+import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink;
+import org.apache.iotdb.db.pipe.sink.protocol.thrift.sync.IoTDBDataRegionSyncSink;
+import org.apache.iotdb.db.pipe.source.dataregion.IoTDBDataRegionSource;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeDataRegionAssigner;
+import org.apache.iotdb.db.pipe.source.schemaregion.IoTDBSchemaRegionSource;
+import org.apache.iotdb.metrics.config.MetricConfig;
+import org.apache.iotdb.metrics.config.MetricConfigDescriptor;
+import org.apache.iotdb.metrics.metricsets.IMetricSet;
+import org.apache.iotdb.metrics.type.IMetric;
+import org.apache.iotdb.metrics.type.Rate;
+import org.apache.iotdb.metrics.type.Timer;
+import org.apache.iotdb.metrics.utils.MetricInfo;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Pipe metrics must survive a metric service restart, which drops and rebinds all metrics. */
+public class PipeMetricsRestartTest {
+
+ private static final String PIPE = "pipe";
+ private static final long CREATION_TIME = 1L;
+ private static final String PIPE_ID = PIPE + "_" + CREATION_TIME;
+ private static final String TASK_ID = "task";
+ private static final int REGION_ID = 1;
+
+ private final MetricService service = MetricService.getInstance();
+ private final MetricConfig config = MetricConfigDescriptor.getInstance().getMetricConfig();
+ private final List<IMetricSet> boundMetricSets = new ArrayList<>();
+ private MetricLevel originalLevel;
+ private String originalReporters;
+
+ @Before
+ public void setUp() {
+ originalLevel = config.getMetricLevel();
+ originalReporters =
+ config.getMetricReporterList().stream().map(Enum::name).collect(Collectors.joining(","));
+ config.setMetricLevel(MetricLevel.IMPORTANT);
+ config.setMetricReporterList("");
+ service.startService();
+ }
+
+ @After
+ public void tearDown() {
+ boundMetricSets.forEach(service::removeMetricSet);
+ service.stopService();
+ config.setMetricLevel(originalLevel);
+ config.setMetricReporterList(originalReporters);
+ }
+
+ @Test
+ public void testSinglePipeMetricsKeepOperators() {
+ final PipeDataNodeSinglePipeMetrics metrics = PipeDataNodeSinglePipeMetrics.getInstance();
+ bind(metrics);
+ final IoTDBDataRegionSource source = mockDataRegionSource();
+ metrics.register(source);
+ try {
+ metrics.increaseTsFileEventCount(PIPE, CREATION_TIME);
+ assertEquals(1, count(Metric.PIPE_DATANODE_REMAINING_EVENT_COUNT));
+
+ service.restartService();
+ assertEquals(1, count(Metric.PIPE_DATANODE_REMAINING_EVENT_COUNT));
+ // The operator holding the state of the pipe is kept
+ assertEquals(1L, (long) metrics.getRemainingEventAndTime(PIPE, CREATION_TIME).getLeft());
+ // and records into the timer of the current binding
+ metrics.updateTsFileTransferTimer(PIPE, CREATION_TIME, 10);
+ assertEquals(1, ((Timer) get(Metric.PIPE_TSFILE_EVENT_TRANSFER_TIME)).getCount());
+ } finally {
+ metrics.deregister(PIPE_ID);
+ }
+ assertEquals(0, count(Metric.PIPE_DATANODE_REMAINING_EVENT_COUNT));
+ assertTrue(metrics.remainingEventAndTimeOperatorMap.isEmpty());
+ }
+
+ @Test
+ public void testDataRegionSourceMetrics() {
+ final PipeDataRegionSourceMetrics metrics = PipeDataRegionSourceMetrics.getInstance();
+ bind(metrics);
+ metrics.register(mockDataRegionSource());
+ try {
+ service.restartService();
+ assertEquals(1, count(Metric.UNPROCESSED_TABLET_COUNT));
+ metrics.markTabletEvent(TASK_ID);
+ assertEquals(1, ((Rate) get(Metric.PIPE_EXTRACTOR_TABLET_SUPPLY)).getCount());
+ } finally {
+ metrics.deregister(TASK_ID);
+ }
+ assertEquals(0, count(Metric.UNPROCESSED_TABLET_COUNT));
+ }
+
+ @Test
+ public void testDataRegionSinkMetrics() {
+ final PipeDataRegionSinkMetrics metrics = PipeDataRegionSinkMetrics.getInstance();
+ bind(metrics);
+ final PipeSinkSubtask subtask = mockSinkSubtask();
+ metrics.register(subtask);
+ try {
+ service.restartService();
+ assertEquals(1, count(Metric.UNTRANSFERRED_TABLET_COUNT));
+ assertSame(get(Metric.PIPE_COMPRESSION_TIME), metrics.getCompressionTimer(TASK_ID));
+ // The histograms pushed into the subtask are replaced as well
+ verify(subtask, times(2)).setTabletBatchSizeHistogram(any());
+ } finally {
+ metrics.deregister(TASK_ID);
+ }
+ assertEquals(0, count(Metric.UNTRANSFERRED_TABLET_COUNT));
+ }
+
+ @Test
+ public void testSinksUseCompressionTimerOfCurrentBinding() throws Exception {
+ final PipeDataRegionSinkMetrics metrics = PipeDataRegionSinkMetrics.getInstance();
+ bind(metrics);
+ metrics.register(mockSinkSubtask());
+ try (final IoTDBDataRegionSyncSink syncSink = new IoTDBDataRegionSyncSink();
+ final IoTDBDataRegionAsyncSink asyncSink = new IoTDBDataRegionAsyncSink();
+ final IoTDBDataRegionAirGapSink airGapSink = new IoTDBDataRegionAirGapSink()) {
+ final Method airGapCompress =
+ IoTDBDataRegionAirGapSink.class.getDeclaredMethod("compressIfNeeded", byte[].class);
+ airGapCompress.setAccessible(true);
+ for (final IoTDBSink sink : new IoTDBSink[] {syncSink, asyncSink, airGapSink}) {
+ setField(sink, "sinkTaskId", TASK_ID);
+ }
+
+ for (int i = 0; i < 2; i++) {
+ final Timer timer = metrics.getCompressionTimer(TASK_ID);
+ syncSink.compressIfNeeded(newReq());
+ asyncSink.compressIfNeeded(newReq());
+ airGapCompress.invoke(airGapSink, (Object) new byte[1]);
+ for (final IoTDBSink sink : new IoTDBSink[] {syncSink, asyncSink, airGapSink}) {
+ assertSame(timer, getField(sink, "compressionTimer"));
+ }
+ service.restartService();
+ assertNotSame(timer, metrics.getCompressionTimer(TASK_ID));
+ }
+ } finally {
+ metrics.deregister(TASK_ID);
+ }
+ }
+
+ @Test
+ public void testSchemaRegionSinkMetrics() {
+ final PipeSchemaRegionSinkMetrics metrics = PipeSchemaRegionSinkMetrics.getInstance();
+ bind(metrics);
+ final PipeSinkSubtask subtask = mockSinkSubtask();
+ metrics.register(subtask);
+ try {
+ service.restartService();
+ assertEquals(1, count(Metric.PIPE_CONNECTOR_SCHEMA_TRANSFER));
+ verify(subtask, times(2)).setSchemaBatchSizeHistogram(any());
+ } finally {
+ metrics.deregister(TASK_ID);
+ }
+ assertEquals(0, count(Metric.PIPE_CONNECTOR_SCHEMA_TRANSFER));
+ }
+
+ @Test
+ public void testSchemaRegionSourceMetrics() {
+ final PipeSchemaRegionSourceMetrics metrics = PipeSchemaRegionSourceMetrics.getInstance();
+ bind(metrics);
+ final IoTDBSchemaRegionSource source = Mockito.mock(IoTDBSchemaRegionSource.class);
+ when(source.getTaskID()).thenReturn(TASK_ID);
+ when(source.getPipeName()).thenReturn(PIPE);
+ when(source.getRegionId()).thenReturn(REGION_ID);
+ when(source.getCreationTime()).thenReturn(CREATION_TIME);
+ metrics.register(source);
+ try {
+ service.restartService();
+ assertEquals(1, count(Metric.UNTRANSFERRED_SCHEMA_COUNT));
+ } finally {
+ metrics.deregister(TASK_ID);
+ }
+ assertEquals(0, count(Metric.UNTRANSFERRED_SCHEMA_COUNT));
+ }
+
+ @Test
+ public void testProcessorMetrics() {
+ final PipeProcessorMetrics metrics = PipeProcessorMetrics.getInstance();
+ bind(metrics);
+ final PipeProcessorSubtask subtask = Mockito.mock(PipeProcessorSubtask.class);
+ when(subtask.getTaskID()).thenReturn(TASK_ID);
+ when(subtask.getPipeName()).thenReturn(PIPE);
+ when(subtask.getRegionId()).thenReturn(REGION_ID);
+ when(subtask.getCreationTime()).thenReturn(CREATION_TIME);
+ metrics.register(subtask);
+ try {
+ service.restartService();
+ metrics.markTabletEvent(TASK_ID);
+ assertEquals(1, ((Rate) get(Metric.PIPE_PROCESSOR_TABLET_PROCESS)).getCount());
+ } finally {
+ metrics.deregister(TASK_ID);
+ }
+ assertEquals(0, count(Metric.PIPE_PROCESSOR_TABLET_PROCESS));
+ }
+
+ /**
+ * A subtask may deregister while the metric service restart is unbinding the processor metrics.
+ * The restart must neither fail on it nor drop the metrics of the other subtasks.
+ */
+ @Test
+ public void testProcessorMetricsWithConcurrentDeregistration() throws Exception {
+ final PipeProcessorMetrics metrics = PipeProcessorMetrics.getInstance();
+ bind(metrics);
+ final AtomicBoolean armed = new AtomicBoolean(false);
+ final AtomicReference<PipeProcessorSubtask> paused = new AtomicReference<>();
+ final CountDownLatch pausedLatch = new CountDownLatch(1);
+ final CountDownLatch resumeLatch = new CountDownLatch(1);
+ final PipeProcessorSubtask first =
+ mockBlockingProcessorSubtask("first", armed, paused, pausedLatch, resumeLatch);
+ final PipeProcessorSubtask second =
+ mockBlockingProcessorSubtask("second", armed, paused, pausedLatch, resumeLatch);
+ metrics.register(first);
+ metrics.register(second);
+
+ // Pause the restart while it unbinds the first subtask, and deregister the other one meanwhile
+ armed.set(true);
+ final Thread restart = new Thread(service::restartService);
+ restart.start();
+ Thread deregistration = null;
+ try {
+ assertTrue(pausedLatch.await(30, TimeUnit.SECONDS));
+ final PipeProcessorSubtask other = paused.get() == first ? second : first;
+ deregistration = new Thread(() -> metrics.deregister(other.getTaskID()));
+ deregistration.start();
+ // Either the deregistration completes at once, or it waits for the unbinding to complete
+ final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30);
+ while (deregistration.getState() != Thread.State.TERMINATED
+ && deregistration.getState() != Thread.State.BLOCKED
+ && deregistration.getState() != Thread.State.WAITING
+ && System.nanoTime() < deadline) {
+ Thread.sleep(10);
+ }
+ } finally {
+ resumeLatch.countDown();
+ restart.join(TimeUnit.SECONDS.toMillis(30));
+ if (deregistration != null) {
+ deregistration.join(TimeUnit.SECONDS.toMillis(30));
+ }
+ }
+
+ final String remainingPipe = paused.get().getPipeName();
+ try {
+ assertEquals(
+ 1,
+ service.getAllMetrics().keySet().stream()
+ .filter(
+ info ->
+ Metric.PIPE_PROCESSOR_TABLET_PROCESS.toString().equals(info.getName())
+ && remainingPipe.equals(info.getTags().get("name")))
+ .count());
+ assertEquals(1, count(Metric.PIPE_PROCESSOR_TABLET_PROCESS));
+ } finally {
+ metrics.deregister(paused.get().getTaskID());
+ }
+ }
+
+ private static PipeProcessorSubtask mockBlockingProcessorSubtask(
+ final String name,
+ final AtomicBoolean armed,
+ final AtomicReference<PipeProcessorSubtask> paused,
+ final CountDownLatch pausedLatch,
+ final CountDownLatch resumeLatch) {
+ final PipeProcessorSubtask subtask = Mockito.mock(PipeProcessorSubtask.class);
+ when(subtask.getTaskID()).thenReturn(name);
+ when(subtask.getRegionId()).thenReturn(REGION_ID);
+ when(subtask.getCreationTime()).thenReturn(CREATION_TIME);
+ when(subtask.getPipeName())
+ .thenAnswer(
+ invocation -> {
+ // Only the first call after arming pauses, which is during the unbinding
+ if (armed.get() && paused.compareAndSet(null, subtask)) {
+ pausedLatch.countDown();
+ resumeLatch.await(30, TimeUnit.SECONDS);
+ }
+ return name;
+ });
+ return subtask;
+ }
+
+ @Test
+ public void testAssignerMetrics() {
+ final PipeAssignerMetrics metrics = PipeAssignerMetrics.getInstance();
+ bind(metrics);
+ final PipeDataRegionAssigner assigner = Mockito.mock(PipeDataRegionAssigner.class);
+ when(assigner.getDataRegionId()).thenReturn(REGION_ID);
+ metrics.register(assigner);
+ try {
+ service.restartService();
+ assertEquals(1, count(Metric.UNASSIGNED_TABLET_COUNT));
+ } finally {
+ metrics.deregister(REGION_ID);
+ }
+ assertEquals(0, count(Metric.UNASSIGNED_TABLET_COUNT));
+ }
+
+ @Test
+ public void testTsFileToTabletsMetrics() {
+ final PipeTsFileToTabletsMetrics metrics = PipeTsFileToTabletsMetrics.getInstance();
+ bind(metrics);
+ metrics.register(mockDataRegionSource());
+ try {
+ service.restartService();
+ metrics.recordTsFileToTabletTime(PIPE_ID, 10);
+ assertEquals(1, ((Timer) get(Metric.PIPE_TSFILE_TO_TABLETS_TIME)).getCount());
+ } finally {
+ metrics.deregister(PIPE_ID);
+ }
+ assertEquals(0, count(Metric.PIPE_TSFILE_TO_TABLETS_TIME));
+ }
+
+ private void bind(final IMetricSet metricSet) {
+ service.addMetricSet(metricSet);
+ boundMetricSets.add(metricSet);
+ }
+
+ private static IoTDBDataRegionSource mockDataRegionSource() {
+ final IoTDBDataRegionSource source = Mockito.mock(IoTDBDataRegionSource.class);
+ when(source.getTaskID()).thenReturn(TASK_ID);
+ when(source.getPipeName()).thenReturn(PIPE);
+ when(source.getRegionId()).thenReturn(REGION_ID);
+ when(source.getCreationTime()).thenReturn(CREATION_TIME);
+ return source;
+ }
+
+ private static PipeSinkSubtask mockSinkSubtask() {
+ final PipeSinkSubtask subtask = Mockito.mock(PipeSinkSubtask.class);
+ when(subtask.getTaskID()).thenReturn(TASK_ID);
+ when(subtask.getPipeName()).thenReturn(PIPE);
+ when(subtask.getAttributeSortedString()).thenReturn("sink");
+ when(subtask.getCreationTime()).thenReturn(CREATION_TIME);
+ return subtask;
+ }
+
+ private static TPipeTransferReq newReq() {
+ final TPipeTransferReq req = new TPipeTransferReq();
+ req.body = ByteBuffer.wrap(new byte[1]);
+ return req;
+ }
+
+ private long count(final Metric metric) {
+ return service.getAllMetrics().keySet().stream()
+ .filter(info -> metric.toString().equals(info.getName()))
+ .count();
+ }
+
+ private IMetric get(final Metric metric) {
+ for (final Map.Entry<MetricInfo, IMetric> entry : service.getAllMetrics().entrySet()) {
+ if (metric.toString().equals(entry.getKey().getName())) {
+ return entry.getValue();
+ }
+ }
+ throw new AssertionError(metric + " is not registered");
+ }
+
+ private static void setField(final IoTDBSink sink, final String name, final Object value)
+ throws ReflectiveOperationException {
+ final Field field = IoTDBSink.class.getDeclaredField(name);
+ field.setAccessible(true);
+ field.set(sink, value);
+ }
+
+ private static Object getField(final IoTDBSink sink, final String name)
+ throws ReflectiveOperationException {
+ final Field field = IoTDBSink.class.getDeclaredField(name);
+ field.setAccessible(true);
+ return field.get(sink);
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionListenerMetricsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionListenerMetricsTest.java
new file mode 100644
index 00000000000..2a8e8b8091d
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionListenerMetricsTest.java
@@ -0,0 +1,116 @@
+/*
+ * 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.iotdb.db.pipe.metric.schema;
+
+import org.apache.iotdb.commons.service.metric.MetricService;
+import org.apache.iotdb.commons.service.metric.enums.Metric;
+import org.apache.iotdb.commons.service.metric.enums.Tag;
+import org.apache.iotdb.db.pipe.source.schemaregion.SchemaRegionListeningQueue;
+import org.apache.iotdb.metrics.config.MetricConfig;
+import org.apache.iotdb.metrics.config.MetricConfigDescriptor;
+import org.apache.iotdb.metrics.type.AutoGauge;
+import org.apache.iotdb.metrics.type.IMetric;
+import org.apache.iotdb.metrics.utils.MetricInfo;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+public class PipeSchemaRegionListenerMetricsTest {
+
+ private static final int SCHEMA_REGION_ID = 7;
+
+ private final PipeSchemaRegionListenerMetrics metrics =
+ PipeSchemaRegionListenerMetrics.getInstance();
+ private final MetricService service = MetricService.getInstance();
+ private final MetricConfig config = MetricConfigDescriptor.getInstance().getMetricConfig();
+ private MetricLevel originalLevel;
+ private String originalReporters;
+
+ @Before
+ public void setUp() {
+ originalLevel = config.getMetricLevel();
+ originalReporters =
+ config.getMetricReporterList().stream().map(Enum::name).collect(Collectors.joining(","));
+ config.setMetricLevel(MetricLevel.IMPORTANT);
+ config.setMetricReporterList("");
+ service.startService();
+ service.addMetricSet(metrics);
+ }
+
+ @After
+ public void tearDown() {
+ metrics.deregister(SCHEMA_REGION_ID);
+ service.removeMetricSet(metrics);
+ service.stopService();
+ config.setMetricLevel(originalLevel);
+ config.setMetricReporterList(originalReporters);
+ }
+
+ @Test
+ public void testQueueSizeSurvivesMetricServiceRestart() {
+ final SchemaRegionListeningQueue queue = Mockito.mock(SchemaRegionListeningQueue.class);
+ Mockito.when(queue.getSize()).thenReturn(3L);
+ metrics.register(queue, SCHEMA_REGION_ID);
+ assertQueueSize(3);
+
+ for (final MetricLevel level :
+ new MetricLevel[] {MetricLevel.ALL, MetricLevel.CORE, MetricLevel.IMPORTANT}) {
+ config.setMetricLevel(level);
+ service.restartService();
+ if (level == MetricLevel.CORE) {
+ assertNull(getQueueSizeGauge());
+ } else {
+ assertQueueSize(3);
+ }
+ }
+
+ metrics.deregister(SCHEMA_REGION_ID);
+ assertNull(getQueueSizeGauge());
+ service.restartService();
+ assertNull(getQueueSizeGauge());
+ }
+
+ private AutoGauge getQueueSizeGauge() {
+ for (final Map.Entry<MetricInfo, IMetric> entry : service.getAllMetrics().entrySet()) {
+ if (Metric.PIPE_SCHEMA_LINKED_QUEUE_SIZE.toString().equals(entry.getKey().getName())
+ && String.valueOf(SCHEMA_REGION_ID)
+ .equals(entry.getKey().getTags().get(Tag.REGION.toString()))) {
+ return (AutoGauge) entry.getValue();
+ }
+ }
+ return null;
+ }
+
+ private void assertQueueSize(final double expected) {
+ final AutoGauge gauge = getQueueSizeGauge();
+ assertNotNull(gauge);
+ assertEquals(expected, gauge.getValue(), 0);
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/metric/SchemaMemMetricTableTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/metric/SchemaMemMetricTableTest.java
new file mode 100644
index 00000000000..294f5df6301
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/metric/SchemaMemMetricTableTest.java
@@ -0,0 +1,133 @@
+/*
+ * 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.iotdb.db.schemaengine.metric;
+
+import org.apache.iotdb.commons.service.metric.MetricService;
+import org.apache.iotdb.db.schemaengine.rescon.MemSchemaEngineStatistics;
+import org.apache.iotdb.db.schemaengine.rescon.MemSchemaRegionStatistics;
+import org.apache.iotdb.metrics.config.MetricConfig;
+import org.apache.iotdb.metrics.config.MetricConfigDescriptor;
+import org.apache.iotdb.metrics.metricsets.IMetricSet;
+import org.apache.iotdb.metrics.type.AutoGauge;
+import org.apache.iotdb.metrics.type.IMetric;
+import org.apache.iotdb.metrics.utils.MetricInfo;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * The per-table device number gauges must be exported next to the other schema engine and schema
+ * region metrics, and bound again when the metric service restarts.
+ */
+public class SchemaMemMetricTableTest {
+
+ private static final String TABLE = "t1";
+
+ private final MetricService service = MetricService.getInstance();
+ private final MetricConfig config = MetricConfigDescriptor.getInstance().getMetricConfig();
+ private MetricLevel originalLevel;
+ private String originalReporters;
+
+ @Before
+ public void setUp() {
+ originalLevel = config.getMetricLevel();
+ originalReporters =
+ config.getMetricReporterList().stream().map(Enum::name).collect(Collectors.joining(","));
+ config.setMetricLevel(MetricLevel.IMPORTANT);
+ config.setMetricReporterList("");
+ service.startService();
+ }
+
+ @After
+ public void tearDown() {
+ service.stopService();
+ config.setMetricLevel(originalLevel);
+ config.setMetricReporterList(originalReporters);
+ }
+
+ @Test
+ public void testSchemaEngineTableMetrics() {
+ final Map<String, Long> table2DevicesNum = new ConcurrentHashMap<>();
+ final MemSchemaEngineStatistics statistics = mock(MemSchemaEngineStatistics.class);
+ when(statistics.getTable2DevicesNumMap()).thenReturn(table2DevicesNum);
+ when(statistics.getTableDeviceNumber(TABLE)).thenReturn(3L);
+ final SchemaEngineMemMetric metric = new SchemaEngineMemMetric(statistics);
+ checkTableMetrics(metric, metric::bindTableMetrics, table2DevicesNum);
+ }
+
+ @Test
+ public void testSchemaRegionTableMetrics() {
+ final Map<String, Long> table2DevicesNum = new ConcurrentHashMap<>();
+ final MemSchemaRegionStatistics statistics = mock(MemSchemaRegionStatistics.class);
+ when(statistics.getSchemaRegionId()).thenReturn(1);
+ when(statistics.getTable2DevicesNumMap()).thenReturn(table2DevicesNum);
+ when(statistics.getTableDevicesNumber(TABLE)).thenReturn(3L);
+ final SchemaRegionMemMetric metric = new SchemaRegionMemMetric(statistics, "db");
+ checkTableMetrics(metric, metric::bindTableMetrics, table2DevicesNum);
+ }
+
+ private void checkTableMetrics(
+ final IMetricSet metric,
+ final Consumer<String> bindTableMetrics,
+ final Map<String, Long> table2DevicesNum) {
+ service.addMetricSet(metric);
+ try {
+ // A table binds its metrics when its first device is added
+ table2DevicesNum.put(TABLE, 3L);
+ bindTableMetrics.accept(TABLE);
+ assertEquals(3, getTableGauge().getValue(), 0);
+
+ service.restartService();
+ assertEquals(3, getTableGauge().getValue(), 0);
+ } finally {
+ service.removeMetricSet(metric);
+ }
+ assertNull(findTableGauge());
+ }
+
+ private AutoGauge getTableGauge() {
+ final IMetric metric = findTableGauge();
+ if (metric == null) {
+ throw new AssertionError("The gauge of table " + TABLE + " is not registered");
+ }
+ return (AutoGauge) metric;
+ }
+
+ private IMetric findTableGauge() {
+ for (final Map.Entry<MetricInfo, IMetric> entry : service.getAllMetrics().entrySet()) {
+ if (TABLE.equals(entry.getKey().getTags().get(SchemaEngineMemMetric.TABLE))) {
+ return entry.getValue();
+ }
+ }
+ return null;
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsRestartTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsRestartTest.java
new file mode 100644
index 00000000000..554bdd9940d
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsRestartTest.java
@@ -0,0 +1,290 @@
+/*
+ * 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.iotdb.db.service.metrics;
+
+import org.apache.iotdb.commons.consensus.DataRegionId;
+import org.apache.iotdb.commons.service.metric.MetricService;
+import org.apache.iotdb.commons.service.metric.enums.Metric;
+import org.apache.iotdb.commons.service.metric.enums.Tag;
+import org.apache.iotdb.db.protocol.thrift.handler.RPCServiceThriftHandlerMetrics;
+import org.apache.iotdb.db.storageengine.load.metrics.ActiveLoadingFilesNumberMetricsSet;
+import org.apache.iotdb.db.storageengine.load.metrics.LoadTsFileMemMetricSet;
+import org.apache.iotdb.metrics.config.MetricConfig;
+import org.apache.iotdb.metrics.config.MetricConfigDescriptor;
+import org.apache.iotdb.metrics.metricsets.IMetricSet;
+import org.apache.iotdb.metrics.type.AutoGauge;
+import org.apache.iotdb.metrics.type.Counter;
+import org.apache.iotdb.metrics.type.Gauge;
+import org.apache.iotdb.metrics.type.IMetric;
+import org.apache.iotdb.metrics.utils.MetricInfo;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+
+/** DataNode metrics must survive a metric service restart, which drops and rebinds all metrics. */
+public class DataNodeMetricsRestartTest {
+
+ private final MetricService service = MetricService.getInstance();
+ private final MetricConfig config = MetricConfigDescriptor.getInstance().getMetricConfig();
+ private final List<IMetricSet> boundMetricSets = new ArrayList<>();
+ private MetricLevel originalLevel;
+ private String originalReporters;
+
+ @Before
+ public void setUp() {
+ originalLevel = config.getMetricLevel();
+ originalReporters =
+ config.getMetricReporterList().stream().map(Enum::name).collect(Collectors.joining(","));
+ config.setMetricLevel(MetricLevel.IMPORTANT);
+ config.setMetricReporterList("");
+ service.startService();
+ }
+
+ @After
+ public void tearDown() {
+ boundMetricSets.forEach(service::removeMetricSet);
+ service.stopService();
+ config.setMetricLevel(originalLevel);
+ config.setMetricReporterList(originalReporters);
+ }
+
+ @Test
+ public void testWritingMetricsRecordedValues() {
+ final WritingMetrics metrics = WritingMetrics.getInstance();
+ bind(metrics);
+ metrics.recordFlushThreshold(100);
+ metrics.recordRejectThreshold(200);
+ metrics.recordWALQueueMaxMemorySize(300);
+ metrics.recordActiveMemTableCount("1", 3);
+ metrics.recordActiveTimePartitionCount(2);
+ // A region that does not exist any more
+ metrics.recordActiveMemTableCount("2", 1);
+ metrics.recordActiveMemTableCount("2", -1);
+ try {
+ service.restartService();
+ assertEquals(
+ 100,
+ ((Gauge)
+ get(
+ Metric.MEMTABLE_THRESHOLD,
+ Tag.TYPE.toString(),
+ WritingMetrics.FLUSH_THRESHOLD))
+ .getValue());
+ assertEquals(
+ 200,
+ ((Gauge)
+ get(
+ Metric.MEMTABLE_THRESHOLD,
+ Tag.TYPE.toString(),
+ WritingMetrics.REJECT_THRESHOLD))
+ .getValue());
+ assertEquals(
+ 300,
+ ((Gauge)
+ get(
+ Metric.WAL_QUEUE_MEM_COST,
+ Tag.NAME.toString(),
+ WritingMetrics.WAL_QUEUE_MAX_MEM_COST))
+ .getValue());
+ assertEquals(
+ 3,
+ ((Counter)
+ get(
+ Metric.ACTIVE_MEMTABLE_COUNT,
+ Tag.REGION.toString(),
+ new DataRegionId(1).toString()))
+ .getCount());
+ assertEquals(2, ((Counter) get(Metric.ACTIVE_TIME_PARTITION_COUNT)).getCount());
+ assertEquals(
+ 0,
+ count(
+ Metric.ACTIVE_MEMTABLE_COUNT, Tag.REGION.toString(), new DataRegionId(2).toString()));
+ } finally {
+ metrics.recordActiveMemTableCount("1", -3);
+ metrics.recordActiveTimePartitionCount(-2);
+ }
+ }
+
+ @Test
+ public void testActiveMemTableCountIsRecordedToTheCounterOfTheRegion() {
+ final WritingMetrics metrics = WritingMetrics.getInstance();
+ final DataRegionId dataRegionId = new DataRegionId(3);
+ metrics.createActiveMemtableCounterMetrics(dataRegionId);
+ metrics.recordActiveMemTableCount(String.valueOf(dataRegionId.getId()), 2);
+ try {
+ // The counter created with the region is the one recorded to and removed with the region
+ assertEquals(1, countActiveMemTableCounters(dataRegionId));
+ assertEquals(
+ 2,
+ ((Counter)
+ get(Metric.ACTIVE_MEMTABLE_COUNT, Tag.REGION.toString(), dataRegionId.toString()))
+ .getCount());
+ } finally {
+ metrics.recordActiveMemTableCount(String.valueOf(dataRegionId.getId()), -2);
+ metrics.removeActiveMemtableCounterMetrics(dataRegionId);
+ }
+ assertEquals(0, countActiveMemTableCounters(dataRegionId));
+ }
+
+ @Test
+ public void testLoadTsFileOtherMemory() {
+ final LoadTsFileMemMetricSet metrics = LoadTsFileMemMetricSet.getInstance();
+ bind(metrics);
+ metrics.updateOtherMemory(100);
+ try {
+ service.restartService();
+ assertEquals(100, getLoadTsFileOtherMemory());
+ metrics.updateOtherMemory(-40);
+ assertEquals(60, getLoadTsFileOtherMemory());
+ } finally {
+ metrics.updateOtherMemory(-60);
+ }
+ }
+
+ @Test
+ public void testActiveLoadingFilesCounters() {
+ final ActiveLoadingFilesNumberMetricsSet metrics =
+ ActiveLoadingFilesNumberMetricsSet.getInstance();
+ bind(metrics);
+ metrics.updatePendingDirList(Collections.singleton("pendingDir"));
+ metrics.updateFailedDir("failedDir");
+ metrics.increaseQueuingFileCounter(2);
+ try {
+ service.restartService();
+ metrics.updatePendingFileCounterInDir("pendingDir", 3);
+ metrics.updateTotalFailedFileCounter(4);
+ assertEquals(
+ 3,
+ ((Counter)
+ get(
+ Metric.ACTIVE_LOADING_FILES_NUMBER,
+ Tag.TYPE.toString(),
+ "pending - pendingDir"))
+ .getCount());
+ assertEquals(
+ 4,
+ ((Counter)
+ get(
+ Metric.ACTIVE_LOADING_FILES_NUMBER,
+ Tag.TYPE.toString(),
+ "failed - failedDir"))
+ .getCount());
+ assertEquals(
+ 2,
+ ((Counter) get(Metric.ACTIVE_LOADING_FILES_NUMBER, Tag.TYPE.toString(), "queuing"))
+ .getCount());
+ } finally {
+ metrics.increaseQueuingFileCounter(-2);
+ }
+ }
+
+ @Test
+ public void testRpcServiceThriftHandlerMetrics() {
+ final RPCServiceThriftHandlerMetrics metrics = RPCServiceThriftHandlerMetrics.getInstance();
+ bind(metrics);
+ bind(new RPCServiceThriftHandlerMetrics(new AtomicLong(5)));
+
+ service.restartService();
+ metrics.recordMemoryUsage(7);
+ assertEquals(5, ((AutoGauge) get(Metric.THRIFT_CONNECTIONS)).getValue(), 0);
+ assertEquals(7, ((Gauge) get(Metric.THRIFT_RPC_MEMORY_USAGE)).getValue());
+ }
+
+ @Test
+ public void testCacheMetrics() {
+ final CacheMetrics metrics = CacheMetrics.getInstance();
+ bind(metrics);
+
+ service.restartService();
+ metrics.record(true, CacheMetrics.DATABASE_CACHE_NAME);
+ assertEquals(
+ 1,
+ ((Counter)
+ get(
+ Metric.CACHE,
+ Tag.NAME.toString(),
+ CacheMetrics.DATABASE_CACHE_NAME,
+ Tag.TYPE.toString(),
+ "hit"))
+ .getCount());
+ }
+
+ private void bind(final IMetricSet metricSet) {
+ service.addMetricSet(metricSet);
+ boundMetricSets.add(metricSet);
+ }
+
+ private long getLoadTsFileOtherMemory() {
+ return ((Gauge)
+ get(
+ Metric.LOAD_MEM,
+ Tag.NAME.toString(),
+ LoadTsFileMemMetricSet.LOAD_TSFILE_OTHER_MEMORY))
+ .getValue();
+ }
+
+ private long countActiveMemTableCounters(final DataRegionId dataRegionId) {
+ // Match both forms of the region tag
+ return service.getAllMetrics().keySet().stream()
+ .filter(
+ info ->
+ Metric.ACTIVE_MEMTABLE_COUNT.toString().equals(info.getName())
+ && (dataRegionId.toString().equals(info.getTags().get(Tag.REGION.toString()))
+ || String.valueOf(dataRegionId.getId())
+ .equals(info.getTags().get(Tag.REGION.toString()))))
+ .count();
+ }
+
+ private long count(final Metric metric, final String... tags) {
+ return service.getAllMetrics().keySet().stream()
+ .filter(info -> metric.toString().equals(info.getName()) && hasTags(info, tags))
+ .count();
+ }
+
+ private IMetric get(final Metric metric, final String... tags) {
+ for (final Map.Entry<MetricInfo, IMetric> entry : service.getAllMetrics().entrySet()) {
+ if (metric.toString().equals(entry.getKey().getName()) && hasTags(entry.getKey(), tags)) {
+ return entry.getValue();
+ }
+ }
+ throw new AssertionError(metric + Arrays.toString(tags) + " is not registered");
+ }
+
+ private static boolean hasTags(final MetricInfo info, final String... tags) {
+ for (int i = 0; i < tags.length; i += 2) {
+ if (!tags[i + 1].equals(info.getTags().get(tags[i]))) {
+ return false;
+ }
+ }
+ return true;
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/metrics/file/TsFileMetricsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/metrics/file/TsFileMetricsTest.java
new file mode 100644
index 00000000000..bf846ec1fe6
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/metrics/file/TsFileMetricsTest.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.iotdb.db.service.metrics.file;
+
+import org.apache.iotdb.commons.service.metric.MetricService;
+import org.apache.iotdb.commons.service.metric.enums.Tag;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.metrics.config.MetricConfig;
+import org.apache.iotdb.metrics.config.MetricConfigDescriptor;
+import org.apache.iotdb.metrics.type.Gauge;
+import org.apache.iotdb.metrics.type.IMetric;
+import org.apache.iotdb.metrics.utils.MetricInfo;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.io.File;
+import java.util.Collections;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+public class TsFileMetricsTest {
+
+ private static final String DATABASE = "root.tsfile_metrics_test";
+ private static final String REGION = "1";
+
+ private final TsFileMetrics metrics = new TsFileMetrics();
+ private final MetricService service = MetricService.getInstance();
+ private final MetricConfig config = MetricConfigDescriptor.getInstance().getMetricConfig();
+ private MetricLevel originalLevel;
+ private String originalReporters;
+
+ @Before
+ public void setUp() {
+ originalLevel = config.getMetricLevel();
+ originalReporters =
+ config.getMetricReporterList().stream().map(Enum::name).collect(Collectors.joining(","));
+ config.setMetricLevel(MetricLevel.IMPORTANT);
+ config.setMetricReporterList("");
+ service.startService();
+ service.addMetricSet(metrics);
+ }
+
+ @After
+ public void tearDown() {
+ service.removeMetricSet(metrics);
+ service.stopService();
+ config.setMetricLevel(originalLevel);
+ config.setMetricReporterList(originalReporters);
+ }
+
+ @Test
+ public void testFileGaugesSurviveMetricServiceRestart() {
+ metrics.addTsFile(mockTsFile("1-1-0-0.tsfile", 100));
+ assertFileGauges(1, 100);
+
+ for (final MetricLevel level :
+ new MetricLevel[] {MetricLevel.ALL, MetricLevel.IMPORTANT, MetricLevel.IMPORTANT}) {
+ config.setMetricLevel(level);
+ service.restartService();
+ assertFileGauges(1, 100);
+ }
+
+ // Updates after the restart must reach the gauges that are currently registered.
+ final TsFileResource second = mockTsFile("2-2-0-0.tsfile", 50);
+ metrics.addTsFile(second);
+ assertFileGauges(2, 150);
+ Mockito.when(second.markAsUnrecordedByMetric()).thenReturn(true);
+ metrics.deleteFile(Collections.singletonList(second));
+ assertFileGauges(1, 100);
+ }
+
+ @Test
+ public void testFileGaugesAreCreatedWhenLevelIsEnabledByRestart() {
+ config.setMetricLevel(MetricLevel.OFF);
+ service.restartService();
+ metrics.addTsFile(mockTsFile("1-1-0-0.tsfile", 100));
+
+ config.setMetricLevel(MetricLevel.IMPORTANT);
+ service.restartService();
+ assertFileGauges(1, 100);
+ }
+
+ @Test
+ public void testFileGaugesSurviveMetricServiceStopAndStart() {
+ metrics.addTsFile(mockTsFile("1-1-0-0.tsfile", 100));
+ service.stopService();
+ // Files recorded while the metric service is stopped are exported once it starts again.
+ metrics.addTsFile(mockTsFile("2-2-0-0.tsfile", 50));
+ service.startService();
+ assertFileGauges(2, 150);
+ }
+
+ private static TsFileResource mockTsFile(final String fileName, final long size) {
+ final TsFileResource resource = Mockito.mock(TsFileResource.class);
+ Mockito.when(resource.markAsRecordedByMetric()).thenReturn(true);
+ Mockito.when(resource.getTsFileSize()).thenReturn(size);
+ Mockito.when(resource.isSeq()).thenReturn(true);
+ Mockito.when(resource.getDatabaseName()).thenReturn(DATABASE);
+ Mockito.when(resource.getDataRegionId()).thenReturn(REGION);
+ Mockito.when(resource.getTsFile()).thenReturn(new File(fileName));
+ return resource;
+ }
+
+ private void assertFileGauges(final long count, final long size) {
+ assertEquals(count, getGauge("file_global_count", Tag.REGION.toString(), REGION));
+ assertEquals(size, getGauge("file_global_size", Tag.REGION.toString(), REGION));
+ assertEquals(count, getGauge("file_level_count", "level", "0"));
+ assertEquals(size, getGauge("file_level_size", "level", "0"));
+ }
+
+ private long getGauge(final String name, final String tagKey, final String tagValue) {
+ Gauge gauge = null;
+ for (final Map.Entry<MetricInfo, IMetric> entry : service.getAllMetrics().entrySet()) {
+ final Map<String, String> tags = entry.getKey().getTags();
+ if (name.equals(entry.getKey().getName())
+ && "seq".equals(tags.get(Tag.NAME.toString()))
+ && tagValue.equals(tags.get(tagKey))) {
+ gauge = (Gauge) entry.getValue();
+ }
+ }
+ assertNotNull(name, gauge);
+ return gauge.getValue();
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DatabaseMemMetricsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DatabaseMemMetricsTest.java
new file mode 100644
index 00000000000..799cf8a5f70
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DatabaseMemMetricsTest.java
@@ -0,0 +1,144 @@
+/*
+ * 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.iotdb.db.storageengine.dataregion;
+
+import org.apache.iotdb.commons.service.metric.MetricService;
+import org.apache.iotdb.commons.service.metric.enums.Metric;
+import org.apache.iotdb.commons.service.metric.enums.Tag;
+import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessorInfo;
+import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessorInfoMetrics;
+import org.apache.iotdb.metrics.config.MetricConfig;
+import org.apache.iotdb.metrics.config.MetricConfigDescriptor;
+import org.apache.iotdb.metrics.metricsets.IMetricSet;
+import org.apache.iotdb.metrics.type.AutoGauge;
+import org.apache.iotdb.metrics.type.IMetric;
+import org.apache.iotdb.metrics.utils.MetricInfo;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * A database has several data regions and TsFileProcessors, whose memory is exported as a single
+ * gauge of the database.
+ */
+public class DatabaseMemMetricsTest {
+
+ private static final String DATABASE = "root.db";
+
+ private final MetricService service = MetricService.getInstance();
+ private final MetricConfig config = MetricConfigDescriptor.getInstance().getMetricConfig();
+ private MetricLevel originalLevel;
+ private String originalReporters;
+
+ @Before
+ public void setUp() {
+ originalLevel = config.getMetricLevel();
+ originalReporters =
+ config.getMetricReporterList().stream().map(Enum::name).collect(Collectors.joining(","));
+ config.setMetricLevel(MetricLevel.IMPORTANT);
+ config.setMetricReporterList("");
+ service.startService();
+ }
+
+ @After
+ public void tearDown() {
+ service.stopService();
+ config.setMetricLevel(originalLevel);
+ config.setMetricReporterList(originalReporters);
+ }
+
+ @Test
+ public void testDatabaseMemSumsUpTheDataRegions() {
+ final IMetricSet first = new DataRegionMetrics(mockDataRegion(100));
+ final IMetricSet second = new DataRegionMetrics(mockDataRegion(200));
+ checkSumOfTwoOwners(first, second, "database_" + DATABASE, 300, 100);
+ }
+
+ @Test
+ public void testChunkMetadataMemSumsUpTheTsFileProcessors() {
+ final IMetricSet first = new TsFileProcessorInfoMetrics(DATABASE, mockTsFileProcessorInfo(10));
+ final IMetricSet second = new TsFileProcessorInfoMetrics(DATABASE, mockTsFileProcessorInfo(20));
+ checkSumOfTwoOwners(first, second, "chunkMetaData_" + DATABASE, 30, 10);
+ }
+
+ private void checkSumOfTwoOwners(
+ final IMetricSet first,
+ final IMetricSet second,
+ final String name,
+ final long sum,
+ final long firstValue) {
+ service.addMetricSet(first);
+ service.addMetricSet(second);
+ try {
+ assertEquals(sum, getMem(name).getValue(), 0);
+ service.restartService();
+ assertEquals(sum, getMem(name).getValue(), 0);
+
+ // The gauge stays while the database has any owner left
+ service.removeMetricSet(second);
+ assertEquals(firstValue, getMem(name).getValue(), 0);
+ } finally {
+ service.removeMetricSet(first);
+ service.removeMetricSet(second);
+ }
+ assertNull(findMem(name));
+ }
+
+ private static DataRegion mockDataRegion(final long memCost) {
+ final DataRegion dataRegion = mock(DataRegion.class);
+ when(dataRegion.getDatabaseName()).thenReturn(DATABASE);
+ when(dataRegion.getMemCost()).thenReturn(memCost);
+ return dataRegion;
+ }
+
+ private static TsFileProcessorInfo mockTsFileProcessorInfo(final long memCost) {
+ final TsFileProcessorInfo tsFileProcessorInfo = mock(TsFileProcessorInfo.class);
+ when(tsFileProcessorInfo.getMemCost()).thenReturn(memCost);
+ return tsFileProcessorInfo;
+ }
+
+ private AutoGauge getMem(final String name) {
+ final IMetric metric = findMem(name);
+ if (metric == null) {
+ throw new AssertionError(Metric.MEM + " " + name + " is not registered");
+ }
+ return (AutoGauge) metric;
+ }
+
+ private IMetric findMem(final String name) {
+ for (final Map.Entry<MetricInfo, IMetric> entry : service.getAllMetrics().entrySet()) {
+ if (Metric.MEM.toString().equals(entry.getKey().getName())
+ && name.equals(entry.getKey().getTags().get(Tag.NAME.toString()))) {
+ return entry.getValue();
+ }
+ }
+ return null;
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/SubscriptionMetricsRestartTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/SubscriptionMetricsRestartTest.java
new file mode 100644
index 00000000000..a2142ad61fd
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/SubscriptionMetricsRestartTest.java
@@ -0,0 +1,188 @@
+/*
+ * 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.iotdb.db.subscription.metric;
+
+import org.apache.iotdb.commons.consensus.DataRegionId;
+import org.apache.iotdb.commons.service.metric.MetricService;
+import org.apache.iotdb.commons.service.metric.enums.Metric;
+import org.apache.iotdb.db.subscription.broker.SubscriptionPrefetchingQueue;
+import org.apache.iotdb.db.subscription.broker.consensus.ConsensusPrefetchingQueue;
+import org.apache.iotdb.metrics.config.MetricConfig;
+import org.apache.iotdb.metrics.config.MetricConfigDescriptor;
+import org.apache.iotdb.metrics.metricsets.IMetricSet;
+import org.apache.iotdb.metrics.type.AutoGauge;
+import org.apache.iotdb.metrics.type.IMetric;
+import org.apache.iotdb.metrics.type.Rate;
+import org.apache.iotdb.metrics.utils.MetricInfo;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.when;
+
+/** Subscription metrics must survive a metric service restart, which drops all metrics. */
+public class SubscriptionMetricsRestartTest {
+
+ private static final String QUEUE_ID = "consumer_group_topic";
+
+ private final MetricService service = MetricService.getInstance();
+ private final MetricConfig config = MetricConfigDescriptor.getInstance().getMetricConfig();
+ private final List<IMetricSet> boundMetricSets = new ArrayList<>();
+ private MetricLevel originalLevel;
+ private String originalReporters;
+
+ @Before
+ public void setUp() {
+ originalLevel = config.getMetricLevel();
+ originalReporters =
+ config.getMetricReporterList().stream().map(Enum::name).collect(Collectors.joining(","));
+ config.setMetricLevel(MetricLevel.IMPORTANT);
+ config.setMetricReporterList("");
+ service.startService();
+ }
+
+ @After
+ public void tearDown() {
+ boundMetricSets.forEach(service::removeMetricSet);
+ service.stopService();
+ config.setMetricLevel(originalLevel);
+ config.setMetricReporterList(originalReporters);
+ }
+
+ @Test
+ public void testPrefetchingQueueMetrics() {
+ final SubscriptionPrefetchingQueueMetrics metrics =
+ SubscriptionPrefetchingQueueMetrics.getInstance();
+ bind(metrics);
+ final SubscriptionPrefetchingQueue queue = Mockito.mock(SubscriptionPrefetchingQueue.class);
+ when(queue.getPrefetchingQueueId()).thenReturn(QUEUE_ID);
+ when(queue.getSubscriptionUncommittedEventCount()).thenReturn(3L);
+ metrics.register(queue);
+ try {
+ service.restartService();
+ assertEquals(3, ((AutoGauge) get(Metric.SUBSCRIPTION_UNCOMMITTED_EVENT_COUNT)).getValue(), 0);
+ metrics.mark(QUEUE_ID, 5);
+ assertEquals(5, ((Rate) get(Metric.SUBSCRIPTION_EVENT_TRANSFER)).getCount());
+ } finally {
+ metrics.deregister(QUEUE_ID);
+ }
+ assertEquals(0, count(Metric.SUBSCRIPTION_UNCOMMITTED_EVENT_COUNT));
+ assertEquals(0, count(Metric.SUBSCRIPTION_EVENT_TRANSFER));
+ }
+
+ @Test
+ public void testConsensusPrefetchingQueueMetrics() {
+ final ConsensusSubscriptionPrefetchingQueueMetrics metrics =
+ ConsensusSubscriptionPrefetchingQueueMetrics.getInstance();
+ bind(metrics);
+ final ConsensusPrefetchingQueue queue = Mockito.mock(ConsensusPrefetchingQueue.class);
+ when(queue.getPrefetchingQueueId()).thenReturn(QUEUE_ID);
+ when(queue.getConsensusGroupId()).thenReturn(new DataRegionId(1));
+ when(queue.getLag()).thenReturn(7L);
+ metrics.register(queue);
+ try {
+ service.restartService();
+ assertEquals(7, ((AutoGauge) get(Metric.SUBSCRIPTION_CONSENSUS_LAG)).getValue(), 0);
+ metrics.mark(QUEUE_ID, new DataRegionId(1).toString(), 5);
+ assertEquals(5, ((Rate) get(Metric.SUBSCRIPTION_EVENT_TRANSFER)).getCount());
+ } finally {
+ metrics.deregister(queue);
+ }
+ assertEquals(0, count(Metric.SUBSCRIPTION_CONSENSUS_LAG));
+ assertEquals(0, count(Metric.SUBSCRIPTION_EVENT_TRANSFER));
+ }
+
+ @Test
+ public void testPipeAndConsensusQueuesCoexist() {
+ final SubscriptionPrefetchingQueueMetrics pipeMetrics =
+ SubscriptionPrefetchingQueueMetrics.getInstance();
+ final ConsensusSubscriptionPrefetchingQueueMetrics consensusMetrics =
+ ConsensusSubscriptionPrefetchingQueueMetrics.getInstance();
+ bind(pipeMetrics);
+ bind(consensusMetrics);
+ // Whichever kind of queue registers first must not stop the other kind from exporting metrics
+ final ConsensusPrefetchingQueue firstConsensusQueue = mockConsensusQueue("consensus_topic", 1);
+ final SubscriptionPrefetchingQueue pipeQueue = Mockito.mock(SubscriptionPrefetchingQueue.class);
+ when(pipeQueue.getPrefetchingQueueId()).thenReturn(QUEUE_ID);
+ when(pipeQueue.getSubscriptionUncommittedEventCount()).thenReturn(3L);
+ final ConsensusPrefetchingQueue secondConsensusQueue = mockConsensusQueue("consensus_topic", 2);
+ consensusMetrics.register(firstConsensusQueue);
+ pipeMetrics.register(pipeQueue);
+ consensusMetrics.register(secondConsensusQueue);
+ try {
+ assertEquals(3, count(Metric.SUBSCRIPTION_UNCOMMITTED_EVENT_COUNT));
+ pipeMetrics.mark(QUEUE_ID, 5);
+ consensusMetrics.mark("consensus_topic", new DataRegionId(1).toString(), 7);
+ consensusMetrics.mark("consensus_topic", new DataRegionId(2).toString(), 11);
+ assertEquals(
+ 23,
+ service.getAllMetrics().entrySet().stream()
+ .filter(
+ entry ->
+ Metric.SUBSCRIPTION_EVENT_TRANSFER
+ .toString()
+ .equals(entry.getKey().getName()))
+ .mapToLong(entry -> ((Rate) entry.getValue()).getCount())
+ .sum());
+ } finally {
+ pipeMetrics.deregister(QUEUE_ID);
+ consensusMetrics.deregister(firstConsensusQueue);
+ consensusMetrics.deregister(secondConsensusQueue);
+ }
+ assertEquals(0, count(Metric.SUBSCRIPTION_UNCOMMITTED_EVENT_COUNT));
+ }
+
+ private static ConsensusPrefetchingQueue mockConsensusQueue(
+ final String queueId, final int regionId) {
+ final ConsensusPrefetchingQueue queue = Mockito.mock(ConsensusPrefetchingQueue.class);
+ when(queue.getPrefetchingQueueId()).thenReturn(queueId);
+ when(queue.getConsensusGroupId()).thenReturn(new DataRegionId(regionId));
+ return queue;
+ }
+
+ private void bind(final IMetricSet metricSet) {
+ service.addMetricSet(metricSet);
+ boundMetricSets.add(metricSet);
+ }
+
+ private long count(final Metric metric) {
+ return service.getAllMetrics().keySet().stream()
+ .filter(info -> metric.toString().equals(info.getName()))
+ .count();
+ }
+
+ private IMetric get(final Metric metric) {
+ for (final Map.Entry<MetricInfo, IMetric> entry : service.getAllMetrics().entrySet()) {
+ if (metric.toString().equals(entry.getKey().getName())) {
+ return entry.getValue();
+ }
+ }
+ throw new AssertionError(metric + " is not registered");
+ }
+}
diff --git a/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/jvm/JvmGcMetrics.java b/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/jvm/JvmGcMetrics.java
index ee65e7eff13..3ddf7af9d3e 100644
--- a/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/jvm/JvmGcMetrics.java
+++ b/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/jvm/JvmGcMetrics.java
@@ -43,9 +43,11 @@ import java.lang.management.ManagementFactory;
import java.lang.management.MemoryPoolMXBean;
import java.lang.management.MemoryType;
import java.lang.management.MemoryUsage;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
@@ -59,6 +61,10 @@ public class JvmGcMetrics implements IMetricSet, AutoCloseable {
private String oldGenPoolName;
private String nonGenerationalMemoryPool;
private final Map<String, AtomicLong> lastGcTotalDurationMap = new ConcurrentHashMap<>();
+ // The action and cause tags of the GC pause timers created by the GC notification listeners
+ private final Set<List<String>> gcPauseTimerTags = ConcurrentHashMap.newKeySet();
+ // The ZGC counters created by the GC notification listeners
+ private final Set<String> zgcCounterNames = ConcurrentHashMap.newKeySet();
public JvmGcMetrics() {
for (MemoryPoolMXBean mbean : ManagementFactory.getMemoryPoolMXBeans()) {
@@ -167,6 +173,21 @@ public class JvmGcMetrics implements IMetricSet, AutoCloseable {
// long live heap pool includes old gen heap pool and non-generation heap pool.
final AtomicLong longLivedHeapPoolSizeAfterGc = new AtomicLong();
+ // The listeners create the GC pause timers and the ZGC counters on the first GC, so create the
+ // known ones again when binding again, e.g. when the metric service restarts
+ for (List<String> tags : gcPauseTimerTags) {
+ metricService.getOrCreateTimer(
+ SystemMetric.JVM_GC_PAUSE.toString(),
+ MetricLevel.CORE,
+ "action",
+ tags.get(0),
+ "cause",
+ tags.get(1));
+ }
+ for (String name : zgcCounterNames) {
+ metricService.getOrCreateCounter(name, MetricLevel.CORE);
+ }
+
// start watching for GC notifications
for (GarbageCollectorMXBean mbean : ManagementFactory.getGarbageCollectorMXBeans()) {
if (!(mbean instanceof NotificationEmitter)) {
@@ -210,6 +231,7 @@ public class JvmGcMetrics implements IMetricSet, AutoCloseable {
"cause",
gcCause);
timer.update(duration, TimeUnit.MILLISECONDS);
+ gcPauseTimerTags.add(Arrays.asList(gcAction, gcCause));
// add support for ZGC
if (mbean.getName().equals("ZGC Cycles")) {
@@ -217,11 +239,13 @@ public class JvmGcMetrics implements IMetricSet, AutoCloseable {
metricService.getOrCreateCounter(
SystemMetric.JVM_ZGC_CYCLES_COUNT.toString(), MetricLevel.CORE);
cyclesCount.inc();
+ zgcCounterNames.add(SystemMetric.JVM_ZGC_CYCLES_COUNT.toString());
} else if (mbean.getName().equals("ZGC Pauses")) {
Counter pausesCount =
metricService.getOrCreateCounter(
SystemMetric.JVM_ZGC_PAUSES_COUNT.toString(), MetricLevel.CORE);
pausesCount.inc();
+ zgcCounterNames.add(SystemMetric.JVM_ZGC_PAUSES_COUNT.toString());
}
// Update promotion and allocation counters
@@ -309,6 +333,10 @@ public class JvmGcMetrics implements IMetricSet, AutoCloseable {
return;
}
+ // Stop the listeners added by bindTo, otherwise they keep running after the metric service
+ // restarts, next to the listeners added by the next bindTo
+ close();
+
metricService.remove(MetricType.AUTO_GAUGE, SystemMetric.JVM_GC_MAX_DATA_SIZE_BYTES.toString());
metricService.remove(
MetricType.AUTO_GAUGE, SystemMetric.JVM_GC_LIVE_DATA_SIZE_BYTES.toString());
@@ -330,51 +358,18 @@ public class JvmGcMetrics implements IMetricSet, AutoCloseable {
}
}
- // start watching for GC notifications
- for (GarbageCollectorMXBean mbean : ManagementFactory.getGarbageCollectorMXBeans()) {
- if (!(mbean instanceof NotificationEmitter)) {
- continue;
- }
- NotificationListener notificationListener =
- (notification, ref) -> {
- CompositeData cd = (CompositeData) notification.getUserData();
- GarbageCollectionNotificationInfo notificationInfo =
- GarbageCollectionNotificationInfo.from(cd);
-
- String gcCause = notificationInfo.getGcCause();
- String gcAction = notificationInfo.getGcAction();
- metricService.remove(
- MetricType.TIMER,
- SystemMetric.JVM_GC_PAUSE.toString(),
- "action",
- gcAction,
- "cause",
- gcCause);
-
- if (mbean.getName().equals("ZGC Cycles")) {
- metricService.remove(
- MetricType.COUNTER, SystemMetric.JVM_ZGC_CYCLES_COUNT.toString());
- } else if (mbean.getName().equals("ZGC Pauses")) {
- metricService.remove(
- MetricType.COUNTER, SystemMetric.JVM_ZGC_PAUSES_COUNT.toString());
- }
- };
- NotificationEmitter notificationEmitter = (NotificationEmitter) mbean;
- notificationEmitter.addNotificationListener(
- notificationListener,
- notification ->
- notification
- .getType()
- .equals(GarbageCollectionNotificationInfo.GARBAGE_COLLECTION_NOTIFICATION),
- null);
- notificationListenerCleanUpRunnables.add(
- () -> {
- try {
- notificationEmitter.removeNotificationListener(notificationListener);
- } catch (ListenerNotFoundException ignore) {
- // do nothing
- }
- });
+ // Keep the tags and names, the next bindTo creates the metrics again
+ for (List<String> tags : gcPauseTimerTags) {
+ metricService.remove(
+ MetricType.TIMER,
+ SystemMetric.JVM_GC_PAUSE.toString(),
+ "action",
+ tags.get(0),
+ "cause",
+ tags.get(1));
+ }
+ for (String name : zgcCounterNames) {
+ metricService.remove(MetricType.COUNTER, name);
}
}
@@ -420,6 +415,7 @@ public class JvmGcMetrics implements IMetricSet, AutoCloseable {
@Override
public void close() {
notificationListenerCleanUpRunnables.forEach(Runnable::run);
+ notificationListenerCleanUpRunnables.clear();
}
enum GcGenerationAge {
diff --git a/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/system/SystemMetrics.java b/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/system/SystemMetrics.java
index e25a1cb4fbd..ce447ac482c 100644
--- a/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/system/SystemMetrics.java
+++ b/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/system/SystemMetrics.java
@@ -42,6 +42,7 @@ import java.io.IOException;
import java.io.InputStreamReader;
import java.lang.management.ManagementFactory;
import java.nio.file.FileStore;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -61,7 +62,8 @@ public class SystemMetrics implements IMetricSet {
static final String SYSTEM = "system";
private final com.sun.management.OperatingSystemMXBean osMxBean;
- private Set<FileStore> fileStores = new HashSet<>();
+ private volatile Set<FileStore> fileStores = new HashSet<>();
+ private volatile List<String> diskDirs = Collections.emptyList();
private static final String FAILED_TO_STATISTIC = "Failed to statistic the size of {}, because";
public SystemMetrics() {
@@ -69,6 +71,7 @@ public class SystemMetrics implements IMetricSet {
}
public void setDiskDirs(List<String> diskDirs) {
+ this.diskDirs = diskDirs;
if (!MetricConfigDescriptor.getInstance()
.getMetricConfig()
.getMetricLevel()
@@ -292,6 +295,11 @@ public class SystemMetrics implements IMetricSet {
}
private void collectSystemDiskInfo(AbstractMetricService metricService) {
+ // The file stores are skipped when the metric level is OFF, so collect them if the metric
+ // service restarts with another level
+ if (fileStores.isEmpty() && !CONFIG.getMetricLevel().equals(MetricLevel.OFF)) {
+ fileStores = getFileStores(diskDirs);
+ }
metricService.createAutoGauge(
SystemMetric.SYS_DISK_TOTAL_SPACE.toString(),
MetricLevel.CORE,
@@ -331,7 +339,8 @@ public class SystemMetrics implements IMetricSet {
SystemMetric.SYS_DISK_AVAILABLE_SPACE.toString(),
SystemTag.NAME.toString(),
SYSTEM);
- fileStores.clear();
+ // Keep the file stores: they are also read by the disk load sampling, and are only collected
+ // again when the disk dirs change, not when the metric service restarts
}
public long getSystemDiskTotalSpace() {
diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java
index e8646e005a3..0aafc358427 100644
--- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java
+++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java
@@ -566,8 +566,6 @@ public final class PipeMessages {
// ===================== PipeEventCommitMetrics =====================
- public static final String FAILED_TO_UNBIND_COMMIT_METRICS =
- "Failed to unbind from pipe event commit metrics, event committer map not empty";
// ===================== PipePhantomReferenceManager =====================
diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/ServiceMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/ServiceMessages.java
index 8c8bfbc0d29..6b2462b9d2a 100644
--- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/ServiceMessages.java
+++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/ServiceMessages.java
@@ -117,6 +117,8 @@ public final class ServiceMessages {
"MetricService try to restart.";
public static final String METRIC_SERVICE_REBIND_METRIC_SET =
"MetricService rebind metricSet: {}";
+ public static final String LOG_FAILED_TO_REBIND_METRIC_SET_ARG_2C440B71 =
+ "Failed to rebind metric set {}";
public static final String METRIC_SERVICE_RESTART_SUCCESSFULLY =
"MetricService restart successfully.";
public static final String METRIC_SERVICE_TRY_TO_STOP = "MetricService try to stop.";
diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java
index c6da4c6082e..9746b71500d 100644
--- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java
+++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java
@@ -546,8 +546,6 @@ public final class PipeMessages {
// ===================== PipeEventCommitMetrics =====================
- public static final String FAILED_TO_UNBIND_COMMIT_METRICS =
- "从 pipe 事件提交指标解绑失败，事件提交器映射不为空";
// ===================== PipePhantomReferenceManager =====================
diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/ServiceMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/ServiceMessages.java
index a96ae6f449d..66ce5ebfaf1 100644
--- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/ServiceMessages.java
+++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/ServiceMessages.java
@@ -117,6 +117,8 @@ public final class ServiceMessages {
"指标服务尝试重启。";
public static final String METRIC_SERVICE_REBIND_METRIC_SET =
"指标服务重新绑定指标集：{}";
+ public static final String LOG_FAILED_TO_REBIND_METRIC_SET_ARG_2C440B71 =
+ "重新绑定指标集 {} 失败";
public static final String METRIC_SERVICE_RESTART_SUCCESSFULLY =
"指标服务重启成功。";
public static final String METRIC_SERVICE_TRY_TO_STOP = "指标服务尝试停止。";
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadPoolMetrics.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadPoolMetrics.java
index 0523daf05a0..37566ccd740 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadPoolMetrics.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadPoolMetrics.java
@@ -29,14 +29,12 @@ import org.apache.iotdb.metrics.utils.SystemTag;
import java.util.HashMap;
import java.util.Map;
-import java.util.Optional;
@SuppressWarnings("java:S6548")
public class ThreadPoolMetrics implements IMetricSet {
private AbstractMetricService metricService;
- private final Map<String, IThreadPoolMBean> notRegisteredPoolMap = new HashMap<>();
- private final Map<String, IThreadPoolMBean> registeredPoolMap = new HashMap<>();
+ private final Map<String, IThreadPoolMBean> poolMap = new HashMap<>();
public static ThreadPoolMetrics getInstance() {
return ThreadPoolMetricsHolder.INSTANCE;
@@ -45,72 +43,53 @@ public class ThreadPoolMetrics implements IMetricSet {
private ThreadPoolMetrics() {}
public synchronized void registerThreadPool(IThreadPoolMBean pool, String name) {
- if (metricService == null) {
- notRegisteredPoolMap.put(name, pool);
- } else {
- registeredPoolMap.put(name, pool);
- registerThreadPoolMetrics(name);
+ poolMap.put(name, pool);
+ if (metricService != null) {
+ registerThreadPoolMetrics(name, pool);
}
}
- private void registerThreadPoolMetrics(String name) {
+ private void registerThreadPoolMetrics(String name, IThreadPoolMBean pool) {
metricService.createAutoGauge(
SystemMetric.THREAD_POOL_ACTIVE_THREAD_COUNT.toString(),
MetricLevel.IMPORTANT,
- registeredPoolMap,
- map ->
- Optional.ofNullable(registeredPoolMap.get(name))
- .map(IThreadPoolMBean::getActiveCount)
- .orElse(0),
+ pool,
+ IThreadPoolMBean::getActiveCount,
SystemTag.POOL_NAME.toString(),
name);
metricService.createAutoGauge(
SystemMetric.THREAD_POOL_CORE_SIZE.toString(),
MetricLevel.IMPORTANT,
- registeredPoolMap,
- map ->
- Optional.ofNullable(registeredPoolMap.get(name))
- .map(IThreadPoolMBean::getCorePoolSize)
- .orElse(0),
+ pool,
+ IThreadPoolMBean::getCorePoolSize,
SystemTag.POOL_NAME.toString(),
name);
metricService.createAutoGauge(
SystemMetric.THREAD_POOL_WAITING_TASK_COUNT.toString(),
MetricLevel.IMPORTANT,
- registeredPoolMap,
- map ->
- Optional.ofNullable(registeredPoolMap.get(name))
- .map(IThreadPoolMBean::getQueueLength)
- .orElse(0),
+ pool,
+ IThreadPoolMBean::getQueueLength,
SystemTag.POOL_NAME.toString(),
name);
metricService.createAutoGauge(
SystemMetric.THREAD_POOL_DONE_TASK_COUNT.toString(),
MetricLevel.IMPORTANT,
- registeredPoolMap,
- map ->
- Optional.ofNullable(registeredPoolMap.get(name))
- .map(IThreadPoolMBean::getCompletedTaskCount)
- .orElse(0L),
+ pool,
+ IThreadPoolMBean::getCompletedTaskCount,
SystemTag.POOL_NAME.toString(),
name);
metricService.createAutoGauge(
SystemMetric.THREAD_POOL_LARGEST_POOL_SIZE.toString(),
MetricLevel.IMPORTANT,
- registeredPoolMap,
- map ->
- Optional.ofNullable(registeredPoolMap.get(name))
- .map(IThreadPoolMBean::getLargestPoolSize)
- .orElse(0),
+ pool,
+ IThreadPoolMBean::getLargestPoolSize,
SystemTag.POOL_NAME.toString(),
name);
}
- public synchronized void unRegisterThreadPool(String name) {
- if (metricService == null) {
- notRegisteredPoolMap.remove(name);
- } else {
- registeredPoolMap.remove(name);
+ public synchronized void unRegisterThreadPool(IThreadPoolMBean pool, String name) {
+ // A pool with the same name may have been registered while the old pool was shutting down.
+ if (poolMap.remove(name, pool) && metricService != null) {
unRegisterThreadPoolMetrics(name);
}
}
@@ -146,20 +125,21 @@ public class ThreadPoolMetrics implements IMetricSet {
@Override
public synchronized void bindTo(AbstractMetricService metricService) {
this.metricService = metricService;
- for (Map.Entry<String, IThreadPoolMBean> entry : notRegisteredPoolMap.entrySet()) {
- registerThreadPoolMetrics(entry.getKey());
+ for (Map.Entry<String, IThreadPoolMBean> entry : poolMap.entrySet()) {
+ registerThreadPoolMetrics(entry.getKey(), entry.getValue());
}
- registeredPoolMap.putAll(notRegisteredPoolMap);
- notRegisteredPoolMap.clear();
}
@Override
public synchronized void unbindFrom(AbstractMetricService metricService) {
- for (Map.Entry<String, IThreadPoolMBean> entry : registeredPoolMap.entrySet()) {
- unRegisterThreadPoolMetrics(entry.getKey());
+ if (this.metricService != metricService) {
+ return;
}
- registeredPoolMap.clear();
- notRegisteredPoolMap.clear();
+ // Keep live pools registered so a metric service restart can bind them again.
+ for (String name : poolMap.keySet()) {
+ unRegisterThreadPoolMetrics(name);
+ }
+ this.metricService = null;
}
private static class ThreadPoolMetricsHolder {
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedScheduledExecutorService.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedScheduledExecutorService.java
index 05aa8047739..524ced4858a 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedScheduledExecutorService.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedScheduledExecutorService.java
@@ -80,13 +80,13 @@ public class WrappedScheduledExecutorService
public void shutdown() {
service.shutdown();
JMXService.deregisterMBean(mbeanName);
- ThreadPoolMetrics.getInstance().unRegisterThreadPool(this.mbeanName);
+ ThreadPoolMetrics.getInstance().unRegisterThreadPool(this, this.mbeanName);
}
@Override
public List<Runnable> shutdownNow() {
JMXService.deregisterMBean(mbeanName);
- ThreadPoolMetrics.getInstance().unRegisterThreadPool(this.mbeanName);
+ ThreadPoolMetrics.getInstance().unRegisterThreadPool(this, this.mbeanName);
return service.shutdownNow();
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadExecutorService.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadExecutorService.java
index 604a2ae438b..f51792e5f9a 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadExecutorService.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadExecutorService.java
@@ -56,13 +56,13 @@ public class WrappedSingleThreadExecutorService
public void shutdown() {
service.shutdown();
JMXService.deregisterMBean(mbeanName);
- ThreadPoolMetrics.getInstance().unRegisterThreadPool(this.mbeanName);
+ ThreadPoolMetrics.getInstance().unRegisterThreadPool(this, this.mbeanName);
}
@Override
public List<Runnable> shutdownNow() {
JMXService.deregisterMBean(mbeanName);
- ThreadPoolMetrics.getInstance().unRegisterThreadPool(this.mbeanName);
+ ThreadPoolMetrics.getInstance().unRegisterThreadPool(this, this.mbeanName);
return service.shutdownNow();
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadScheduledExecutor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadScheduledExecutor.java
index 406a00371f4..04330f9728c 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadScheduledExecutor.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadScheduledExecutor.java
@@ -80,13 +80,13 @@ public class WrappedSingleThreadScheduledExecutor
public void shutdown() {
service.shutdown();
JMXService.deregisterMBean(mbeanName);
- ThreadPoolMetrics.getInstance().unRegisterThreadPool(this.mbeanName);
+ ThreadPoolMetrics.getInstance().unRegisterThreadPool(this, this.mbeanName);
}
@Override
public List<Runnable> shutdownNow() {
JMXService.deregisterMBean(mbeanName);
- ThreadPoolMetrics.getInstance().unRegisterThreadPool(this.mbeanName);
+ ThreadPoolMetrics.getInstance().unRegisterThreadPool(this, this.mbeanName);
return service.shutdownNow();
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedThreadPoolExecutor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedThreadPoolExecutor.java
index 52780294882..e2df59a6922 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedThreadPoolExecutor.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedThreadPoolExecutor.java
@@ -97,13 +97,13 @@ public class WrappedThreadPoolExecutor extends ThreadPoolExecutor
public void shutdown() {
super.shutdown();
JMXService.deregisterMBean(mbeanName);
- ThreadPoolMetrics.getInstance().unRegisterThreadPool(this.mbeanName);
+ ThreadPoolMetrics.getInstance().unRegisterThreadPool(this, this.mbeanName);
}
@Override
public List<Runnable> shutdownNow() {
JMXService.deregisterMBean(mbeanName);
- ThreadPoolMetrics.getInstance().unRegisterThreadPool(this.mbeanName);
+ ThreadPoolMetrics.getInstance().unRegisterThreadPool(this, this.mbeanName);
return super.shutdownNow();
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/metric/PipeEventCommitMetrics.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/metric/PipeEventCommitMetrics.java
index 05737e5643c..7c3ba736ce4 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/metric/PipeEventCommitMetrics.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/metric/PipeEventCommitMetrics.java
@@ -47,7 +47,7 @@ public class PipeEventCommitMetrics implements IMetricSet {
//////////////////////////// bindTo & unbindFrom (metric framework) ////////////////////////////
@Override
- public void bindTo(final AbstractMetricService metricService) {
+ public synchronized void bindTo(final AbstractMetricService metricService) {
this.metricService = metricService;
final ImmutableSet<String> committerKeys = ImmutableSet.copyOf(eventCommitterMap.keySet());
for (String committerKey : committerKeys) {
@@ -73,14 +73,11 @@ public class PipeEventCommitMetrics implements IMetricSet {
}
@Override
- public void unbindFrom(final AbstractMetricService metricService) {
- final ImmutableSet<String> committerKeys = ImmutableSet.copyOf(eventCommitterMap.keySet());
- for (String committerKey : committerKeys) {
- deregister(committerKey);
- }
- if (!eventCommitterMap.isEmpty()) {
- LOGGER.warn(PipeMessages.FAILED_TO_UNBIND_COMMIT_METRICS);
- }
+ public synchronized void unbindFrom(final AbstractMetricService metricService) {
+ // Keep the committers registered: they register only once, so a metric service restart
+ // must be able to bind them again.
+ // Synchronized with the (de)registrations, which may remove a registration being unbound.
+ ImmutableSet.copyOf(eventCommitterMap.keySet()).forEach(this::removeMetrics);
}
private void removeMetrics(final String committerKey) {
@@ -100,7 +97,8 @@ public class PipeEventCommitMetrics implements IMetricSet {
//////////////////////////// register & deregister (pipe integration) ////////////////////////////
- public void register(final PipeEventCommitter eventCommitter, final String committerKey) {
+ public synchronized void register(
+ final PipeEventCommitter eventCommitter, final String committerKey) {
if (Objects.isNull(eventCommitter)) {
return;
}
@@ -111,7 +109,7 @@ public class PipeEventCommitMetrics implements IMetricSet {
}
}
- public void deregister(final String committerKey) {
+ public synchronized void deregister(final String committerKey) {
if (!eventCommitterMap.containsKey(committerKey)) {
LOGGER.warn(PipeMessages.FAILED_TO_DEREGISTER_COMMIT_METRICS, committerKey);
return;
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/JvmGcMonitorMetrics.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/JvmGcMonitorMetrics.java
index 0bdd863950d..11feb13c43d 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/JvmGcMonitorMetrics.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/JvmGcMonitorMetrics.java
@@ -88,16 +88,14 @@ public class JvmGcMonitorMetrics implements IMetricSet {
curData,
GcData::getGcTimePercentage);
- monitorStartTime = System.currentTimeMillis();
- // Set start time's accumulated GC Time
- curData.setAccumulatedGcTime(getTotalGCTime());
- // current collect time: startTime + start delay(50ms)
- gcDataBuf[startIdx].setValues(monitorStartTime + TimeUnit.MILLISECONDS.toMillis(50), 0);
+ startMonitoring(System.currentTimeMillis(), getTotalGCTime());
scheduledGcMonitorFuture =
ScheduledExecutorUtil.safelyScheduleWithFixedDelay(
scheduledGCInfoMonitor,
this::scheduledMonitoring,
- TimeUnit.MILLISECONDS.toMillis(50), // to prevent / ZERO exception
+ // The first sample covers a whole interval, as the later ones do, so the percentage is
+ // never calculated over a tiny window
+ SLEEP_INTERVAL_MS,
SLEEP_INTERVAL_MS,
TimeUnit.MILLISECONDS);
}
@@ -113,8 +111,21 @@ public class JvmGcMonitorMetrics implements IMetricSet {
}
}
+ /**
+ * Starts a new observation. The buffer may still hold the samples of a previous binding, e.g.
+ * before the metric service restarts, so start again from an empty buffer.
+ */
+ synchronized void startMonitoring(long startTime, long totalGcTime) {
+ monitorStartTime = startTime;
+ // Set start time's accumulated GC Time
+ curData.setAccumulatedGcTime(totalGcTime);
+ startIdx = 0;
+ endIdx = 0;
+ gcDataBuf[startIdx].setValues(startTime, 0);
+ }
+
private void scheduledMonitoring() {
- calculateGCTimePercentageWithinObservedInterval();
+ calculateGCTimePercentageWithinObservedInterval(System.currentTimeMillis(), getTotalGCTime());
// Alert if necessary
if (alertHandler != null && curData.getGcTimePercentage() > MAX_GC_TIME_PERCENTAGE) {
@@ -130,12 +141,11 @@ public class JvmGcMonitorMetrics implements IMetricSet {
return totalGcTime;
}
- private void calculateGCTimePercentageWithinObservedInterval() {
+ synchronized void calculateGCTimePercentageWithinObservedInterval(
+ long curTime, long totalGcTime) {
long prevTotalGcTime = curData.getAccumulatedGcTime();
- long totalGcTime = getTotalGCTime();
long gcTimeWithinSleepInterval = totalGcTime - prevTotalGcTime;
- long curTime = System.currentTimeMillis();
long gcMonitorRunTime = curTime - monitorStartTime;
endIdx = (endIdx + 1) % bufSize;
@@ -153,10 +163,10 @@ public class JvmGcMonitorMetrics implements IMetricSet {
// in our observation window.
long gcTimeWithinObservationWindow =
Math.min(gcDataBuf[startIdx].gcPause, gcDataBuf[startIdx].ts - startObsWindowTs);
- if (startIdx != endIdx) {
- for (int i = (startIdx + 1) % bufSize; i != endIdx; i = (i + 1) % bufSize) {
- gcTimeWithinObservationWindow += gcDataBuf[i].gcPause;
- }
+ // The entries after the start one are all within the window, including the latest one
+ for (int i = startIdx; i != endIdx; ) {
+ i = (i + 1) % bufSize;
+ gcTimeWithinObservationWindow += gcDataBuf[i].gcPause;
}
curData.update(
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/MetricService.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/MetricService.java
index 4af78e3d9cf..69de6df79d2 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/MetricService.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/MetricService.java
@@ -130,8 +130,17 @@ public class MetricService extends AbstractMetricService implements MetricServic
for (IMetricSet metricSet : metricSets) {
LOGGER.info(
ServiceMessages.METRIC_SERVICE_REBIND_METRIC_SET, metricSet.getClass().getName());
- metricSet.unbindFrom(this);
- metricSet.bindTo(this);
+ // Every metric was dropped by the restart, so a failed metric set must not stop the others
+ // from being bound again
+ try {
+ metricSet.unbindFrom(this);
+ metricSet.bindTo(this);
+ } catch (Exception e) {
+ LOGGER.warn(
+ ServiceMessages.LOG_FAILED_TO_REBIND_METRIC_SET_ARG_2C440B71,
+ metricSet.getClass().getName(),
+ e);
+ }
}
}
LOGGER.info(ServiceMessages.METRIC_SERVICE_RESTART_SUCCESSFULLY);
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
index c6e0213630f..5bcafffaa14 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
@@ -116,6 +116,9 @@ public enum Metric {
LEADER_QUANTITY("leader_quantity"),
SCHEMA_REGION("schema_region"),
SCHEMA_ENGINE("schema_engine"),
+ // The per-table metrics have an extra table tag, so they cannot share the names above
+ SCHEMA_REGION_TABLE("schema_region_table"),
+ SCHEMA_ENGINE_TABLE("schema_engine_table"),
// query engine related
QUERY_PLAN_COST("query_plan_cost"),
OPERATOR_EXECUTION_COST("operator_execution_cost"),
diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/concurrent/ThreadPoolMetricsTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/concurrent/ThreadPoolMetricsTest.java
new file mode 100644
index 00000000000..c65554fba34
--- /dev/null
+++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/concurrent/ThreadPoolMetricsTest.java
@@ -0,0 +1,157 @@
+/*
+ * 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.iotdb.commons.concurrent;
+
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.commons.service.metric.MetricService;
+import org.apache.iotdb.metrics.config.MetricConfig;
+import org.apache.iotdb.metrics.config.MetricConfigDescriptor;
+import org.apache.iotdb.metrics.type.AutoGauge;
+import org.apache.iotdb.metrics.type.IMetric;
+import org.apache.iotdb.metrics.utils.MetricInfo;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+import org.apache.iotdb.metrics.utils.SystemMetric;
+import org.apache.iotdb.metrics.utils.SystemTag;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class ThreadPoolMetricsTest {
+
+ private static final String POOL_PREFIX = "ThreadPoolMetricsTest-";
+
+ private final ThreadPoolMetrics metrics = ThreadPoolMetrics.getInstance();
+ private final MetricService service = MetricService.getInstance();
+ private final MetricConfig config = MetricConfigDescriptor.getInstance().getMetricConfig();
+ private final List<ExecutorService> pools = new ArrayList<>();
+ private MetricLevel originalLevel;
+ private String originalReporters;
+
+ @Before
+ public void setUp() {
+ originalLevel = config.getMetricLevel();
+ originalReporters =
+ config.getMetricReporterList().stream().map(Enum::name).collect(Collectors.joining(","));
+ config.setMetricLevel(MetricLevel.IMPORTANT);
+ config.setMetricReporterList("");
+ service.startService();
+ service.addMetricSet(metrics);
+ }
+
+ @After
+ public void tearDown() {
+ pools.forEach(ExecutorService::shutdownNow);
+ service.removeMetricSet(metrics);
+ service.stopService();
+ config.setMetricLevel(originalLevel);
+ config.setMetricReporterList(originalReporters);
+ }
+
+ @Test
+ public void testMetricsSurviveMetricServiceRestart() {
+ createPool("restart", 2);
+ assertPoolMetrics("restart", 2);
+
+ for (MetricLevel level :
+ new MetricLevel[] {
+ MetricLevel.ALL, MetricLevel.CORE, MetricLevel.IMPORTANT, MetricLevel.IMPORTANT
+ }) {
+ config.setMetricLevel(level);
+ service.restartService();
+ if (level == MetricLevel.CORE) {
+ assertTrue(getPoolMetrics("restart").isEmpty());
+ } else {
+ assertPoolMetrics("restart", 2);
+ }
+ }
+ }
+
+ @Test
+ public void testShutdownPoolIsNotRestoredByRestart() {
+ ExecutorService pool = createPool("shutdown", 1);
+ assertPoolMetrics("shutdown", 1);
+
+ pool.shutdown();
+ assertTrue(getPoolMetrics("shutdown").isEmpty());
+ service.restartService();
+ assertTrue(getPoolMetrics("shutdown").isEmpty());
+ }
+
+ @Test
+ public void testRegistrationWhileUnbound() {
+ service.removeMetricSet(metrics);
+ createPool("unbound", 3);
+ assertTrue(getPoolMetrics("unbound").isEmpty());
+
+ service.addMetricSet(metrics);
+ assertPoolMetrics("unbound", 3);
+ }
+
+ @Test
+ public void testShuttingDownSupersededPoolKeepsReplacementMetrics() {
+ ExecutorService first = createPool("replaced", 1);
+ ExecutorService replacement = createPool("replaced", 4);
+
+ first.shutdown();
+ assertPoolMetrics("replaced", 4);
+ service.restartService();
+ assertPoolMetrics("replaced", 4);
+
+ replacement.shutdown();
+ assertTrue(getPoolMetrics("replaced").isEmpty());
+ }
+
+ private ExecutorService createPool(String name, int size) {
+ ExecutorService pool = IoTDBThreadPoolFactory.newFixedThreadPool(size, POOL_PREFIX + name);
+ pools.add(pool);
+ return pool;
+ }
+
+ private Map<String, Double> getPoolMetrics(String name) {
+ String poolName =
+ String.format(
+ "%s:%s=%s",
+ IoTDBConstant.IOTDB_THREADPOOL_JMX_NAME, IoTDBConstant.JMX_TYPE, POOL_PREFIX + name);
+ Map<String, Double> poolMetrics = new HashMap<>();
+ for (Map.Entry<MetricInfo, IMetric> entry : service.getAllMetrics().entrySet()) {
+ if (poolName.equals(entry.getKey().getTags().get(SystemTag.POOL_NAME.toString()))) {
+ poolMetrics.put(entry.getKey().getName(), ((AutoGauge) entry.getValue()).getValue());
+ }
+ }
+ return poolMetrics;
+ }
+
+ private void assertPoolMetrics(String name, int coreSize) {
+ Map<String, Double> poolMetrics = getPoolMetrics(name);
+ assertEquals(5, poolMetrics.size());
+ assertEquals(coreSize, poolMetrics.get(SystemMetric.THREAD_POOL_CORE_SIZE.toString()), 0);
+ }
+}
diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/metric/PipeEventCommitMetricsTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/metric/PipeEventCommitMetricsTest.java
new file mode 100644
index 00000000000..ed6b18d8bd0
--- /dev/null
+++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/metric/PipeEventCommitMetricsTest.java
@@ -0,0 +1,99 @@
+/*
+ * 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.iotdb.commons.pipe.metric;
+
+import org.apache.iotdb.commons.pipe.agent.task.progress.PipeEventCommitter;
+import org.apache.iotdb.commons.service.metric.MetricService;
+import org.apache.iotdb.commons.service.metric.enums.Metric;
+import org.apache.iotdb.metrics.config.MetricConfig;
+import org.apache.iotdb.metrics.config.MetricConfigDescriptor;
+import org.apache.iotdb.metrics.type.AutoGauge;
+import org.apache.iotdb.metrics.type.IMetric;
+import org.apache.iotdb.metrics.utils.MetricInfo;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+public class PipeEventCommitMetricsTest {
+
+ private static final String COMMITTER_KEY = "committer";
+
+ private final PipeEventCommitMetrics metrics = PipeEventCommitMetrics.getInstance();
+ private final MetricService service = MetricService.getInstance();
+ private final MetricConfig config = MetricConfigDescriptor.getInstance().getMetricConfig();
+ private MetricLevel originalLevel;
+ private String originalReporters;
+
+ @Before
+ public void setUp() {
+ originalLevel = config.getMetricLevel();
+ originalReporters =
+ config.getMetricReporterList().stream().map(Enum::name).collect(Collectors.joining(","));
+ config.setMetricLevel(MetricLevel.IMPORTANT);
+ config.setMetricReporterList("");
+ service.startService();
+ service.addMetricSet(metrics);
+ }
+
+ @After
+ public void tearDown() {
+ service.removeMetricSet(metrics);
+ service.stopService();
+ config.setMetricLevel(originalLevel);
+ config.setMetricReporterList(originalReporters);
+ }
+
+ @Test
+ public void testCommitQueueSizeSurvivesMetricServiceRestart() {
+ final PipeEventCommitter committer = Mockito.mock(PipeEventCommitter.class);
+ Mockito.when(committer.getPipeName()).thenReturn("pipe");
+ Mockito.when(committer.getRegionId()).thenReturn(1);
+ Mockito.when(committer.commitQueueSize()).thenReturn(3L);
+ metrics.register(committer, COMMITTER_KEY);
+ try {
+ service.restartService();
+ final AutoGauge gauge = getCommitQueueSizeGauge();
+ assertNotNull(gauge);
+ assertEquals(3, gauge.getValue(), 0);
+ } finally {
+ metrics.deregister(COMMITTER_KEY);
+ }
+ assertNull(getCommitQueueSizeGauge());
+ }
+
+ private AutoGauge getCommitQueueSizeGauge() {
+ for (final Map.Entry<MetricInfo, IMetric> entry : service.getAllMetrics().entrySet()) {
+ if (Metric.PIPE_EVENT_COMMIT_QUEUE_SIZE.toString().equals(entry.getKey().getName())) {
+ return (AutoGauge) entry.getValue();
+ }
+ }
+ return null;
+ }
+}
diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/metric/JvmGcMonitorMetricsTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/metric/JvmGcMonitorMetricsTest.java
new file mode 100644
index 00000000000..799df7960c6
--- /dev/null
+++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/metric/JvmGcMonitorMetricsTest.java
@@ -0,0 +1,72 @@
+/*
+ * 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.iotdb.commons.service.metric;
+
+import org.junit.Test;
+
+import static org.apache.iotdb.commons.service.metric.JvmGcMonitorMetrics.OBSERVATION_WINDOW_MS;
+import static org.apache.iotdb.commons.service.metric.JvmGcMonitorMetrics.SLEEP_INTERVAL_MS;
+import static org.junit.Assert.assertEquals;
+
+public class JvmGcMonitorMetricsTest {
+
+ private static final long START_TIME = 1_000_000L;
+
+ private final JvmGcMonitorMetrics monitor = JvmGcMonitorMetrics.getInstance();
+
+ /** Each sample spends the given GC time within its interval, returns the latest percentage. */
+ private long sample(final long startTime, final int samples, final long gcTimePerSample) {
+ long totalGcTime = monitor.getGcData().getAccumulatedGcTime();
+ for (int i = 1; i <= samples; i++) {
+ totalGcTime += gcTimePerSample;
+ monitor.calculateGCTimePercentageWithinObservedInterval(
+ startTime + i * SLEEP_INTERVAL_MS, totalGcTime);
+ }
+ return monitor.getGcData().getGcTimePercentage();
+ }
+
+ @Test
+ public void testLatestSampleIsCounted() {
+ monitor.startMonitoring(START_TIME, 0);
+ // 300 ms GC in the first 3 s
+ assertEquals(10, sample(START_TIME, 1, SLEEP_INTERVAL_MS / 10));
+ }
+
+ @Test
+ public void testWindowIsFullyCountedWhileSliding() {
+ monitor.startMonitoring(START_TIME, 0);
+ final int samplesToFillTheWindow = (int) (OBSERVATION_WINDOW_MS / SLEEP_INTERVAL_MS);
+ // Keep spending 10% of the time in GC, after the window is full and while it wraps the buffer
+ assertEquals(10, sample(START_TIME, samplesToFillTheWindow, SLEEP_INTERVAL_MS / 10));
+ final long windowFullTime = START_TIME + samplesToFillTheWindow * SLEEP_INTERVAL_MS;
+ assertEquals(10, sample(windowFullTime, 3 * samplesToFillTheWindow, SLEEP_INTERVAL_MS / 10));
+ }
+
+ @Test
+ public void testRestartDropsTheSamplesOfThePreviousRun() {
+ monitor.startMonitoring(START_TIME, 0);
+ // Half of the time in GC before the restart
+ sample(START_TIME, 3, SLEEP_INTERVAL_MS / 2);
+
+ final long restartTime = START_TIME + 4 * SLEEP_INTERVAL_MS;
+ monitor.startMonitoring(restartTime, monitor.getGcData().getAccumulatedGcTime());
+ assertEquals(0, sample(restartTime, 1, 0));
+ }
+}
diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/metric/MetricServiceRestartTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/metric/MetricServiceRestartTest.java
new file mode 100644
index 00000000000..8b93cb13aa3
--- /dev/null
+++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/metric/MetricServiceRestartTest.java
@@ -0,0 +1,218 @@
+/*
+ * 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.iotdb.commons.service.metric;
+
+import org.apache.iotdb.metrics.AbstractMetricService;
+import org.apache.iotdb.metrics.config.MetricConfig;
+import org.apache.iotdb.metrics.config.MetricConfigDescriptor;
+import org.apache.iotdb.metrics.metricsets.IMetricSet;
+import org.apache.iotdb.metrics.metricsets.jvm.JvmGcMetrics;
+import org.apache.iotdb.metrics.metricsets.system.SystemMetrics;
+import org.apache.iotdb.metrics.type.IMetric;
+import org.apache.iotdb.metrics.type.Timer;
+import org.apache.iotdb.metrics.utils.MetricInfo;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+import org.apache.iotdb.metrics.utils.MetricType;
+import org.apache.iotdb.metrics.utils.SystemMetric;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class MetricServiceRestartTest {
+
+ private static final String METRIC = "metric_service_restart_test";
+
+ private final MetricService service = MetricService.getInstance();
+ private final MetricConfig config = MetricConfigDescriptor.getInstance().getMetricConfig();
+ private final List<IMetricSet> boundMetricSets = new ArrayList<>();
+ private MetricLevel originalLevel;
+ private String originalReporters;
+
+ @Before
+ public void setUp() {
+ originalLevel = config.getMetricLevel();
+ originalReporters =
+ config.getMetricReporterList().stream().map(Enum::name).collect(Collectors.joining(","));
+ config.setMetricLevel(MetricLevel.IMPORTANT);
+ config.setMetricReporterList("");
+ service.startService();
+ }
+
+ @After
+ public void tearDown() {
+ boundMetricSets.forEach(service::removeMetricSet);
+ service.stopService();
+ config.setMetricLevel(originalLevel);
+ config.setMetricReporterList(originalReporters);
+ }
+
+ @Test
+ public void testFailedMetricSetDoesNotStopOthersFromRebinding() {
+ // The hash codes make the failed set iterated before the other one
+ final FailingMetricSet failing = new FailingMetricSet();
+ final GaugeMetricSet gauge = new GaugeMetricSet();
+ bind(failing);
+ bind(gauge);
+ assertTrue(service.getAllMetrics().keySet().stream().anyMatch(this::isTestMetric));
+
+ failing.fail = true;
+ service.restartService();
+ failing.fail = false;
+ assertTrue(service.getAllMetrics().keySet().stream().anyMatch(this::isTestMetric));
+ }
+
+ @Test
+ public void testSystemDiskMetricsSurviveRestart() {
+ final SystemMetrics metrics = new SystemMetrics();
+ metrics.setDiskDirs(Collections.singletonList(System.getProperty("java.io.tmpdir")));
+ bind(metrics);
+ assertTrue(metrics.getSystemDiskTotalSpace() > 0);
+
+ service.restartService();
+ assertTrue(metrics.getSystemDiskTotalSpace() > 0);
+ }
+
+ @Test
+ public void testSystemDiskMetricsAreCollectedWhenLevelIsEnabledByRestart() {
+ config.setMetricLevel(MetricLevel.OFF);
+ final SystemMetrics metrics = new SystemMetrics();
+ metrics.setDiskDirs(Collections.singletonList(System.getProperty("java.io.tmpdir")));
+ bind(metrics);
+ assertEquals(0, metrics.getSystemDiskTotalSpace());
+
+ config.setMetricLevel(MetricLevel.IMPORTANT);
+ service.restartService();
+ assertTrue(metrics.getSystemDiskTotalSpace() > 0);
+ }
+
+ @Test
+ public void testGcPauseTimersKeepCountingAfterRestart() {
+ bind(new JvmGcMetrics());
+ service.restartService();
+ service.restartService();
+
+ // The listeners of previous bindings used to reset the timers on every GC
+ System.gc();
+ await().atMost(30, TimeUnit.SECONDS).until(() -> getExplicitGcCount() >= 1);
+ System.gc();
+ await().atMost(30, TimeUnit.SECONDS).until(() -> getExplicitGcCount() >= 2);
+ }
+
+ @Test
+ public void testGcPauseTimersAreBoundAgainBeforeTheNextGc() {
+ bind(new JvmGcMetrics());
+ System.gc();
+ await().atMost(30, TimeUnit.SECONDS).until(() -> getExplicitGcCount() >= 1);
+
+ service.restartService();
+ // The timers used to be created again only by the next GC with the same cause
+ assertTrue(
+ service.getAllMetrics().keySet().stream()
+ .anyMatch(
+ info ->
+ SystemMetric.JVM_GC_PAUSE.toString().equals(info.getName())
+ && "System.gc()".equals(info.getTags().get("cause"))));
+ }
+
+ private void bind(final IMetricSet metricSet) {
+ service.addMetricSet(metricSet);
+ boundMetricSets.add(metricSet);
+ }
+
+ private boolean isTestMetric(final MetricInfo info) {
+ return METRIC.equals(info.getName());
+ }
+
+ private long getExplicitGcCount() {
+ long count = 0;
+ for (final Map.Entry<MetricInfo, IMetric> entry : service.getAllMetrics().entrySet()) {
+ if (SystemMetric.JVM_GC_PAUSE.toString().equals(entry.getKey().getName())
+ && "System.gc()".equals(entry.getKey().getTags().get("cause"))) {
+ count += ((Timer) entry.getValue()).getCount();
+ }
+ }
+ return count;
+ }
+
+ private static class FailingMetricSet implements IMetricSet {
+
+ private volatile boolean fail = false;
+
+ @Override
+ public void bindTo(final AbstractMetricService metricService) {
+ if (fail) {
+ throw new IllegalStateException("bind failure for test");
+ }
+ }
+
+ @Override
+ public void unbindFrom(final AbstractMetricService metricService) {
+ // do nothing
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ return this == o;
+ }
+
+ @Override
+ public int hashCode() {
+ return 0;
+ }
+ }
+
+ private static class GaugeMetricSet implements IMetricSet {
+
+ private final AtomicLong value = new AtomicLong(1);
+
+ @Override
+ public void bindTo(final AbstractMetricService metricService) {
+ metricService.createAutoGauge(METRIC, MetricLevel.IMPORTANT, value, AtomicLong::get);
+ }
+
+ @Override
+ public void unbindFrom(final AbstractMetricService metricService) {
+ metricService.remove(MetricType.AUTO_GAUGE, METRIC);
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ return this == o;
+ }
+
+ @Override
+ public int hashCode() {
+ return 1;
+ }
+ }
+}