Skip to content

Commit 103c7a7

Browse files
m-nagarajanm-nagarajan
committed
[da-vinci][server] Add OTel metrics to DIVStatsReporter
Add dual Tehuti + OTel recording to AggVersionedDIVStats, consolidating 8 Tehuti AsyncGauge sensors into 4 OTel counter metrics with dimension-based differentiation. OTel metrics: - ingestion.div.message.count (ASYNC_COUNTER_FOR_HIGH_PERF_CASES) Dimensions: STORE, CLUSTER, VERSION_ROLE, VENICE_DIV_RESULT - ingestion.div.offset.rewind.count (COUNTER) Dimensions: STORE, CLUSTER, VERSION_ROLE, VENICE_DIV_SEVERITY - ingestion.div.producer.failure.count (COUNTER) Dimensions: STORE, CLUSTER, VERSION_ROLE - ingestion.div.producer.failure.benign_count (COUNTER) Dimensions: STORE, CLUSTER, VERSION_ROLE New dimension enums: VeniceDIVResult, VeniceDIVSeverity New metric entity enum: DIVOtelMetricEntity (registered in ServerMetricEntity) Enhanced OtelVersionedStatsUtils.classifyVersion with null-safe handling
1 parent b961f65 commit 103c7a7

16 files changed

Lines changed: 1042 additions & 21 deletions

File tree

