This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch cp/af225846-dev13 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 9a42e6e77c03f09e838ff479b2e033af9846b88d Author: Caideyipi <[email protected]> AuthorDate: Mon Nov 2 15:17:18 2026 +0800 [Pipe] Fix remaining-event metric identities (#19071) * [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); + } +}
