Skip to content

Commit 37eaa4c

Browse files
m-nagarajanm-nagarajan
committed
addOtelMetricsForKafkaConsumerServiceStats
1 parent 5e16b32 commit 37eaa4c

13 files changed

Lines changed: 1257 additions & 88 deletions

File tree

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

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -143,7 +143,8 @@ protected KafkaConsumerService(
143143
kafkaClusterAlias,
144144
this::getMaxElapsedTimeMSSinceLastPollInConsumerPool,
145145
metadataRepository,
146-
isUnregisterMetricForDeletedStoreEnabled);
146+
isUnregisterMetricForDeletedStoreEnabled,
147+
serverConfig.getClusterName());
147148

148149
VeniceProperties properties = new VeniceProperties(consumerProperties);
149150
PubSubConsumerAdapterContext.Builder contextBuilder =
@@ -403,14 +404,16 @@ private AggKafkaConsumerServiceStats createAggKafkaConsumerServiceStats(
403404
String kafkaClusterAlias,
404405
LongSupplier getMaxElapsedTimeSinceLastPollInConsumerPool,
405406
ReadOnlyStoreRepository metadataRepository,
406-
boolean isUnregisterMetricForDeletedStoreEnabled) {
407+
boolean isUnregisterMetricForDeletedStoreEnabled,
408+
String veniceClusterName) {
407409
String nameWithKafkaClusterAlias = "kafka_consumer_service_for_" + kafkaClusterAlias;
408410
return new AggKafkaConsumerServiceStats(
409411
nameWithKafkaClusterAlias,
410412
metricsRepository,
411413
metadataRepository,
412414
getMaxElapsedTimeSinceLastPollInConsumerPool,
413-
isUnregisterMetricForDeletedStoreEnabled);
415+
isUnregisterMetricForDeletedStoreEnabled,
416+
veniceClusterName);
414417
}
415418

416419
@Override
@@ -555,6 +558,8 @@ final void recordPartitionsPerConsumerSensor() {
555558
totalPartitions += subscribedPartitionCount;
556559
minPartitionsPerConsumer = Math.min(minPartitionsPerConsumer, subscribedPartitionCount);
557560
maxPartitionsPerConsumer = Math.max(maxPartitionsPerConsumer, subscribedPartitionCount);
561+
// Record raw per-consumer partition count to OTel histogram (asymmetric: Tehuti uses pre-computed gauges)
562+
aggStats.recordTotalPartitionAssignmentForOtel(subscribedPartitionCount);
558563
}
559564
int avgPartitionsPerConsumer = totalPartitions / consumerToConsumptionTask.size();
560565

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

