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

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

@@ -142,6 +142,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, LongCounter> guageMap = new VeniceConcurrentHashMap<>();
 
   MetricExporter getOtlpHttpMetricExporter(VeniceMetricsConfig metricsConfig) {
     OtlpHttpMetricExporterBuilder exporterBuilder =
@@ -252,6 +253,19 @@ public LongCounter createCounter(MetricEntity metricEntity) {
     });
   }
 
+  public LongCounter createGuage(MetricEntity metricEntity) {
+    if (!emitOpenTelemetryMetrics()) {
+      return null;
+    }
+    return guageMap.computeIfAbsent(metricEntity.getMetricName(), key -> {
+      String fullMetricName = getFullMetricName(getMetricPrefix(metricEntity), metricEntity.getMetricName());
+      LongCounterBuilder builder = meter.counterBuilder(fullMetricName)
+          .setUnit(metricEntity.getUnit().name())
+          .setDescription(metricEntity.getDescription());
+      return builder.build();
+    });
+  }
+
   public Object createInstrument(MetricEntity metricEntity) {
     MetricType metricType = metricEntity.getMetricType();
     switch (metricType) {
@@ -261,6 +275,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/JobRunStatus.java

@@ -0,0 +1,25 @@
+package com.linkedin.venice.stats.dimensions;
+
+public enum JobRunStatus implements VeniceDimensionInterface {
+  FAILED, SUCCESS;
+
+  private final String status;
+
+  JobRunStatus() {
+    this.status = 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.JOB_EXECUTION_STATUS;
+  }
+
+  @Override
+  public String getDimensionValue() {
+    return this.status;
+  }
+}

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

@@ -0,0 +1,25 @@
+package com.linkedin.venice.stats.dimensions;
+
+public enum LogCompactionSelectionReason implements VeniceDimensionInterface {
+  TIME_SINCE_LAST_VERSION_CREATION_EXCEEDS_THRESHOLD;
+
+  private final String reason;
+
+  LogCompactionSelectionReason() {
+    this.reason = 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.LOG_COMPACTION_SELECTION_REASON;
+  }
+
+  @Override
+  public String getDimensionValue() {
+    return this.reason;
+  }
+}

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_STORE_TRIGGER_SOURCE;
+  }
+
+  @Override
+  public String getDimensionValue() {
+    return this.triggerSource;
+  }
+}

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

