[all] OpenTelemetry metrics integration in venice-router

build.gradle

@@ -54,6 +54,7 @@ def hadoopVersion = '2.10.2'
 def apacheSparkVersion = '3.3.3'
 def antlrVersion = '4.8'
 def scala = '2.12'
+def openTelemetryVersion = '1.43.0'
 
 ext.libraries = [
     alpnAgent: "org.mortbay.jetty.alpn:jetty-alpn-agent:${alpnAgentVersion}",
@@ -141,6 +142,11 @@ ext.libraries = [
     zkclient: 'com.101tec:zkclient:0.7', // For Kafka AdminUtils
     zookeeper: 'org.apache.zookeeper:zookeeper:3.6.3',
     zstd: 'com.github.luben:zstd-jni:1.5.2-3',
+    opentelemetryApi: "io.opentelemetry:opentelemetry-api:${openTelemetryVersion}",
+    opentelemetrySdk: "io.opentelemetry:opentelemetry-sdk:${openTelemetryVersion}",
+    opentelemetryExporterLogging: "io.opentelemetry:opentelemetry-exporter-logging:${openTelemetryVersion}",
+    opentelemetryExporterOtlp: "io.opentelemetry:opentelemetry-exporter-otlp:${openTelemetryVersion}",
+    opentelemetryExporterCommon: "io.opentelemetry:opentelemetry-exporter-common:${openTelemetryVersion}"
 ]
 
 group = 'com.linkedin.venice'

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

@@ -23,10 +23,12 @@ public AggHostLevelIngestionStats(
       boolean unregisterMetricForDeletedStoreEnabled,
       Time time) {
     super(
+        serverConfig.getClusterName(),
         metricsRepository,
         new HostLevelStoreIngestionStatsSupplier(serverConfig, ingestionTaskMap, time),
         metadataRepository,
-        unregisterMetricForDeletedStoreEnabled);
+        unregisterMetricForDeletedStoreEnabled,
+        false);
   }
 
   static class HostLevelStoreIngestionStatsSupplier implements StatsSupplier<HostLevelIngestionStats> {
@@ -44,14 +46,15 @@ static class HostLevelStoreIngestionStatsSupplier implements StatsSupplier<HostL
     }
 
     @Override
