Skip to content

Commit 55a09c7

Browse files
m-nagarajanm-nagarajan
authored
[da-vinci] Add OTel metrics to AggVersionedDaVinciRecordTransformerStats (linkedin#2725)
Add 2 OTel metrics to AggVersionedDaVinciRecordTransformerStats with a VENICE_RECORD_TRANSFORMER_OPERATION dimension (PUT/DELETE): - record_transformer.latency (HISTOGRAM) — DaVinci record transformer operation latency - record_transformer.error_count (COUNTER) — DaVinci record transformer operation error count
1 parent a59a126 commit 55a09c7

10 files changed

Lines changed: 505 additions & 4 deletions

File tree

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

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

3+
import static com.linkedin.davinci.stats.DaVinciRecordTransformerOtelMetricEntity.RECORD_TRANSFORMER_ERROR_COUNT;
4+
import static com.linkedin.davinci.stats.DaVinciRecordTransformerOtelMetricEntity.RECORD_TRANSFORMER_LATENCY;
5+
6+
import com.google.common.annotations.VisibleForTesting;
37
import com.linkedin.davinci.config.VeniceServerConfig;
48
import com.linkedin.venice.meta.ReadOnlyStoreRepository;
9+
import com.linkedin.venice.stats.OpenTelemetryMetricsSetup;
10+
import com.linkedin.venice.stats.VeniceOpenTelemetryMetricsRepository;
11+
import com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions;
12+
import com.linkedin.venice.stats.dimensions.VeniceRecordTransformerOperation;
13+
import com.linkedin.venice.stats.metrics.MetricEntityStateOneEnum;
14+
import com.linkedin.venice.utils.concurrent.VeniceConcurrentHashMap;
515
import io.tehuti.metrics.MetricsRepository;
16+
import java.util.HashMap;
17+
import java.util.Map;
618

719

820
/**
9-
* The store level stats for {@link com.linkedin.davinci.client.DaVinciRecordTransformer}
21+
* The store level stats for {@link com.linkedin.davinci.client.DaVinciRecordTransformer}.
22+
* OTel metrics are recorded directly here (separate API) because Tehuti uses the Reporter
23+
* layer ({@link DaVinciRecordTransformerStatsReporter}) with AsyncGauge polling, while OTel
24+
* records at the point of the call.
1025
*/
1126
public class AggVersionedDaVinciRecordTransformerStats
1227
extends AbstractVeniceAggVersionedStats<DaVinciRecordTransformerStats, DaVinciRecordTransformerStatsReporter> {
28+
private final VeniceOpenTelemetryMetricsRepository otelRepository;
29+
private final Map<VeniceMetricsDimensions, String> baseDimensionsMap;
30+
private final boolean emitOtelMetrics;
31+
32+
/**
33+
* Per-store OTel metric state for latency. Bounded by the number of stores on this host.
34+
* Entries created lazily inside {@link #recordOtelLatency}, removed in
35+
* {@link #handleStoreDeleted(String)}.
36+
*/
37+
private final Map<String, MetricEntityStateOneEnum<VeniceRecordTransformerOperation>> latencyPerStore =
38+
new VeniceConcurrentHashMap<>();
39+
40+
/**
41+
* Per-store OTel metric state for error count. Same bounding and lifecycle as latencyPerStore.
42+
*/
43+
private final Map<String, MetricEntityStateOneEnum<VeniceRecordTransformerOperation>> errorCountPerStore =
44+
new VeniceConcurrentHashMap<>();
45+
1346
public AggVersionedDaVinciRecordTransformerStats(
1447
MetricsRepository metricsRepository,
1548
ReadOnlyStoreRepository metadataRepository,
@@ -20,21 +53,90 @@ public AggVersionedDaVinciRecordTransformerStats(
2053
DaVinciRecordTransformerStats::new,
2154
DaVinciRecordTransformerStatsReporter::new,
2255
serverConfig.isUnregisterMetricForDeletedStoreEnabled());
56+
57+
OpenTelemetryMetricsSetup.OpenTelemetryMetricsSetupInfo otelData =
58+
OpenTelemetryMetricsSetup.builder(metricsRepository).setClusterName(serverConfig.getClusterName()).build();
59+
this.otelRepository = otelData.getOtelRepository();
60+
this.baseDimensionsMap = otelData.getBaseDimensionsMap();
61+
this.emitOtelMetrics = otelData.emitOpenTelemetryMetrics();
62+
}
63+
64+
@Override
65+
public void handleStoreDeleted(String storeName) {
66+
try {
67+
super.handleStoreDeleted(storeName);
68+
} finally {
69+
latencyPerStore.remove(storeName);
70+
errorCountPerStore.remove(storeName);
71+
}
2372
}
2473

2574
public void recordPutLatency(String storeName, int version, double value, long timestamp) {
2675
recordVersionedAndTotalStat(storeName, version, stat -> stat.recordPutLatency(value, timestamp));
76+
recordOtelLatency(storeName, value, VeniceRecordTransformerOperation.PUT);
2777
}
2878

2979
public void recordDeleteLatency(String storeName, int version, double value, long timestamp) {
3080
recordVersionedAndTotalStat(storeName, version, stat -> stat.recordDeleteLatency(value, timestamp));
81+
recordOtelLatency(storeName, value, VeniceRecordTransformerOperation.DELETE);
3182
}
3283

3384
public void recordPutError(String storeName, int version, long timestamp) {
3485
recordVersionedAndTotalStat(storeName, version, stat -> stat.recordPutError(timestamp));
86+
recordOtelErrorCount(storeName, VeniceRecordTransformerOperation.PUT);
3587
}
3688

3789
public void recordDeleteError(String storeName, int version, long timestamp) {
3890
recordVersionedAndTotalStat(storeName, version, stat -> stat.recordDeleteError(timestamp));
91+
recordOtelErrorCount(storeName, VeniceRecordTransformerOperation.DELETE);
92+
}
93+
94+
private void recordOtelLatency(String storeName, double value, VeniceRecordTransformerOperation operation) {
95+
if (!emitOtelMetrics) {
96+
return;
97+
}
98+
latencyPerStore.computeIfAbsent(storeName, k -> createPerStoreMetric(k, RECORD_TRANSFORMER_LATENCY))
99+
.record(value, operation);
100+
}
101+
102+
private void recordOtelErrorCount(String storeName, VeniceRecordTransformerOperation operation) {
103+
if (!emitOtelMetrics) {
104+
return;
105+
}
106+
errorCountPerStore.computeIfAbsent(storeName, k -> createPerStoreMetric(k, RECORD_TRANSFORMER_ERROR_COUNT))
107+
.record(1, operation);
108+
}
109+
110+
@VisibleForTesting
111+
boolean hasLatencyMetricFor(String storeName) {
112+
return latencyPerStore.containsKey(storeName);
113+
}
114+
115+
@VisibleForTesting
116+
boolean hasErrorCountMetricFor(String storeName) {
117+
return errorCountPerStore.containsKey(storeName);
118+
}
119+
120+
@VisibleForTesting
121+
int latencyStoreCount() {
122+
return latencyPerStore.size();
123+
}
124+
125+
@VisibleForTesting
126+
int errorCountStoreCount() {
127+
return errorCountPerStore.size();
128+
}
129+
130+
private MetricEntityStateOneEnum<VeniceRecordTransformerOperation> createPerStoreMetric(
131+
String storeName,
132+
DaVinciRecordTransformerOtelMetricEntity metricEntity) {
133+
Map<VeniceMetricsDimensions, String> storeDimensionsMap = new HashMap<>(baseDimensionsMap);
134+
storeDimensionsMap
135+
.put(VeniceMetricsDimensions.VENICE_STORE_NAME, OpenTelemetryMetricsSetup.sanitizeStoreName(storeName));
136+
return MetricEntityStateOneEnum.create(
137+
metricEntity.getMetricEntity(),
138+
otelRepository,
139+
storeDimensionsMap,
140+
VeniceRecordTransformerOperation.class);
39141
}
40142
}

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

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
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_RECORD_TRANSFORMER_OPERATION;
5+
import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_STORE_NAME;
6+
import static com.linkedin.venice.utils.Utils.setOf;
7+
8+
import com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions;
9+
import com.linkedin.venice.stats.metrics.MetricEntity;
10+
import com.linkedin.venice.stats.metrics.MetricType;
11+
import com.linkedin.venice.stats.metrics.MetricUnit;
12+
import com.linkedin.venice.stats.metrics.ModuleMetricEntityInterface;
13+
import java.util.Set;
14+
15+
16+
/**
17+
* OTel metric entity definitions recorded from {@link AggVersionedDaVinciRecordTransformerStats}.
18+
* Tracks DaVinci record transformer latency and error counts by operation (put/delete).
19+
*/
20+
public enum DaVinciRecordTransformerOtelMetricEntity implements ModuleMetricEntityInterface {
21+
RECORD_TRANSFORMER_LATENCY(
22+
"record_transformer.latency", MetricType.HISTOGRAM, MetricUnit.MILLISECOND,
23+
"DaVinci record transformer operation latency",
24+
setOf(VENICE_CLUSTER_NAME, VENICE_STORE_NAME, VENICE_RECORD_TRANSFORMER_OPERATION)
25+
),
26+
27+
RECORD_TRANSFORMER_ERROR_COUNT(
28+
"record_transformer.error_count", MetricType.COUNTER, MetricUnit.NUMBER,
29+
"DaVinci record transformer operation error count",
30+
setOf(VENICE_CLUSTER_NAME, VENICE_STORE_NAME, VENICE_RECORD_TRANSFORMER_OPERATION)
31+
);
32+
33+
private final MetricEntity metricEntity;
34+
35+
DaVinciRecordTransformerOtelMetricEntity(
36+
String metricName,
37+
MetricType metricType,
38+
MetricUnit unit,
39+
String description,
40+
Set<VeniceMetricsDimensions> dimensions) {
41+
this.metricEntity = new MetricEntity(metricName, metricType, unit, description, dimensions);
42+
}
43+
44+
@Override
45+
public MetricEntity getMetricEntity() {
46+
return metricEntity;
47+
}
48+
}

‎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
@@ -50,7 +50,8 @@ public static List<Class<? extends ModuleMetricEntityInterface>> getMetricEntity
5050
ServerLoadOtelMetricEntity.class,
5151
RocksDBStatsOtelMetricEntity.class,
5252
BackupVersionOptimizationOtelMetricEntity.class,
53-
ParticipantStateTransitionOtelMetricEntity.class);
53+
ParticipantStateTransitionOtelMetricEntity.class,
54+
DaVinciRecordTransformerOtelMetricEntity.class);
5455
}
5556

5657
public static final Collection<MetricEntity> SERVER_METRIC_ENTITIES =

0 commit comments

Comments
 (0)