[controller] Cleanup log compaction metrics and add some tests

m-nagarajan · m-nagarajan · commit 03baefb71fc6 · 2025-09-20T18:52:00.000-07:00
diff --git a/internal/venice-client-common/src/main/java/com/linkedin/venice/stats/dimensions/StoreRepushTriggerSource.java b/internal/venice-client-common/src/main/java/com/linkedin/venice/stats/dimensions/StoreRepushTriggerSource.java
@@ -1,11 +1,11 @@
 package com.linkedin.venice.stats.dimensions;
 
-public enum RepushStoreTriggerSource implements VeniceDimensionInterface {
-  MANUAL, SCHEDULED;
+public enum StoreRepushTriggerSource implements VeniceDimensionInterface {
+  MANUAL, SCHEDULED_FOR_LOG_COMPACTION;
 
   private final String triggerSource;
 
-  RepushStoreTriggerSource() {
+  StoreRepushTriggerSource() {
     this.triggerSource = name().toLowerCase();
   }
 
@@ -15,7 +15,7 @@ public enum RepushStoreTriggerSource implements VeniceDimensionInterface {
    */
   @Override
   public VeniceMetricsDimensions getDimensionName() {
-    return VeniceMetricsDimensions.REPUSH_TRIGGER_SOURCE;
+    return VeniceMetricsDimensions.STORE_REPUSH_TRIGGER_SOURCE;
   }
 
   @Override
diff --git a/internal/venice-client-common/src/main/java/com/linkedin/venice/stats/dimensions/VeniceMetricsDimensions.java b/internal/venice-client-common/src/main/java/com/linkedin/venice/stats/dimensions/VeniceMetricsDimensions.java
@@ -36,8 +36,8 @@ public enum VeniceMetricsDimensions {
   /** {@link RequestRetryAbortReason} */
   VENICE_REQUEST_RETRY_ABORT_REASON("venice.request.retry_abort_reason"),
 
-  /** {@link RepushStoreTriggerSource} */
-  REPUSH_TRIGGER_SOURCE("repush.trigger.source");
+  /** {@link StoreRepushTriggerSource} */
+  STORE_REPUSH_TRIGGER_SOURCE("store.repush.trigger.source");
 
   private final String[] dimensionName = new String[VeniceOpenTelemetryMetricNamingFormat.SIZE];
 
diff --git a/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/StoreRepushTriggerSourceTest.java b/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/StoreRepushTriggerSourceTest.java
@@ -0,0 +1,24 @@
+package com.linkedin.venice.stats.dimensions;
+
+import com.linkedin.venice.utils.CollectionUtils;
+import java.util.Map;
+
+
+public class StoreRepushTriggerSourceTest extends VeniceDimensionInterfaceTest<StoreRepushTriggerSource> {
+  protected StoreRepushTriggerSourceTest() {
+    super(StoreRepushTriggerSource.class);
+  }
+
+  @Override
+  protected VeniceMetricsDimensions expectedDimensionName() {
+    return VeniceMetricsDimensions.STORE_REPUSH_TRIGGER_SOURCE;
+  }
+
+  @Override
+  protected Map<StoreRepushTriggerSource, String> expectedDimensionValueMapping() {
+    return CollectionUtils.<StoreRepushTriggerSource, String>mapBuilder()
+        .put(StoreRepushTriggerSource.MANUAL, "manual")
+        .put(StoreRepushTriggerSource.SCHEDULED_FOR_LOG_COMPACTION, "scheduled_for_log_compaction")
+        .build();
+  }
+}
diff --git a/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/StreamProgressTest.java b/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/StreamProgressTest.java
@@ -0,0 +1,25 @@
+package com.linkedin.venice.stats.dimensions;
+
+import com.linkedin.venice.utils.CollectionUtils;
+import java.util.Map;
+
+
+public class StreamProgressTest extends VeniceDimensionInterfaceTest<StreamProgress> {
+  protected StreamProgressTest() {
+    super(StreamProgress.class);
+  }
+
+  @Override
+  protected VeniceMetricsDimensions expectedDimensionName() {
+    return VeniceMetricsDimensions.VENICE_STREAM_PROGRESS;
+  }
+
+  @Override
+  protected Map<StreamProgress, String> expectedDimensionValueMapping() {
+    return CollectionUtils.<StreamProgress, String>mapBuilder()
+        .put(StreamProgress.FIRST, "FirstRecord")
+        .put(StreamProgress.PCT_50, "50thPercentileRecord")
+        .put(StreamProgress.PCT_90, "90thPercentileRecord")
+        .build();
+  }
+}
diff --git a/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/VeniceMetricsDimensionsTest.java b/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/dimensions/VeniceMetricsDimensionsTest.java
@@ -43,8 +43,8 @@ public void testGetDimensionNameInSnakeCase() {
         case VENICE_STREAM_PROGRESS:
           assertEquals(dimension.getDimensionName(format), "venice.stream.progress");
           break;
-        case REPUSH_TRIGGER_SOURCE:
-          assertEquals(dimension.getDimensionName(format), "repush.trigger.source");
+        case STORE_REPUSH_TRIGGER_SOURCE:
+          assertEquals(dimension.getDimensionName(format), "store.repush.trigger.source");
           break;
         default:
           throw new IllegalArgumentException("Unknown dimension: " + dimension);
@@ -87,8 +87,8 @@ public void testGetDimensionNameInCamelCase() {
         case VENICE_STREAM_PROGRESS:
           assertEquals(dimension.getDimensionName(format), "venice.stream.progress");
           break;
-        case REPUSH_TRIGGER_SOURCE:
-          assertEquals(dimension.getDimensionName(format), "repush.trigger.source");
+        case STORE_REPUSH_TRIGGER_SOURCE:
+          assertEquals(dimension.getDimensionName(format), "store.repush.trigger.source");
           break;
         default:
           throw new IllegalArgumentException("Unknown dimension: " + dimension);
@@ -131,8 +131,8 @@ public void testGetDimensionNameInPascalCase() {
         case VENICE_STREAM_PROGRESS:
           assertEquals(dimension.getDimensionName(format), "Venice.Stream.Progress");
           break;
-        case REPUSH_TRIGGER_SOURCE:
-          assertEquals(dimension.getDimensionName(format), "Repush.Trigger.Source");
+        case STORE_REPUSH_TRIGGER_SOURCE:
+          assertEquals(dimension.getDimensionName(format), "Store.Repush.Trigger.Source");
           break;
         default:
           throw new IllegalArgumentException("Unknown dimension: " + dimension);
diff --git a/services/venice-controller/src/main/java/com/linkedin/venice/controller/logcompaction/LogCompactionService.java b/services/venice-controller/src/main/java/com/linkedin/venice/controller/logcompaction/LogCompactionService.java
@@ -7,7 +7,7 @@
 import com.linkedin.venice.controllerapi.RepushJobResponse;
 import com.linkedin.venice.meta.StoreInfo;
 import com.linkedin.venice.service.AbstractVeniceService;
-import com.linkedin.venice.stats.dimensions.RepushStoreTriggerSource;
+import com.linkedin.venice.stats.dimensions.StoreRepushTriggerSource;
 import com.linkedin.venice.utils.LogContext;
 import java.time.Instant;
 import java.time.ZoneId;
@@ -109,8 +109,11 @@ public void run() {
     private void compactStoresInClusters() {
       for (StoreInfo storeInfo: admin.getStoresForCompaction(clusterName)) {
         try {
-          RepushJobResponse response = admin
-              .repushStore(new RepushJobRequest(clusterName, storeInfo.getName(), RepushStoreTriggerSource.SCHEDULED));
+          RepushJobResponse response = admin.repushStore(
+              new RepushJobRequest(
+                  clusterName,
+                  storeInfo.getName(),
+                  StoreRepushTriggerSource.SCHEDULED_FOR_LOG_COMPACTION));
           LOGGER.info(
               "Succeeded to trigger log compaction for store: {} in cluster: {} | execution ID: {}",
               response.getName(),
diff --git a/services/venice-controller/src/main/java/com/linkedin/venice/controller/repush/RepushJobRequest.java b/services/venice-controller/src/main/java/com/linkedin/venice/controller/repush/RepushJobRequest.java
@@ -4,7 +4,7 @@
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.SOURCE_REGION;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.STORE_NAME;
 
-import com.linkedin.venice.stats.dimensions.RepushStoreTriggerSource;
+import com.linkedin.venice.stats.dimensions.StoreRepushTriggerSource;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -13,9 +13,9 @@ public class RepushJobRequest {
   private final String clusterName;
   private final String storeName;
   private final String sourceRegion;
-  private final RepushStoreTriggerSource triggerSource;
+  private final StoreRepushTriggerSource triggerSource;
 
-  public RepushJobRequest(String clusterName, String storeName, RepushStoreTriggerSource triggerSource) {
+  public RepushJobRequest(String clusterName, String storeName, StoreRepushTriggerSource triggerSource) {
     this.clusterName = clusterName;
     this.storeName = storeName;
     this.sourceRegion = null; // default to null if not specified
@@ -26,7 +26,7 @@ public RepushJobRequest(
       String clusterName,
       String storeName,
       String sourceRegion,
-      RepushStoreTriggerSource triggerSource) {
+      StoreRepushTriggerSource triggerSource) {
     this.clusterName = clusterName;
     this.storeName = storeName;
     this.sourceRegion = sourceRegion;
@@ -45,7 +45,7 @@ public String getSourceRegion() {
     return sourceRegion;
   }
 
-  public RepushStoreTriggerSource getTriggerSource() {
+  public StoreRepushTriggerSource getTriggerSource() {
     return triggerSource;
   }
 
diff --git a/services/venice-controller/src/main/java/com/linkedin/venice/controller/server/StoresRoutes.java b/services/venice-controller/src/main/java/com/linkedin/venice/controller/server/StoresRoutes.java
@@ -116,7 +116,7 @@
 import com.linkedin.venice.pubsub.api.PubSubTopic;
 import com.linkedin.venice.pubsub.api.exceptions.PubSubTopicDoesNotExistException;
 import com.linkedin.venice.pubsub.manager.TopicManager;
-import com.linkedin.venice.stats.dimensions.RepushStoreTriggerSource;
+import com.linkedin.venice.stats.dimensions.StoreRepushTriggerSource;
 import com.linkedin.venice.systemstore.schemas.StoreProperties;
 import com.linkedin.venice.utils.Utils;
 import java.util.ArrayList;
@@ -1112,7 +1112,7 @@ public void internalHandle(Request request, RepushJobResponse veniceResponse) {
         try {
           veniceResponse.copyValueOf(
               admin.repushStore(
-                  new RepushJobRequest(clusterName, storeName, sourceRegion, RepushStoreTriggerSource.MANUAL)));
+                  new RepushJobRequest(clusterName, storeName, sourceRegion, StoreRepushTriggerSource.MANUAL)));
         } catch (Exception e) {
           veniceResponse.setError("Failed to compact store: " + storeName, e);
         }
diff --git a/services/venice-controller/src/main/java/com/linkedin/venice/controller/stats/ControllerMetricEntity.java b/services/venice-controller/src/main/java/com/linkedin/venice/controller/stats/ControllerMetricEntity.java
@@ -1,11 +1,12 @@
 package com.linkedin.venice.controller.stats;
 
-import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.REPUSH_TRIGGER_SOURCE;
+import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.STORE_REPUSH_TRIGGER_SOURCE;
 import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_CLUSTER_NAME;
 import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_RESPONSE_STATUS_CODE_CATEGORY;
 import static com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions.VENICE_STORE_NAME;
 import static com.linkedin.venice.utils.Utils.setOf;
 
+import com.linkedin.venice.annotation.VisibleForTesting;
 import com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions;
 import com.linkedin.venice.stats.metrics.MetricEntity;
 import com.linkedin.venice.stats.metrics.MetricType;
@@ -15,34 +16,41 @@
 
 
 public enum ControllerMetricEntity implements ModuleMetricEntityInterface {
-  REPUSH_CALL_COUNT(
-      MetricType.COUNTER, MetricUnit.NUMBER, "Count of all calls to a controller endpoint",
-      setOf(VENICE_STORE_NAME, VENICE_RESPONSE_STATUS_CODE_CATEGORY, VENICE_CLUSTER_NAME, REPUSH_TRIGGER_SOURCE)
+  STORE_REPUSH_CALL_COUNT(
+      "store.repush.call_count", MetricType.COUNTER, MetricUnit.NUMBER, "Count of all requests to repush a store",
+      setOf(VENICE_STORE_NAME, VENICE_RESPONSE_STATUS_CODE_CATEGORY, VENICE_CLUSTER_NAME, STORE_REPUSH_TRIGGER_SOURCE)
   ),
-  COMPACTION_ELIGIBLE_STATE(
-      MetricType.GAUGE, MetricUnit.NUMBER,
-      "This metric indicates the duration from when a store is first nominated for compaction until the store is compacted successfully."
-          + " When a store is nominated for scheduled compaction and remains uncompacted, this metric will be at 1."
-          + " When the store is compacted, this metric will return to 0.",
+
+  /** log compaction related metrics */
+  STORE_COMPACTION_NOMINATED_COUNT(
+      "store.compaction.nominated_count", MetricType.COUNTER, MetricUnit.NUMBER,
+      "Count of stores nominated for scheduled compaction", setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME)
+  ),
+  STORE_COMPACTION_ELIGIBLE_STATE(
+      "store.compaction.eligible_state", MetricType.GAUGE, MetricUnit.NUMBER,
+      "State of the compaction eligibility: shows the duration from compaction nomination to triggered to completion. stays 1 after nomination and becomes 0 when the compaction is compacted",
       setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME)
   ),
-  STORE_NOMINATED_FOR_COMPACTION_COUNT(
-      MetricType.COUNTER, MetricUnit.NUMBER, "Count of stores nominated for scheduled compaction",
+  STORE_COMPACTION_TRIGGERED_COUNT(
+      "store.compaction.triggered_count", MetricType.COUNTER, MetricUnit.NUMBER,
+      "Count of log compaction repush triggered for a store after it becomes eligible",
       setOf(VENICE_STORE_NAME, VENICE_CLUSTER_NAME)
   );
 
   private final MetricEntity metricEntity;
   private final String metricName;
 
   ControllerMetricEntity(
+      String metricName,
       MetricType metricType,
       MetricUnit unit,
       String description,
       Set<VeniceMetricsDimensions> dimensionsList) {
-    this.metricName = this.name().toLowerCase();
+    this.metricName = metricName;
     this.metricEntity = new MetricEntity(metricName, metricType, unit, description, dimensionsList);
   }
 
+  @VisibleForTesting
   public String getMetricName() {
     return metricName;
   }
diff --git a/services/venice-controller/src/main/java/com/linkedin/venice/controller/stats/LogCompactionStats.java b/services/venice-controller/src/main/java/com/linkedin/venice/controller/stats/LogCompactionStats.java
@@ -7,7 +7,7 @@
 import com.linkedin.venice.stats.VeniceMetricsConfig;
 import com.linkedin.venice.stats.VeniceMetricsRepository;
 import com.linkedin.venice.stats.VeniceOpenTelemetryMetricsRepository;
-import com.linkedin.venice.stats.dimensions.RepushStoreTriggerSource;
+import com.linkedin.venice.stats.dimensions.StoreRepushTriggerSource;
 import com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions;
 import com.linkedin.venice.stats.dimensions.VeniceResponseStatusCategory;
 import com.linkedin.venice.stats.metrics.MetricEntityStateGeneric;
@@ -51,23 +51,23 @@ public LogCompactionStats(MetricsRepository metricsRepository, String clusterNam
     }
 
     repushCallCountMetric = MetricEntityStateGeneric.create(
-        ControllerMetricEntity.REPUSH_CALL_COUNT.getMetricEntity(),
+        ControllerMetricEntity.STORE_REPUSH_CALL_COUNT.getMetricEntity(),
         otelRepository,
         this::registerSensor,
         ControllerTehutiMetricNameEnum.REPUSH_CALL_COUNT,
         Collections.singletonList(new OccurrenceRate()),
         baseDimensionsMap);
 
     compactionEligibleMetric = MetricEntityStateGeneric.create(
-        ControllerMetricEntity.COMPACTION_ELIGIBLE_STATE.getMetricEntity(),
+        ControllerMetricEntity.STORE_COMPACTION_ELIGIBLE_STATE.getMetricEntity(),
         otelRepository,
         this::registerSensor,
         ControllerTehutiMetricNameEnum.COMPACTION_ELIGIBLE_STATE,
         Collections.singletonList(new Gauge()),
         baseDimensionsMap);
 
     storeNominatedForCompactionCountMetric = MetricEntityStateGeneric.create(
-        ControllerMetricEntity.STORE_NOMINATED_FOR_COMPACTION_COUNT.getMetricEntity(),
+        ControllerMetricEntity.STORE_COMPACTION_NOMINATED_COUNT.getMetricEntity(),
         otelRepository,
         this::registerSensor,
         ControllerTehutiMetricNameEnum.STORE_NOMINATED_FOR_COMPACTION_COUNT,
@@ -77,12 +77,12 @@ public LogCompactionStats(MetricsRepository metricsRepository, String clusterNam
 
   public void recordRepushStoreCall(
       String storeName,
-      RepushStoreTriggerSource triggerSource,
+      StoreRepushTriggerSource triggerSource,
       VeniceResponseStatusCategory executionStatus) {
     repushCallCountMetric.record(
         1,
         getDimensionsBuilder().put(VeniceMetricsDimensions.VENICE_STORE_NAME, storeName)
-            .put(VeniceMetricsDimensions.REPUSH_TRIGGER_SOURCE, triggerSource.getDimensionValue())
+            .put(VeniceMetricsDimensions.STORE_REPUSH_TRIGGER_SOURCE, triggerSource.getDimensionValue())
             .put(VeniceMetricsDimensions.VENICE_RESPONSE_STATUS_CODE_CATEGORY, executionStatus.getDimensionValue())
             .build());
   }
@@ -113,11 +113,11 @@ private ImmutableMap.Builder<VeniceMetricsDimensions, String> getDimensionsBuild
   }
 
   enum ControllerTehutiMetricNameEnum implements TehutiMetricNameEnum {
-    /** for {@link ControllerMetricEntity#REPUSH_CALL_COUNT} */
+    /** for {@link ControllerMetricEntity#STORE_REPUSH_CALL_COUNT} */
     REPUSH_CALL_COUNT,
-    /** for {@link ControllerMetricEntity#COMPACTION_ELIGIBLE_STATE} */
+    /** for {@link ControllerMetricEntity#STORE_COMPACTION_ELIGIBLE_STATE} */
     COMPACTION_ELIGIBLE_STATE,
-    /** for {@link ControllerMetricEntity#STORE_NOMINATED_FOR_COMPACTION_COUNT} */
+    /** for {@link ControllerMetricEntity#STORE_COMPACTION_NOMINATED_COUNT} */
     STORE_NOMINATED_FOR_COMPACTION_COUNT;
 
     private final String metricName;
diff --git a/services/venice-controller/src/test/java/com/linkedin/venice/controller/TestVeniceHelixAdmin.java b/services/venice-controller/src/test/java/com/linkedin/venice/controller/TestVeniceHelixAdmin.java
@@ -67,7 +67,7 @@
 import com.linkedin.venice.pubsub.manager.TopicManager;
 import com.linkedin.venice.pushmonitor.ExecutionStatus;
 import com.linkedin.venice.pushstatushelper.PushStatusStoreWriter;
-import com.linkedin.venice.stats.dimensions.RepushStoreTriggerSource;
+import com.linkedin.venice.stats.dimensions.StoreRepushTriggerSource;
 import com.linkedin.venice.stats.dimensions.VeniceResponseStatusCategory;
 import com.linkedin.venice.system.store.MetaStoreWriter;
 import com.linkedin.venice.utils.DataProviderUtils;
@@ -1323,9 +1323,9 @@ public void testRepushStore(boolean manualRepush, boolean responseFailure, boole
     RepushJobRequest mockRequest = mock(RepushJobRequest.class);
     when(mockRequest.getClusterName()).thenReturn(clusterName);
     when(mockRequest.getStoreName()).thenReturn(storeName);
-    RepushStoreTriggerSource repushStoreTriggerSource =
-        manualRepush ? RepushStoreTriggerSource.MANUAL : RepushStoreTriggerSource.SCHEDULED;
-    when(mockRequest.getTriggerSource()).thenReturn(repushStoreTriggerSource);
+    StoreRepushTriggerSource storeRepushTriggerSource =
+        manualRepush ? StoreRepushTriggerSource.MANUAL : StoreRepushTriggerSource.SCHEDULED_FOR_LOG_COMPACTION;
+    when(mockRequest.getTriggerSource()).thenReturn(storeRepushTriggerSource);
 
     RepushJobResponse mockRepushJobResponse = mock(RepushJobResponse.class);
     when(mockRepushJobResponse.isError()).thenReturn(responseFailure);
@@ -1355,7 +1355,7 @@ public void testRepushStore(boolean manualRepush, boolean responseFailure, boole
         ? VeniceResponseStatusCategory.FAIL
         : VeniceResponseStatusCategory.SUCCESS;
     verify(mockLogCompactionStats, Mockito.times(1))
-        .recordRepushStoreCall(storeName, repushStoreTriggerSource, expectedResponseCategory);
+        .recordRepushStoreCall(storeName, storeRepushTriggerSource, expectedResponseCategory);
   }
 
   @Test
diff --git a/services/venice-controller/src/test/java/com/linkedin/venice/controller/logcompaction/CompactionManagerTest.java b/services/venice-controller/src/test/java/com/linkedin/venice/controller/logcompaction/CompactionManagerTest.java
@@ -19,7 +19,7 @@
 import com.linkedin.venice.meta.HybridStoreConfig;
 import com.linkedin.venice.meta.StoreInfo;
 import com.linkedin.venice.meta.Version;
-import com.linkedin.venice.stats.dimensions.RepushStoreTriggerSource;
+import com.linkedin.venice.stats.dimensions.StoreRepushTriggerSource;
 import com.linkedin.venice.utils.Utils;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -285,7 +285,7 @@ public void testRepushStoreWithNullResponse() throws Exception {
     RepushJobRequest repushJobRequest = new RepushJobRequest(
         TEST_CLUSTER_NAME_1,
         Utils.getUniqueString(TEST_STORE_NAME_PREFIX),
-        RepushStoreTriggerSource.MANUAL);
+        StoreRepushTriggerSource.MANUAL);
 
     // Call the repushStore method and expect a VeniceException
     testCompactionManager.repushStore(repushJobRequest);
diff --git a/services/venice-controller/src/test/java/com/linkedin/venice/controller/stats/LogCompactionStatsTest.java b/services/venice-controller/src/test/java/com/linkedin/venice/controller/stats/LogCompactionStatsTest.java