‎clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/KafkaStoreIngestionService.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -421,7 +421,8 @@ public void handleStoreDeleted(Store store) {
421421
AggVersionedDIVStats versionedDIVStats = new AggVersionedDIVStats(
422422
metricsRepository,
423423
metadataRepo,
424-
serverConfig.isUnregisterMetricForDeletedStoreEnabled());
424+
serverConfig.isUnregisterMetricForDeletedStoreEnabled(),
425+
serverConfig.getClusterName());
425426
this.versionedIngestionStats = new AggVersionedIngestionStats(metricsRepository, metadataRepo, serverConfig);
426427
if (serverConfig.isDedicatedDrainerQueueEnabled()) {
427428
this.storeBufferService = new SeparatedStoreBufferService(serverConfig, metricsRepository);

‎clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/AggVersionedDIVStats.java‎

Lines changed: 151 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,33 +1,92 @@
11
package com.linkedin.davinci.stats;
22

3+
import static com.linkedin.davinci.stats.OtelVersionedStatsUtils.classifyVersion;
34
import static com.linkedin.venice.meta.Store.NON_EXISTING_VERSION;
5+
import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_STORE_NAME;
46

57
import com.linkedin.venice.exceptions.validation.CorruptDataException;
68
import com.linkedin.venice.exceptions.validation.DataValidationException;
79
import com.linkedin.venice.exceptions.validation.DuplicateDataException;
810
import com.linkedin.venice.exceptions.validation.MissingDataException;
911
import com.linkedin.venice.meta.ReadOnlyStoreRepository;
12+
import com.linkedin.venice.server.VersionRole;
13+
import com.linkedin.venice.stats.OpenTelemetryMetricsSetup;
14+
import com.linkedin.venice.stats.VeniceOpenTelemetryMetricsRepository;
15+
import com.linkedin.venice.stats.dimensions.VeniceDIVResult;
16+
import com.linkedin.venice.stats.dimensions.VeniceDIVSeverity;
17+
import com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions;
18+
import com.linkedin.venice.stats.metrics.MetricEntityStateOneEnum;
19+
import com.linkedin.venice.stats.metrics.MetricEntityStateTwoEnums;
1020
import com.linkedin.venice.utils.Utils;
21+
import com.linkedin.venice.utils.concurrent.VeniceConcurrentHashMap;
1122
import io.tehuti.metrics.MetricsRepository;
1223
import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
1324
import it.unimi.dsi.fastutil.ints.IntSet;
25+
import java.util.Collections;
26+
import java.util.HashMap;
27+
import java.util.Map;
1428
import java.util.concurrent.atomic.AtomicLong;
1529
import java.util.function.BiConsumer;
1630
import java.util.function.Function;
1731
import java.util.function.IntConsumer;
1832

1933

34+
/**
35+
* Aggregated versioned DIV stats with dual Tehuti + OTel recording.
36+
*
37+
* <p><b>Recording architecture:</b> Each public recording method (e.g., {@link #recordSuccessMsg})
38+
* records to <b>both</b> Tehuti (via {@code recordVersionedAndTotalStat} into total + per-version
39+
* {@link DIVStats} objects) and OTel (once per call, with store/cluster/version-role dimensions).
40+
* OTel totals are derived at query time by aggregating across the version-role dimension —
41+
* no separate OTel recording for total stats.
42+
*
43+
* <p><b>Version classification:</b> The version number passed to each recording method is classified
44+
* as CURRENT, FUTURE, or BACKUP for the OTel {@code VERSION_ROLE} dimension. Versions not matching
45+
* the registered current or future version default to BACKUP.
46+
*/
2047
public class AggVersionedDIVStats extends AbstractVeniceAggVersionedStats<DIVStats, DIVStatsReporter> {
48+
private final boolean emitOtelMetrics;
49+
private final VeniceOpenTelemetryMetricsRepository otelRepository;
50+
private final Map<VeniceMetricsDimensions, String> baseDimensionsMap;
51+
52+
/**
53+
* Per-store OTel metric state maps. Each map grows lazily via {@code computeIfAbsent} and is bounded
54+
* by the number of stores the server is actively ingesting. Entries are removed when a store is
55+
* deleted via {@link #handleStoreDeleted(String)}. These maps are OTel-only; Tehuti recording is
56+
* handled by the parent class via {@code recordVersionedAndTotalStat}.
57+
*/
58+
private final Map<String, MetricEntityStateTwoEnums<VersionRole, VeniceDIVResult>> messageCountPerStore =
59+
new VeniceConcurrentHashMap<>();
60+
private final Map<String, MetricEntityStateTwoEnums<VersionRole, VeniceDIVSeverity>> offsetRewindCountPerStore =
61+
new VeniceConcurrentHashMap<>();
62+
private final Map<String, MetricEntityStateOneEnum<VersionRole>> producerFailureCountPerStore =
63+
new VeniceConcurrentHashMap<>();
64+
private final Map<String, MetricEntityStateOneEnum<VersionRole>> benignProducerFailureCountPerStore =
65+
new VeniceConcurrentHashMap<>();
66+
67+
/**
68+
* Per-store version info for classifying versions as CURRENT, FUTURE, or BACKUP.
69+
* Updated via {@link #onVersionInfoUpdated(String, int, int)}.
70+
*/
71+
private final Map<String, OtelVersionedStatsUtils.VersionInfo> versionInfoMap = new VeniceConcurrentHashMap<>();
72+
2173
public AggVersionedDIVStats(
2274
MetricsRepository metricsRepository,
2375
ReadOnlyStoreRepository metadataRepository,
24-
boolean unregisterMetricForDeletedStoreEnabled) {
76+
boolean unregisterMetricForDeletedStoreEnabled,
77+
String clusterName) {
2578
super(
2679
metricsRepository,
2780
metadataRepository,
2881
DIVStats::new,
2982
DIVStatsReporter::new,
3083
unregisterMetricForDeletedStoreEnabled);
84+
85+
OpenTelemetryMetricsSetup.OpenTelemetryMetricsSetupInfo otelData =
86+
OpenTelemetryMetricsSetup.builder(metricsRepository).setClusterName(clusterName).build();
87+
this.emitOtelMetrics = otelData.emitOpenTelemetryMetrics();
88+
this.otelRepository = otelData.getOtelRepository();
89+
this.baseDimensionsMap = Collections.unmodifiableMap(otelData.getBaseDimensionsMap());
3190
}
3291

3392
public void recordException(String storeName, int version, DataValidationException e) {
@@ -42,34 +101,68 @@ public void recordException(String storeName, int version, DataValidationExcepti
42101

43102
public void recordDuplicateMsg(String storeName, int version) {
44103
recordVersionedAndTotalStat(storeName, version, DIVStats::recordDuplicateMsg);
104+
recordOtelMessageCount(storeName, version, VeniceDIVResult.DUPLICATE);
45105
}
46106

47107
public void recordMissingMsg(String storeName, int version) {
48108
recordVersionedAndTotalStat(storeName, version, DIVStats::recordMissingMsg);
109+
recordOtelMessageCount(storeName, version, VeniceDIVResult.MISSING);
49110
}
50111

51112
public void recordCorruptedMsg(String storeName, int version) {
52113
recordVersionedAndTotalStat(storeName, version, DIVStats::recordCorruptedMsg);
114+
recordOtelMessageCount(storeName, version, VeniceDIVResult.CORRUPTED);
53115
}
54116

55117
public void recordSuccessMsg(String storeName, int version) {
56118
recordVersionedAndTotalStat(storeName, version, DIVStats::recordSuccessMsg);
119+
recordOtelMessageCount(storeName, version, VeniceDIVResult.SUCCESS);
57120
}
58121

59122
public void recordBenignLeaderOffsetRewind(String storeName, int version) {
60123
recordVersionedAndTotalStat(storeName, version, DIVStats::recordBenignLeaderOffsetRewind);
124+
recordOtelOffsetRewindCount(storeName, version, VeniceDIVSeverity.BENIGN);
61125
}
62126

63127
public void recordPotentiallyLossyLeaderOffsetRewind(String storeName, int version) {
64128
recordVersionedAndTotalStat(storeName, version, DIVStats::recordPotentiallyLossyLeaderOffsetRewind);
129+
recordOtelOffsetRewindCount(storeName, version, VeniceDIVSeverity.POTENTIALLY_LOSSY);
65130
}
66131

67132
public void recordLeaderProducerFailure(String storeName, int version) {
68133
recordVersionedAndTotalStat(storeName, version, DIVStats::recordLeaderProducerFailure);
134+
recordOtelOneEnumMetric(
135+
storeName,
136+
version,
137+
producerFailureCountPerStore,
138+
DIVOtelMetricEntity.PRODUCER_FAILURE_COUNT);
69139
}
70140

71141
public void recordBenignLeaderProducerFailure(String storeName, int version) {
72142
recordVersionedAndTotalStat(storeName, version, DIVStats::recordBenignLeaderProducerFailure);
143+
recordOtelOneEnumMetric(
144+
storeName,
145+
version,
146+
benignProducerFailureCountPerStore,
147+
DIVOtelMetricEntity.BENIGN_PRODUCER_FAILURE_COUNT);
148+
}
149+
150+
@Override
151+
protected void onVersionInfoUpdated(String storeName, int currentVersion, int futureVersion) {
152+
versionInfoMap.put(storeName, new OtelVersionedStatsUtils.VersionInfo(currentVersion, futureVersion));
153+
}
154+
155+
@Override
156+
public void handleStoreDeleted(String storeName) {
157+
try {
158+
super.handleStoreDeleted(storeName);
159+
} finally {
160+
messageCountPerStore.remove(storeName);
161+
offsetRewindCountPerStore.remove(storeName);
162+
producerFailureCountPerStore.remove(storeName);
163+
benignProducerFailureCountPerStore.remove(storeName);
164+
versionInfoMap.remove(storeName);
165+
}
73166
}
74167

75168
@Override
@@ -124,4 +217,61 @@ private void resetTotalStats(
124217
existingVersions.forEach(versionConsumer);
125218
Utils.computeIfNotNull(getTotalStats(storeName), stat -> statsUpdater.accept(stat, totalStatCount.get()));
126219
}
220+
221+
// --- OTel recording helpers ---
222+
223+
private Map<VeniceMetricsDimensions, String> buildStoreDimensionsMap(String storeName) {
224+
Map<VeniceMetricsDimensions, String> map = new HashMap<>(baseDimensionsMap);
225+
map.put(VENICE_STORE_NAME, OpenTelemetryMetricsSetup.sanitizeStoreName(storeName));
226+
return Collections.unmodifiableMap(map);
227+
}
228+
229+
private void recordOtelMessageCount(String storeName, int version, VeniceDIVResult result) {
230+
if (!emitOtelMetrics) {
231+
return;
232+
}
233+
VersionRole role = classifyVersion(version, versionInfoMap.get(storeName));
234+
messageCountPerStore.computeIfAbsent(
235+
storeName,
236+
k -> MetricEntityStateTwoEnums.create(
237+
DIVOtelMetricEntity.MESSAGE_COUNT.getMetricEntity(),
238+
otelRepository,
239+
buildStoreDimensionsMap(k),
240+
VersionRole.class,
241+
VeniceDIVResult.class))
242+
.record(1, role, result);
243+
}
244+
245+
private void recordOtelOffsetRewindCount(String storeName, int version, VeniceDIVSeverity severity) {
246+
if (!emitOtelMetrics) {
247+
return;
248+
}
249+
VersionRole role = classifyVersion(version, versionInfoMap.get(storeName));
250+
offsetRewindCountPerStore.computeIfAbsent(
251+
storeName,
252+
k -> MetricEntityStateTwoEnums.create(
253+
DIVOtelMetricEntity.OFFSET_REWIND_COUNT.getMetricEntity(),
254+
otelRepository,
255+
buildStoreDimensionsMap(k),
256+
VersionRole.class,
257+
VeniceDIVSeverity.class))
258+
.record(1, role, severity);
259+
}
260+
261+
private void recordOtelOneEnumMetric(
262+
String storeName,
263+
int version,
264+
Map<String, MetricEntityStateOneEnum<VersionRole>> perStoreMap,
265+
DIVOtelMetricEntity metricEntity) {
266+
if (!emitOtelMetrics) {
267+
return;
268+
}
269+
VersionRole role = classifyVersion(version, versionInfoMap.get(storeName));
270+
perStoreMap
271+
.computeIfAbsent(
272+
storeName,
273+
k -> MetricEntityStateOneEnum
274+
.create(metricEntity.getMetricEntity(), otelRepository, buildStoreDimensionsMap(k), VersionRole.class))
275+
.record(1, role);
276+
}
127277
}

‎clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/DIVOtelMetricEntity.java‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
package com.linkedin.davinci.stats;
2+
3+
import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_CLUSTER_NAME;
4+
import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_DIV_RESULT;
5+
import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_DIV_SEVERITY;
6+
import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_STORE_NAME;
7+
import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_VERSION_ROLE;
8+
import static com.linkedin.venice.utils.Utils.setOf;
9+
10+
import com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions;
11+
import com.linkedin.venice.stats.metrics.MetricEntity;
12+
import com.linkedin.venice.stats.metrics.MetricType;
13+
import com.linkedin.venice.stats.metrics.MetricUnit;
14+
import com.linkedin.venice.stats.metrics.ModuleMetricEntityInterface;
15+
import java.util.Set;
16+
17+
18+
/**
19+
* OTel metric entity definitions for Data Integrity Validation (DIV) stats.
20+
* Consolidates 8 Tehuti AsyncGauge sensors into 4 OTel counter metrics.
21+
*/
22+
public enum DIVOtelMetricEntity implements ModuleMetricEntityInterface {
23+
// Recorded per-message on the leader ingestion path — uses async counter for high throughput.
24+
MESSAGE_COUNT(
25+
"ingestion.div.message.count", MetricType.ASYNC_COUNTER_FOR_HIGH_PERF_CASES, MetricUnit.NUMBER,
26+
"Count of DIV message validation results",
27+
setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE, VENICE_DIV_RESULT)
28+
),
29+
30+
OFFSET_REWIND_COUNT(
31+
"ingestion.div.offset.rewind.count", MetricType.COUNTER, MetricUnit.NUMBER,
32+
"Count of leader offset rewind events",
33+
setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE, VENICE_DIV_SEVERITY)
34+
),
35+
36+
PRODUCER_FAILURE_COUNT(
37+
"ingestion.div.producer.failure.count", MetricType.COUNTER, MetricUnit.NUMBER,
38+
"Count of leader producer failures", setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE)
39+
),
40+
41+
// Not mutually exclusive with PRODUCER_FAILURE_COUNT — a single produce operation can increment both counters.
42+
BENIGN_PRODUCER_FAILURE_COUNT(
43+
"ingestion.div.producer.failure.benign_count", MetricType.COUNTER, MetricUnit.NUMBER,
44+
"Count of benign leader producer failures", setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE)
45+
);
46+
47+
private final MetricEntity metricEntity;
48+
49+
DIVOtelMetricEntity(
50+
String metricName,
51+
MetricType metricType,
52+
MetricUnit unit,
53+
String description,
54+
Set<VeniceMetricsDimensions> dimensions) {
55+
this.metricEntity = new MetricEntity(metricName, metricType, unit, description, dimensions);
56+
}
57+
58+
@Override
59+
public MetricEntity getMetricEntity() {
60+
return metricEntity;
61+
}
62+
}

