maintenance: define VictoriaMetrics label collisions (#4286)

Co-authored-by: Duansg <siguoduan@gmail.com>

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<CollectRep.Field> fields = metricsData.getFields();
         Map<String, String> customLabels = metricsData.getLabels();
         List<String> fieldNames = fields.stream().map(CollectRep.Field::getName).collect(Collectors.toList());
+        Set<String> 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<String> findLabelCollisions(Map<String, String> customLabels, List<String> fieldNames) {
+        if (customLabels == null || customLabels.isEmpty()) {
+            return Set.of();
+        }
+        Set<String> 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<String, List<Value>> getHistoryMetricData(String instance, String app, String metrics, String metric,
                                                          String history) {

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<VictoriaMetricsDataStorage.VictoriaMetricsContent> 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<String, String> customizedLabels = VictoriaMetricsDataStorage.withoutManagedLabels(
+                metricsData.getLabels(), managedLabelCollisions);
         Map<String, String> 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();
+                                VictoriaMetricsDataStorage.addCustomizedLabels(labels, customizedLabels);
-                                if (!ObjectUtils.isEmpty(customizedLabels)) {
-                                    labels.putAll(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<String> 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()) {

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<String> 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<VictoriaMetricsDataStorage.VictoriaMetricsContent> 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<String> managedLabelCollisions = findManagedLabelCollisions(metricsData.getLabels());
+        if (!managedLabelCollisions.isEmpty()) {
+            recordIgnoredLabelCollisions(metricsData.getId(), managedLabelCollisions);
+        }
+        Map<String, String> customizedLabels = withoutManagedLabels(
+                metricsData.getLabels(), managedLabelCollisions);
         Map<String, String> 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();
+                            addCustomizedLabels(labels, customizedLabels);
-                            if (!ObjectUtils.isEmpty(customizedLabels)) {
-                                labels.putAll(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<String, String> labels, Map<String, String> customizedLabels) {
+        if (ObjectUtils.isEmpty(customizedLabels)) {
+            return;
+        }
+        labels.putAll(customizedLabels);
+    }
+
+    private void recordIgnoredLabelCollisions(long monitorId, Set<String> 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<String> findManagedLabelCollisions(Map<String, String> customizedLabels) {
+        if (ObjectUtils.isEmpty(customizedLabels)) {
+            return Set.of();
+        }
+        Set<String> collisions = new TreeSet<>();
+        for (String key : customizedLabels.keySet()) {
+            if (key != null && MANAGED_LABEL_KEYS.contains(key)) {
+                collisions.add(key);
+            }
+        }
+        return collisions;
+    }
+
+    static Map<String, String> withoutManagedLabels(
+            Map<String, String> customizedLabels, Set<String> managedLabelCollisions) {
+        if (ObjectUtils.isEmpty(customizedLabels)) {
+            return Map.of();
+        }
+        if (managedLabelCollisions.isEmpty()) {
+            return customizedLabels;
+        }
+        Map<String, String> safeLabels = new HashMap<>(customizedLabels);
+        managedLabelCollisions.forEach(safeLabels::remove);
+        return safeLabels;
+    }
+
     @Override
     public void destroy() {
         if (metricsFlushTimer != null && !metricsFlushTimer.isStop()) {

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<Result<WriteOk, Err>> 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<GreptimeDB> mockedStatic = mockStatic(GreptimeDB.class)) {
             mockedStatic.when(() -> GreptimeDB.create(any())).thenReturn(greptimeDb);
-
+            @SuppressWarnings("unchecked")
             Result<WriteOk, Err> mockResult = mock(Result.class);
             when(mockResult.isOk()).thenReturn(true);
-            CompletableFuture<Result<WriteOk, Err>> mockFuture = CompletableFuture.completedFuture(mockResult);
+            when(greptimeDb.write(any(Table.class)))
-            when(greptimeDb.write(any(Table.class))).thenReturn(mockFuture);
+                    .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<String> columnNames = getColumnSchemas(tableCaptor.getValue()).stream()
-
-            List<RowData.ColumnSchema> columnSchemas = getColumnSchemas(capturedTable);
-            List<String> columnNames = columnSchemas.stream()
                     .map(RowData.ColumnSchema::getColumnName)
-                    .collect(Collectors.toList());
+                    .toList();
-            assertEquals(5, columnNames.size());
+            // 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<RowData.ColumnSchema> envColumnSchemas = columnSchemas.stream()
+    @Test
-                    .filter(columnSchema -> "env".equals(columnSchema.getColumnName()))
+    void testSaveDataAcceptsNonConflictingCustomLabels() throws Exception {
-                    .collect(Collectors.toList());
+        try (MockedStatic<GreptimeDB> mockedStatic = mockStatic(GreptimeDB.class)) {
-            assertEquals(1, envColumnSchemas.size());
+            mockedStatic.when(() -> GreptimeDB.create(any())).thenReturn(greptimeDb);
-            assertEquals(Common.SemanticType.TAG, envColumnSchemas.get(0).getSemanticType());
+            @SuppressWarnings("unchecked")
+            Result<WriteOk, Err> 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<RowData.Row> rows = getRows(capturedTable);
+            ArgumentCaptor<Table> tableCaptor = ArgumentCaptor.forClass(Table.class);
-            assertEquals(1, rows.size());
+            greptimeDbDataStorage.saveData(metricsData);
-            RowData.Row row = rows.get(0);
+
-            assertEquals("prod", row.getValuesList().get(4).getStringValue());
+            verify(greptimeDb).write(tableCaptor.capture());
+            List<RowData.ColumnSchema> schemas = getColumnSchemas(tableCaptor.getValue());
+            assertTrue(schemas.stream().anyMatch(schema -> "env".equals(schema.getColumnName())));
         }
     }
 

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<String> 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<String> columnValues = List.of("server-test-01", "68.7");

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?