Skip to content

Commit 0027ab6

Browse files
m-nagarajanm-nagarajan
authored
[da-vinci] Add OTel metrics to StoreBufferServiceStats (linkedin#2648)
- Add joint Tehuti + OTel metrics for all 6 sensors ## Metrics Mapping - `total_memory_usage` → `drainer.memory.used` (`ASYNC_GAUGE`) - `total_remaining_memory` → `drainer.memory.remaining` (`ASYNC_GAUGE`) - `max_memory_usage_per_writer` → `drainer.writer.memory.max_used` (`ASYNC_GAUGE`) - `min_memory_usage_per_writer` → `drainer.writer.memory.min_used` (`ASYNC_GAUGE`) - `internal_processing_latency` → `drainer.record.processing.time` (`MIN_MAX_COUNT_SUM`) - `internal_processing_error` → `drainer.record.processing.error_count` (`COUNTER`) ## Dimensions - Memory metrics: `CLUSTER_NAME` + `DRAINER_TYPE` (sorted / unsorted) ## Key Design Decisions - Joint API (`AsyncMetricEntityStateBase` / `MetricEntityStateBase`) binds Tehuti and OTel to every `.record()` call - Per-store `VeniceConcurrentHashMap` used for latency and error metrics (bounded by active stores) - `STORE_NAME` sanitized once at source (`StoreBufferService`) using `sanitizeStoreName()` and was Initialized to `UNKNOWN_STORE_NAME` - `addCustomDimension(VeniceDimensionInterface)` added to `OpenTelemetryMetricsSetup.Builder` for adding component-specific dimensions like `DRAINER_TYPE` and avoids bloating builder with rarely used setters
1 parent 062f261 commit 0027ab6

19 files changed

Lines changed: 861 additions & 44 deletions

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

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -429,7 +429,8 @@ public void handleStoreDeleted(Store store) {
429429
serverConfig.getClusterName());
430430
this.versionedIngestionStats = new AggVersionedIngestionStats(metricsRepository, metadataRepo, serverConfig);
431431
if (serverConfig.isDedicatedDrainerQueueEnabled()) {
432-
this.storeBufferService = new SeparatedStoreBufferService(serverConfig, metricsRepository);
432+
this.storeBufferService =
433+
new SeparatedStoreBufferService(serverConfig, metricsRepository, serverConfig.getClusterName());
433434
} else {
434435
this.storeBufferService = new StoreBufferService(
435436
serverConfig.getStoreWriterNumber(),
@@ -438,7 +439,8 @@ public void handleStoreDeleted(Store store) {
438439
serverConfig.isStoreWriterBufferAfterLeaderLogicEnabled(),
439440
serverConfig.getLogContext(),
440441
metricsRepository,
441-
true);
442+
true,
443+
serverConfig.getClusterName());
442444
}
443445
this.kafkaMessageEnvelopeSchemaReader = kafkaMessageEnvelopeSchemaReader;
444446

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

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,10 @@ public class SeparatedStoreBufferService extends AbstractStoreBufferService {
2323
private final int sortedPoolSize;
2424
private final int unsortedPoolSize;
2525

26-
SeparatedStoreBufferService(VeniceServerConfig serverConfig, MetricsRepository metricsRepository) {
26+
SeparatedStoreBufferService(
27+
VeniceServerConfig serverConfig,
28+
MetricsRepository metricsRepository,
29+
String clusterName) {
2730
this(
2831
serverConfig.getDrainerPoolSizeSortedInput(),
2932
serverConfig.getDrainerPoolSizeUnsortedInput(),
@@ -34,15 +37,17 @@ public class SeparatedStoreBufferService extends AbstractStoreBufferService {
3437
serverConfig.isStoreWriterBufferAfterLeaderLogicEnabled(),
3538
serverConfig.getLogContext(),
3639
metricsRepository,
37-
true),
40+
true,
41+
clusterName),
3842
new StoreBufferService(
3943
serverConfig.getDrainerPoolSizeUnsortedInput(),
4044
serverConfig.getStoreWriterBufferMemoryCapacity(),
4145
serverConfig.getStoreWriterBufferNotifyDelta(),
4246
serverConfig.isStoreWriterBufferAfterLeaderLogicEnabled(),
4347
serverConfig.getLogContext(),
4448
metricsRepository,
45-
false));
49+
false,
50+
clusterName));
4651
LOGGER.info(
4752
"Created separated store buffer service with {} sorted drainers and {} unsorted drainers queues with capacity of {}",
4853
sortedPoolSize,

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

Lines changed: 24 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import com.linkedin.venice.pubsub.api.PubSubPosition;
2222
import com.linkedin.venice.pubsub.api.PubSubTopic;
2323
import com.linkedin.venice.pubsub.api.PubSubTopicPartition;
24+
import com.linkedin.venice.stats.OpenTelemetryMetricsSetup;
2425
import com.linkedin.venice.utils.DaemonThreadFactory;
2526
import com.linkedin.venice.utils.LogContext;
2627
import com.linkedin.venice.utils.Utils;
@@ -84,7 +85,8 @@ public StoreBufferService(
8485
boolean queueLeaderWrites,
8586
LogContext logContext,
8687
MetricsRepository metricsRepository,
87-
boolean sorted) {
88+
boolean sorted,
89+
String clusterName) {
8890
this(
8991
drainerNum,
9092
bufferCapacityPerDrainer,
@@ -93,7 +95,8 @@ public StoreBufferService(
9395
null,
9496
logContext,
9597
metricsRepository,
96-
sorted);
98+
sorted,
99+
clusterName);
97100
}
98101

99102
/**
@@ -106,7 +109,16 @@ public StoreBufferService(
106109
boolean queueLeaderWrites,
107110
StoreBufferServiceStats stats,
108111
LogContext logContext) {
109-
this(drainerNum, bufferCapacityPerDrainer, bufferNotifyDelta, queueLeaderWrites, stats, logContext, null, true);
112+
this(
113+
drainerNum,
114+
bufferCapacityPerDrainer,
115+
bufferNotifyDelta,
116+
queueLeaderWrites,
117+
stats,
118+
logContext,
119+
null,
120+
true,
121+
null);
110122
}
111123

112124
/**
@@ -124,7 +136,8 @@ private StoreBufferService(
124136
StoreBufferServiceStats stats,
125137
LogContext logContext,
126138
MetricsRepository metricsRepository,
127-
boolean sorted) {
139+
boolean sorted,
140+
String clusterName) {
128141
this.logContext = logContext;
129142
this.drainerNum = drainerNum;
130143
this.blockingQueueArr = new ArrayList<>();
@@ -139,6 +152,8 @@ private StoreBufferService(
139152
: new StoreBufferServiceStats(
140153
Objects.requireNonNull(metricsRepository),
141154
sorted ? "StoreBufferServiceSorted" : "StoreBufferServiceUnsorted",
155+
clusterName,
156+
sorted,
142157
this::getTotalMemoryUsage,
143158
this::getTotalRemainingMemory,
144159
this::getMaxMemoryUsagePerDrainer,
@@ -759,6 +774,7 @@ public void run() {
759774
LeaderProducedRecordContext leaderProducedRecordContext = null;
760775
StoreIngestionTask ingestionTask = null;
761776
CompletableFuture<Void> recordPersistedFuture = null;
777+
String storeName = OpenTelemetryMetricsSetup.UNKNOWN_STORE_NAME;
762778
while (isRunning.get()) {
763779
try {
764780
node = blockingQueue.take();
@@ -768,6 +784,8 @@ public void run() {
768784
leaderProducedRecordContext = node.getLeaderProducedRecordContext();
769785
ingestionTask = node.getIngestionTask();
770786
recordPersistedFuture = node.getQueuedRecordPersistedFuture();
787+
storeName =
788+
OpenTelemetryMetricsSetup.sanitizeStoreName(ingestionTask != null ? ingestionTask.getStoreName() : null);
771789

772790
long startTime = System.currentTimeMillis();
773791

@@ -797,7 +815,7 @@ public void run() {
797815
recordPersistedFuture.complete(null);
798816
}
799817
long latencyInMS = System.currentTimeMillis() - startTime;
800-
this.stats.recordInternalProcessingLatency(latencyInMS);
818+
this.stats.recordInternalProcessingLatency(latencyInMS, storeName);
801819
topicToTimeSpent.compute(consumerRecord.getTopicPartition(), (K, V) -> (V == null ? 0 : V) + latencyInMS);
802820
} catch (Throwable e) {
803821
if (e instanceof InterruptedException) {
@@ -820,7 +838,7 @@ public void run() {
820838
logBuilder.append(consumerRecordString);
821839
}
822840
LOGGER.error(logBuilder.toString(), e);
823-
stats.recordInternalProcessingError();
841+
stats.recordInternalProcessingError(storeName);
824842

825843
/**
826844
* Catch all the thrown exception and store it in {@link StoreIngestionTask#lastWorkerException}.

‎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
@@ -41,7 +41,8 @@ public static List<Class<? extends ModuleMetricEntityInterface>> getMetricEntity
4141
RocksDBMemoryOtelMetricEntity.class,
4242
DIVOtelMetricEntity.class,
4343
ServerReadQuotaOtelMetricEntity.class,
44-
ServerConnectionOtelMetricEntity.class);
44+
ServerConnectionOtelMetricEntity.class,
45+
StoreBufferServiceOtelMetricEntity.class);
4546
}
4647

4748
public static final Collection<MetricEntity> SERVER_METRIC_ENTITIES =

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

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,60 @@
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_DRAINER_TYPE;
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+
/** OTel metric entities for store buffer service (drainer queue) tracking. */
17+
public enum StoreBufferServiceOtelMetricEntity implements ModuleMetricEntityInterface {
18+
MEMORY_USED(
19+
"drainer.memory.used", MetricType.ASYNC_GAUGE, MetricUnit.BYTES, "Total memory used across all drainer queues",
20+
setOf(VENICE_CLUSTER_NAME, VENICE_DRAINER_TYPE)
21+
),
22+
MEMORY_REMAINING(
23+
"drainer.memory.remaining", MetricType.ASYNC_GAUGE, MetricUnit.BYTES,
24+
"Total remaining memory capacity across all drainer queues", setOf(VENICE_CLUSTER_NAME, VENICE_DRAINER_TYPE)
25+
),
26+
MEMORY_USED_PER_WRITER_MAX(
27+
"drainer.writer.memory.max_used", MetricType.ASYNC_GAUGE, MetricUnit.BYTES,
28+
"Maximum memory used by any single drainer writer", setOf(VENICE_CLUSTER_NAME, VENICE_DRAINER_TYPE)
29+
),
30+
MEMORY_USED_PER_WRITER_MIN(
31+
"drainer.writer.memory.min_used", MetricType.ASYNC_GAUGE, MetricUnit.BYTES,
32+
"Minimum memory used by any single drainer writer", setOf(VENICE_CLUSTER_NAME, VENICE_DRAINER_TYPE)
33+
),
34+
PROCESSING_TIME(
35+
"drainer.record.processing.time", MetricType.MIN_MAX_COUNT_SUM_AGGREGATIONS, MetricUnit.MILLISECOND,
36+
"Time spent processing each record in the drainer",
37+
setOf(VENICE_CLUSTER_NAME, VENICE_DRAINER_TYPE, VENICE_STORE_NAME)
38+
),
39+
PROCESSING_ERROR_COUNT(
40+
"drainer.record.processing.error_count", MetricType.COUNTER, MetricUnit.NUMBER,
41+
"Count of errors encountered while processing records in the drainer",
42+
setOf(VENICE_CLUSTER_NAME, VENICE_DRAINER_TYPE, VENICE_STORE_NAME)
43+
);
44+
45+
private final MetricEntity metricEntity;
46+
47+
StoreBufferServiceOtelMetricEntity(
48+
String metricName,
49+
MetricType metricType,
50+
MetricUnit unit,
51+
String description,
52+
Set<VeniceMetricsDimensions> dimensions) {
53+
this.metricEntity = new MetricEntity(metricName, metricType, unit, description, dimensions);
54+
}
55+
56+
@Override
57+
public MetricEntity getMetricEntity() {
58+
return metricEntity;
59+
}
60+
}

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

Lines changed: 122 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -1,53 +1,150 @@
11
package com.linkedin.davinci.stats;
22

33
import com.linkedin.venice.stats.AbstractVeniceStats;
4+
import com.linkedin.venice.stats.OpenTelemetryMetricsSetup;
5+
import com.linkedin.venice.stats.VeniceOpenTelemetryMetricsRepository;
6+
import com.linkedin.venice.stats.dimensions.VeniceDrainerType;
7+
import com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions;
8+
import com.linkedin.venice.stats.metrics.AsyncMetricEntityStateBase;
9+
import com.linkedin.venice.stats.metrics.MetricEntity;
10+
import com.linkedin.venice.stats.metrics.MetricEntityStateBase;
11+
import com.linkedin.venice.stats.metrics.TehutiMetricNameEnum;
12+
import com.linkedin.venice.utils.concurrent.VeniceConcurrentHashMap;
13+
import io.opentelemetry.api.common.Attributes;
14+
import io.tehuti.metrics.MeasurableStat;
415
import io.tehuti.metrics.MetricsRepository;
5-
import io.tehuti.metrics.Sensor;
616
import io.tehuti.metrics.stats.AsyncGauge;
717
import io.tehuti.metrics.stats.Avg;
818
import io.tehuti.metrics.stats.Max;
919
import io.tehuti.metrics.stats.OccurrenceRate;
20+
import java.util.Arrays;
21+
import java.util.Collections;
22+
import java.util.HashMap;
23+
import java.util.List;
24+
import java.util.Map;
1025
import java.util.function.LongSupplier;
1126

1227

1328
public class StoreBufferServiceStats extends AbstractVeniceStats {
14-
private final Sensor totalMemoryUsageSensor;
15-
private final Sensor totalRemainingMemorySensor;
16-
private final Sensor maxMemoryUsagePerWriterSensor;
17-
private final Sensor minMemoryUsagePerWriterSensor;
18-
private final Sensor internalProcessingLatencySensor;
19-
private final Sensor internalProcessingErrorSensor;
29+
enum TehutiMetricName implements TehutiMetricNameEnum {
30+
TOTAL_MEMORY_USAGE, TOTAL_REMAINING_MEMORY, MAX_MEMORY_USAGE_PER_WRITER, MIN_MEMORY_USAGE_PER_WRITER,
31+
INTERNAL_PROCESSING_LATENCY, INTERNAL_PROCESSING_ERROR;
32+
}
33+
34+
private final VeniceOpenTelemetryMetricsRepository otelRepository;
35+
private final Map<VeniceMetricsDimensions, String> baseDimensionsMap;
36+
private final Attributes baseAttributes;
37+
38+
/**
39+
* Per-store latency metric states. Bounded by the number of active stores on this server (typically < 100).
40+
* All stores share a single Tehuti sensor (registered once via {@code registerSensorIfAbsent}); per-store
41+
* structure exists for OTel dimensions only. Entries are never evicted; bounded cardinality makes this safe.
42+
*/
43+
private final VeniceConcurrentHashMap<String, MetricEntityStateBase> latencyPerStore =
44+
new VeniceConcurrentHashMap<>();
45+
46+
/**
47+
* Per-store error metric states. Same bounding and lifecycle as {@link #latencyPerStore}.
48+
*/
49+
private final VeniceConcurrentHashMap<String, MetricEntityStateBase> errorPerStore = new VeniceConcurrentHashMap<>();
2050

2151
public StoreBufferServiceStats(
2252
MetricsRepository metricsRepository,
2353
String metricNamePrefix,
54+
String clusterName,
55+
boolean sorted,
2456
LongSupplier totalMemoryUsageSupplier,
2557
LongSupplier totalRemainingMemorySupplier,
2658
LongSupplier maxMemoryUsagePerDrainerSupplier,
2759
LongSupplier minMemoryUsagePerDrainerSupplier) {
2860
super(metricsRepository, metricNamePrefix);
29-
totalMemoryUsageSensor = registerSensor(
30-
new AsyncGauge((ignored, ignored2) -> totalMemoryUsageSupplier.getAsLong(), "total_memory_usage"));
31-
totalRemainingMemorySensor = registerSensor(
32-
new AsyncGauge((ignored, ignored2) -> totalRemainingMemorySupplier.getAsLong(), "total_remaining_memory"));
33-
maxMemoryUsagePerWriterSensor = registerSensor(
34-
new AsyncGauge(
35-
(ignored, ignored2) -> maxMemoryUsagePerDrainerSupplier.getAsLong(),
36-
"max_memory_usage_per_writer"));
37-
minMemoryUsagePerWriterSensor = registerSensor(
38-
new AsyncGauge(
39-
(ignored, ignored2) -> minMemoryUsagePerDrainerSupplier.getAsLong(),
40-
"min_memory_usage_per_writer"));
4161

42-
internalProcessingLatencySensor = registerSensor("internal_processing_latency", new Avg(), new Max());
43-
internalProcessingErrorSensor = registerSensor("internal_processing_error", new OccurrenceRate());
62+
VeniceDrainerType bufferType = sorted ? VeniceDrainerType.SORTED : VeniceDrainerType.UNSORTED;
63+
OpenTelemetryMetricsSetup.OpenTelemetryMetricsSetupInfo otelData =
64+
OpenTelemetryMetricsSetup.builder(metricsRepository)
65+
.setClusterName(clusterName)
66+
.addCustomDimension(bufferType)
67+
.build();
68+
this.otelRepository = otelData.getOtelRepository();
69+
this.baseDimensionsMap = otelData.getBaseDimensionsMap();
70+
// All 4 memory metrics share the same dimension set {CLUSTER_NAME, STORE_BUFFER_SERVICE_TYPE},
71+
// so baseAttributes built from any one of them is valid for all four.
72+
this.baseAttributes = otelData.getBaseAttributes();
73+
74+
// Memory metrics (#1-4): joint Tehuti+OTel AsyncGauge.
75+
// Return values are intentionally discarded — the gauge callback is registered internally
76+
// by the Tehuti sensor and OTel SDK during create(). No per-recording state is needed.
77+
registerMemoryGauge(
78+
StoreBufferServiceOtelMetricEntity.MEMORY_USED,
79+
TehutiMetricName.TOTAL_MEMORY_USAGE,
80+
totalMemoryUsageSupplier);
81+
registerMemoryGauge(
82+
StoreBufferServiceOtelMetricEntity.MEMORY_REMAINING,
83+
TehutiMetricName.TOTAL_REMAINING_MEMORY,
84+
totalRemainingMemorySupplier);
85+
registerMemoryGauge(
86+
StoreBufferServiceOtelMetricEntity.MEMORY_USED_PER_WRITER_MAX,
87+
TehutiMetricName.MAX_MEMORY_USAGE_PER_WRITER,
88+
maxMemoryUsagePerDrainerSupplier);
89+
registerMemoryGauge(
90+
StoreBufferServiceOtelMetricEntity.MEMORY_USED_PER_WRITER_MIN,
91+
TehutiMetricName.MIN_MEMORY_USAGE_PER_WRITER,
92+
minMemoryUsagePerDrainerSupplier);
93+
}
94+
95+
private void registerMemoryGauge(
96+
StoreBufferServiceOtelMetricEntity metricEntity,
97+
TehutiMetricName tehutiName,
98+
LongSupplier supplier) {
99+
AsyncMetricEntityStateBase.create(
100+
metricEntity.getMetricEntity(),
101+
otelRepository,
102+
this::registerSensorIfAbsent,
103+
tehutiName,
104+
Collections.singletonList(new AsyncGauge((ig, ig2) -> supplier.getAsLong(), tehutiName.getMetricName())),
105+
baseDimensionsMap,
106+
baseAttributes,
107+
supplier);
108+
}
109+
110+
private MetricEntityStateBase createPerStoreState(
111+
String storeName,
112+
MetricEntity metricEntity,
113+
TehutiMetricNameEnum tehutiName,
114+
List<MeasurableStat> tehutiStats) {
115+
Map<VeniceMetricsDimensions, String> dims = new HashMap<>(baseDimensionsMap);
116+
dims.put(VeniceMetricsDimensions.VENICE_STORE_NAME, storeName);
117+
dims = Collections.unmodifiableMap(dims);
118+
Attributes attrs = otelRepository != null ? otelRepository.createAttributes(metricEntity, dims) : null;
119+
return MetricEntityStateBase
120+
.create(metricEntity, otelRepository, this::registerSensorIfAbsent, tehutiName, tehutiStats, dims, attrs);
121+
}
122+
123+
private MetricEntityStateBase getOrCreateLatencyState(String storeName) {
124+
return latencyPerStore.computeIfAbsent(
125+
storeName,
126+
k -> createPerStoreState(
127+
k,
128+
StoreBufferServiceOtelMetricEntity.PROCESSING_TIME.getMetricEntity(),
129+
TehutiMetricName.INTERNAL_PROCESSING_LATENCY,
130+
Arrays.asList(new Avg(), new Max())));
131+
}
132+
133+
private MetricEntityStateBase getOrCreateErrorState(String storeName) {
134+
return errorPerStore.computeIfAbsent(
135+
storeName,
136+
k -> createPerStoreState(
137+
k,
138+
StoreBufferServiceOtelMetricEntity.PROCESSING_ERROR_COUNT.getMetricEntity(),
139+
TehutiMetricName.INTERNAL_PROCESSING_ERROR,
140+
Collections.singletonList(new OccurrenceRate())));
44141
}
45142

46-
public void recordInternalProcessingError() {
47-
internalProcessingErrorSensor.record();
143+
public void recordInternalProcessingLatency(long latency, String storeName) {
144+
getOrCreateLatencyState(storeName).record(latency);
48145
}
49146

50-
public void recordInternalProcessingLatency(long latency) {
51-
internalProcessingLatencySensor.record(latency);
147+
public void recordInternalProcessingError(String storeName) {
148+
getOrCreateErrorState(storeName).record(1);
52149
}
53150
}

0 commit comments

Comments
 (0)