‎clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/OtelVersionedStatsUtils.java‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,14 +37,19 @@ public int getFutureVersion() {
3737

3838
/**
3939
* Classifies a version as CURRENT, FUTURE, or BACKUP.
40+
* Returns {@link VersionRole#BACKUP} when {@code versionInfo} is null (e.g., store
41+
* not yet registered in a per-store version info map).
4042
*
4143
* @param version The version number to classify
42-
* @param versionInfo The current/future version info
44+
* @param versionInfo The current/future version info, or null
4345
* @return {@link VersionRole#CURRENT} if version matches currentVersion,
4446
* {@link VersionRole#FUTURE} if version matches futureVersion,
45-
* {@link VersionRole#BACKUP} otherwise
47+
* {@link VersionRole#BACKUP} otherwise or if versionInfo is null
4648
*/
4749
public static VersionRole classifyVersion(int version, VersionInfo versionInfo) {
50+
if (versionInfo == null) {
51+
return VersionRole.BACKUP;
52+
}
4853
if (version == versionInfo.currentVersion) {
4954
return VersionRole.CURRENT;
5055
} else if (version == versionInfo.futureVersion) {

‎clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ServerMetricEntity.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,8 @@ public static List<Class<? extends ModuleMetricEntityInterface>> getMetricEntity
3838
HeartbeatMonitoringOtelMetricEntity.class,
3939
BlobTransferOtelMetricEntity.class,
4040
KafkaConsumerServiceOtelMetricEntity.class,
41-
RocksDBMemoryOtelMetricEntity.class);
41+
RocksDBMemoryOtelMetricEntity.class,
42+
DIVOtelMetricEntity.class);
4243
}
4344

