This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new deb408af3a0 [Pipe] Fix remaining-event metric identities (#19071)
(#19082)
deb408af3a0 is described below
commit deb408af3a0b33ffc7b366d8cc8ad663a79f3b14
Author: Caideyipi <[email protected]>
AuthorDate: Tue Nov 3 09:51:34 2026 +0800
[Pipe] Fix remaining-event metric identities (#19071) (#19082)
* [Pipe] Fix ConfigRegion remaining event count lookup
* [Pipe] Preserve full pipe names in remaining metrics
(cherry picked from commit af225846393dfc0da265afa80a3bb37f98e85ef0)
---
.../PipeTemporaryMetaInCoordinatorMetrics.java | 42 ++--
.../source/PipeConfigRegionSourceMetrics.java | 21 +-
.../metric/PipeConfigNodeMetricsRestartTest.java | 235 +++++++++++++++++++++
3 files changed, 268 insertions(+), 30 deletions(-)
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 8ebc9d22b3c..220536b8701 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
@@ -57,7 +57,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);
}
@@ -68,34 +68,34 @@ public class PipeTemporaryMetaInCoordinatorMetrics
implements IMetricSet {
private void createAutoGauge(final String pipeID) {
final PipeTemporaryMetaInCoordinator pipeTemporaryMeta =
pipeTemporaryMetaMap.get(pipeID);
- final String[] pipeNameAndCreationTime = pipeID.split("_");
+ final int separatorIndex = pipeID.lastIndexOf('_');
+ final String pipeName = pipeID.substring(0, separatorIndex);
+ final String creationTime = pipeID.substring(separatorIndex + 1);
metricService.createAutoGauge(
Metric.PIPE_GLOBAL_REMAINING_EVENT_COUNT.toString(),
MetricLevel.IMPORTANT,
pipeTemporaryMeta,
PipeTemporaryMetaInCoordinator::getGlobalRemainingEvents,
Tag.NAME.toString(),
- pipeNameAndCreationTime[0],
+ pipeName,
Tag.CREATION_TIME.toString(),
- pipeNameAndCreationTime[1]);
+ creationTime);
metricService.createAutoGauge(
Metric.PIPE_GLOBAL_REMAINING_TIME.toString(),
MetricLevel.IMPORTANT,
pipeTemporaryMeta,
PipeTemporaryMetaInCoordinator::getGlobalRemainingTime,
Tag.NAME.toString(),
- pipeNameAndCreationTime[0],
+ pipeName,
Tag.CREATION_TIME.toString(),
- pipeNameAndCreationTime[1]);
+ creationTime);
}
@Override
- public void unbindFrom(final AbstractMetricService metricService) {
-
ImmutableSet.copyOf(pipeTemporaryMetaMap.keySet()).forEach(this::deregister);
- if (!pipeTemporaryMetaMap.isEmpty()) {
- LOGGER.warn(
- "Failed to unbind from pipe temporary meta metrics,
PipeTemporaryMeta map not empty");
- }
+ 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) {
@@ -103,27 +103,28 @@ public class PipeTemporaryMetaInCoordinatorMetrics
implements IMetricSet {
}
private void removeAutoGauge(final String pipeID) {
- final String[] pipeNameAndCreationTime = pipeID.split("_");
+ final int separatorIndex = pipeID.lastIndexOf('_');
+ final String pipeName = pipeID.substring(0, separatorIndex);
+ final String creationTime = pipeID.substring(separatorIndex + 1);
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.PIPE_GLOBAL_REMAINING_EVENT_COUNT.toString(),
Tag.NAME.toString(),
- pipeNameAndCreationTime[0],
+ pipeName,
Tag.CREATION_TIME.toString(),
- pipeNameAndCreationTime[1]);
+ creationTime);
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.PIPE_GLOBAL_REMAINING_TIME.toString(),
Tag.NAME.toString(),
- pipeNameAndCreationTime[0],
+ pipeName,
Tag.CREATION_TIME.toString(),
- pipeNameAndCreationTime[1]);
- pipeTemporaryMetaMap.remove(pipeID);
+ creationTime);
}
//////////////////////////// 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(
@@ -133,7 +134,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(
"Failed to deregister pipe temporary meta metrics,
PipeTemporaryMeta({}) does not exist",
@@ -143,6 +144,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/source/PipeConfigRegionSourceMetrics.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/source/PipeConfigRegionSourceMetrics.java
index 59e204fd276..8ebca581816 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
@@ -35,6 +35,8 @@ import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
+import static
org.apache.iotdb.db.protocol.client.ConfigNodeInfo.CONFIG_REGION_ID;
+
public class PipeConfigRegionSourceMetrics implements IMetricSet {
private static final Logger LOGGER =
LoggerFactory.getLogger(PipeConfigRegionSourceMetrics.class);
@@ -46,7 +48,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);
}
@@ -69,12 +71,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(
- "Failed to unbind from pipe config region extractor metrics,
extractor map not empty");
- }
+ 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) {
@@ -95,7 +96,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)) {
@@ -103,7 +104,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(
"Failed to deregister pipe config region extractor metrics,
IoTDBConfigRegionExtractor({}) does not exist",
@@ -119,7 +120,7 @@ public class PipeConfigRegionSourceMetrics implements
IMetricSet {
//////////////////////////// Show pipes ////////////////////////////
public long getRemainingEventCount(final String pipeName, final long
creationTime) {
- final String taskID = pipeName + "_" + creationTime;
+ final String taskID = pipeName + "_" + CONFIG_REGION_ID.getId() + "_" +
creationTime;
final IoTDBConfigRegionSource extractor = extractorMap.get(taskID);
// Do not print log to allow collection when config region extractor does
not exists
if (Objects.isNull(extractor)) {
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..137297833c9
--- /dev/null
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/metric/PipeConfigNodeMetricsRestartTest.java
@@ -0,0 +1,235 @@
+/*
+ * 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.commons.service.metric.enums.Tag;
+import
org.apache.iotdb.confignode.manager.pipe.metric.overview.PipeTemporaryMetaInCoordinatorMetrics;
+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.AutoGauge;
+import org.apache.iotdb.metrics.type.IMetric;
+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.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+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);
+ final IoTDBConfigRegionSource source = mockConfigRegionSource();
+ when(source.getTaskID()).thenReturn(PIPE + "_0_" + CREATION_TIME);
+ when(source.getUnTransferredEventCount()).thenReturn(7L);
+ metrics.register(source);
+ try {
+ assertEquals(7L, metrics.getRemainingEventCount(PIPE, CREATION_TIME));
+ assertEquals(0L, metrics.getRemainingEventCount(PIPE, CREATION_TIME +
1));
+ assertEquals(0L, metrics.getRemainingEventCount("other_pipe",
CREATION_TIME));
+ service.restartService();
+ assertEquals(1, count(Metric.UNTRANSFERRED_CONFIG_COUNT));
+ when(source.getUnTransferredEventCount()).thenReturn(3L);
+ assertEquals(3L, metrics.getRemainingEventCount(PIPE, CREATION_TIME));
+ } finally {
+ metrics.deregister(source.getTaskID());
+ }
+ assertEquals(0, count(Metric.UNTRANSFERRED_CONFIG_COUNT));
+ assertEquals(0L, metrics.getRemainingEventCount(PIPE, CREATION_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));
+ }
+
+ @Test
+ public void testTemporaryMetaMetricsWithUnderscores() {
+ final PipeTemporaryMetaInCoordinatorMetrics metrics =
+ PipeTemporaryMetaInCoordinatorMetrics.getInstance();
+ bind(metrics);
+ final String[] pipeNames = {
+ "pipe_with_underscores", "pipe_with_underscores",
"pipe_with_another_suffix_", "_pipe"
+ };
+ final long[] creationTimes = {CREATION_TIME, CREATION_TIME + 1,
CREATION_TIME, CREATION_TIME};
+ final List<String> registeredPipeIds = new ArrayList<>();
+ final List<PipeTemporaryMetaInCoordinator> temporaryMetas = new
ArrayList<>();
+ try {
+ for (int i = 0; i < pipeNames.length; i++) {
+ final PipeStaticMeta staticMeta = Mockito.mock(PipeStaticMeta.class);
+ when(staticMeta.getPipeName()).thenReturn(pipeNames[i]);
+ when(staticMeta.getCreationTime()).thenReturn(creationTimes[i]);
+ final PipeMeta pipeMeta = Mockito.mock(PipeMeta.class);
+ when(pipeMeta.getStaticMeta()).thenReturn(staticMeta);
+ final PipeTemporaryMetaInCoordinator temporaryMeta =
+ Mockito.mock(PipeTemporaryMetaInCoordinator.class);
+ when(temporaryMeta.getGlobalRemainingEvents()).thenReturn(7L + i);
+ when(temporaryMeta.getGlobalRemainingTime()).thenReturn(10.0 + i);
+ when(pipeMeta.getTemporaryMeta()).thenReturn(temporaryMeta);
+ registeredPipeIds.add(pipeNames[i] + "_" + creationTimes[i]);
+ temporaryMetas.add(temporaryMeta);
+ metrics.register(pipeMeta);
+ }
+ assertEquals(pipeNames.length,
count(Metric.PIPE_GLOBAL_REMAINING_EVENT_COUNT));
+ assertEquals(pipeNames.length, count(Metric.PIPE_GLOBAL_REMAINING_TIME));
+ for (int i = 0; i < pipeNames.length; i++) {
+ assertTemporaryMetaValues(pipeNames[i], creationTimes[i], 7L + i, 10.0
+ i);
+ }
+ service.restartService();
+ when(temporaryMetas.get(0).getGlobalRemainingEvents()).thenReturn(3L);
+ when(temporaryMetas.get(0).getGlobalRemainingTime()).thenReturn(4.0);
+ assertTemporaryMetaValues(pipeNames[0], creationTimes[0], 3L, 4.0);
+ metrics.deregister(registeredPipeIds.remove(0));
+ assertEquals(pipeNames.length - 1,
count(Metric.PIPE_GLOBAL_REMAINING_EVENT_COUNT));
+ assertEquals(pipeNames.length - 1,
count(Metric.PIPE_GLOBAL_REMAINING_TIME));
+ assertNull(get(Metric.PIPE_GLOBAL_REMAINING_EVENT_COUNT, pipeNames[0],
creationTimes[0]));
+ assertNull(get(Metric.PIPE_GLOBAL_REMAINING_TIME, pipeNames[0],
creationTimes[0]));
+ for (int i = 1; i < pipeNames.length; i++) {
+ assertTemporaryMetaValues(pipeNames[i], creationTimes[i], 7L + i, 10.0
+ i);
+ }
+ service.restartService();
+ assertEquals(pipeNames.length - 1,
count(Metric.PIPE_GLOBAL_REMAINING_EVENT_COUNT));
+ assertEquals(pipeNames.length - 1,
count(Metric.PIPE_GLOBAL_REMAINING_TIME));
+ assertNull(get(Metric.PIPE_GLOBAL_REMAINING_EVENT_COUNT, pipeNames[0],
creationTimes[0]));
+ assertNull(get(Metric.PIPE_GLOBAL_REMAINING_TIME, pipeNames[0],
creationTimes[0]));
+ for (int i = 1; i < pipeNames.length; i++) {
+ assertTemporaryMetaValues(pipeNames[i], creationTimes[i], 7L + i, 10.0
+ i);
+ }
+ } finally {
+ registeredPipeIds.forEach(metrics::deregister);
+ }
+ assertEquals(0, count(Metric.PIPE_GLOBAL_REMAINING_EVENT_COUNT));
+ assertEquals(0, count(Metric.PIPE_GLOBAL_REMAINING_TIME));
+ }
+
+ private void assertTemporaryMetaValues(
+ final String pipeName,
+ final long creationTime,
+ final long remainingEvents,
+ final double remainingTime) {
+ final IMetric eventGauge =
+ get(Metric.PIPE_GLOBAL_REMAINING_EVENT_COUNT, pipeName, creationTime);
+ final IMetric timeGauge = get(Metric.PIPE_GLOBAL_REMAINING_TIME, pipeName,
creationTime);
+ assertNotNull(eventGauge);
+ assertNotNull(timeGauge);
+ assertEquals(remainingEvents, ((AutoGauge) eventGauge).getValue(), 0.0);
+ assertEquals(remainingTime, ((AutoGauge) timeGauge).getValue(), 0.0);
+ }
+
+ 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, final String pipeName, final long
creationTime) {
+ return service.getAllMetrics().entrySet().stream()
+ .filter(entry -> metric.toString().equals(entry.getKey().getName()))
+ .filter(entry ->
pipeName.equals(entry.getKey().getTags().get(Tag.NAME.toString())))
+ .filter(
+ entry ->
+ String.valueOf(creationTime)
+
.equals(entry.getKey().getTags().get(Tag.CREATION_TIME.toString())))
+ .map(Map.Entry::getValue)
+ .findFirst()
+ .orElse(null);
+ }
+}
