feat: Add admin reindex API to rebuild ES indices from database

core/src/main/java/com/netflix/conductor/dao/ExecutionDAO.java

@@ -195,6 +195,41 @@ List<WorkflowModel> getWorkflowsByCorrelationId(
      */
     boolean canSearchAcrossWorkflows();
 
+    /**
+     * Get all workflow ids with pagination. Used for reindexing.
+     *
+     * @param offset start offset
+     * @param limit max number of ids to return
+     * @return list of workflow ids
+     */
+    default List<String> getAllWorkflowIds(int offset, int limit) {
+        throw new UnsupportedOperationException(
+                getClass().getSimpleName() + " does not support getAllWorkflowIds");
+    }
+
+    /**
+     * Get the total count of workflows. Used for reindex progress tracking.
+     *
+     * @return total number of workflows
+     */
+    default long getWorkflowCount() {
+        throw new UnsupportedOperationException(
+                getClass().getSimpleName() + " does not support getWorkflowCount");
+    }
+
+    /**
+     * Return workflow ids whose workflow_id is strictly greater than {@code cursor}, ordered by
+     * workflow_id, up to {@code limit} entries. Pass an empty string as the initial cursor.
+     *
+     * @param cursor exclusive lower bound (empty string to start from the beginning)
+     * @param limit max number of ids to return
+     * @return list of workflow ids
+     */
+    default List<String> getAllWorkflowIdsAfter(String cursor, int limit) {
+        throw new UnsupportedOperationException(
+                getClass().getSimpleName() + " does not support getAllWorkflowIdsAfter");
+    }
+
     // Events
 
     /**

core/src/main/java/com/netflix/conductor/service/AdminService.java

@@ -69,4 +69,18 @@ boolean verifyAndRepairWorkflowConsistency(
      * @return map of event queues
      */
     Map<String, ?> getEventQueues(boolean verbose);
+
+    /**
+     * Start an async reindex job. Returns immediately. Use getReindexStatus() to track progress.
+     *
+     * @return initial status map
+     */
+    Map<String, Object> startReindex();
+
+    /**
+     * Get the current reindex job status and progress.
+     *
+     * @return status map with state, processed, errors, total fields
+     */
+    Map<String, Object> getReindexStatus();
 }

core/src/main/java/com/netflix/conductor/service/AdminServiceImpl.java

@@ -17,41 +17,90 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.info.BuildProperties;
 import org.springframework.stereotype.Service;
 
 import com.netflix.conductor.annotations.Audit;
 import com.netflix.conductor.annotations.Trace;
 import com.netflix.conductor.common.metadata.tasks.Task;
+import com.netflix.conductor.common.run.TaskSummary;
+import com.netflix.conductor.common.run.WorkflowSummary;
 import com.netflix.conductor.core.config.ConductorProperties;
 import com.netflix.conductor.core.events.EventQueueManager;
 import com.netflix.conductor.core.reconciliation.WorkflowRepairService;
 import com.netflix.conductor.core.utils.Utils;
+import com.netflix.conductor.dao.ExecutionDAO;
+import com.netflix.conductor.dao.IndexDAO;
 import com.netflix.conductor.dao.QueueDAO;