@@ -11,6 +11,7 @@
 
 public enum VeniceMetricsDimensions {
   VENICE_STORE_NAME("venice.store.name"), VENICE_CLUSTER_NAME("venice.cluster.name"),
+  VENICE_STORE_VERSION("venice.store.version"),
 
   /** {@link com.linkedin.venice.read.RequestType} */
   VENICE_REQUEST_METHOD("venice.request.method"),
@@ -28,7 +29,16 @@ public enum VeniceMetricsDimensions {
   VENICE_REQUEST_RETRY_TYPE("venice.request.retry_type"),
 
   /** {@link RequestRetryAbortReason} */
-  VENICE_REQUEST_RETRY_ABORT_REASON("venice.request.retry_abort_reason");
+  VENICE_REQUEST_RETRY_ABORT_REASON("venice.request.retry_abort_reason"),
+
+  /** {@link JobRunStatus} */
+  JOB_EXECUTION_STATUS("job.execution.status"),
+
+  /** {@link RepushStoreTriggerSource} */
+  REPUSH_STORE_TRIGGER_SOURCE("repush.store.trigger.source"),
+
+  /** {@link LogCompactionSelectionReason} */
+  LOG_COMPACTION_SELECTION_REASON("log.compaction.selection.reason");
 
   private final String[] dimensionName = new String[VeniceOpenTelemetryMetricNamingFormat.SIZE];
 

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 UpDownCounter: A counter that can be added to or subtracted from. Emits the absolute value of the metric value.
+   */
+  GAUGE;
 }

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

@@ -61,13 +61,15 @@
 import com.linkedin.venice.controller.VeniceControllerContext;
 import com.linkedin.venice.controller.VeniceHelixAdmin;
 import com.linkedin.venice.controller.kafka.consumer.AdminConsumerService;
+import com.linkedin.venice.controller.stats.ControllerMetricEntity;
 import com.linkedin.venice.controller.supersetschema.SupersetSchemaGenerator;
 import com.linkedin.venice.d2.D2Server;
 import com.linkedin.venice.meta.PersistenceType;
 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.stats.metrics.MetricEntity;
 import com.linkedin.venice.utils.PropertyBuilder;
 import com.linkedin.venice.utils.SslUtils;
 import com.linkedin.venice.utils.TestUtils;
@@ -77,6 +79,7 @@
 import java.io.File;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -100,6 +103,12 @@ public class VeniceControllerWrapper extends ProcessWrapper {
   public static final String PARENT_D2_CLUSTER_NAME = "ParentControllerD2Cluster";
   public static final String PARENT_D2_SERVICE_NAME = "ParentController";
 
+  public static final String CONTROLLER_SERVICE_METRIC_PREFIX = "controller";
+  public static final Collection<MetricEntity> CONTROLLER_SERVICE_METRIC_ENTITIES = Collections.unmodifiableList(
+      Arrays.stream(ControllerMetricEntity.values())
+          .map(ControllerMetricEntity::getMetricEntity)
+          .collect(Collectors.toList()));
+
   public static final String SUPERSET_SCHEMA_GENERATOR = "SupersetSchemaGenerator";
 
   public static final double DEFAULT_STORAGE_ENGINE_OVERHEAD_RATIO = 0.85d;
@@ -354,7 +363,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,
+          CONTROLLER_SERVICE_METRIC_PREFIX,
+          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/HelixVeniceClusterResources.java

@@ -230,7 +230,7 @@ public HelixVeniceClusterResources(
     }
 
     if (config.isParent() && config.isLogCompactionSchedulingEnabled()) {
-      this.logCompactionService = new LogCompactionService(admin, clusterName, config);
+      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,7 @@
 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.system.store.ControllerClientBackedSystemSchemaInitializer;
 import com.linkedin.venice.utils.LogContext;
 import com.linkedin.venice.utils.PropertyBuilder;
@@ -46,10 +48,13 @@
 import io.grpc.ServerInterceptor;
 import io.tehuti.metrics.MetricsRepository;
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.ThreadPoolExecutor;
+import java.util.stream.Collectors;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -61,6 +66,11 @@ 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_METRIC_PREFIX = "controller";
+  public static final Collection<MetricEntity> CONTROLLER_SERVICE_METRIC_ENTITIES = Collections.unmodifiableList(
+      Arrays.stream(ControllerMetricEntity.values())
+          .map(ControllerMetricEntity::getMetricEntity)
+          .collect(Collectors.toList()));
 
   // 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,13 @@ 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.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;

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

@@ -43,6 +43,7 @@
 import com.linkedin.venice.PushJobCheckpoints;
 import com.linkedin.venice.SSLConfig;
 import com.linkedin.venice.acl.DynamicAccessController;
+import com.linkedin.venice.annotation.VisibleForTesting;
 import com.linkedin.venice.client.store.AvroSpecificStoreClient;
 import com.linkedin.venice.client.store.ClientConfig;
 import com.linkedin.venice.client.store.ClientFactory;
@@ -71,6 +72,7 @@
 import com.linkedin.venice.controller.stats.AddVersionLatencyStats;
 import com.linkedin.venice.controller.stats.DeadStoreStats;
 import com.linkedin.venice.controller.stats.DisabledPartitionStats;
+import com.linkedin.venice.controller.stats.LogCompactionStats;
 import com.linkedin.venice.controller.stats.PushJobStatusStats;
 import com.linkedin.venice.controllerapi.AdminOperationProtocolVersionControllerResponse;
 import com.linkedin.venice.controllerapi.ControllerClient;
@@ -267,6 +269,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.function.Consumer;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import javax.annotation.Nonnull;
 import org.apache.avro.Schema;
@@ -627,16 +630,22 @@ public VeniceHelixAdmin(
         new DataRecoveryManager(this, icProvider, pubSubTopicRepository, participantStoreClientsManager);
 
     if (multiClusterConfigs.isLogCompactionEnabled()) {
-      // TODO LC: extends interchangeable with implements?
       Class<? extends RepushOrchestrator> repushOrchestratorClass =
           ReflectUtils.loadClass(multiClusterConfigs.getRepushOrchestratorClassName());
       try {
         RepushOrchestrator repushOrchestrator = ReflectUtils.callConstructor(
             repushOrchestratorClass,
             new Class[] { VeniceProperties.class },
             new Object[] { multiClusterConfigs.getRepushOrchestratorConfigs() });
-        compactionManager =
-            new CompactionManager(repushOrchestrator, multiClusterConfigs.getTimeSinceLastLogCompactionThresholdMS());
+        compactionManager = new CompactionManager(
+            repushOrchestrator,
+            multiClusterConfigs.getTimeSinceLastLogCompactionThresholdMS(),
+            multiClusterConfigs.getClusters()
+                .stream()
+                .collect(
+                    Collectors.toMap(
+                        Function.identity(),
+                        clusterName -> new LogCompactionStats(metricsRepository, clusterName))));
       } catch (Exception e) {
         LOGGER.error("Failed to enable " + LogCompactionService.class.getSimpleName(), e);
         throw new VeniceException(e);
@@ -9230,8 +9239,8 @@ public boolean isClusterWipeAllowed(String clusterName) {
     return multiClusterConfigs.getControllerConfig(clusterName).isClusterWipeAllowed();
   }
 
-  // Visible for testing
-  VeniceControllerMultiClusterConfig getMultiClusterConfigs() {
+  @VisibleForTesting
+  public VeniceControllerMultiClusterConfig getMultiClusterConfigs() {
     return multiClusterConfigs;
   }