Lines changed: 19 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -22,11 +22,12 @@ public AggKafkaConsumerServiceStats(
2222
MetricsRepository metricsRepository,
2323
ReadOnlyStoreRepository metadataRepository,
2424
LongSupplier getMaxElapsedTimeSinceLastPollInConsumerPool,
25-
boolean isUnregisterMetricForDeletedStoreEnabled) {
25+
boolean isUnregisterMetricForDeletedStoreEnabled,
26+
String veniceClusterName) {
2627
super(
2728
regionName,
2829
metricsRepository,
29-
new KafkaConsumerServiceStatsSupplier(getMaxElapsedTimeSinceLastPollInConsumerPool),
30+
new KafkaConsumerServiceStatsSupplier(getMaxElapsedTimeSinceLastPollInConsumerPool, veniceClusterName),
3031
metadataRepository,
3132
isUnregisterMetricForDeletedStoreEnabled,
3233
true);
@@ -84,11 +85,24 @@ public void recordTotalSubscribedPartitionsNum(int count) {
8485
totalStats.recordSubscribedPartitionsNum(count);
8586
}
8687

88+
/**
89+
* Records a single per-consumer partition count to the OTel partition assignment histogram
90+
* on the total stats instance. Called for each consumer in the pool from
91+
* {@link com.linkedin.davinci.kafka.consumer.KafkaConsumerService#recordPartitionsPerConsumerSensor()}.
92+
*/
93+
public void recordTotalPartitionAssignmentForOtel(int partitionCount) {
94+
totalStats.recordPartitionAssignmentForOtel(partitionCount);
95+
}
96+
8797
static class KafkaConsumerServiceStatsSupplier implements StatsSupplier<KafkaConsumerServiceStats> {
8898
private final LongSupplier getMaxElapsedTimeSinceLastPollInConsumerPool;
99+
private final String veniceClusterName;
89100

90-
KafkaConsumerServiceStatsSupplier(LongSupplier getMaxElapsedTimeSinceLastPollInConsumerPool) {
101+
KafkaConsumerServiceStatsSupplier(
102+
LongSupplier getMaxElapsedTimeSinceLastPollInConsumerPool,
103+
String veniceClusterName) {
91104
this.getMaxElapsedTimeSinceLastPollInConsumerPool = getMaxElapsedTimeSinceLastPollInConsumerPool;
105+
this.veniceClusterName = veniceClusterName;
92106
}
93107

94108
@Override
@@ -107,7 +121,8 @@ public KafkaConsumerServiceStats get(
107121
storeName,
108122
getMaxElapsedTimeSinceLastPollInConsumerPool,
109123
totalStats,
110-
SystemTime.INSTANCE);
124+
SystemTime.INSTANCE,
125+
veniceClusterName);
111126
}
112127
}
113128
}

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

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,110 @@
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_CONSUMER_POOL_ACTION;
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 for {@link KafkaConsumerServiceStats}.
18+
* Uses {@code pubsub} (not {@code kafka}) in metric names to align with Venice's PubSub abstraction.
19+
* Tehuti names are preserved as-is for backward compatibility.
20+
*/
21+
public enum KafkaConsumerServiceOtelMetricEntity implements ModuleMetricEntityInterface {
22+
POLL_BYTES(
23+
"ingestion.pubsub.consumer.poll.bytes", MetricType.MIN_MAX_COUNT_SUM_AGGREGATIONS, MetricUnit.BYTES,
24+
"Byte size of polled PubSub messages per poll request", setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME)
25+
),
26+
27+
POLL_RECORD_COUNT(
28+
"ingestion.pubsub.consumer.poll.record_count", MetricType.MIN_MAX_COUNT_SUM_AGGREGATIONS, MetricUnit.NUMBER,
29+
"Number of records returned per poll request", setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME)
30+
),
31+
32+
POLL_COUNT(
33+
"ingestion.pubsub.consumer.poll.count", MetricType.ASYNC_COUNTER_FOR_HIGH_PERF_CASES, MetricUnit.NUMBER,
34+
"Total count of poll requests to the PubSub consumer", setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME)
35+
),
36+
37+
POLL_TIME(
38+
"ingestion.pubsub.consumer.poll.time", MetricType.HISTOGRAM, MetricUnit.MILLISECOND,
39+
"Latency of PubSub consumer poll requests", setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME)
40+
),
41+
42+
POLL_NON_EMPTY_COUNT(
43+
"ingestion.pubsub.consumer.poll.non_empty_count", MetricType.ASYNC_COUNTER_FOR_HIGH_PERF_CASES, MetricUnit.NUMBER,
44+
"Count of poll requests that returned at least one record", setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME)
45+
),
46+
47+
POLL_ERROR_COUNT(
48+
"ingestion.pubsub.consumer.poll.error_count", MetricType.COUNTER, MetricUnit.NUMBER,
49+
"Count of PubSub consumer poll errors", setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME)
50+
),
51+
52+
PRODUCE_TO_WRITE_BUFFER_TIME(
53+
"ingestion.pubsub.consumer.produce_to_write_buffer_time", MetricType.HISTOGRAM, MetricUnit.MILLISECOND,
54+
"Latency of producing consumed records to the write buffer", setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME)
55+
),
56+
57+
TOPIC_DELETED_COUNT(
58+
"ingestion.pubsub.consumer.topic.deleted_count", MetricType.COUNTER, MetricUnit.NUMBER,
59+
"Count of detected deleted topics", setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME)
60+
),
61+
62+
TOPIC_NO_INGESTION_COUNT(
63+
"ingestion.pubsub.consumer.topic.no_ingestion_count", MetricType.COUNTER, MetricUnit.NUMBER,
64+
"Count of topic-partitions with no running ingestion task", setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME)
65+
),
66+
67+
/**
68+
* Latency of consumer pool actions (subscribe, update assignment).
69+
* Shared OTel instrument differentiated by {@link com.linkedin.venice.stats.dimensions.VeniceConsumerPoolAction}.
70+
*/
71+
CONSUMER_ACTION_TIME(
72+
"ingestion.pubsub.consumer.consumer_action.time", MetricType.HISTOGRAM, MetricUnit.MILLISECOND,
73+
"Latency of consumer pool actions (subscribe, update assignment)",
74+
setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME, VENICE_CONSUMER_POOL_ACTION)
75+
),
76+
77+
POOL_IDLE_TIME(
78+
"ingestion.pubsub.consumer.pool.idle_time", MetricType.GAUGE, MetricUnit.MILLISECOND,
79+
"Maximum idle time of the consumer pool since last successful poll", setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME)
80+
),
81+
82+
POLL_TIME_SINCE_LAST_SUCCESS(
83+
"ingestion.pubsub.consumer.poll.time_since_last_success", MetricType.ASYNC_GAUGE, MetricUnit.MILLISECOND,
84+
"Maximum elapsed time since the last successful poll across all consumers in the pool",
85+
setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME)
86+
),
87+
88+
/** Raw per-consumer partition assignment counts. OTel-only (Tehuti uses 4 pre-computed gauges). */
89+
PARTITION_ASSIGNMENT_COUNT(
90+
"ingestion.pubsub.consumer.partition_assignment.count", MetricType.MIN_MAX_COUNT_SUM_AGGREGATIONS,
91+
MetricUnit.NUMBER, "Raw per-consumer partition assignment counts across the consumer pool",
92+
setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME)
93+
);
94+
95+
private final MetricEntity metricEntity;
96+
97+
KafkaConsumerServiceOtelMetricEntity(
98+
String metricName,
99+
MetricType metricType,
100+
MetricUnit unit,
101+
String description,
102+
Set<VeniceMetricsDimensions> dimensions) {
103+
this.metricEntity = new MetricEntity(metricName, metricType, unit, description, dimensions);
104+
}
105+
106+
@Override
107+
public MetricEntity getMetricEntity() {
108+
return metricEntity;
109+
}
110+
}

0 commit comments

Comments
 (0)