From 4c4c9eed08624ad0f071ec52961bd3a18845f33b Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 2 Nov 2026 10:47:41 +0800 Subject: [PATCH 1/2] [Pipe] Fix ConfigRegion remaining event count lookup --- .../source/PipeConfigRegionSourceMetrics.java | 4 +++- .../metric/PipeConfigNodeMetricsRestartTest.java | 13 +++++++++++-- 2 files changed, 14 insertions(+), 3 deletions(-) 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 44dcdf2c522de..d96b00fd2e0df 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 @@ -36,6 +36,8 @@ import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; +import static org.apache.iotdb.commons.client.AbstractConfigNodeInfo.CONFIG_REGION_ID; + public class PipeConfigRegionSourceMetrics implements IMetricSet { private static final Logger LOGGER = LoggerFactory.getLogger(PipeConfigRegionSourceMetrics.class); @@ -117,7 +119,7 @@ public synchronized void deregister(final String taskID) { //////////////////////////// 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 index 0bd5e52aff1f5..51a42cbbd2aff 100644 --- 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 @@ -87,14 +87,23 @@ public void tearDown() { public void testConfigRegionSourceMetrics() { final PipeConfigRegionSourceMetrics metrics = PipeConfigRegionSourceMetrics.getInstance(); bind(metrics); - metrics.register(mockConfigRegionSource()); + 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(TASK_ID); + metrics.deregister(source.getTaskID()); } assertEquals(0, count(Metric.UNTRANSFERRED_CONFIG_COUNT)); + assertEquals(0L, metrics.getRemainingEventCount(PIPE, CREATION_TIME)); } @Test From 6f952119ffd0abc0ed55125978d665a067fd8ef0 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 2 Nov 2026 14:22:36 +0800 Subject: [PATCH 2/2] [Pipe] Preserve full pipe names in remaining metrics --- ...PipeTemporaryMetaInCoordinatorMetrics.java | 24 ++--- .../PipeConfigNodeMetricsRestartTest.java | 90 +++++++++++++++++++ 2 files changed, 104 insertions(+), 10 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 c4e3ca285e2f7..38b52fe83e18a 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 @@ -69,25 +69,27 @@ private void createMetrics(final String pipeID) { 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 @@ -102,21 +104,23 @@ private void removeMetrics(final String pipeID) { } 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]); + creationTime); } //////////////////////////// register & deregister (pipe integration) //////////////////////////// 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 index 51a42cbbd2aff..63f499e46d218 100644 --- 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 @@ -24,6 +24,7 @@ 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.agent.task.PipeConfigNodeSubtask; import org.apache.iotdb.confignode.manager.pipe.metric.overview.PipeConfigNodeRemainingTimeMetrics; import org.apache.iotdb.confignode.manager.pipe.metric.overview.PipeTemporaryMetaInCoordinatorMetrics; @@ -33,6 +34,7 @@ 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; @@ -49,6 +51,8 @@ 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. */ @@ -163,6 +167,79 @@ public void testTemporaryMetaInCoordinatorMetrics() { 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 registeredPipeIds = new ArrayList<>(); + final List 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); @@ -190,4 +267,17 @@ private IMetric get(final Metric metric) { } throw new AssertionError(metric + " is not registered"); } + + 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); + } }