Add index readiness API for marking index as READY

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<Object> updateIndexMapping(
         }
     }
 
-      /**
+/**
      * Update index metadata in the specified cluster.
      * PUT /{clusterId}/{index}/_metadata
      */
@@ -231,4 +232,31 @@ public ResponseEntity<Object> 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<Object> 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()));
+        }
+    }
 }

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();
+    }
+}
+

src/main/java/io/clustercontroller/indices/IndexManager.java

@@ -791,6 +791,18 @@ private void deepMerge(Map<String, Object> target, Map<String, Object> 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
      */

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<String, List<Map<String, ShardLevelStats>>> 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<Map<String, ShardLevelStats>> indexShards = stats.getIndices().getShards().get(indexName);
+        if (indexShards == null) {
+            return -1;
+        }
+
+        String shardIdStr = String.valueOf(shardId);
+        for (Map<String, ShardLevelStats> 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<Map<String, ShardLevelStats>> indexShards = stats.getIndices().getShards().get(indexName);
+        if (indexShards == null) {
+            return false;
+        }
+
+        String shardIdStr = String.valueOf(shardId);
+        for (Map<String, ShardLevelStats> shardMap : indexShards) {
+            ShardLevelStats shardStats = shardMap.get(shardIdStr);
+            if (shardStats != null && shardStats.getSeqNo() != null) {
+                SeqNoStats seqNo = shardStats.getSeqNo();
+                return seqNo.getGlobalCheckpoint() == seqNo.getLocalCheckpoint();
+            }
+        }
+
+        return false;
+    }
 } 

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<String, SearchUnitActualState> 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<Integer> readyShardIds = new HashSet<>();
+
+            for (Map.Entry<String, SearchUnitActualState> 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<Integer> 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);
+        }
+    }
 }

src/main/java/io/clustercontroller/store/MetadataStore.java

@@ -276,6 +276,19 @@ public interface MetadataStore {
      */
     List<Alias> 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
     // =================================================================