This is an automated email from the ASF dual-hosted git repository.
Duansg pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hertzbeat.git
The following commit(s) were added to refs/heads/master by this push:
new db73a43f9e maintenance: define VictoriaMetrics label collisions (#4286)
db73a43f9e is described below
commit db73a43f9e34e3252b908d461157261f08f6c965
Author: Logic <[email protected]>
AuthorDate: Wed Aug 26 00:46:27 2026 +0800
maintenance: define VictoriaMetrics label collisions (#4286)
Co-authored-by: Duansg <[email protected]>
---
.../tsdb/greptime/GreptimeDbDataStorage.java | 35 +++++++
.../tsdb/vm/VictoriaMetricsClusterDataStorage.java | 30 +++++-
.../tsdb/vm/VictoriaMetricsDataStorage.java | 68 ++++++++++++-
.../tsdb/greptime/GreptimeDbDataStorageTest.java | 54 ++++++-----
.../tsdb/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<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) {
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<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();
- 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<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()) {
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<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();
- 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<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()) {
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<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))).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<RowData.ColumnSchema> columnSchemas =
getColumnSchemas(capturedTable);
- List<String> columnNames = columnSchemas.stream()
+ List<String> 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());
+ }
+ }
+
+ @Test
+ void testSaveDataAcceptsNonConflictingCustomLabels() 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);
+ 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.ColumnSchema> envColumnSchemas =
columnSchemas.stream()
- .filter(columnSchema ->
"env".equals(columnSchema.getColumnName()))
- .collect(Collectors.toList());
- assertEquals(1, envColumnSchemas.size());
- assertEquals(Common.SemanticType.TAG,
envColumnSchemas.get(0).getSemanticType());
+ ArgumentCaptor<Table> tableCaptor =
ArgumentCaptor.forClass(Table.class);
+ greptimeDbDataStorage.saveData(metricsData);
- List<RowData.Row> rows = getRows(capturedTable);
- assertEquals(1, rows.size());
- 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())));
}
}
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<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");
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?
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
