From db73a43f9e34e3252b908d461157261f08f6c965 Mon Sep 17 00:00:00 2001 From: Logic Date: Wed, 26 Aug 2026 00:46:27 +0800 Subject: [PATCH] maintenance: define VictoriaMetrics label collisions (#4286) Co-authored-by: Duansg --- .../tsdb/greptime/GreptimeDbDataStorage.java | 35 ++++++ .../vm/VictoriaMetricsClusterDataStorage.java | 30 ++++- .../tsdb/vm/VictoriaMetricsDataStorage.java | 68 ++++++++++- .../greptime/GreptimeDbDataStorageTest.java | 54 +++++---- .../vm/VictoriaMetricsDataStorageTest.java | 107 +++++++++++++++++- home/docs/start/victoria-metrics-init.md | 19 ++++ 6 files changed, 280 insertions(+), 33 deletions(-) diff --git a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeDbDataStorage.java b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeDbDataStorage.java index 5ccbe4b84a..ca3b8cf142 100644 --- a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeDbDataStorage.java +++ b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeDbDataStorage.java @@ -45,9 +45,12 @@ import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; +import java.util.TreeSet; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.function.BiConsumer; import java.util.function.Function; import java.util.stream.Collectors; @@ -113,6 +116,7 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { private final RestTemplate restTemplate; private final GreptimeSqlQueryExecutor greptimeSqlQueryExecutor; + private final AtomicLong ignoredLabelCollisionCount = new AtomicLong(); public GreptimeDbDataStorage(GreptimeProperties greptimeProperties, @Qualifier(WarehouseConstants.GREPTIME_QUERY_REST_TEMPLATE) @@ -164,6 +168,16 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { List fields = metricsData.getFields(); Map customLabels = metricsData.getLabels(); List fieldNames = fields.stream().map(CollectRep.Field::getName).collect(Collectors.toList()); + Set labelCollisions = findLabelCollisions(customLabels, fieldNames); + if (!labelCollisions.isEmpty()) { + long previousCount = ignoredLabelCollisionCount.getAndAdd(labelCollisions.size()); + long ignoredCount = previousCount + labelCollisions.size(); + if (shouldLogLabelCollisions(previousCount, ignoredCount)) { + log.warn("[warehouse greptime] ignore custom labels {} from metrics data {} because " + + "the keys are storage-managed; cumulative ignored labels: {}.", + labelCollisions, metricsData.getId(), ignoredCount); + } + } fields.forEach(field -> { if (field.getLabel()) { tableSchemaBuilder.addTag(field.getName(), DataType.String); @@ -237,6 +251,27 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage { } } + private Set findLabelCollisions(Map customLabels, List fieldNames) { + if (customLabels == null || customLabels.isEmpty()) { + return Set.of(); + } + Set collisions = new TreeSet<>(); + for (String key : customLabels.keySet()) { + if (LABEL_KEY_INSTANCE.equals(key) || LABEL_KEY_TS.equals(key) || fieldNames.contains(key)) { + collisions.add(key); + } + } + return collisions; + } + + private boolean shouldLogLabelCollisions(long previousCount, long currentCount) { + return previousCount == 0 || previousCount / 100 < currentCount / 100; + } + + long getIgnoredLabelCollisionCount() { + return ignoredLabelCollisionCount.get(); + } + @Override public Map> getHistoryMetricData(String instance, String app, String metrics, String metric, String history) { diff --git a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorage.java b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorage.java index ae93a1e7e2..3777bb1830 100644 --- a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorage.java +++ b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorage.java @@ -33,9 +33,11 @@ import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -69,7 +71,6 @@ import org.springframework.http.HttpMethod; import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; import org.springframework.stereotype.Component; -import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; import org.springframework.web.client.RestTemplate; import org.springframework.web.util.UriComponentsBuilder; @@ -107,6 +108,7 @@ public class VictoriaMetricsClusterDataStorage extends AbstractHistoryDataStorag private final VictoriaMetricsSelectProperties vmSelectProps; private final RestTemplate restTemplate; private final BlockingQueue metricsBufferQueue; + private final AtomicLong ignoredLabelCollisionCount = new AtomicLong(); private HashedWheelTimer metricsFlushTimer = null; private MetricsFlushTask metricsFlushtask = null; @@ -186,6 +188,13 @@ public class VictoriaMetricsClusterDataStorage extends AbstractHistoryDataStorag metricsData.getId(), metricsData.getApp(), metricsData.getMetrics()); return; } + var managedLabelCollisions = + VictoriaMetricsDataStorage.findManagedLabelCollisions(metricsData.getLabels()); + if (!managedLabelCollisions.isEmpty()) { + recordIgnoredLabelCollisions(metricsData.getId(), managedLabelCollisions); + } + Map customizedLabels = VictoriaMetricsDataStorage.withoutManagedLabels( + metricsData.getLabels(), managedLabelCollisions); Map defaultLabels = Maps.newHashMapWithExpectedSize(8); defaultLabels.put(MONITOR_METRICS_KEY, metricsData.getMetrics()); boolean isPrometheusAuto; @@ -243,10 +252,7 @@ public class VictoriaMetricsClusterDataStorage extends AbstractHistoryDataStorag } labels.put(LABEL_KEY_MONITOR_ID, String.valueOf(metricsData.getId())); // add customized labels as identifier - var customizedLabels = metricsData.getLabels(); - if (!ObjectUtils.isEmpty(customizedLabels)) { - labels.putAll(customizedLabels); - } + VictoriaMetricsDataStorage.addCustomizedLabels(labels, customizedLabels); VictoriaMetricsDataStorage.VictoriaMetricsContent content = VictoriaMetricsDataStorage.VictoriaMetricsContent.builder() .metric(new HashMap<>(labels)) .values(new Double[]{entry.getValue()}) @@ -276,6 +282,20 @@ public class VictoriaMetricsClusterDataStorage extends AbstractHistoryDataStorag } } + private void recordIgnoredLabelCollisions(long monitorId, Set collisions) { + long previousCount = ignoredLabelCollisionCount.getAndAdd(collisions.size()); + long ignoredCount = previousCount + collisions.size(); + if (previousCount == 0 || previousCount / 100 < ignoredCount / 100) { + log.warn("[warehouse victoria-metrics-cluster] ignore custom labels {} from metrics data {} because " + + "the keys are HertzBeat-managed; cumulative ignored labels: {}.", + collisions, monitorId, ignoredCount); + } + } + + long getIgnoredLabelCollisionCount() { + return ignoredLabelCollisionCount.get(); + } + @Override public void destroy() { if (metricsFlushTimer != null && !metricsFlushTimer.isStop()) { diff --git a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorage.java b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorage.java index 5714e42186..5a80c9824c 100644 --- a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorage.java +++ b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorage.java @@ -32,10 +32,13 @@ import java.util.HashMap; import java.util.LinkedList; import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.TreeSet; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; import java.util.zip.GZIPOutputStream; import com.google.common.collect.Maps; @@ -98,12 +101,19 @@ public class VictoriaMetricsDataStorage extends AbstractHistoryDataStorage { private static final String SPILT = "_"; private static final String MONITOR_METRICS_KEY = "__metrics__"; private static final String MONITOR_METRIC_KEY = "__metric__"; + private static final Set MANAGED_LABEL_KEYS = Set.of( + LABEL_KEY_NAME, + LABEL_KEY_MONITOR_ID, + MONITOR_METRICS_KEY, + MONITOR_METRIC_KEY, + LABEL_KEY_INSTANCE); private static final long MAX_WAIT_MS = 500L; private static final int MAX_RETRIES = 3; private final VictoriaMetricsProperties victoriaMetricsProp; private final RestTemplate restTemplate; private final BlockingQueue metricsBufferQueue; + private final AtomicLong ignoredLabelCollisionCount = new AtomicLong(); private HashedWheelTimer metricsFlushTimer = null; private final VictoriaMetricsProperties.InsertConfig insertConfig; @@ -170,6 +180,12 @@ public class VictoriaMetricsDataStorage extends AbstractHistoryDataStorage { metricsData.getId(), metricsData.getApp(), metricsData.getMetrics()); return; } + Set managedLabelCollisions = findManagedLabelCollisions(metricsData.getLabels()); + if (!managedLabelCollisions.isEmpty()) { + recordIgnoredLabelCollisions(metricsData.getId(), managedLabelCollisions); + } + Map customizedLabels = withoutManagedLabels( + metricsData.getLabels(), managedLabelCollisions); Map defaultLabels = Maps.newHashMapWithExpectedSize(8); defaultLabels.put(MONITOR_METRICS_KEY, metricsData.getMetrics()); boolean isPrometheusAuto = false; @@ -226,10 +242,7 @@ public class VictoriaMetricsDataStorage extends AbstractHistoryDataStorage { } labels.put(LABEL_KEY_MONITOR_ID, String.valueOf(metricsData.getId())); // add customized labels as identifier - var customizedLabels = metricsData.getLabels(); - if (!ObjectUtils.isEmpty(customizedLabels)) { - labels.putAll(customizedLabels); - } + addCustomizedLabels(labels, customizedLabels); VictoriaMetricsContent content = VictoriaMetricsContent.builder() .metric(new HashMap<>(labels)) .values(new Double[]{entry.getValue()}) @@ -255,6 +268,53 @@ public class VictoriaMetricsDataStorage extends AbstractHistoryDataStorage { sendVictoriaMetrics(contentList); } + static void addCustomizedLabels(Map labels, Map customizedLabels) { + if (ObjectUtils.isEmpty(customizedLabels)) { + return; + } + labels.putAll(customizedLabels); + } + + private void recordIgnoredLabelCollisions(long monitorId, Set collisions) { + long previousCount = ignoredLabelCollisionCount.getAndAdd(collisions.size()); + long ignoredCount = previousCount + collisions.size(); + if (previousCount == 0 || previousCount / 100 < ignoredCount / 100) { + log.warn("[warehouse victoria-metrics] ignore custom labels {} from metrics data {} because " + + "the keys are HertzBeat-managed; cumulative ignored labels: {}.", + collisions, monitorId, ignoredCount); + } + } + + long getIgnoredLabelCollisionCount() { + return ignoredLabelCollisionCount.get(); + } + + static Set findManagedLabelCollisions(Map customizedLabels) { + if (ObjectUtils.isEmpty(customizedLabels)) { + return Set.of(); + } + Set collisions = new TreeSet<>(); + for (String key : customizedLabels.keySet()) { + if (key != null && MANAGED_LABEL_KEYS.contains(key)) { + collisions.add(key); + } + } + return collisions; + } + + static Map withoutManagedLabels( + Map customizedLabels, Set managedLabelCollisions) { + if (ObjectUtils.isEmpty(customizedLabels)) { + return Map.of(); + } + if (managedLabelCollisions.isEmpty()) { + return customizedLabels; + } + Map safeLabels = new HashMap<>(customizedLabels); + managedLabelCollisions.forEach(safeLabels::remove); + return safeLabels; + } + @Override public void destroy() { if (metricsFlushTimer != null && !metricsFlushTimer.isStop()) { diff --git a/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeDbDataStorageTest.java b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeDbDataStorageTest.java index 91e4ccac62..2e1a3ba980 100644 --- a/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeDbDataStorageTest.java +++ b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeDbDataStorageTest.java @@ -38,7 +38,6 @@ import io.greptime.models.Err; import io.greptime.models.Result; import io.greptime.models.Table; import io.greptime.models.WriteOk; -import io.greptime.v1.Common; import io.greptime.v1.RowData; import java.lang.reflect.Field; import java.util.ArrayList; @@ -48,7 +47,6 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; -import java.util.stream.Collectors; import org.apache.hertzbeat.common.constants.CommonConstants; import org.apache.hertzbeat.common.entity.arrow.ArrowCell; import org.apache.hertzbeat.common.entity.arrow.RowWrapper; @@ -133,7 +131,6 @@ class GreptimeDbDataStorageTest { when(mockResult.isOk()).thenReturn(true); CompletableFuture> mockFuture = CompletableFuture.completedFuture(mockResult); when(greptimeDb.write(any(Table.class))).thenReturn(mockFuture); - greptimeDbDataStorage = new GreptimeDbDataStorage(greptimeProperties, restTemplate, greptimeSqlQueryExecutor); // Test with valid metrics data @@ -157,15 +154,14 @@ class GreptimeDbDataStorageTest { } @Test - void testSaveDataWithCustomLabels() throws Exception { + void testSaveDataSkipsCustomLabelCollisionsWithoutDroppingMetrics() throws Exception { try (MockedStatic mockedStatic = mockStatic(GreptimeDB.class)) { mockedStatic.when(() -> GreptimeDB.create(any())).thenReturn(greptimeDb); - + @SuppressWarnings("unchecked") Result mockResult = mock(Result.class); when(mockResult.isOk()).thenReturn(true); - CompletableFuture> mockFuture = CompletableFuture.completedFuture(mockResult); - when(greptimeDb.write(any(Table.class))).thenReturn(mockFuture); - + when(greptimeDb.write(any(Table.class))) + .thenReturn(CompletableFuture.completedFuture(mockResult)); greptimeDbDataStorage = new GreptimeDbDataStorage(greptimeProperties, restTemplate, greptimeSqlQueryExecutor); CollectRep.MetricsData metricsData = createMockMetricsData(true); @@ -181,27 +177,39 @@ class GreptimeDbDataStorageTest { greptimeDbDataStorage.saveData(metricsData); verify(greptimeDb).write(tableCaptor.capture()); - Table capturedTable = tableCaptor.getValue(); - - List columnSchemas = getColumnSchemas(capturedTable); - List columnNames = columnSchemas.stream() + List columnNames = getColumnSchemas(tableCaptor.getValue()).stream() .map(RowData.ColumnSchema::getColumnName) - .collect(Collectors.toList()); - assertEquals(5, columnNames.size()); + .toList(); + // The fixture already contains an `instance` metric field in addition to the + // storage identity tag; the conflicting custom label must not add a third column. assertEquals(2, Collections.frequency(columnNames, "instance")); assertEquals(1, Collections.frequency(columnNames, "ts")); assertEquals(1, Collections.frequency(columnNames, "usage")); + assertEquals(1, Collections.frequency(columnNames, "env")); + assertEquals(3, greptimeDbDataStorage.getIgnoredLabelCollisionCount()); + } + } - List envColumnSchemas = columnSchemas.stream() - .filter(columnSchema -> "env".equals(columnSchema.getColumnName())) - .collect(Collectors.toList()); - assertEquals(1, envColumnSchemas.size()); - assertEquals(Common.SemanticType.TAG, envColumnSchemas.get(0).getSemanticType()); + @Test + void testSaveDataAcceptsNonConflictingCustomLabels() throws Exception { + try (MockedStatic mockedStatic = mockStatic(GreptimeDB.class)) { + mockedStatic.when(() -> GreptimeDB.create(any())).thenReturn(greptimeDb); + @SuppressWarnings("unchecked") + Result mockResult = mock(Result.class); + when(mockResult.isOk()).thenReturn(true); + when(greptimeDb.write(any(Table.class))) + .thenReturn(CompletableFuture.completedFuture(mockResult)); + greptimeDbDataStorage = new GreptimeDbDataStorage( + greptimeProperties, restTemplate, greptimeSqlQueryExecutor); + CollectRep.MetricsData metricsData = createMockMetricsData(true); + when(metricsData.getLabels()).thenReturn(Map.of("env", "prod")); - List rows = getRows(capturedTable); - assertEquals(1, rows.size()); - RowData.Row row = rows.get(0); - assertEquals("prod", row.getValuesList().get(4).getStringValue()); + ArgumentCaptor tableCaptor = ArgumentCaptor.forClass(Table.class); + greptimeDbDataStorage.saveData(metricsData); + + verify(greptimeDb).write(tableCaptor.capture()); + List schemas = getColumnSchemas(tableCaptor.getValue()); + assertTrue(schemas.stream().anyMatch(schema -> "env".equals(schema.getColumnName()))); } } diff --git a/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorageTest.java b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorageTest.java index f11fdb905d..dd0ac7d9f3 100644 --- a/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorageTest.java +++ b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorageTest.java @@ -32,6 +32,7 @@ import org.apache.hertzbeat.common.constants.MetricDataConstants; import org.apache.hertzbeat.common.entity.arrow.ArrowCell; import org.apache.hertzbeat.common.entity.arrow.RowWrapper; import org.apache.hertzbeat.common.entity.message.CollectRep; +import org.apache.hertzbeat.common.util.JsonUtil; import org.awaitility.Awaitility; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -48,16 +49,20 @@ import org.springframework.http.HttpEntity; import org.springframework.http.HttpMethod; import org.springframework.http.HttpStatus; import org.springframework.http.ResponseEntity; +import org.springframework.boot.test.system.CapturedOutput; +import org.springframework.boot.test.system.OutputCaptureExtension; import org.springframework.web.client.RestTemplate; import java.util.List; +import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; /** * Test case for {@link VictoriaMetricsDataStorage} */ -@ExtendWith(MockitoExtension.class) +@ExtendWith({MockitoExtension.class, OutputCaptureExtension.class}) @MockitoSettings(strictness = Strictness.LENIENT) class VictoriaMetricsDataStorageTest { @@ -73,6 +78,7 @@ class VictoriaMetricsDataStorageTest { private VictoriaMetricsDataStorage victoriaMetricsDataStorage; private final AtomicInteger postForEntityCount = new AtomicInteger(0); + private final AtomicReference lastPayload = new AtomicReference<>(); @BeforeEach void setUp() { @@ -97,6 +103,10 @@ class VictoriaMetricsDataStorageTest { eq(String.class) )).thenAnswer(invocation -> { postForEntityCount.incrementAndGet(); + HttpEntity httpEntity = invocation.getArgument(1); + if (httpEntity.getBody() instanceof String payload) { + lastPayload.set(payload); + } return responseEntity; }); } @@ -184,6 +194,99 @@ class VictoriaMetricsDataStorageTest { .isGreaterThanOrEqualTo(threadCount * writeSize / bufferSize)); } + @Test + void customLabelsKeepJobButCannotOverrideStorageInstance() { + when(victoriaMetricsProperties.insert()).thenReturn(new VictoriaMetricsProperties.InsertConfig( + 1, Integer.MAX_VALUE, new VictoriaMetricsProperties.Compression(false))); + CollectRep.MetricsData metricsData = generateMockedMetricsData(); + when(metricsData.getLabels()).thenReturn(Map.of( + "job", "custom-job", + "instance", "custom-instance", + "region", "west")); + victoriaMetricsDataStorage = new VictoriaMetricsDataStorage(victoriaMetricsProperties, restTemplate); + + victoriaMetricsDataStorage.saveData(metricsData); + + Awaitility.await() + .atMost(5, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(postForEntityCount.get()).isEqualTo(1)); + VictoriaMetricsDataStorage.VictoriaMetricsContent content = + JsonUtil.fromJson(lastPayload.get().trim(), VictoriaMetricsDataStorage.VictoriaMetricsContent.class); + assertThat(content.getMetric()) + .containsEntry("job", "custom-job") + .containsEntry("instance", "storage-instance") + .containsEntry("region", "west"); + assertThat(victoriaMetricsDataStorage.getIgnoredLabelCollisionCount()).isEqualTo(1); + } + + @Test + void managedLabelCollisionsAreSkippedWithoutDroppingTheBatch(CapturedOutput output) { + when(victoriaMetricsProperties.insert()).thenReturn(new VictoriaMetricsProperties.InsertConfig( + 1, Integer.MAX_VALUE, new VictoriaMetricsProperties.Compression(false))); + CollectRep.MetricsData metricsData = generateMockedMetricsData(); + when(metricsData.getLabels()).thenReturn(Map.of( + "__name__", "custom-name", + "__monitor_id__", "custom-monitor")); + victoriaMetricsDataStorage = new VictoriaMetricsDataStorage(victoriaMetricsProperties, restTemplate); + + victoriaMetricsDataStorage.saveData(metricsData); + + Awaitility.await() + .atMost(5, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(postForEntityCount.get()).isEqualTo(1)); + VictoriaMetricsDataStorage.VictoriaMetricsContent content = + JsonUtil.fromJson(lastPayload.get().trim(), VictoriaMetricsDataStorage.VictoriaMetricsContent.class); + assertThat(content.getMetric()) + .containsEntry("__monitor_id__", "0") + .doesNotContainEntry("__name__", "custom-name") + .doesNotContainEntry("__monitor_id__", "custom-monitor"); + assertThat(victoriaMetricsDataStorage.getIgnoredLabelCollisionCount()).isEqualTo(2); + assertThat(output.getAll()) + .contains("__name__") + .contains("__monitor_id__") + .doesNotContain("custom-name") + .doesNotContain("custom-monitor"); + } + + @Test + void clusterWriterUsesTheSameNonDestructiveCollisionPolicy() { + when(responseEntity.getStatusCode()).thenReturn(HttpStatus.OK); + when(responseEntity.getBody()).thenReturn("{\"status\":\"success\"}"); + when(restTemplate.postForEntity( + startsWith("http://vm-insert"), + any(HttpEntity.class), + eq(String.class))) + .thenAnswer(invocation -> { + HttpEntity httpEntity = invocation.getArgument(1); + lastPayload.set((String) httpEntity.getBody()); + return responseEntity; + }); + VictoriaMetricsClusterProperties clusterProperties = new VictoriaMetricsClusterProperties( + true, + "0", + new VictoriaMetricsInsertProperties("http://vm-insert", null, null, 10, 0), + new VictoriaMetricsSelectProperties("http://vm-select", null, null)); + CollectRep.MetricsData metricsData = generateMockedMetricsData(); + when(metricsData.getLabels()).thenReturn(Map.of( + "instance", "custom-instance", + "region", "west")); + VictoriaMetricsClusterDataStorage clusterStorage = + new VictoriaMetricsClusterDataStorage(clusterProperties, restTemplate); + + try { + clusterStorage.saveData(metricsData); + + VictoriaMetricsDataStorage.VictoriaMetricsContent content = JsonUtil.fromJson( + lastPayload.get().trim(), VictoriaMetricsDataStorage.VictoriaMetricsContent.class); + assertThat(content.getMetric()) + .containsEntry("instance", "storage-instance") + .containsEntry("region", "west"); + assertThat(clusterStorage.getIgnoredLabelCollisionCount()).isEqualTo(1); + } finally { + clusterStorage.destroy(); + } + } + @AfterEach void stop() { if (victoriaMetricsDataStorage != null) { @@ -200,6 +303,8 @@ class VictoriaMetricsDataStorageTest { when(mockMetricsData.getTime()).thenReturn(System.currentTimeMillis()); when(mockMetricsData.getCode()).thenReturn(CollectRep.Code.SUCCESS); when(mockMetricsData.getApp()).thenReturn("app"); + when(mockMetricsData.getInstance()).thenReturn("storage-instance"); + when(mockMetricsData.getLabels()).thenReturn(Map.of()); CollectRep.ValueRow mockValueRow = Mockito.mock(CollectRep.ValueRow.class); List columnValues = List.of("server-test-01", "68.7"); diff --git a/home/docs/start/victoria-metrics-init.md b/home/docs/start/victoria-metrics-init.md index eb0f3265a9..3754686f71 100644 --- a/home/docs/start/victoria-metrics-init.md +++ b/home/docs/start/victoria-metrics-init.md @@ -148,6 +148,25 @@ warehouse: Once configured, restart HertzBeat to connect to the VictoriaMetrics cluster. +### Custom Label Collision Policy + +Monitor custom labels keep their existing Prometheus semantics when HertzBeat +writes to VictoriaMetrics: + +- `job` and ordinary custom labels continue to use their configured values. +- `instance`, `__name__`, `__monitor_id__`, `__metrics__`, and `__metric__` are + managed by HertzBeat. If a monitor supplies one of these custom-label keys, + HertzBeat ignores only the conflicting label, stores the remaining metrics, + and reports the key name through a rate-limited warning and cumulative count. + Label values are not written to the warning. + +Before upgrading, inspect monitor custom labels and rename managed keys. If an +existing monitor uses a custom `instance` value for a separate identity, move +that value to a distinct label such as `target_instance`; new samples use the +HertzBeat monitor instance. Existing VictoriaMetrics series are not rewritten. +No migration is needed for monitors that use `job` or other ordinary custom +labels. + ### FAQ 1. Do both the time series databases need to be configured? Can they both be used?