+import com.netflix.conductor.model.TaskModel;
+import com.netflix.conductor.model.WorkflowModel;
+
+import com.google.common.util.concurrent.RateLimiter;
 
 @Audit
 @Trace
 @Service
 public class AdminServiceImpl implements AdminService {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(AdminServiceImpl.class);
+
+    // ---- reindex state ----
+    private enum ReindexState {
+        IDLE,
+        RUNNING,
+        COMPLETED,
+        FAILED
+    }
+
+    private final AtomicReference<ReindexState> reindexState =
+            new AtomicReference<>(ReindexState.IDLE);
+    private final AtomicLong reindexProcessed = new AtomicLong(0);
+    private final AtomicLong reindexErrors = new AtomicLong(0);
+    private final AtomicLong reindexTotal = new AtomicLong(0);
+    private volatile String reindexMessage = "";
+    private volatile RateLimiter reindexRateLimiter = null;
+    private final ExecutorService reindexExecutor =
+            Executors.newSingleThreadExecutor(
+                    r -> {
+                        Thread t = new Thread(r, "reindex-worker");
+                        t.setDaemon(true);
+                        return t;
+                    });
+
     private final ConductorProperties properties;
     private final ExecutionService executionService;
     private final QueueDAO queueDAO;
+    private final ExecutionDAO executionDAO;
+    private final IndexDAO indexDAO;
     private final WorkflowRepairService workflowRepairService;
     private final EventQueueManager eventQueueManager;
     private final BuildProperties buildProperties;
 
+    @Value("${conductor.reindex.rateLimitPerSecond:0}")
+    private double reindexRateLimitPerSecond;
+
     public AdminServiceImpl(
             ConductorProperties properties,
             ExecutionService executionService,
             QueueDAO queueDAO,
+            ExecutionDAO executionDAO,
+            IndexDAO indexDAO,
             Optional<WorkflowRepairService> workflowRepairService,
             Optional<EventQueueManager> eventQueueManager,
             Optional<BuildProperties> buildProperties) {
         this.properties = properties;
         this.executionService = executionService;
         this.queueDAO = queueDAO;
+        this.executionDAO = executionDAO;
+        this.indexDAO = indexDAO;
         this.workflowRepairService = workflowRepairService.orElse(null);
         this.eventQueueManager = eventQueueManager.orElse(null);
         this.buildProperties = buildProperties.orElse(null);
@@ -135,4 +184,120 @@ public String requeueSweep(String workflowId) {
         }
         return (verbose ? eventQueueManager.getQueueSizes() : eventQueueManager.getQueues());
     }
+
+    @Override
+    public Map<String, Object> startReindex() {
+        // Fail fast if the DAO doesn't support reindexing
+        try {
+            executionDAO.getWorkflowCount();
+            executionDAO.getAllWorkflowIdsAfter("", 1);
+        } catch (UnsupportedOperationException e) {
+            Map<String, Object> result = new HashMap<>();
+            result.put("state", "UNSUPPORTED");
+            result.put(
+                    "message",
+                    "Reindex is not supported by the current persistence backend ("
+                            + executionDAO.getClass().getSimpleName()
+                            + ")");
+            return result;
+        }
+
+        if (!reindexState.compareAndSet(ReindexState.IDLE, ReindexState.RUNNING)
+                && !reindexState.compareAndSet(ReindexState.COMPLETED, ReindexState.RUNNING)
+                && !reindexState.compareAndSet(ReindexState.FAILED, ReindexState.RUNNING)) {
+            Map<String, Object> result = new HashMap<>();
+            result.put("state", "ALREADY_RUNNING");
+            result.put("message", "A reindex job is already in progress");
+            return result;
+        }
+
+        // Reset counters
+        reindexProcessed.set(0);
+        reindexErrors.set(0);
+        reindexTotal.set(0);
+        reindexMessage = "Starting...";
+        reindexRateLimiter =
+                reindexRateLimitPerSecond > 0
+                        ? RateLimiter.create(reindexRateLimitPerSecond)
+                        : null;
+
+        reindexExecutor.submit(this::doReindex);
+
+        Map<String, Object> result = new HashMap<>();
+        result.put("state", "STARTED");
+        result.put(
+                "message",
+                "Reindex job started. Use GET /api/admin/reindex/status to track progress.");
+        return result;
+    }
+
+    @Override
+    public Map<String, Object> getReindexStatus() {
+        Map<String, Object> result = new HashMap<>();
+        result.put("state", reindexState.get().name());
+        result.put("processed", reindexProcessed.get());
+        result.put("errors", reindexErrors.get());
+        result.put("total", reindexTotal.get());
+        result.put("message", reindexMessage);
+        return result;
+    }
+
+    private void doReindex() {
+        LOGGER.info("Reindex job started");
+        int batchSize = 100;
+        String cursor = "";
+
+        try {
+            // Count total using lightweight COUNT query (no heap allocation)
+            long total = executionDAO.getWorkflowCount();
+            reindexTotal.set(total);
+            reindexMessage = "Indexing 0 / " + total;
+            LOGGER.info("Reindex: {} workflows to process", total);
+
+            while (true) {
+                List<String> workflowIds = executionDAO.getAllWorkflowIdsAfter(cursor, batchSize);
+                if (workflowIds.isEmpty()) {
+                    break;
+                }
+                cursor = workflowIds.get(workflowIds.size() - 1);
+
+                for (String workflowId : workflowIds) {
+                    if (reindexRateLimiter != null) {
+                        reindexRateLimiter.acquire();
+                    }
+                    try {
+                        WorkflowModel wfModel = executionDAO.getWorkflow(workflowId, true);
+                        if (wfModel == null) {
+                            LOGGER.warn("Workflow {} not found, skipping", workflowId);
+                            reindexErrors.incrementAndGet();
+                            continue;
+                        }
+                        indexDAO.indexWorkflow(new WorkflowSummary(wfModel.toWorkflow()));
+                        for (TaskModel task : wfModel.getTasks()) {
+                            indexDAO.indexTask(new TaskSummary(task.toTask()));
+                        }
+                        long done = reindexProcessed.incrementAndGet();
+                        reindexMessage = "Indexing " + done + " / " + reindexTotal.get();
+                    } catch (Exception e) {
+                        reindexErrors.incrementAndGet();
+                        LOGGER.error("Failed to reindex workflow {}", workflowId, e);
+                    }
+                }
+
+                LOGGER.info("Reindex progress: {}/{}", reindexProcessed.get(), reindexTotal.get());
+            }
+
+            reindexMessage =
+                    "Completed. processed="
+                            + reindexProcessed.get()
+                            + ", errors="
+                            + reindexErrors.get();
+            reindexState.set(ReindexState.COMPLETED);
+            LOGGER.info("Reindex job completed. {}", reindexMessage);
+        } catch (Exception e) {
+            reindexMessage = "Failed: " + e.getMessage();
+            reindexState.set(ReindexState.FAILED);
+            LOGGER.error("Reindex job failed", e);
+        }
+    }
 }