diff --git a/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java b/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java
index 3414e23..df6722d 100644
--- a/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java
+++ b/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java
@@ -1,5 +1,6 @@
 package io.clustercontroller.allocation;
 
+import io.clustercontroller.enums.NodeRole;
 import io.clustercontroller.enums.ShardState;
 import io.clustercontroller.metrics.MetricsProvider;
 import io.clustercontroller.models.Alias;
@@ -21,8 +22,8 @@
 import java.util.Map;
 import java.util.Set;
 
-import static io.clustercontroller.metrics.MetricsConstants.UPDATE_ACTUAL_ALLOCATION_FAILURES_METRIC_NAME;
-import static io.clustercontroller.metrics.MetricsUtils.buildMetricsTags;
+import static io.clustercontroller.metrics.MetricsConstants.*;
+import static io.clustercontroller.metrics.MetricsUtils.*;
 
 /**
  * ActualAllocationUpdater - Responsible for aggregating search unit actual states into index shard actual allocations
@@ -87,6 +88,12 @@ public void updateActualAllocations(String clusterId) throws Exception {
         // NEW: Update coordinator goal states based on planned allocations
         int coordinatorUpdates = updateCoordinatorGoalStates(clusterId, searchUnits);
         
+        // Emit shard distribution and doc count metrics
+        emitShardDistributionMetrics(clusterId, searchUnits, actualAllocations);
+        
+        // Cleanup stale gauges for deleted indices/shards/nodes
+        metricsProvider.cleanupStaleGauges();
+        
         log.info("ActualAllocationUpdater - Completed actual allocation update with {} updates and {} coordinator goal state updates", 
                     totalUpdates, coordinatorUpdates);
     }
@@ -662,4 +669,306 @@ private boolean isNodeTimestampRecent(SearchUnitActualState actualState, String
         return true;
     }
     
+    /**
+     * Emit shard distribution and doc count metrics for monitoring and visualization.
+     * 
+     * Metrics emitted:
+     * 1. shard_replica_count - Per shard: number of nodes hosting this shard in STARTED state
+     * 2. node_shard_count - Per node: number of shards this node hosts
+     * 3. shard_total_allocation_count - Total nodes allocated for this shard (planned)
+     * 4. shard_primary_doc_count - Doc count from primary for each shard
+     * 5. shard_replica_doc_count - Doc count from each replica (tagged with nodeName)
+     * 6. shard_primary_deleted_doc_count - Deleted doc count from primary
+     * 7. shard_replica_lag_docs - (Primary doc count - Replica doc count) per replica
+     * 8. index_total_doc_count - Sum of all primary doc counts for an index
+     */
+    private void emitShardDistributionMetrics(String clusterId, List<SearchUnit> searchUnits, 
+            Map<String, Map<String, Set<String>>> actualAllocations) {
+        log.debug("ActualAllocationUpdater - Emitting shard distribution metrics for cluster {}", clusterId);
+        
+        // Map to track shard count per node per index: indexName -> (nodeName -> count)
+        Map<String, Map<String, Integer>> nodeShardCountsByIndex = new HashMap<>();
+        // Map to track node roles: nodeName -> role
+        Map<String, String> nodeRoles = new HashMap<>();
+        // Map to track primary doc counts per shard: indexName -> shardId -> docCount
+        Map<String, Map<String, Long>> primaryDocCounts = new HashMap<>();
+        // Map to track index total doc counts: indexName -> totalDocCount
+        Map<String, Long> indexTotalDocCounts = new HashMap<>();
+        
+        // First pass: collect all actual states and build data structures
+        Map<String, SearchUnitActualState> actualStates = new HashMap<>();
+        for (SearchUnit searchUnit : searchUnits) {
+            if (isCoordinatorNode(searchUnit)) {
+                continue;
+            }
+            String unitName = searchUnit.getName();
+            try {
+                SearchUnitActualState actualState = metadataStore.getSearchUnitActualState(clusterId, unitName);
+                if (actualState != null && isNodeTimestampRecent(actualState, unitName)) {
+                    actualStates.put(unitName, actualState);
+                    
+                    // Track node role
+                    String role = searchUnit.getRole() != null ? searchUnit.getRole().toUpperCase() : "UNKNOWN";
+                    nodeRoles.put(unitName, role);
+                }
+            } catch (Exception e) {
+                log.warn("ActualAllocationUpdater - Error getting actual state for metrics, unit {}: {}", 
+                    unitName, e.getMessage());
+            }
+        }
+        
+        // Process each index and shard
+        for (Map.Entry<String, Map<String, Set<String>>> indexEntry : actualAllocations.entrySet()) {
+            String indexName = indexEntry.getKey();
+            long indexTotalDocs = 0;
+            
+            for (Map.Entry<String, Set<String>> shardEntry : indexEntry.getValue().entrySet()) {
+                String shardId = shardEntry.getKey();
+                Set<String> allocatedUnits = shardEntry.getValue();
+                
+                // 1. Count primary and search nodes separately using NodeRole enum
+                int primaryCount = 0;
+                int searchCount = 0;
+                for (String unitName : allocatedUnits) {
+                    String roleStr = nodeRoles.get(unitName);
+                    NodeRole role = NodeRole.fromString(roleStr);
+                    if (role == NodeRole.PRIMARY) {
+                        primaryCount++;
+                    } else if (role == NodeRole.REPLICA) {
+                        searchCount++;
+                    }
+                }
+                
+                // Emit shard_primary_started_count - number of primary nodes in STARTED state
+                metricsProvider.gauge(
+                    SHARD_PRIMARY_STARTED_COUNT_METRIC_NAME,
+                    primaryCount,
+                    buildMetricsTags(clusterId, indexName, shardId)
+                );
+                
+                // Emit shard_replica_started_count - number of search/replica nodes in STARTED state
+                int replicaCount = searchCount;
+                metricsProvider.gauge(
+                    SHARD_REPLICA_STARTED_COUNT_METRIC_NAME,
+                    replicaCount,
+                    buildMetricsTags(clusterId, indexName, shardId)
+                );
+                
+                // 3. Emit shard_total_allocation_count and shard_pending_allocation_count
+                try {
+                    ShardAllocation plannedAllocation = metadataStore.getPlannedAllocation(clusterId, indexName, shardId);
+                    if (plannedAllocation != null) {
+                        int totalPlanned = plannedAllocation.getIngestSUs().size() + plannedAllocation.getSearchSUs().size();
+                        metricsProvider.gauge(
+                            SHARD_TOTAL_ALLOCATION_COUNT_METRIC_NAME,
+                            totalPlanned,
+                            buildMetricsTags(clusterId, indexName, shardId)
+                        );
+                        
+                        // Emit pending allocation count (planned - started)
+                        int totalStarted = primaryCount + replicaCount;
+                        int pendingCount = Math.max(0, totalPlanned - totalStarted);
+                        metricsProvider.gauge(
+                            SHARD_PENDING_ALLOCATION_COUNT_METRIC_NAME,
+                            pendingCount,
+                            buildMetricsTags(clusterId, indexName, shardId)
+                        );
+                    }
+                } catch (Exception e) {
+                    log.debug("ActualAllocationUpdater - Error getting planned allocation for metrics: {}", e.getMessage());
+                }
+                
+                // Find primary and collect doc counts
+                long primaryDocCount = -1;
+                long primaryDeletedDocCount = -1;
+                int shardIdInt = Integer.parseInt(shardId);
+                
+                for (String unitName : allocatedUnits) {
+                    // Track shard count per node per index
+                    nodeShardCountsByIndex
+                        .computeIfAbsent(indexName, k -> new HashMap<>())
+                        .merge(unitName, 1, Integer::sum);
+                    
+                    SearchUnitActualState actualState = actualStates.get(unitName);
+                    if (actualState == null) {
+                        continue;
+                    }
+                    
+                    // Get doc count for this shard
+                    long docCount = actualState.getShardDocCount(indexName, shardIdInt);
+                    
+                    // Check if this is a primary node
+                    Map<String, List<SearchUnitActualState.ShardRoutingInfo>> nodeRouting = actualState.getNodeRouting();
+                    if (nodeRouting != null && nodeRouting.containsKey(indexName)) {
+                        List<SearchUnitActualState.ShardRoutingInfo> shards = nodeRouting.get(indexName);
+                        if (shards != null) {
+                            for (SearchUnitActualState.ShardRoutingInfo shard : shards) {
+                                if (shard.getShardId() == shardIdInt) {
+                                    if (shard.isPrimary()) {
+                                        // This is the primary - emit primary doc count
+                                        primaryDocCount = docCount;
+                                        
+                                        // Get deleted doc count if available
+                                        if (actualState.getStats() != null 
+                                            && actualState.getStats().getIndices() != null
+                                            && actualState.getStats().getIndices().getShards() != null) {
+                                            List<Map<String, SearchUnitActualState.ShardLevelStats>> indexShards = 
+                                                actualState.getStats().getIndices().getShards().get(indexName);
+                                            if (indexShards != null) {
+                                                for (Map<String, SearchUnitActualState.ShardLevelStats> shardMap : indexShards) {
+                                                    SearchUnitActualState.ShardLevelStats stats = shardMap.get(shardId);
+                                                    if (stats != null && stats.getDocs() != null) {
+                                                        primaryDeletedDocCount = stats.getDocs().getDeleted();
+                                                    }
+                                                }
+                                            }
+                                        }
+                                        
+                                        // 4. Emit shard_primary_doc_count (tagged with node name for consistency)
+                                        if (primaryDocCount >= 0) {
+                                            metricsProvider.gauge(
+                                                SHARD_PRIMARY_DOC_COUNT_METRIC_NAME,
+                                                primaryDocCount,
+                                                buildMetricsTagsWithNode(clusterId, indexName, shardId, unitName)
+                                            );
+                                            
+                                            // Store for lag calculation
+                                            primaryDocCounts
+                                                .computeIfAbsent(indexName, k -> new HashMap<>())
+                                                .put(shardId, primaryDocCount);
+                                            
+                                            // Add to index total
+                                            indexTotalDocs += primaryDocCount;
+                                        }
+                                        
+                                        // 6. Emit shard_primary_deleted_doc_count (tagged with node name for consistency)
+                                        if (primaryDeletedDocCount >= 0) {
+                                            metricsProvider.gauge(
+                                                SHARD_PRIMARY_DELETED_DOC_COUNT_METRIC_NAME,
+                                                primaryDeletedDocCount,
+                                                buildMetricsTagsWithNode(clusterId, indexName, shardId, unitName)
+                                            );
+                                        }
+                                    } else {
+                                        // This is a replica - emit replica doc count
+                                        if (docCount >= 0) {
+                                            // 5. Emit shard_replica_doc_count (tagged with node name)
+                                            metricsProvider.gauge(
+                                                SHARD_REPLICA_DOC_COUNT_METRIC_NAME,
+                                                docCount,
+                                                buildMetricsTagsWithNode(clusterId, indexName, shardId, unitName)
+                                            );
+                                        }
+                                    }
+                                    break;
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+            
+            // 8. Emit index_total_doc_count
+            if (indexTotalDocs > 0) {
+                metricsProvider.gauge(
+                    INDEX_TOTAL_DOC_COUNT_METRIC_NAME,
+                    indexTotalDocs,
+                    buildIndexMetricsTags(clusterId, indexName)
+                );
+            }
+            indexTotalDocCounts.put(indexName, indexTotalDocs);
+        }
+        
+        // Second pass: emit replica lag metrics (now that we have primary doc counts)
+        for (Map.Entry<String, Map<String, Set<String>>> indexEntry : actualAllocations.entrySet()) {
+            String indexName = indexEntry.getKey();
+            Map<String, Long> shardPrimaryDocs = primaryDocCounts.get(indexName);
+            if (shardPrimaryDocs == null) {
+                continue;
+            }
+            
+            for (Map.Entry<String, Set<String>> shardEntry : indexEntry.getValue().entrySet()) {
+                String shardId = shardEntry.getKey();
+                Long primaryDocCount = shardPrimaryDocs.get(shardId);
+                if (primaryDocCount == null || primaryDocCount < 0) {
+                    continue;
+                }
+                
+                int shardIdInt = Integer.parseInt(shardId);
+                
+                for (String unitName : shardEntry.getValue()) {
+                    SearchUnitActualState actualState = actualStates.get(unitName);
+                    if (actualState == null) {
+                        continue;
+                    }
+                    
+                    // Check if this is a replica
+                    Map<String, List<SearchUnitActualState.ShardRoutingInfo>> nodeRouting = actualState.getNodeRouting();
+                    if (nodeRouting != null && nodeRouting.containsKey(indexName)) {
+                        List<SearchUnitActualState.ShardRoutingInfo> shards = nodeRouting.get(indexName);
+                        if (shards != null) {
+                            for (SearchUnitActualState.ShardRoutingInfo shard : shards) {
+                                if (shard.getShardId() == shardIdInt && !shard.isPrimary()) {
+                                    long replicaDocCount = actualState.getShardDocCount(indexName, shardIdInt);
+                                    if (replicaDocCount >= 0) {
+                                        // 7. Emit shard_replica_lag_docs
+                                        long lag = primaryDocCount - replicaDocCount;
+                                        metricsProvider.gauge(
+                                            SHARD_REPLICA_LAG_DOCS_METRIC_NAME,
+                                            lag,
+                                            buildMetricsTagsWithNode(clusterId, indexName, shardId, unitName)
+                                        );
+                                    }
+                                    break;
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        }
+        
+        // 2. Emit node_shard_count for each node per index
+        int totalNodeMetrics = 0;
+        Map<String, Integer> nodeTotalShardCounts = new HashMap<>();
+        
+        for (Map.Entry<String, Map<String, Integer>> indexEntry : nodeShardCountsByIndex.entrySet()) {
+            String indexName = indexEntry.getKey();
+            Map<String, Integer> nodeCounts = indexEntry.getValue();
+            
+            for (Map.Entry<String, Integer> nodeEntry : nodeCounts.entrySet()) {
+                String nodeName = nodeEntry.getKey();
+                int shardCount = nodeEntry.getValue();
+                String role = nodeRoles.getOrDefault(nodeName, "UNKNOWN");
+                
+                // Emit per-index node shard count
+                metricsProvider.gauge(
+                    NODE_SHARD_COUNT_METRIC_NAME,
+                    shardCount,
+                    buildNodeMetricsTagsWithIndex(clusterId, indexName, nodeName, role)
+                );
+                totalNodeMetrics++;
+                
+                // Accumulate total shards per node across all indices
+                nodeTotalShardCounts.merge(nodeName, shardCount, Integer::sum);
+            }
+        }
+        
+        // 3. Emit node_total_shard_count for aggregate node load
+        for (Map.Entry<String, Integer> nodeEntry : nodeTotalShardCounts.entrySet()) {
+            String nodeName = nodeEntry.getKey();
+            int totalShards = nodeEntry.getValue();
+            String role = nodeRoles.getOrDefault(nodeName, "UNKNOWN");
+            
+            metricsProvider.gauge(
+                NODE_TOTAL_SHARD_COUNT_METRIC_NAME,
+                totalShards,
+                buildNodeMetricsTags(clusterId, nodeName, role)
+            );
+        }
+        
+        log.info("ActualAllocationUpdater - Emitted metrics for {} indices, {} node-index combinations, {} total nodes", 
+            actualAllocations.size(), totalNodeMetrics, nodeTotalShardCounts.size());
+    }
+    
 }
diff --git a/src/main/java/io/clustercontroller/metrics/MetricsConstants.java b/src/main/java/io/clustercontroller/metrics/MetricsConstants.java
index a394c4b..bdf4e6b 100644
--- a/src/main/java/io/clustercontroller/metrics/MetricsConstants.java
+++ b/src/main/java/io/clustercontroller/metrics/MetricsConstants.java
@@ -4,16 +4,41 @@
  * Constants for metrics names and tags used in the Cluster Controller.
  */
 public class MetricsConstants {
+    // Existing metrics
     public final static String ROLLING_UPDATE_PROGRESS_PERCENTAGE_METRIC_NAME = "rolling_update_progress_percentage";
     public final static String ROLLING_UPDATE_TRANSIT_NODES_PERCENTAGE_METRIC_NAME = "rolling_update_transit_nodes_percentage";
     public final static String PLANNED_INGEST_SU_ALLOCATION_METRIC_NAME = "planned_ingest_su_allocation";
     public final static String PLANNED_SEARCH_SU_ALLOCATION_METRIC_NAME = "planned_search_su_allocation";
     public final static String UPDATE_ACTUAL_ALLOCATION_FAILURES_METRIC_NAME = "update_actual_allocation_failures_count";
     public final static String DISCOVERY_CLEANED_STALE_SEARCH_UNITS_COUNT_METRIC_NAME = "discovery_cleaned_stale_search_units_count";
+    
+    // Shard distribution metrics (bin-packing visualization)
+    public final static String SHARD_PRIMARY_STARTED_COUNT_METRIC_NAME = "shard_primary_started_count";
+    public final static String SHARD_REPLICA_STARTED_COUNT_METRIC_NAME = "shard_replica_started_count";
+    public final static String NODE_SHARD_COUNT_METRIC_NAME = "node_shard_count";
+    public final static String NODE_TOTAL_SHARD_COUNT_METRIC_NAME = "node_total_shard_count";
+    public final static String SHARD_TOTAL_ALLOCATION_COUNT_METRIC_NAME = "shard_total_allocation_count";
+    
+    // Doc count metrics
+    public final static String SHARD_PRIMARY_DOC_COUNT_METRIC_NAME = "shard_primary_doc_count";
+    public final static String SHARD_REPLICA_DOC_COUNT_METRIC_NAME = "shard_replica_doc_count";
+    public final static String SHARD_PRIMARY_DELETED_DOC_COUNT_METRIC_NAME = "shard_primary_deleted_doc_count";
+    
+    // Replication lag metrics
+    public final static String SHARD_REPLICA_LAG_DOCS_METRIC_NAME = "shard_replica_lag_docs";
+    
+    // Allocation convergence metrics
+    public final static String SHARD_PENDING_ALLOCATION_COUNT_METRIC_NAME = "shard_pending_allocation_count";
+    
+    // Index level metrics
+    public final static String INDEX_TOTAL_DOC_COUNT_METRIC_NAME = "index_total_doc_count";
+    
+    // Tags
     public final static String CLUSTER_ID_TAG = "clusterId";
     public final static String INDEX_NAME_TAG = "indexName";
     public final static String SHARD_ID_TAG = "shardId";
     public final static String ROLE_TAG = "role";
+    public final static String NODE_NAME_TAG = "nodeName";
 
     private MetricsConstants() {}
 }
diff --git a/src/main/java/io/clustercontroller/metrics/MetricsProvider.java b/src/main/java/io/clustercontroller/metrics/MetricsProvider.java
index f298203..1fc3f4f 100644
--- a/src/main/java/io/clustercontroller/metrics/MetricsProvider.java
+++ b/src/main/java/io/clustercontroller/metrics/MetricsProvider.java
@@ -21,10 +21,12 @@
 public class MetricsProvider {
     private static final double[] TIMER_PERCENTILES = {0.5, 0.9, 0.99};
     private static final String HOST_NAME_TAG = "hostname";
+    private static final long GAUGE_TTL_MS = 5 * 60 * 1000; // 5 minutes
 
     private final MeterRegistry registry;
     private final String hostname;
     private final Map<String, AtomicDouble> gaugeCache = new ConcurrentHashMap<>();
+    private final Map<String, Long> gaugeLastUpdated = new ConcurrentHashMap<>();
 
     @Autowired
     public MetricsProvider(
@@ -60,6 +62,10 @@ public Counter counter(String name, Map<String, String> tags) {
     public AtomicDouble gauge(String name, double value, Map<String, String> tags) {
         tags.put(HOST_NAME_TAG, hostname);
         String cacheKey = buildCacheKey(name, tags);
+        
+        // Track last update time for TTL-based cleanup
+        gaugeLastUpdated.put(cacheKey, System.currentTimeMillis());
+        
         AtomicDouble gauge = gaugeCache.computeIfAbsent(cacheKey, k -> {
             AtomicDouble gaugeValue = new AtomicDouble(value);
             Gauge.builder(name, gaugeValue::get)
@@ -117,5 +123,33 @@ private String[] mapToTagArray(Map<String, String> tags) {
         }
         return tagArray;
     }
+
+    /**
+     * Cleans up stale gauges that haven't been updated within the TTL period.
+     * Sets stale gauge values to 0 to indicate the metric is no longer active.
+     * This handles cleanup for deleted indices, shards, and nodes automatically.
+     */
+    public void cleanupStaleGauges() {
+        long now = System.currentTimeMillis();
+        int cleanedCount = 0;
+        
+        for (Map.Entry<String, Long> entry : gaugeLastUpdated.entrySet()) {
+            String cacheKey = entry.getKey();
+            long lastUpdated = entry.getValue();
+            
+            if (now - lastUpdated > GAUGE_TTL_MS) {
+                AtomicDouble gauge = gaugeCache.get(cacheKey);
+                if (gauge != null) {
+                    gauge.set(0);
+                    cleanedCount++;
+                }
+                gaugeLastUpdated.remove(cacheKey);
+            }
+        }
+        
+        if (cleanedCount > 0) {
+            log.info("Cleaned up {} stale gauges (TTL: {}ms)", cleanedCount, GAUGE_TTL_MS);
+        }
+    }
 }
 
diff --git a/src/main/java/io/clustercontroller/metrics/MetricsUtils.java b/src/main/java/io/clustercontroller/metrics/MetricsUtils.java
index 18e95e7..b146780 100644
--- a/src/main/java/io/clustercontroller/metrics/MetricsUtils.java
+++ b/src/main/java/io/clustercontroller/metrics/MetricsUtils.java
@@ -7,6 +7,7 @@
 
 import static io.clustercontroller.metrics.MetricsConstants.CLUSTER_ID_TAG;
 import static io.clustercontroller.metrics.MetricsConstants.INDEX_NAME_TAG;
+import static io.clustercontroller.metrics.MetricsConstants.NODE_NAME_TAG;
 import static io.clustercontroller.metrics.MetricsConstants.ROLE_TAG;
 import static io.clustercontroller.metrics.MetricsConstants.SHARD_ID_TAG;
 
@@ -44,4 +45,69 @@ public static Map<String, String> buildMetricsTags(String clusterId, String inde
         tags.put(SHARD_ID_TAG, shardId);
         return tags;
     }
+    
+    /**
+     * Builds a map of metrics tags including cluster ID, index name, shard ID, and node name.
+     * Used for replica-level metrics where we need to identify specific nodes.
+     *
+     * @param clusterId the cluster ID
+     * @param indexName the index name
+     * @param shardId the shard ID
+     * @param nodeName the node name
+     * @return a map of metrics tags
+     */
+    public static Map<String, String> buildMetricsTagsWithNode(String clusterId, String indexName, String shardId, String nodeName) {
+        Map<String, String> tags = buildMetricsTags(clusterId, indexName, shardId);
+        tags.put(NODE_NAME_TAG, nodeName);
+        return tags;
+    }
+    
+    /**
+     * Builds a map of metrics tags for node-level metrics.
+     *
+     * @param clusterId the cluster ID
+     * @param nodeName the node name
+     * @param role the node role
+     * @return a map of metrics tags
+     */
+    public static Map<String, String> buildNodeMetricsTags(String clusterId, String nodeName, String role) {
+        Map<String, String> tags = new HashMap<>();
+        tags.put(CLUSTER_ID_TAG, clusterId);
+        tags.put(NODE_NAME_TAG, nodeName);
+        tags.put(ROLE_TAG, role);
+        return tags;
+    }
+    
+    /**
+     * Builds a map of metrics tags for node-level metrics with index context.
+     * Used for tracking per-node per-index shard distribution.
+     *
+     * @param clusterId the cluster ID
+     * @param indexName the index name
+     * @param nodeName the node name
+     * @param role the node role
+     * @return a map of metrics tags
+     */
+    public static Map<String, String> buildNodeMetricsTagsWithIndex(String clusterId, String indexName, String nodeName, String role) {
+        Map<String, String> tags = new HashMap<>();
+        tags.put(CLUSTER_ID_TAG, clusterId);
+        tags.put(INDEX_NAME_TAG, indexName);
+        tags.put(NODE_NAME_TAG, nodeName);
+        tags.put(ROLE_TAG, role);
+        return tags;
+    }
+    
+    /**
+     * Builds a map of metrics tags for index-level metrics.
+     *
+     * @param clusterId the cluster ID
+     * @param indexName the index name
+     * @return a map of metrics tags
+     */
+    public static Map<String, String> buildIndexMetricsTags(String clusterId, String indexName) {
+        Map<String, String> tags = new HashMap<>();
+        tags.put(CLUSTER_ID_TAG, clusterId);
+        tags.put(INDEX_NAME_TAG, indexName);
+        return tags;
+    }
 }
diff --git a/src/main/java/io/clustercontroller/orchestration/RollingUpdateOrchestrationStrategy.java b/src/main/java/io/clustercontroller/orchestration/RollingUpdateOrchestrationStrategy.java
index 9c69303..be5ec2e 100644
--- a/src/main/java/io/clustercontroller/orchestration/RollingUpdateOrchestrationStrategy.java
+++ b/src/main/java/io/clustercontroller/orchestration/RollingUpdateOrchestrationStrategy.java
@@ -80,6 +80,9 @@ public void orchestrate(String clusterId) {
                 }
             }
             
+            // Cleanup stale gauges for deleted indices/shards
+            metricsProvider.cleanupStaleGauges();
+            
             log.info("Completed rolling update orchestration for cluster: {}", clusterId);
         } catch (Exception e) {
             log.error("Failed to orchestrate goal states for cluster {}: {}", clusterId, e.getMessage(), e);