Skip to content

maintenance: define VictoriaMetrics label collisions #4286

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

Open
zqr10159 wants to merge 5 commits into
base: master
Choose a base branch
from
Open

maintenance: define VictoriaMetrics label collisions #4286

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
8d03fda
maintenance: preserve VictoriaMetrics label identity
zqr10159 Jul 30, 2026
0d24ce0
maintenance: define VictoriaMetrics label collisions
zqr10159 Jul 30, 2026
b87ff44
Make metric label collisions observable
zqr10159 Aug 4, 2026
31abf4d
Merge remote-tracking branch 'apache/master' into maintenance/vm-labe…
zqr10159 Aug 9, 2026
4918d3c
Merge remote-tracking branch 'apache/master' into maintenance/vm-labe…
zqr10159 Aug 9, 2026
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
29 changes: 29 additions & 0 deletions ...ava/org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeDbDataStorage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,9 +45,12 @@
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;
Expand Down Expand Up @@ -109,6 +112,7 @@ public class GreptimeDbDataStorage extends AbstractHistoryDataStorage {
private final RestTemplate restTemplate;

private final GreptimeSqlQueryExecutor greptimeSqlQueryExecutor;
private final AtomicLong rejectedLabelCollisionCount = new AtomicLong();

public GreptimeDbDataStorage(GreptimeProperties greptimeProperties,
@Qualifier(WarehouseConstants.GREPTIME_QUERY_REST_TEMPLATE)
Expand Down Expand Up @@ -160,6 +164,14 @@ public void saveData(CollectRep.MetricsData metricsData) {
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 rejectedCount = rejectedLabelCollisionCount.incrementAndGet();
log.error("[warehouse greptime] reject metrics data {} because custom labels contain "
+ "storage-managed keys {}; cumulative rejected batches: {}.",
metricsData.getId(), labelCollisions, rejectedCount);
return;
}
fields.forEach(field -> {
if (field.getLabel()) {
tableSchemaBuilder.addTag(field.getName(), DataType.String);
Expand Down Expand Up @@ -233,6 +245,23 @@ public void saveData(CollectRep.MetricsData metricsData) {
}
}

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;
}

long getRejectedLabelCollisionCount() {
return rejectedLabelCollisionCount.get();
}

@Override
public Map<String, List<Value>> getHistoryMetricData(String instance, String app, String metrics, String metric,
String history) {
Expand Down
21 changes: 16 additions & 5 deletions ...g/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
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;

Expand Down Expand Up @@ -69,7 +70,6 @@
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;
Expand Down Expand Up @@ -107,6 +107,7 @@ public class VictoriaMetricsClusterDataStorage extends AbstractHistoryDataStorag
private final VictoriaMetricsSelectProperties vmSelectProps;
private final RestTemplate restTemplate;
private final BlockingQueue<VictoriaMetricsDataStorage.VictoriaMetricsContent> metricsBufferQueue;
private final AtomicLong rejectedLabelCollisionCount = new AtomicLong();

private HashedWheelTimer metricsFlushTimer = null;
private MetricsFlushTask metricsFlushtask = null;
Expand Down Expand Up @@ -186,6 +187,15 @@ public void saveData(CollectRep.MetricsData metricsData) {
metricsData.getId(), metricsData.getApp(), metricsData.getMetrics());
return;
}
var managedLabelCollisions =
VictoriaMetricsDataStorage.findManagedLabelCollisions(metricsData.getLabels());
if (!managedLabelCollisions.isEmpty()) {
long rejectedCount = rejectedLabelCollisionCount.incrementAndGet();
log.error("[warehouse victoria-metrics] reject metrics data {} because custom labels contain "
+ "HertzBeat-managed keys {}; cumulative rejected batches: {}.",
metricsData.getId(), managedLabelCollisions, rejectedCount);
return;
}
Map<String, String> defaultLabels = Maps.newHashMapWithExpectedSize(8);
defaultLabels.put(MONITOR_METRICS_KEY, metricsData.getMetrics());
boolean isPrometheusAuto;
Expand Down Expand Up @@ -243,10 +253,7 @@ public void saveData(CollectRep.MetricsData metricsData) {
}
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, metricsData.getLabels());
VictoriaMetricsDataStorage.VictoriaMetricsContent content = VictoriaMetricsDataStorage.VictoriaMetricsContent.builder()
.metric(new HashMap<>(labels))
.values(new Double[]{entry.getValue()})
Expand Down Expand Up @@ -276,6 +283,10 @@ public void saveData(CollectRep.MetricsData metricsData) {
}
}

long getRejectedLabelCollisionCount() {
return rejectedLabelCollisionCount.get();
}

@Override
public void destroy() {
if (metricsFlushTimer != null && !metricsFlushTimer.isStop()) {
Expand Down
42 changes: 38 additions & 4 deletions ...java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,10 +32,13 @@
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;
Expand Down Expand Up @@ -98,12 +101,18 @@ 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);
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 rejectedLabelCollisionCount = new AtomicLong();

private HashedWheelTimer metricsFlushTimer = null;
private final VictoriaMetricsProperties.InsertConfig insertConfig;
Expand Down Expand Up @@ -170,6 +179,14 @@ public void saveData(CollectRep.MetricsData metricsData) {
metricsData.getId(), metricsData.getApp(), metricsData.getMetrics());
return;
}
Set<String> managedLabelCollisions = findManagedLabelCollisions(metricsData.getLabels());
if (!managedLabelCollisions.isEmpty()) {
long rejectedCount = rejectedLabelCollisionCount.incrementAndGet();
log.error("[warehouse victoria-metrics] reject metrics data {} because custom labels contain "
+ "HertzBeat-managed keys {}; cumulative rejected batches: {}.",
metricsData.getId(), managedLabelCollisions, rejectedCount);
return;
}
Map<String, String> defaultLabels = Maps.newHashMapWithExpectedSize(8);
defaultLabels.put(MONITOR_METRICS_KEY, metricsData.getMetrics());
boolean isPrometheusAuto = false;
Expand Down Expand Up @@ -226,10 +243,7 @@ public void saveData(CollectRep.MetricsData metricsData) {
}
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, metricsData.getLabels());
VictoriaMetricsContent content = VictoriaMetricsContent.builder()
.metric(new HashMap<>(labels))
.values(new Double[]{entry.getValue()})
Expand All @@ -255,6 +269,26 @@ public void saveData(CollectRep.MetricsData metricsData) {
sendVictoriaMetrics(contentList);
}

