[controller] Add metric for disabled partition map.

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