From 9bbca1fc26751a6759096bfb50007d3148b225a2 Mon Sep 17 00:00:00 2001 From: Krina Patel Date: Thu, 18 Dec 2025 13:20:48 -0500 Subject: [PATCH] Add index readiness API for marking index as READY --- .../api/handlers/IndexHandler.java | 30 +++- .../responses/IndexReadinessResponse.java | 41 ++++++ .../indices/IndexManager.java | 12 ++ .../models/SearchUnitActualState.java | 128 +++++++++++++++++- .../store/EtcdMetadataStore.java | 91 ++++++++++++- .../store/MetadataStore.java | 13 ++ 6 files changed, 312 insertions(+), 3 deletions(-) create mode 100644 src/main/java/io/clustercontroller/api/models/responses/IndexReadinessResponse.java diff --git a/src/main/java/io/clustercontroller/api/handlers/IndexHandler.java b/src/main/java/io/clustercontroller/api/handlers/IndexHandler.java index 701eefc..29c57a8 100644 --- a/src/main/java/io/clustercontroller/api/handlers/IndexHandler.java +++ b/src/main/java/io/clustercontroller/api/handlers/IndexHandler.java @@ -2,6 +2,7 @@ import io.clustercontroller.api.models.requests.IndexRequest; import io.clustercontroller.api.models.responses.ErrorResponse; +import io.clustercontroller.api.models.responses.IndexReadinessResponse; import io.clustercontroller.api.models.responses.IndexResponse; import io.clustercontroller.indices.IndexManager; import com.fasterxml.jackson.databind.ObjectMapper; @@ -206,7 +207,7 @@ public ResponseEntity updateIndexMapping( } } - /** +/** * Update index metadata in the specified cluster. * PUT /{clusterId}/{index}/_metadata */ @@ -231,4 +232,31 @@ public ResponseEntity updateIndexMetadata( return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(ErrorResponse.internalError(e.getMessage())); } } + + /** + * Check if an index is ready (shards have docs ingested and are replicated). + * GET /{clusterId}/{index}/_ready + * + * This endpoint is designed to be polled by workflows (e.g., Cadence) to determine + * when an index is ready to serve traffic after creation and ingestion. + */ + @GetMapping("/{index}/_ready") + public ResponseEntity isIndexReady( + @PathVariable String clusterId, + @PathVariable String index) { + try { + log.info("Checking readiness for index '{}' in cluster '{}'", index, clusterId); + boolean isReady = indexManager.isIndexReady(clusterId, index); + + if (isReady) { + return ResponseEntity.ok(IndexReadinessResponse.ready(index)); + } else { + return ResponseEntity.ok(IndexReadinessResponse.notReady(index, + "Index is not ready - waiting for docs to be ingested and replication to complete")); + } + } catch (Exception e) { + log.error("Error checking readiness for index '{}' in cluster '{}': {}", index, clusterId, e.getMessage()); + return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(ErrorResponse.internalError(e.getMessage())); + } + } } \ No newline at end of file diff --git a/src/main/java/io/clustercontroller/api/models/responses/IndexReadinessResponse.java b/src/main/java/io/clustercontroller/api/models/responses/IndexReadinessResponse.java new file mode 100644 index 0000000..1a19b6a --- /dev/null +++ b/src/main/java/io/clustercontroller/api/models/responses/IndexReadinessResponse.java @@ -0,0 +1,41 @@ +package io.clustercontroller.api.models.responses; + +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.databind.PropertyNamingStrategies; +import com.fasterxml.jackson.databind.annotation.JsonNaming; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +/** + * Response model for index readiness check. + */ +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +@JsonInclude(JsonInclude.Include.NON_NULL) +@JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy.class) +public class IndexReadinessResponse { + private String index; + private boolean ready; + private String message; + + public static IndexReadinessResponse ready(String indexName) { + return IndexReadinessResponse.builder() + .index(indexName) + .ready(true) + .message("Index is ready!") + .build(); + } + + public static IndexReadinessResponse notReady(String indexName, String reason) { + return IndexReadinessResponse.builder() + .index(indexName) + .ready(false) + .message(reason) + .build(); + } +} + diff --git a/src/main/java/io/clustercontroller/indices/IndexManager.java b/src/main/java/io/clustercontroller/indices/IndexManager.java index 0ccca34..e003002 100644 --- a/src/main/java/io/clustercontroller/indices/IndexManager.java +++ b/src/main/java/io/clustercontroller/indices/IndexManager.java @@ -791,6 +791,18 @@ private void deepMerge(Map target, Map source) { } } + /** + * Check if an index is ready (all shards have docs ingested and are replicated). + * + * @param clusterId the cluster identifier + * @param indexName the name of the index to check + * @return true if all shards have docs > 0 and are replicated + */ + public boolean isIndexReady(String clusterId, String indexName) throws Exception { + log.info("Checking if index '{}' is ready in cluster '{}'", indexName, clusterId); + return metadataStore.isIndexReady(clusterId, indexName); + } + /** * Data class to hold parsed create index request */ diff --git a/src/main/java/io/clustercontroller/models/SearchUnitActualState.java b/src/main/java/io/clustercontroller/models/SearchUnitActualState.java index 25a8e16..4c6fd2c 100644 --- a/src/main/java/io/clustercontroller/models/SearchUnitActualState.java +++ b/src/main/java/io/clustercontroller/models/SearchUnitActualState.java @@ -82,7 +82,9 @@ public class SearchUnitActualState { @JsonProperty("cluster_name") private String clusterName; // "search-cluster", "analytics-cluster", etc. - + // Stats reported by the dataplane (doc counts, etc.) + @JsonProperty("stats") + private IndicesStats stats; public SearchUnitActualState() { this.nodeRouting = new HashMap<>(); @@ -181,4 +183,128 @@ public boolean isPrimary() { return role != null && Constants.ROLE_PRIMARY.equalsIgnoreCase(role); } } + + /** + * Stats reported by the dataplane containing index and shard level metrics + */ + @Data + @JsonIgnoreProperties(ignoreUnknown = true) + public static class IndicesStats { + @JsonProperty("indices") + private IndicesContainer indices; + } + + /** + * Container for index-level stats + */ + @Data + @JsonIgnoreProperties(ignoreUnknown = true) + public static class IndicesContainer { + @JsonProperty("docs") + private DocsStats docs; + + @JsonProperty("shards") + private Map>> shards; // indexName -> list of shard stats + } + + /** + * Document count statistics + */ + @Data + @JsonIgnoreProperties(ignoreUnknown = true) + public static class DocsStats { + @JsonProperty("count") + private long count; + + @JsonProperty("deleted") + private long deleted; + } + + /** + * Shard-level statistics including docs and sequence numbers + */ + @Data + @JsonIgnoreProperties(ignoreUnknown = true) + public static class ShardLevelStats { + @JsonProperty("docs") + private DocsStats docs; + + @JsonProperty("seq_no") + private SeqNoStats seqNo; + } + + /** + * Sequence number stats for tracking replication progress + */ + @Data + @JsonIgnoreProperties(ignoreUnknown = true) + public static class SeqNoStats { + @JsonProperty("max_seq_no") + private long maxSeqNo; + + @JsonProperty("local_checkpoint") + private long localCheckpoint; + + @JsonProperty("global_checkpoint") + private long globalCheckpoint; + } + + // ========== UTILITY METHODS FOR STATS ========== + + /** + * Get the document count for a specific index and shard from the stats. + * + * @param indexName the name of the index + * @param shardId the shard ID + * @return the document count, or -1 if not found + */ + public long getShardDocCount(String indexName, int shardId) { + if (stats == null || stats.getIndices() == null || stats.getIndices().getShards() == null) { + return -1; + } + + List> indexShards = stats.getIndices().getShards().get(indexName); + if (indexShards == null) { + return -1; + } + + String shardIdStr = String.valueOf(shardId); + for (Map shardMap : indexShards) { + ShardLevelStats shardStats = shardMap.get(shardIdStr); + if (shardStats != null && shardStats.getDocs() != null) { + return shardStats.getDocs().getCount(); + } + } + + return -1; + } + + /** + * Check if a shard's data is fully replicated (global checkpoint equals local checkpoint). + * + * @param indexName the name of the index + * @param shardId the shard ID + * @return true if replicated, false if not or if data not available + */ + public boolean isShardReplicated(String indexName, int shardId) { + if (stats == null || stats.getIndices() == null || stats.getIndices().getShards() == null) { + return false; + } + + List> indexShards = stats.getIndices().getShards().get(indexName); + if (indexShards == null) { + return false; + } + + String shardIdStr = String.valueOf(shardId); + for (Map shardMap : indexShards) { + ShardLevelStats shardStats = shardMap.get(shardIdStr); + if (shardStats != null && shardStats.getSeqNo() != null) { + SeqNoStats seqNo = shardStats.getSeqNo(); + return seqNo.getGlobalCheckpoint() == seqNo.getLocalCheckpoint(); + } + } + + return false; + } } \ No newline at end of file diff --git a/src/main/java/io/clustercontroller/store/EtcdMetadataStore.java b/src/main/java/io/clustercontroller/store/EtcdMetadataStore.java index c0b6d18..1e13194 100644 --- a/src/main/java/io/clustercontroller/store/EtcdMetadataStore.java +++ b/src/main/java/io/clustercontroller/store/EtcdMetadataStore.java @@ -1397,7 +1397,7 @@ public void deletePrefix(String clusterId, String prefix) throws Exception { } } - /** +/** * Get the controller ID assigned to a cluster. */ @Override @@ -1456,4 +1456,93 @@ public ClusterInformation.Version getClusterVersion(String clusterId) throws Exc throw e; } } + + // ================================================================= + // INDEX READINESS OPERATIONS + // ================================================================= + + @Override + public boolean isIndexReady(String clusterId, String indexName) throws Exception { + log.debug("Checking if index {} is ready in cluster {}", indexName, clusterId); + + try { + // 1. Get expected shard count from index config + Index indexConfig = getAllIndexConfigs(clusterId).stream() + .filter(i -> indexName.equals(i.getIndexName())) + .findFirst() + .orElse(null); + + if (indexConfig == null) { + log.debug("Index {} not found in cluster {}, not ready", indexName, clusterId); + return false; + } + + int expectedShardCount = indexConfig.getSettings().getNumberOfShards(); + log.debug("Index {} expects {} shards", indexName, expectedShardCount); + + // 2. Get all actual states from all nodes + Map allActualStates = getAllSearchUnitActualStates(clusterId); + + if (allActualStates.isEmpty()) { + log.debug("No search units found for cluster {}, index {} is not ready", clusterId, indexName); + return false; + } + + // 3. Track shard readiness: need docs > 0 + replicated + // (STARTED is implied if docs are ingested and replicated) + Set readyShardIds = new HashSet<>(); + + for (Map.Entry entry : allActualStates.entrySet()) { + String unitName = entry.getKey(); + SearchUnitActualState actualState = entry.getValue(); + + // Check each shard in the stats + for (int shardId = 0; shardId < expectedShardCount; shardId++) { + // Already found a ready copy of this shard on another node + if (readyShardIds.contains(shardId)) { + continue; + } + + // Check 1: Docs are ingested (docCount > 0) + long docCount = actualState.getShardDocCount(indexName, shardId); + if (docCount <= 0) { + continue; + } + + // Check 2: Data is replicated (global_checkpoint == local_checkpoint) + boolean isReplicated = actualState.isShardReplicated(indexName, shardId); + if (!isReplicated) { + log.debug("Index {} shard {} on node {} has docs but not fully replicated", + indexName, shardId, unitName); + continue; + } + + // Shard passes all checks + readyShardIds.add(shardId); + log.debug("Index {} shard {} is ready on node {} (docs={}, replicated)", + indexName, shardId, unitName, docCount); + } + } + + // 4. Check all expected shards are ready + if (readyShardIds.size() < expectedShardCount) { + Set notReadyShards = new HashSet<>(); + for (int i = 0; i < expectedShardCount; i++) { + if (!readyShardIds.contains(i)) { + notReadyShards.add(i); + } + } + log.debug("Index {} is not ready - shards not ready: {}", indexName, notReadyShards); + return false; + } + + log.info("Index {} is ready - all {} shards have docs ingested and are replicated", + indexName, expectedShardCount); + return true; + + } catch (Exception e) { + log.error("Failed to check if index {} is ready: {}", indexName, e.getMessage(), e); + throw new Exception("Failed to check index readiness", e); + } + } } diff --git a/src/main/java/io/clustercontroller/store/MetadataStore.java b/src/main/java/io/clustercontroller/store/MetadataStore.java index eaca7d1..4fcccd8 100644 --- a/src/main/java/io/clustercontroller/store/MetadataStore.java +++ b/src/main/java/io/clustercontroller/store/MetadataStore.java @@ -276,6 +276,19 @@ public interface MetadataStore { */ List getAllAliases(String clusterId) throws Exception; + // ================================================================= + // INDEX READINESS OPERATIONS + // ================================================================= + + /** + * Check if an index is ready (shards have docs ingested and are replicated). + * + * @param clusterId the cluster identifier + * @param indexName the name of the index to check + * @return true if shards have docs > 0 and global_checkpoint == local_checkpoint + */ + boolean isIndexReady(String clusterId, String indexName) throws Exception; + // ================================================================= // CLUSTER OPERATIONS // =================================================================