static void addCustomizedLabels(Map<String, String> labels, Map<String, String> customizedLabels) {
if (ObjectUtils.isEmpty(customizedLabels)) {
return;
}
labels.putAll(customizedLabels);
}

long getRejectedLabelCollisionCount() {
return rejectedLabelCollisionCount.get();
}

static Set<String> findManagedLabelCollisions(Map<String, String> customizedLabels) {
if (ObjectUtils.isEmpty(customizedLabels)) {
return Set.of();
}
Set<String> collisions = new TreeSet<>(customizedLabels.keySet());
collisions.retainAll(MANAGED_LABEL_KEYS);
return collisions;
}

@Override
public void destroy() {
if (metricsFlushTimer != null && !metricsFlushTimer.isStop()) {
Expand Down
54 changes: 24 additions & 30 deletions ...org/apache/hertzbeat/warehouse/store/history/tsdb/greptime/GreptimeDbDataStorageTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,6 @@
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;
Expand All @@ -48,7 +47,6 @@
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;
Expand Down Expand Up @@ -132,7 +130,6 @@ void testSaveData() {
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
Expand All @@ -156,15 +153,10 @@ void testSaveData() {
}

@Test
void testSaveDataWithCustomLabels() throws Exception {
void testSaveDataRejectsCustomLabelCollisions() {
try (MockedStatic<GreptimeDB> mockedStatic = mockStatic(GreptimeDB.class)) {
mockedStatic.when(() -> GreptimeDB.create(any())).thenReturn(greptimeDb);

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);

greptimeDbDataStorage = new GreptimeDbDataStorage(greptimeProperties, restTemplate, greptimeSqlQueryExecutor);

CollectRep.MetricsData metricsData = createMockMetricsData(true);
Expand All @@ -176,31 +168,33 @@ void testSaveDataWithCustomLabels() throws Exception {
when(metricsData.getLabels()).thenReturn(customLabels);
when(metricsData.getInstance()).thenReturn("server1");

greptimeDbDataStorage.saveData(metricsData);

verify(greptimeDb, never()).write(any(Table.class));
assertEquals(1, greptimeDbDataStorage.getRejectedLabelCollisionCount());
}
}

@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"));

ArgumentCaptor<Table> tableCaptor = ArgumentCaptor.forClass(Table.class);
greptimeDbDataStorage.saveData(metricsData);

verify(greptimeDb).write(tableCaptor.capture());
Table capturedTable = tableCaptor.getValue();

List<RowData.ColumnSchema> columnSchemas = getColumnSchemas(capturedTable);
List<String> columnNames = columnSchemas.stream()
.map(RowData.ColumnSchema::getColumnName)
.collect(Collectors.toList());
assertEquals(5, columnNames.size());
assertEquals(2, Collections.frequency(columnNames, "instance"));
assertEquals(1, Collections.frequency(columnNames, "ts"));
assertEquals(1, Collections.frequency(columnNames, "usage"));

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());

List<RowData.Row> rows = getRows(capturedTable);
assertEquals(1, rows.size());
RowData.Row row = rows.get(0);
assertEquals("prod", row.getValuesList().get(4).getStringValue());
List<RowData.ColumnSchema> schemas = getColumnSchemas(tableCaptor.getValue());
assertTrue(schemas.stream().anyMatch(schema -> "env".equals(schema.getColumnName())));
}
}

Expand Down
Loading
Loading