[controller] Add metric for disabled partition map. (#742)

Since disabled metric can cause a cluster to be in undeployable state, this PR is adding an alert to indicate disabled partition per cluster.
The proposed action would be for oncall to investigate what error caused the alert and mitigate by restarting those hosts.



---------

Co-authored-by: Sourav Maji <[email protected]>

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

@@ -166,7 +166,8 @@ public HelixVeniceClusterResources(
         getActiveActiveRealTimeSourceKafkaURLs(config),
         helixAdminClient,
         config,
-        admin.getPushStatusStoreReader().orElse(null));
+        admin.getPushStatusStoreReader().orElse(null),
+        metricsRepository);
 
     this.leakedPushStatusCleanUpService = new LeakedPushStatusCleanUpService(
         clusterName,

services/venice-controller/src/main/java/com/linkedin/venice/controller/stats/DisabledPartitionStats.java

@@ -0,0 +1,21 @@
+package com.linkedin.venice.controller.stats;
+
+import com.linkedin.venice.stats.AbstractVeniceStats;
+import io.tehuti.metrics.MetricsRepository;
+import io.tehuti.metrics.Sensor;
+import io.tehuti.metrics.stats.Count;
+
+
+public class DisabledPartitionStats extends AbstractVeniceStats {
+  private final Sensor disabledPartitionCount;
+
+  public DisabledPartitionStats(MetricsRepository metricsRepository, String name) {
+    super(metricsRepository, name);
+    disabledPartitionCount = registerSensorIfAbsent("disabled_partition_count", new Count());
+  }
+
+  public void recordDisabledPartition() {
+    disabledPartitionCount.record();
+  }
+
+}

services/venice-controller/src/main/java/com/linkedin/venice/pushmonitor/AbstractPushMonitor.java

@@ -10,6 +10,7 @@
 
 import com.linkedin.venice.controller.HelixAdminClient;
 import com.linkedin.venice.controller.VeniceControllerConfig;
+import com.linkedin.venice.controller.stats.DisabledPartitionStats;
 import com.linkedin.venice.exceptions.VeniceException;
 import com.linkedin.venice.exceptions.VeniceNoStoreException;
 import com.linkedin.venice.helix.HelixCustomizedViewOfflinePushRepository;
@@ -33,6 +34,7 @@
 import com.linkedin.venice.utils.concurrent.VeniceConcurrentHashMap;
 import com.linkedin.venice.utils.locks.AutoCloseableLock;
 import com.linkedin.venice.utils.locks.ClusterLockManager;
+import io.tehuti.metrics.MetricsRepository;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -82,6 +84,7 @@ public abstract class AbstractPushMonitor
   private final long offlineJobResourceAssignmentWaitTimeInMilliseconds;
 
   private final PushStatusCollector pushStatusCollector;
+  private final DisabledPartitionStats disabledPartitionStats;
 
   private final boolean isOfflinePushMonitorDaVinciPushStatusEnabled;
 
@@ -98,7 +101,8 @@ public AbstractPushMonitor(
       List<String> activeActiveRealTimeSourceKafkaURLs,
       HelixAdminClient helixAdminClient,
       VeniceControllerConfig controllerConfig,
-      PushStatusStoreReader pushStatusStoreReader) {
+      PushStatusStoreReader pushStatusStoreReader,
+      MetricsRepository metricsRepository) {
     this.clusterName = clusterName;
     this.offlinePushAccessor = offlinePushAccessor;
     this.storeCleaner = storeCleaner;
@@ -110,6 +114,8 @@ public AbstractPushMonitor(
     this.aggregateRealTimeSourceKafkaUrl = aggregateRealTimeSourceKafkaUrl;
     this.activeActiveRealTimeSourceKafkaURLs = activeActiveRealTimeSourceKafkaURLs;
     this.helixAdminClient = helixAdminClient;
+    this.disabledPartitionStats = new DisabledPartitionStats(metricsRepository, clusterName);
+
     this.disableErrorLeaderReplica = controllerConfig.isErrorLeaderReplicaFailOverEnabled();
     this.helixClientThrottler =
         new EventThrottler(10, "push_monitor_helix_client_throttler", false, EventThrottler.BLOCK_STRATEGY);
@@ -755,6 +761,7 @@ public void disableReplica(String instance, int partitionId) {
             kafkaTopic,
             Collections.singletonList(HelixUtils.getPartitionName(kafkaTopic, partitionId)));
         disabledReplicaMap.computeIfAbsent(instance, k -> new HashSet<>()).add(partitionId);
+        disabledPartitionStats.recordDisabledPartition();
       }
 
       @Override

services/venice-controller/src/main/java/com/linkedin/venice/pushmonitor/PartitionStatusBasedPushMonitor.java

@@ -10,6 +10,7 @@
 import com.linkedin.venice.meta.StoreCleaner;
 import com.linkedin.venice.pushstatushelper.PushStatusStoreReader;
 import com.linkedin.venice.utils.locks.ClusterLockManager;
+import io.tehuti.metrics.MetricsRepository;
 import java.util.List;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -36,7 +37,8 @@ public PartitionStatusBasedPushMonitor(
       List<String> childDataCenterKafkaUrls,
       HelixAdminClient helixAdminClient,
       VeniceControllerConfig controllerConfig,
-      PushStatusStoreReader pushStatusStoreReader) {
+      PushStatusStoreReader pushStatusStoreReader,
+      MetricsRepository metricsRepository) {
     super(
         clusterName,
         offlinePushAccessor,
@@ -50,7 +52,8 @@ public PartitionStatusBasedPushMonitor(
         childDataCenterKafkaUrls,
         helixAdminClient,
         controllerConfig,
-        pushStatusStoreReader);
+        pushStatusStoreReader,
+        metricsRepository);
   }
 
   @Override

services/venice-controller/src/main/java/com/linkedin/venice/pushmonitor/PushMonitorDelegator.java

@@ -18,6 +18,7 @@
 import com.linkedin.venice.utils.concurrent.VeniceConcurrentHashMap;
 import com.linkedin.venice.utils.locks.AutoCloseableLock;
 import com.linkedin.venice.utils.locks.ClusterLockManager;
+import io.tehuti.metrics.MetricsRepository;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -55,7 +56,8 @@ public PushMonitorDelegator(
       List<String> activeActiveRealTimeSourceKafkaURLs,
       HelixAdminClient helixAdminClient,
       VeniceControllerConfig controllerConfig,
-      PushStatusStoreReader pushStatusStoreReader) {
+      PushStatusStoreReader pushStatusStoreReader,
+      MetricsRepository metricsRepository) {
     this.clusterName = clusterName;
     this.metadataRepository = metadataRepository;
 
@@ -72,7 +74,8 @@ public PushMonitorDelegator(
         activeActiveRealTimeSourceKafkaURLs,
         helixAdminClient,
         controllerConfig,
-        pushStatusStoreReader);
+        pushStatusStoreReader,
+        metricsRepository);
     this.clusterLockManager = clusterLockManager;
 
     this.topicToPushMonitorMap = new VeniceConcurrentHashMap<>();

services/venice-controller/src/test/java/com/linkedin/venice/pushmonitor/AbstractPushMonitorTest.java

@@ -43,6 +43,8 @@
 import com.linkedin.venice.utils.Utils;
 import com.linkedin.venice.utils.locks.AutoCloseableLock;
 import com.linkedin.venice.utils.locks.ClusterLockManager;
+import io.tehuti.metrics.MetricsRepository;
+import io.tehuti.metrics.Sensor;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.EnumMap;
@@ -70,6 +72,8 @@ public abstract class AbstractPushMonitorTest {
 
   protected VeniceControllerConfig mockControllerConfig;
 
+  protected MetricsRepository mockMetricRepo;
+
   private final static String clusterName = Utils.getUniqueString("test_cluster");
   private final static String aggregateRealTimeSourceKafkaUrl = "aggregate-real-time-source-kafka-url";
   private String storeName;
@@ -94,10 +98,12 @@ public void setUp() {
     mockAccessor = mock(OfflinePushAccessor.class);
     mockStoreCleaner = mock(StoreCleaner.class);
     mockStoreRepo = mock(ReadWriteStoreRepository.class);
+    mockMetricRepo = mock(MetricsRepository.class);
     mockRoutingDataRepo = mock(RoutingDataRepository.class);
     mockPushHealthStats = mock(AggPushHealthStats.class);
     clusterLockManager = new ClusterLockManager(clusterName);
     mockControllerConfig = mock(VeniceControllerConfig.class);
+    when(mockMetricRepo.sensor(anyString(), any())).thenReturn(mock(Sensor.class));
     when(mockControllerConfig.isErrorLeaderReplicaFailOverEnabled()).thenReturn(true);
     when(mockControllerConfig.isDaVinciPushStatusEnabled()).thenReturn(true);
     when(mockControllerConfig.getDaVinciPushStatusScanIntervalInSeconds()).thenReturn(5);

services/venice-controller/src/test/java/com/linkedin/venice/pushmonitor/PartitionStatusBasedPushMonitorTest.java

@@ -61,7 +61,8 @@ protected AbstractPushMonitor getPushMonitor(StoreCleaner storeCleaner) {
         Collections.emptyList(),
         helixAdminClient,
         getMockControllerConfig(),
-        null);
+        null,
+        mockMetricRepo);
   }
 
   @Override
@@ -79,7 +80,8 @@ protected AbstractPushMonitor getPushMonitor(RealTimeTopicSwitcher mockRealTimeT
         Collections.emptyList(),
         mock(HelixAdminClient.class),
         getMockControllerConfig(),
-        null);
+        null,
+        mockMetricRepo);
   }
 
   @Test