Skip to content

[To dev/1.3] [Pipe] Fix remaining-event metric identities #19082

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jt2594838 merged 1 commit into from
Nov 3, 2026
+268 −30
Merged

[To dev/1.3] [Pipe] Fix remaining-event metric identities #19082

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 22 additions & 20 deletions .../iotdb/confignode/manager/pipe/metric/overview/PipeTemporaryMetaInCoordinatorMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand All @@ -68,62 +68,63 @@ 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
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) {
removeAutoGauge(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]);
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(
Expand All @@ -133,7 +134,7 @@ public void register(final PipeMeta pipeMeta) {
}
}

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",
Expand All @@ -143,6 +144,7 @@ public void deregister(final String pipeID) {
if (Objects.nonNull(metricService)) {
removeMetrics(pipeID);
}
pipeTemporaryMetaMap.remove(pipeID);
}

public void handleTemporaryMetaChanges(final Iterable<PipeMeta> pipeMetaList) {
Expand Down
21 changes: 11 additions & 10 deletions ...org/apache/iotdb/confignode/manager/pipe/metric/source/PipeConfigRegionSourceMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,8 @@
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);
Expand All @@ -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);
}
Expand All @@ -69,12 +71,11 @@ private void createAutoGauge(final String taskID) {
}

@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) {
Expand All @@ -95,15 +96,15 @@ private void removeAutoGauge(final String taskID) {

//////////////////////////// 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)) {
createMetrics(taskID);
}
}

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",
Expand All @@ -119,7 +120,7 @@ public 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)) {
Expand Down
Loading
Loading