[controller] integrate OpenTelemetry into Controller & add log compaction metrics

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

@@ -14,6 +14,8 @@
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.api.metrics.DoubleGauge;
+import io.opentelemetry.api.metrics.DoubleGaugeBuilder;
 import io.opentelemetry.api.metrics.DoubleHistogram;
 import io.opentelemetry.api.metrics.DoubleHistogramBuilder;
 import io.opentelemetry.api.metrics.LongCounter;
@@ -142,6 +144,7 @@ public VeniceOpenTelemetryMetricsRepository(VeniceMetricsConfig metricsConfig) {
    */
   private final VeniceConcurrentHashMap<String, DoubleHistogram> histogramMap = new VeniceConcurrentHashMap<>();
   private final VeniceConcurrentHashMap<String, LongCounter> counterMap = new VeniceConcurrentHashMap<>();
+  private final VeniceConcurrentHashMap<String, DoubleGauge> gaugeMap = new VeniceConcurrentHashMap<>();
 
   MetricExporter getOtlpHttpMetricExporter(VeniceMetricsConfig metricsConfig) {
     OtlpHttpMetricExporterBuilder exporterBuilder =
@@ -182,7 +185,7 @@ private void setExponentialHistogramAggregation(SdkMeterProviderBuilder builder,
 
     for (MetricEntity metricEntity: metricsConfig.getMetricEntities()) {
       if (metricEntity.getMetricType() == MetricType.HISTOGRAM) {
-        uniqueHistogramMetricNames.add(getFullMetricName(getMetricPrefix(metricEntity), metricEntity.getMetricName()));
+        uniqueHistogramMetricNames.add(getFullMetricName(metricEntity));
       }
     }
 
@@ -202,8 +205,8 @@ private void setExponentialHistogramAggregation(SdkMeterProviderBuilder builder,
     }
   }
 
-  String getFullMetricName(String metricPrefix, String name) {
-    String fullMetricName = metricPrefix + "." + name;
+  String getFullMetricName(MetricEntity metricEntity) {
+    String fullMetricName = getMetricPrefix(metricEntity) + "." + metricEntity.getMetricName();
     validateMetricName(fullMetricName);
     return transformMetricName(fullMetricName, getMetricFormat());
   }
@@ -227,7 +230,7 @@ public DoubleHistogram createHistogram(MetricEntity metricEntity) {
       return null;
     }
     return histogramMap.computeIfAbsent(metricEntity.getMetricName(), key -> {
-      String fullMetricName = getFullMetricName(getMetricPrefix(metricEntity), metricEntity.getMetricName());
+      String fullMetricName = getFullMetricName(metricEntity);
       DoubleHistogramBuilder builder = meter.histogramBuilder(fullMetricName)
           .setUnit(metricEntity.getUnit().name())
           .setDescription(metricEntity.getDescription());
@@ -244,14 +247,27 @@ public LongCounter createCounter(MetricEntity metricEntity) {
       return null;
     }
     return counterMap.computeIfAbsent(metricEntity.getMetricName(), key -> {
-      String fullMetricName = getFullMetricName(getMetricPrefix(metricEntity), metricEntity.getMetricName());
+      String fullMetricName = getFullMetricName(metricEntity);
       LongCounterBuilder builder = meter.counterBuilder(fullMetricName)
           .setUnit(metricEntity.getUnit().name())
           .setDescription(metricEntity.getDescription());
       return builder.build();
     });
   }
 
+  public DoubleGauge createGuage(MetricEntity metricEntity) {
+    if (!emitOpenTelemetryMetrics()) {
+      return null;
+    }
+    return gaugeMap.computeIfAbsent(metricEntity.getMetricName(), key -> {
+      String fullMetricName = getFullMetricName(metricEntity);
+      DoubleGaugeBuilder builder = meter.gaugeBuilder(fullMetricName)
+          .setUnit(metricEntity.getUnit().name())
+          .setDescription(metricEntity.getDescription());
+      return builder.build();
+    });
+  }
+
   public Object createInstrument(MetricEntity metricEntity) {
     MetricType metricType = metricEntity.getMetricType();
     switch (metricType) {
@@ -261,6 +277,8 @@ public Object createInstrument(MetricEntity metricEntity) {
 
       case COUNTER:
         return createCounter(metricEntity);
+      case GAUGE:
+        return createGuage(metricEntity);
 
       default:
         throw new VeniceException("Unknown metric type: " + metricType);

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

@@ -0,0 +1,25 @@
+package com.linkedin.venice.stats.dimensions;
+
+public enum RepushStoreTriggerSource implements VeniceDimensionInterface {
+  MANUAL, SCHEDULED;
+
+  private final String triggerSource;
+
+  RepushStoreTriggerSource() {
+    this.triggerSource = name().toLowerCase();
+  }
+
+  /**
+   * All the instances of this Enum should have the same dimension name.
+   * Refer {@link VeniceDimensionInterface#getDimensionName()} for more details.
+   */
+  @Override
+  public VeniceMetricsDimensions getDimensionName() {
+    return VeniceMetricsDimensions.REPUSH_TRIGGER_SOURCE;
+  }
+
+  @Override
+  public String getDimensionValue() {
+    return this.triggerSource;
+  }
+}

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

@@ -31,7 +31,10 @@ public enum VeniceMetricsDimensions {
   VENICE_MESSAGE_TYPE("venice.message.type"),
 
   /** {@link RequestRetryAbortReason} */
-  VENICE_REQUEST_RETRY_ABORT_REASON("venice.request.retry_abort_reason");
+  VENICE_REQUEST_RETRY_ABORT_REASON("venice.request.retry_abort_reason"),
+
+  /** {@link RepushStoreTriggerSource} */
+  REPUSH_TRIGGER_SOURCE("repush.trigger.source");
 
   private final String[] dimensionName = new String[VeniceOpenTelemetryMetricNamingFormat.SIZE];
 

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

@@ -5,6 +5,7 @@
 import com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions;
 import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.metrics.DoubleGauge;
 import io.opentelemetry.api.metrics.DoubleHistogram;
 import io.opentelemetry.api.metrics.LongCounter;
 import io.tehuti.metrics.MeasurableStat;
@@ -106,7 +107,9 @@ public void recordOtelMetric(double value, Attributes attributes) {
         case COUNTER:
           ((LongCounter) otelMetric).add((long) value, attributes);
           break;
-
+        case GAUGE:
+          ((DoubleGauge) otelMetric).set((long) value, attributes);
+          break;
         default:
           throw new IllegalArgumentException("Unsupported metric type: " + metricType);
       }

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

@@ -26,5 +26,10 @@ public enum MetricType {
   /**
    * For Counter: A simple counter that can be added to.
    */
-  COUNTER;
+  COUNTER,
+
+  /**
+   * For Gauge: Emits the absolute value of the metric value.
+   */
+  GAUGE;
 }

internal/venice-client-common/src/test/java/com/linkedin/venice/stats/VeniceOpenTelemetryMetricsRepositoryTest.java

@@ -21,6 +21,7 @@
 import com.linkedin.venice.stats.metrics.MetricType;
 import com.linkedin.venice.stats.metrics.MetricUnit;
 import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.metrics.DoubleGauge;
 import io.opentelemetry.api.metrics.DoubleHistogram;
 import io.opentelemetry.api.metrics.LongCounter;
 import io.opentelemetry.sdk.metrics.export.MetricExporter;
@@ -111,7 +112,13 @@ public void testValidateMetricNameWithInvalidName() {
   @Test
   public void testTransformMetricName() {
     when(mockMetricsConfig.getMetricNamingFormat()).thenReturn(SNAKE_CASE);
-    assertEquals(metricsRepository.getFullMetricName("prefix", "metric_name"), "prefix.metric_name");
+    MetricEntity metricEntity = MetricEntity.createInternalMetricEntityWithoutDimensions(
+        "metric_name",
+        MetricType.COUNTER,
+        MetricUnit.NUMBER,
+        "Test metric",
+        "prefix");
+    assertEquals(metricsRepository.getFullMetricName(metricEntity), "prefix.metric_name");
 
     String transformedName =
         transformMetricName("test.test_metric_name", VeniceOpenTelemetryMetricNamingFormat.PASCAL_CASE);
@@ -169,6 +176,12 @@ public void testCreateAndRecordMetricsForAllMetricTypes() {
               "Instrument should be a LongCounter for metric type: " + metricType);
           metricEntityState.recordOtelMetric(value, attributes);
           break;
+        case GAUGE:
+          assertTrue(
+              instrument instanceof DoubleGauge,
+              "Instrument should be a DoubleGauge for metric type: " + metricType);
+          metricEntityState.recordOtelMetric(value, attributes);
+          break;
         default:
           fail("Unsupported metric type: " + metricType);
       }

internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/VeniceMetricsDimensionsTest.java

@@ -34,11 +34,14 @@ public void testGetDimensionNameInSnakeCase() {
         case VENICE_REQUEST_RETRY_TYPE:
           assertEquals(dimension.getDimensionName(format), "venice.request.retry_type");
           break;
+        case VENICE_MESSAGE_TYPE:
+          assertEquals(dimension.getDimensionName(format), "venice.message.type");
+          break;
         case VENICE_REQUEST_RETRY_ABORT_REASON:
           assertEquals(dimension.getDimensionName(format), "venice.request.retry_abort_reason");
           break;
-        case VENICE_MESSAGE_TYPE:
-          assertEquals(dimension.getDimensionName(format), "venice.message.type");
+        case REPUSH_TRIGGER_SOURCE:
+          assertEquals(dimension.getDimensionName(format), "repush.trigger.source");
           break;
         default:
           throw new IllegalArgumentException("Unknown dimension: " + dimension);
@@ -78,6 +81,9 @@ public void testGetDimensionNameInCamelCase() {
         case VENICE_MESSAGE_TYPE:
           assertEquals(dimension.getDimensionName(format), "venice.message.type");
           break;
+        case REPUSH_TRIGGER_SOURCE:
+          assertEquals(dimension.getDimensionName(format), "repush.trigger.source");
+          break;
         default:
           throw new IllegalArgumentException("Unknown dimension: " + dimension);
       }
@@ -116,6 +122,9 @@ public void testGetDimensionNameInPascalCase() {
         case VENICE_MESSAGE_TYPE:
           assertEquals(dimension.getDimensionName(format), "Venice.Message.Type");
           break;
+        case REPUSH_TRIGGER_SOURCE:
+          assertEquals(dimension.getDimensionName(format), "Repush.Trigger.Source");
+          break;
         default:
           throw new IllegalArgumentException("Unknown dimension: " + dimension);
       }

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/VeniceControllerWrapper.java

@@ -67,7 +67,7 @@
 import com.linkedin.venice.pubsub.PubSubClientsFactory;
 import com.linkedin.venice.pubsub.api.PubSubSecurityProtocol;
 import com.linkedin.venice.servicediscovery.ServiceDiscoveryAnnouncer;
-import com.linkedin.venice.stats.TehutiUtils;
+import com.linkedin.venice.stats.VeniceMetricsRepository;
 import com.linkedin.venice.utils.PropertyBuilder;
 import com.linkedin.venice.utils.SslUtils;
 import com.linkedin.venice.utils.TestUtils;
@@ -354,7 +354,12 @@ static StatefulServiceProvider<VeniceControllerWrapper> generateService(VeniceCo
       }
 
       D2Client d2Client = D2TestUtils.getAndStartD2Client(options.getZkAddress());
-      MetricsRepository metricsRepository = TehutiUtils.getMetricsRepository(D2_SERVICE_NAME);
+      MetricsRepository metricsRepository = VeniceMetricsRepository.getVeniceMetricsRepository(
+          D2_SERVICE_NAME,
+          VeniceController.CONTROLLER_SERVICE_METRIC_PREFIX,
+          VeniceController.CONTROLLER_SERVICE_METRIC_ENTITIES,
+          propertiesList.get(0).getAsMap() // TODO repush otel: not sure if properties is accessed this way
+      );
 
       Optional<ClientConfig> consumerClientConfig = Optional.empty();
       Object clientConfig = options.getExtraProperties().get(VeniceServerWrapper.CLIENT_CONFIG_FOR_CONSUMER);

services/venice-controller/build.gradle

@@ -34,6 +34,7 @@ dependencies {
 
   implementation project(':internal:venice-common')
   implementation project(':clients:venice-thin-client')
+  implementation project(':internal:venice-client-common')
 
   implementation libraries.commonsIo
   implementation libraries.fastUtil
@@ -43,11 +44,13 @@ dependencies {
   implementation libraries.spark
   // It's necessary to pull in the most recent version of zkclient explicitly, otherwise Helix won't have it...
   implementation libraries.zkclient
+  implementation libraries.opentelemetryApi
 
   testImplementation project(':services:venice-router')
   testImplementation libraries.avroUtilFastserde
   testImplementation libraries.kafkaClientsTest // TODO: Get rid of Kafka dependency in venice-common (used by TopicCreator)
   testImplementation project(':internal:venice-test-common')
+  testImplementation libraries.openTelemetryTestSdk
 }
 
 jar {

services/venice-controller/src/main/java/com/linkedin/venice/controller/Admin.java

@@ -5,6 +5,7 @@
 import com.linkedin.venice.controller.kafka.consumer.AdminConsumerService;
 import com.linkedin.venice.controller.logcompaction.CompactionManager;
 import com.linkedin.venice.controller.repush.RepushJobRequest;
+import com.linkedin.venice.controller.stats.LogCompactionStats;
 import com.linkedin.venice.controllerapi.NodeReplicasReadinessState;
 import com.linkedin.venice.controllerapi.RepushInfo;
 import com.linkedin.venice.controllerapi.RepushJobResponse;
@@ -48,6 +49,7 @@
 import com.linkedin.venice.utils.LogContext;
 import com.linkedin.venice.utils.Pair;
 import com.linkedin.venice.utils.VeniceProperties;
+import com.linkedin.venice.utils.concurrent.VeniceConcurrentHashMap;
 import com.linkedin.venice.writer.VeniceWriterFactory;
 import java.io.Closeable;
 import java.io.IOException;
@@ -72,6 +74,7 @@ class OfflinePushStatusInfo {
     private String statusDetails;
     private Map<String, String> extraDetails;
     private List<UncompletedPartition> uncompletedPartitions;
+    final Map<String, LogCompactionStats> logCompactionStatsMap = new VeniceConcurrentHashMap<>();
 
     /** N.B.: Test-only constructor ): */
     public OfflinePushStatusInfo(ExecutionStatus executionStatus) {

services/venice-controller/src/main/java/com/linkedin/venice/controller/HelixVeniceClusterResources.java

@@ -230,7 +230,8 @@ public HelixVeniceClusterResources(
     }
 
     if (config.isParent() && config.isLogCompactionSchedulingEnabled()) {
-      this.logCompactionService = new LogCompactionService(admin, clusterName, config);
+      LOGGER.info("Log compaction service is enabled for cluster: {}", clusterName);
+      this.logCompactionService = new LogCompactionService(admin, clusterName, config, metricsRepository);
     } else {
       this.logCompactionService = null;
     }

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceController.java

@@ -18,6 +18,7 @@
 import com.linkedin.venice.controller.server.AdminSparkServer;
 import com.linkedin.venice.controller.server.VeniceControllerGrpcServiceImpl;
 import com.linkedin.venice.controller.server.VeniceControllerRequestHandler;
+import com.linkedin.venice.controller.stats.ControllerMetricEntity;
 import com.linkedin.venice.controller.stats.DeferredVersionSwapStats;
 import com.linkedin.venice.controller.stats.TopicCleanupServiceStats;
 import com.linkedin.venice.controller.supersetschema.SupersetSchemaGenerator;
@@ -35,6 +36,8 @@
 import com.linkedin.venice.service.ICProvider;
 import com.linkedin.venice.servicediscovery.AsyncRetryingServiceDiscoveryAnnouncer;
 import com.linkedin.venice.servicediscovery.ServiceDiscoveryAnnouncer;
+import com.linkedin.venice.stats.metrics.MetricEntity;
+import com.linkedin.venice.stats.metrics.ModuleMetricEntityInterface;
 import com.linkedin.venice.system.store.ControllerClientBackedSystemSchemaInitializer;
 import com.linkedin.venice.utils.LogContext;
 import com.linkedin.venice.utils.PropertyBuilder;
@@ -46,6 +49,7 @@
 import io.grpc.ServerInterceptor;
 import io.tehuti.metrics.MetricsRepository;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
@@ -60,7 +64,10 @@
 public class VeniceController {
   private static final Logger LOGGER = LogManager.getLogger(VeniceController.class);
   private static final String CONTROLLER_GRPC_SERVER_THREAD_NAME = "ControllerGrpcServer";
-  static final String CONTROLLER_SERVICE_NAME = "venice-controller";
+  public static final String CONTROLLER_SERVICE_NAME = "venice-controller";
+  public static final String CONTROLLER_SERVICE_METRIC_PREFIX = "controller";
+  public static final Collection<MetricEntity> CONTROLLER_SERVICE_METRIC_ENTITIES =
+      ModuleMetricEntityInterface.getUniqueMetricEntities(ControllerMetricEntity.class);
 
   // services
   private final VeniceControllerService controllerService;

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceControllerContext.java

@@ -1,5 +1,7 @@
 package com.linkedin.venice.controller;
 
+import static com.linkedin.venice.controller.VeniceController.CONTROLLER_SERVICE_METRIC_ENTITIES;
+import static com.linkedin.venice.controller.VeniceController.CONTROLLER_SERVICE_METRIC_PREFIX;
 import static com.linkedin.venice.controller.VeniceController.CONTROLLER_SERVICE_NAME;
 
 import com.linkedin.d2.balancer.D2Client;
@@ -10,7 +12,7 @@
 import com.linkedin.venice.pubsub.PubSubClientsFactory;
 import com.linkedin.venice.service.ICProvider;
 import com.linkedin.venice.servicediscovery.ServiceDiscoveryAnnouncer;
-import com.linkedin.venice.stats.TehutiUtils;
+import com.linkedin.venice.stats.VeniceMetricsRepository;
 import com.linkedin.venice.utils.VeniceProperties;
 import io.tehuti.metrics.MetricsRepository;
 import java.util.Collections;
@@ -154,7 +156,16 @@ public Builder setPubSubClientsFactory(PubSubClientsFactory pubSubClientsFactory
 
     private void addDefaultValues() {
       if (metricsRepository == null && !isMetricsRepositorySet) {
-        metricsRepository = TehutiUtils.getMetricsRepository(CONTROLLER_SERVICE_NAME);
+
+        metricsRepository = VeniceMetricsRepository.getVeniceMetricsRepository(
+            CONTROLLER_SERVICE_NAME,
+            CONTROLLER_SERVICE_METRIC_PREFIX,
+            CONTROLLER_SERVICE_METRIC_ENTITIES,
+            (propertiesList == null || propertiesList.isEmpty())
+                ? new VeniceProperties().getAsMap()
+                : propertiesList.get(0).getAsMap());
+        // TODO OTel: today, this gets the properties of the first cluster. This is not ideal.
+        // We need to figure out how to build a common controller-specific config/properties list
       }
       if (serviceDiscoveryAnnouncers == null && !isServiceDiscoveryAnnouncerSet) {
         serviceDiscoveryAnnouncers = Collections.emptyList();

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceControllerService.java

@@ -110,7 +110,8 @@ public VeniceControllerService(
           externalSupersetSchemaGenerator,
           pubSubTopicRepository,
           initRoutineForPushJobDetailsSystemStore,
-          initRoutineForHeartbeatSystemStore);
+          initRoutineForHeartbeatSystemStore,
+          metricsRepository);
       LOGGER.info("Controller works as a parent controller.");
     } else {
       this.admin = internalAdmin;