4445
public static final Collection<MetricEntity> SERVER_METRIC_ENTITIES =

‎clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/DIVOtelMetricEntityTest.java‎

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
package com.linkedin.davinci.stats;
2+
3+
import static com.linkedin.davinci.stats.DIVOtelMetricEntity.BENIGN_PRODUCER_FAILURE_COUNT;
4+
import static com.linkedin.davinci.stats.DIVOtelMetricEntity.MESSAGE_COUNT;
5+
import static com.linkedin.davinci.stats.DIVOtelMetricEntity.OFFSET_REWIND_COUNT;
6+
import static com.linkedin.davinci.stats.DIVOtelMetricEntity.PRODUCER_FAILURE_COUNT;
7+
import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_CLUSTER_NAME;
8+
import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_DIV_RESULT;
9+
import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_DIV_SEVERITY;
10+
import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_STORE_NAME;
11+
import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_VERSION_ROLE;
12+
import static com.linkedin.venice.utils.Utils.setOf;
13+
14+
import com.linkedin.venice.stats.metrics.MetricType;
15+
import com.linkedin.venice.stats.metrics.MetricUnit;
16+
import com.linkedin.venice.stats.metrics.ModuleMetricEntityTestFixture;
17+
import com.linkedin.venice.stats.metrics.ModuleMetricEntityTestFixture.MetricEntityExpectation;
18+
import java.util.HashMap;
19+
import java.util.Map;
20+
import org.testng.annotations.Test;
21+
22+
23+
public class DIVOtelMetricEntityTest {
24+
@Test
25+
public void testMetricEntities() {
26+
new ModuleMetricEntityTestFixture<>(DIVOtelMetricEntity.class, expectedDefinitions()).assertAll();
27+
}
28+
29+
private static Map<DIVOtelMetricEntity, MetricEntityExpectation> expectedDefinitions() {
30+
Map<DIVOtelMetricEntity, MetricEntityExpectation> map = new HashMap<>();
31+
map.put(
32+
MESSAGE_COUNT,
33+
new MetricEntityExpectation(
34+
"ingestion.div.message.count",
35+
MetricType.ASYNC_COUNTER_FOR_HIGH_PERF_CASES,
36+
MetricUnit.NUMBER,
37+
"Count of DIV message validation results",
38+
setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE, VENICE_DIV_RESULT)));
39+
map.put(
40+
OFFSET_REWIND_COUNT,
41+
new MetricEntityExpectation(
42+
"ingestion.div.offset.rewind.count",
43+
MetricType.COUNTER,
44+
MetricUnit.NUMBER,
45+
"Count of leader offset rewind events",
46+
setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE, VENICE_DIV_SEVERITY)));
47+
map.put(
48+
PRODUCER_FAILURE_COUNT,
49+
new MetricEntityExpectation(
50+
"ingestion.div.producer.failure.count",
51+
MetricType.COUNTER,
52+
MetricUnit.NUMBER,
53+
"Count of leader producer failures",
54+
setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE)));
55+
map.put(
56+
BENIGN_PRODUCER_FAILURE_COUNT,
57+
new MetricEntityExpectation(
58+
"ingestion.div.producer.failure.benign_count",
59+
MetricType.COUNTER,
60+
MetricUnit.NUMBER,
61+
"Count of benign leader producer failures",
62+
setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_VERSION_ROLE)));
63+
return map;
64+
}
65+
}

0 commit comments

Comments
 (0)