-    public HostLevelIngestionStats get(MetricsRepository metricsRepository, String storeName) {
+    public HostLevelIngestionStats get(MetricsRepository metricsRepository, String storeName, String clusterName) {
       throw new VeniceException("Should not be called.");
     }
 
     @Override
     public HostLevelIngestionStats get(
         MetricsRepository metricsRepository,
         String storeName,
+        String clusterName,
         HostLevelIngestionStats totalStats) {
       return new HostLevelIngestionStats(
           metricsRepository,

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

@@ -28,7 +28,8 @@ public AggKafkaConsumerServiceStats(
         metricsRepository,
         new KafkaConsumerServiceStatsSupplier(getMaxElapsedTimeSinceLastPollInConsumerPool),
         metadataRepository,
-        isUnregisterMetricForDeletedStoreEnabled);
+        isUnregisterMetricForDeletedStoreEnabled,
+        true);
   }
 
   public void recordTotalConsumerIdleTime(double idleTime) {
@@ -107,14 +108,15 @@ static class KafkaConsumerServiceStatsSupplier implements StatsSupplier<KafkaCon
     }
 
     @Override
-    public KafkaConsumerServiceStats get(MetricsRepository metricsRepository, String storeName) {
+    public KafkaConsumerServiceStats get(MetricsRepository metricsRepository, String storeName, String clusterName) {
       throw new VeniceException("Should not be called.");
     }
 
     @Override
     public KafkaConsumerServiceStats get(
         MetricsRepository metricsRepository,
         String storeName,
+        String clusterName,
         KafkaConsumerServiceStats totalStats) {
       return new KafkaConsumerServiceStats(
           metricsRepository,

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

@@ -83,7 +83,7 @@ public void recordRocksDBOpenFailure() {
   }
 
   static class StorageEngineStatsReporter extends AbstractVeniceStatsReporter<StorageEngineStats> {
-    public StorageEngineStatsReporter(MetricsRepository metricsRepository, String storeName) {
+    public StorageEngineStatsReporter(MetricsRepository metricsRepository, String storeName, String clusterName) {
       super(metricsRepository, storeName);
     }
 

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

@@ -14,7 +14,7 @@
  * collection/visualization system.
  */
 public class DIVStatsReporter extends AbstractVeniceStatsReporter<DIVStats> {
-  public DIVStatsReporter(MetricsRepository metricsRepository, String storeName) {
+  public DIVStatsReporter(MetricsRepository metricsRepository, String storeName, String clusterName) {
     super(metricsRepository, storeName);
   }
 

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

@@ -60,7 +60,7 @@
 public class IngestionStatsReporter extends AbstractVeniceStatsReporter<IngestionStats> {
   private static final Logger LOGGER = LogManager.getLogger(IngestionStatsReporter.class);
 
-  public IngestionStatsReporter(MetricsRepository metricsRepository, String storeName) {
+  public IngestionStatsReporter(MetricsRepository metricsRepository, String storeName, String clusterName) {
     super(metricsRepository, storeName);
   }
 

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

@@ -30,10 +30,10 @@ public VeniceVersionedStatsReporter(
     registerSensor("current_version", new AsyncGauge((ignored1, ignored2) -> currentVersion, "current_version"));
     registerSensor("future_version", new AsyncGauge((ignored1, ignored2) -> futureVersion, "future_version"));
 
-    this.currentStatsReporter = statsSupplier.get(metricsRepository, storeName + "_current");
+    this.currentStatsReporter = statsSupplier.get(metricsRepository, storeName + "_current", (String) null);
     if (!isSystemStore) {
-      this.futureStatsReporter = statsSupplier.get(metricsRepository, storeName + "_future");
-      this.totalStatsReporter = statsSupplier.get(metricsRepository, storeName + "_total");
+      this.futureStatsReporter = statsSupplier.get(metricsRepository, storeName + "_future", (String) null);
+      this.totalStatsReporter = statsSupplier.get(metricsRepository, storeName + "_total", (String) null);
     } else {
       this.futureStatsReporter = null;
       this.totalStatsReporter = null;

clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/heartbeat/HeartbeatMonitoringService.java

@@ -68,7 +68,10 @@ public HeartbeatMonitoringService(
         metricsRepository,
         metadataRepository,
         () -> new HeartbeatStat(new MetricConfig(), regionNames),
-        (aMetricsRepository, storeName) -> new HeartbeatStatReporter(aMetricsRepository, storeName, regionNames),
+        (aMetricsRepository, storeName, clusterName) -> new HeartbeatStatReporter(
+            aMetricsRepository,
+            storeName,
+            regionNames),
         leaderHeartbeatTimeStamps,
         followerHeartbeatTimeStamps);
   }

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

@@ -19,7 +19,7 @@ public void testDIVReporterCanReport() {
     metricsRepository.addReporter(reporter);
 
     String storeName = Utils.getUniqueString("store");
-    DIVStatsReporter divStatsReporter = new DIVStatsReporter(metricsRepository, storeName);
+    DIVStatsReporter divStatsReporter = new DIVStatsReporter(metricsRepository, storeName, null);
 
     assertEquals(reporter.query("." + storeName + "--success_msg.DIVStatsGauge").value(), (double) NULL_DIV_STATS.code);
 

internal/venice-client-common/build.gradle

@@ -39,6 +39,11 @@ dependencies {
   implementation libraries.log4j2api
   implementation libraries.zstd
   implementation libraries.conscrypt
+  implementation libraries.opentelemetryApi
+  implementation libraries.opentelemetrySdk
+  implementation libraries.opentelemetryExporterLogging
+  implementation libraries.opentelemetryExporterOtlp
+  implementation libraries.opentelemetryExporterCommon
 
   testImplementation project(':internal:venice-test-common')
   testImplementation project(':clients:venice-thin-client')

internal/venice-client-common/src/main/java/com/linkedin/venice/stats/AbstractVeniceAggStats.java

@@ -11,39 +11,56 @@ public abstract class AbstractVeniceAggStats<T extends AbstractVeniceStats> {
   protected final Map<String, T> storeStats = new VeniceConcurrentHashMap<>();
 
   private StatsSupplier<T> statsFactory;
+
   private final MetricsRepository metricsRepository;
+  private final String clusterName;
 
-  private AbstractVeniceAggStats(MetricsRepository metricsRepository, StatsSupplier<T> statsSupplier, T totalStats) {
+  private AbstractVeniceAggStats(
+      String clusterName,
+      MetricsRepository metricsRepository,
+      StatsSupplier<T> statsSupplier,
+      T totalStats) {
+    this.clusterName = clusterName;
     this.metricsRepository = metricsRepository;
     this.statsFactory = statsSupplier;
     this.totalStats = totalStats;
   }
 
-  public AbstractVeniceAggStats(MetricsRepository metricsRepository, StatsSupplier<T> statsSupplier) {
-    this(metricsRepository, statsSupplier, statsSupplier.get(metricsRepository, STORE_NAME_FOR_TOTAL_STAT, null));
-  }
-
-  public AbstractVeniceAggStats(MetricsRepository metricsRepository) {
+  public AbstractVeniceAggStats(String clusterName, MetricsRepository metricsRepository) {
     this.metricsRepository = metricsRepository;
+    this.clusterName = clusterName;
   }
 
   public void setStatsSupplier(StatsSupplier<T> statsSupplier) {
     this.statsFactory = statsSupplier;
-    this.totalStats = statsSupplier.get(metricsRepository, STORE_NAME_FOR_TOTAL_STAT, null);
+    this.totalStats = statsSupplier.get(metricsRepository, STORE_NAME_FOR_TOTAL_STAT, clusterName, null);
   }
 
+  /**
+   * clusterName is used to create per cluster aggregate stats and {@link com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions} <br>
+   * If perClusterAggregate is true, it will create per cluster aggregates with storeName as "total.<clusterName>"
+   */
   public AbstractVeniceAggStats(
       String clusterName,
       MetricsRepository metricsRepository,
-      StatsSupplier<T> statsSupplier) {
-    this(
+      StatsSupplier<T> statsSupplier,
+      boolean perClusterAggregate) {
+    if (perClusterAggregate && clusterName == null) {
+      throw new IllegalArgumentException("perClusterAggregate cannot be true when clusterName is null");
+    }
+    this.clusterName = clusterName;
+    this.metricsRepository = metricsRepository;
+    this.statsFactory = statsSupplier;
+    this.totalStats = statsSupplier.get(
         metricsRepository,
-        statsSupplier,
-        statsSupplier.get(metricsRepository, STORE_NAME_FOR_TOTAL_STAT + "." + clusterName, null));
+        perClusterAggregate ? STORE_NAME_FOR_TOTAL_STAT + "." + clusterName : STORE_NAME_FOR_TOTAL_STAT,
+        clusterName,
+        null);
   }
 
   public T getStoreStats(String storeName) {
-    return storeStats.computeIfAbsent(storeName, k -> statsFactory.get(metricsRepository, storeName, totalStats));
+    return storeStats
+        .computeIfAbsent(storeName, k -> statsFactory.get(metricsRepository, storeName, clusterName, totalStats));
   }
 
   public T getNullableStoreStats(String storeName) {

internal/venice-client-common/src/main/java/com/linkedin/venice/stats/StatsSupplier.java

@@ -7,15 +7,15 @@ public interface StatsSupplier<T extends AbstractVeniceStats> {
   /**
    * Legacy function, for implementations that do not use total stats in their constructor.
    *
-   * @see #get(MetricsRepository, String, AbstractVeniceStats) which is the only caller.
+   * @see #get(MetricsRepository, String, String, AbstractVeniceStats) which is the only caller.
    */
-  T get(MetricsRepository metricsRepository, String storeName);
+  T get(MetricsRepository metricsRepository, String storeName, String clusterName);
 
   /**
    * This is the function that gets called by {@link AbstractVeniceAggStats}, and concrete classes can
    * optionally implement it in order to be provided with the total stats instance.
    */
-  default T get(MetricsRepository metricsRepository, String storeName, T totalStats) {
-    return get(metricsRepository, storeName);
+  default T get(MetricsRepository metricsRepository, String storeName, String clusterName, T totalStats) {
+    return get(metricsRepository, storeName, clusterName);
   }
 }