From 6558fd8b37144d3b2880510ffda976e724bb1320 Mon Sep 17 00:00:00 2001 From: tjnaik Date: Wed, 28 Jan 2026 23:56:49 -0800 Subject: [PATCH 1/7] Add shard distribution and doc count metrics --- .../allocation/ActualAllocationUpdater.java | 252 +++++++++++++++++- .../metrics/MetricsConstants.java | 20 ++ .../metrics/MetricsUtils.java | 47 ++++ 3 files changed, 317 insertions(+), 2 deletions(-) diff --git a/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java b/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java index 3414e23..00f80d1 100644 --- a/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java +++ b/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java @@ -21,8 +21,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 +87,9 @@ 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); + log.info("ActualAllocationUpdater - Completed actual allocation update with {} updates and {} coordinator goal state updates", totalUpdates, coordinatorUpdates); } @@ -662,4 +665,249 @@ 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 searchUnits, + Map>> actualAllocations) { + log.debug("ActualAllocationUpdater - Emitting shard distribution metrics for cluster {}", clusterId); + + // Map to track shard count per node: nodeName -> count + Map nodeShardCounts = new HashMap<>(); + // Map to track node roles: nodeName -> role + Map nodeRoles = new HashMap<>(); + // Map to track primary doc counts per shard: indexName -> shardId -> docCount + Map> primaryDocCounts = new HashMap<>(); + // Map to track index total doc counts: indexName -> totalDocCount + Map indexTotalDocCounts = new HashMap<>(); + + // First pass: collect all actual states and build data structures + Map 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>> indexEntry : actualAllocations.entrySet()) { + String indexName = indexEntry.getKey(); + long indexTotalDocs = 0; + + for (Map.Entry> shardEntry : indexEntry.getValue().entrySet()) { + String shardId = shardEntry.getKey(); + Set allocatedUnits = shardEntry.getValue(); + + // 1. Emit shard_replica_count - number of nodes hosting this shard in STARTED state + int replicaCount = allocatedUnits.size(); + metricsProvider.gauge( + SHARD_REPLICA_COUNT_METRIC_NAME, + replicaCount, + buildMetricsTags(clusterId, indexName, shardId) + ); + + // 3. Emit shard_total_allocation_count - get from planned allocation + 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) + ); + } + } 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 + nodeShardCounts.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> nodeRouting = actualState.getNodeRouting(); + if (nodeRouting != null && nodeRouting.containsKey(indexName)) { + List 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> indexShards = + actualState.getStats().getIndices().getShards().get(indexName); + if (indexShards != null) { + for (Map shardMap : indexShards) { + SearchUnitActualState.ShardLevelStats stats = shardMap.get(shardId); + if (stats != null && stats.getDocs() != null) { + primaryDeletedDocCount = stats.getDocs().getDeleted(); + } + } + } + } + + // 4. Emit shard_primary_doc_count + if (primaryDocCount >= 0) { + metricsProvider.gauge( + SHARD_PRIMARY_DOC_COUNT_METRIC_NAME, + primaryDocCount, + buildMetricsTags(clusterId, indexName, shardId) + ); + + // 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 + if (primaryDeletedDocCount >= 0) { + metricsProvider.gauge( + SHARD_PRIMARY_DELETED_DOC_COUNT_METRIC_NAME, + primaryDeletedDocCount, + buildMetricsTags(clusterId, indexName, shardId) + ); + } + } 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>> indexEntry : actualAllocations.entrySet()) { + String indexName = indexEntry.getKey(); + Map shardPrimaryDocs = primaryDocCounts.get(indexName); + if (shardPrimaryDocs == null) { + continue; + } + + for (Map.Entry> 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> nodeRouting = actualState.getNodeRouting(); + if (nodeRouting != null && nodeRouting.containsKey(indexName)) { + List 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 + for (Map.Entry nodeEntry : nodeShardCounts.entrySet()) { + String nodeName = nodeEntry.getKey(); + int shardCount = nodeEntry.getValue(); + String role = nodeRoles.getOrDefault(nodeName, "UNKNOWN"); + + metricsProvider.gauge( + NODE_SHARD_COUNT_METRIC_NAME, + shardCount, + buildNodeMetricsTags(clusterId, nodeName, role) + ); + } + + log.info("ActualAllocationUpdater - Emitted metrics for {} indices, {} nodes", + actualAllocations.size(), nodeShardCounts.size()); + } + } diff --git a/src/main/java/io/clustercontroller/metrics/MetricsConstants.java b/src/main/java/io/clustercontroller/metrics/MetricsConstants.java index a394c4b..0b87f0b 100644 --- a/src/main/java/io/clustercontroller/metrics/MetricsConstants.java +++ b/src/main/java/io/clustercontroller/metrics/MetricsConstants.java @@ -4,16 +4,36 @@ * 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_REPLICA_COUNT_METRIC_NAME = "shard_replica_count"; + public final static String NODE_SHARD_COUNT_METRIC_NAME = "node_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"; + + // 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/MetricsUtils.java b/src/main/java/io/clustercontroller/metrics/MetricsUtils.java index 18e95e7..d56d3f0 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,50 @@ public static Map 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 buildMetricsTagsWithNode(String clusterId, String indexName, String shardId, String nodeName) { + Map 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 buildNodeMetricsTags(String clusterId, String nodeName, String role) { + Map 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 index-level metrics. + * + * @param clusterId the cluster ID + * @param indexName the index name + * @return a map of metrics tags + */ + public static Map buildIndexMetricsTags(String clusterId, String indexName) { + Map tags = new HashMap<>(); + tags.put(CLUSTER_ID_TAG, clusterId); + tags.put(INDEX_NAME_TAG, indexName); + return tags; + } } From e8b6288dea4a9ae9ec1b6312eb6550392af5bf03 Mon Sep 17 00:00:00 2001 From: tjnaik Date: Thu, 29 Jan 2026 00:02:55 -0800 Subject: [PATCH 2/7] Add shard_pending_allocation_count metric --- .../allocation/ActualAllocationUpdater.java | 10 +++++++++- .../io/clustercontroller/metrics/MetricsConstants.java | 3 +++ 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java b/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java index 00f80d1..00ed467 100644 --- a/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java +++ b/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java @@ -730,7 +730,7 @@ private void emitShardDistributionMetrics(String clusterId, List sea buildMetricsTags(clusterId, indexName, shardId) ); - // 3. Emit shard_total_allocation_count - get from planned allocation + // 3. Emit shard_total_allocation_count and shard_pending_allocation_count try { ShardAllocation plannedAllocation = metadataStore.getPlannedAllocation(clusterId, indexName, shardId); if (plannedAllocation != null) { @@ -740,6 +740,14 @@ private void emitShardDistributionMetrics(String clusterId, List sea totalPlanned, buildMetricsTags(clusterId, indexName, shardId) ); + + // Emit pending allocation count (planned - started) + int pendingCount = Math.max(0, totalPlanned - replicaCount); + 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()); diff --git a/src/main/java/io/clustercontroller/metrics/MetricsConstants.java b/src/main/java/io/clustercontroller/metrics/MetricsConstants.java index 0b87f0b..2ac4f6d 100644 --- a/src/main/java/io/clustercontroller/metrics/MetricsConstants.java +++ b/src/main/java/io/clustercontroller/metrics/MetricsConstants.java @@ -25,6 +25,9 @@ public class MetricsConstants { // 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"; From 36f297e96e57972c7431d778d96aa3d63104e678 Mon Sep 17 00:00:00 2001 From: tjnaik Date: Thu, 29 Jan 2026 04:43:41 -0800 Subject: [PATCH 3/7] Add TTL-based cleanup for stale gauge metrics --- .../allocation/ActualAllocationUpdater.java | 3 ++ .../metrics/MetricsProvider.java | 34 +++++++++++++++++++ .../RollingUpdateOrchestrationStrategy.java | 3 ++ 3 files changed, 40 insertions(+) diff --git a/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java b/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java index 00ed467..d3d1129 100644 --- a/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java +++ b/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java @@ -90,6 +90,9 @@ public void updateActualAllocations(String clusterId) throws Exception { // 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); } 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 gaugeCache = new ConcurrentHashMap<>(); + private final Map gaugeLastUpdated = new ConcurrentHashMap<>(); @Autowired public MetricsProvider( @@ -60,6 +62,10 @@ public Counter counter(String name, Map tags) { public AtomicDouble gauge(String name, double value, Map 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 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 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/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); From 39f12621ddbcfad842f30eff5a08842e8889b405 Mon Sep 17 00:00:00 2001 From: tjnaik Date: Thu, 29 Jan 2026 05:12:03 -0800 Subject: [PATCH 4/7] Add node_total_shard_count metric and include indexname in node_shard_count --- .../allocation/ActualAllocationUpdater.java | 50 +++++++++++++++---- .../metrics/MetricsConstants.java | 1 + .../metrics/MetricsUtils.java | 19 +++++++ 3 files changed, 59 insertions(+), 11 deletions(-) diff --git a/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java b/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java index d3d1129..65cf0b5 100644 --- a/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java +++ b/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java @@ -685,8 +685,8 @@ private void emitShardDistributionMetrics(String clusterId, List sea Map>> actualAllocations) { log.debug("ActualAllocationUpdater - Emitting shard distribution metrics for cluster {}", clusterId); - // Map to track shard count per node: nodeName -> count - Map nodeShardCounts = new HashMap<>(); + // Map to track shard count per node per index: indexName -> (nodeName -> count) + Map> nodeShardCountsByIndex = new HashMap<>(); // Map to track node roles: nodeName -> role Map nodeRoles = new HashMap<>(); // Map to track primary doc counts per shard: indexName -> shardId -> docCount @@ -762,8 +762,10 @@ private void emitShardDistributionMetrics(String clusterId, List sea int shardIdInt = Integer.parseInt(shardId); for (String unitName : allocatedUnits) { - // Track shard count per node - nodeShardCounts.merge(unitName, 1, Integer::sum); + // 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) { @@ -904,21 +906,47 @@ private void emitShardDistributionMetrics(String clusterId, List sea } } - // 2. Emit node_shard_count for each node - for (Map.Entry nodeEntry : nodeShardCounts.entrySet()) { + // 2. Emit node_shard_count for each node per index + int totalNodeMetrics = 0; + Map nodeTotalShardCounts = new HashMap<>(); + + for (Map.Entry> indexEntry : nodeShardCountsByIndex.entrySet()) { + String indexName = indexEntry.getKey(); + Map nodeCounts = indexEntry.getValue(); + + for (Map.Entry 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 nodeEntry : nodeTotalShardCounts.entrySet()) { String nodeName = nodeEntry.getKey(); - int shardCount = nodeEntry.getValue(); + int totalShards = nodeEntry.getValue(); String role = nodeRoles.getOrDefault(nodeName, "UNKNOWN"); metricsProvider.gauge( - NODE_SHARD_COUNT_METRIC_NAME, - shardCount, + NODE_TOTAL_SHARD_COUNT_METRIC_NAME, + totalShards, buildNodeMetricsTags(clusterId, nodeName, role) ); } - log.info("ActualAllocationUpdater - Emitted metrics for {} indices, {} nodes", - actualAllocations.size(), nodeShardCounts.size()); + 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 2ac4f6d..4e84ad3 100644 --- a/src/main/java/io/clustercontroller/metrics/MetricsConstants.java +++ b/src/main/java/io/clustercontroller/metrics/MetricsConstants.java @@ -15,6 +15,7 @@ public class MetricsConstants { // Shard distribution metrics (bin-packing visualization) public final static String SHARD_REPLICA_COUNT_METRIC_NAME = "shard_replica_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 diff --git a/src/main/java/io/clustercontroller/metrics/MetricsUtils.java b/src/main/java/io/clustercontroller/metrics/MetricsUtils.java index d56d3f0..b146780 100644 --- a/src/main/java/io/clustercontroller/metrics/MetricsUtils.java +++ b/src/main/java/io/clustercontroller/metrics/MetricsUtils.java @@ -78,6 +78,25 @@ public static Map buildNodeMetricsTags(String clusterId, String 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 buildNodeMetricsTagsWithIndex(String clusterId, String indexName, String nodeName, String role) { + Map 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. * From 751e78aaf0e61cbfa23ddc5c7a80d518ae420725 Mon Sep 17 00:00:00 2001 From: tjnaik Date: Thu, 29 Jan 2026 16:37:51 -0800 Subject: [PATCH 5/7] Add nodeName tag to primary doc count metrics for consistency --- .../allocation/ActualAllocationUpdater.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java b/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java index 65cf0b5..669e058 100644 --- a/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java +++ b/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java @@ -802,12 +802,12 @@ private void emitShardDistributionMetrics(String clusterId, List sea } } - // 4. Emit shard_primary_doc_count + // 4. Emit shard_primary_doc_count (tagged with node name for consistency) if (primaryDocCount >= 0) { metricsProvider.gauge( SHARD_PRIMARY_DOC_COUNT_METRIC_NAME, primaryDocCount, - buildMetricsTags(clusterId, indexName, shardId) + buildMetricsTagsWithNode(clusterId, indexName, shardId, unitName) ); // Store for lag calculation @@ -819,12 +819,12 @@ private void emitShardDistributionMetrics(String clusterId, List sea indexTotalDocs += primaryDocCount; } - // 6. Emit shard_primary_deleted_doc_count + // 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, - buildMetricsTags(clusterId, indexName, shardId) + buildMetricsTagsWithNode(clusterId, indexName, shardId, unitName) ); } } else { From 19578d9d0b1cb3d6cc87053f3f04861732f778a4 Mon Sep 17 00:00:00 2001 From: tjnaik Date: Fri, 30 Jan 2026 12:35:28 -0800 Subject: [PATCH 6/7] Split shard_replica_count into primary and replica started counts --- .../allocation/ActualAllocationUpdater.java | 28 ++++++++++++++++--- .../metrics/MetricsConstants.java | 3 +- 2 files changed, 26 insertions(+), 5 deletions(-) diff --git a/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java b/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java index 669e058..d65235f 100644 --- a/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java +++ b/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java @@ -725,10 +725,29 @@ private void emitShardDistributionMetrics(String clusterId, List sea String shardId = shardEntry.getKey(); Set allocatedUnits = shardEntry.getValue(); - // 1. Emit shard_replica_count - number of nodes hosting this shard in STARTED state - int replicaCount = allocatedUnits.size(); + // 1. Count primary and search nodes separately + int primaryCount = 0; + int searchCount = 0; + for (String unitName : allocatedUnits) { + String role = nodeRoles.get(unitName); + if ("PRIMARY".equals(role) || "INGEST".equals(role)) { + primaryCount++; + } else if ("SEARCH".equals(role)) { + 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_COUNT_METRIC_NAME, + SHARD_REPLICA_STARTED_COUNT_METRIC_NAME, replicaCount, buildMetricsTags(clusterId, indexName, shardId) ); @@ -745,7 +764,8 @@ private void emitShardDistributionMetrics(String clusterId, List sea ); // Emit pending allocation count (planned - started) - int pendingCount = Math.max(0, totalPlanned - replicaCount); + int totalStarted = primaryCount + replicaCount; + int pendingCount = Math.max(0, totalPlanned - totalStarted); metricsProvider.gauge( SHARD_PENDING_ALLOCATION_COUNT_METRIC_NAME, pendingCount, diff --git a/src/main/java/io/clustercontroller/metrics/MetricsConstants.java b/src/main/java/io/clustercontroller/metrics/MetricsConstants.java index 4e84ad3..bdf4e6b 100644 --- a/src/main/java/io/clustercontroller/metrics/MetricsConstants.java +++ b/src/main/java/io/clustercontroller/metrics/MetricsConstants.java @@ -13,7 +13,8 @@ public class MetricsConstants { 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_REPLICA_COUNT_METRIC_NAME = "shard_replica_count"; + 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"; From 6482ed86c69d3975dc99fad835c0d66cae80d21b Mon Sep 17 00:00:00 2001 From: tjnaik Date: Fri, 30 Jan 2026 12:42:59 -0800 Subject: [PATCH 7/7] Refactor to use NodeRole enum for role comparison --- .../allocation/ActualAllocationUpdater.java | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java b/src/main/java/io/clustercontroller/allocation/ActualAllocationUpdater.java index d65235f..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; @@ -725,14 +726,15 @@ private void emitShardDistributionMetrics(String clusterId, List sea String shardId = shardEntry.getKey(); Set allocatedUnits = shardEntry.getValue(); - // 1. Count primary and search nodes separately + // 1. Count primary and search nodes separately using NodeRole enum int primaryCount = 0; int searchCount = 0; for (String unitName : allocatedUnits) { - String role = nodeRoles.get(unitName); - if ("PRIMARY".equals(role) || "INGEST".equals(role)) { + String roleStr = nodeRoles.get(unitName); + NodeRole role = NodeRole.fromString(roleStr); + if (role == NodeRole.PRIMARY) { primaryCount++; - } else if ("SEARCH".equals(role)) { + } else if (role == NodeRole.REPLICA) { searchCount++; } }