adding state index initial

Signed-off-by: Amit Galitzky <[email protected]>

Author: amitgalitz

build.gradle

@@ -56,6 +56,7 @@ buildscript {
         opensearch_group = "org.opensearch"
         opensearch_no_snapshot = opensearch_build.replace("-SNAPSHOT","")
         System.setProperty('tests.security.manager', 'false')
+        common_utils_version = System.getProperty("common_utils.version", opensearch_build)
     }
 
     repositories {
@@ -135,6 +136,7 @@ dependencies {
     implementation 'org.junit.jupiter:junit-jupiter:5.10.0'
     implementation "com.google.guava:guava:32.1.3-jre"
     api group: 'org.opensearch', name:'opensearch-ml-client', version: "${opensearch_build}"
+    implementation "org.opensearch:common-utils:${common_utils_version}"
 
     configurations.all {
         resolutionStrategy {

src/main/java/org/opensearch/flowframework/FlowFrameworkPlugin.java

@@ -24,14 +24,13 @@
 import org.opensearch.core.xcontent.NamedXContentRegistry;
 import org.opensearch.env.Environment;
 import org.opensearch.env.NodeEnvironment;
-import org.opensearch.flowframework.indices.GlobalContextHandler;
+import org.opensearch.flowframework.indices.FlowFrameworkIndicesHandler;
 import org.opensearch.flowframework.rest.RestCreateWorkflowAction;
 import org.opensearch.flowframework.rest.RestProvisionWorkflowAction;
 import org.opensearch.flowframework.transport.CreateWorkflowAction;
 import org.opensearch.flowframework.transport.CreateWorkflowTransportAction;
 import org.opensearch.flowframework.transport.ProvisionWorkflowAction;
 import org.opensearch.flowframework.transport.ProvisionWorkflowTransportAction;
-import org.opensearch.flowframework.workflow.CreateIndexStep;
 import org.opensearch.flowframework.workflow.WorkflowProcessSorter;
 import org.opensearch.flowframework.workflow.WorkflowStepFactory;
 import org.opensearch.plugins.ActionPlugin;
@@ -79,10 +78,9 @@ public Collection<Object> createComponents(
         WorkflowStepFactory workflowStepFactory = new WorkflowStepFactory(clusterService, client);
         WorkflowProcessSorter workflowProcessSorter = new WorkflowProcessSorter(workflowStepFactory, threadPool);
 
-        // TODO : Refactor, move system index creation/associated methods outside of the CreateIndexStep
-        GlobalContextHandler globalContextHandler = new GlobalContextHandler(client, new CreateIndexStep(clusterService, client));
+        FlowFrameworkIndicesHandler flowFrameworkIndicesHandler = new FlowFrameworkIndicesHandler(client, clusterService);
 
-        return ImmutableList.of(workflowStepFactory, workflowProcessSorter, globalContextHandler);
+        return ImmutableList.of(workflowStepFactory, workflowProcessSorter, flowFrameworkIndicesHandler);
     }
 
     @Override

src/main/java/org/opensearch/flowframework/common/CommonValue.java

@@ -27,6 +27,12 @@ private CommonValue() {}
     public static final String GLOBAL_CONTEXT_INDEX_MAPPING = "mappings/global-context.json";
     /** Global Context index mapping version */
     public static final Integer GLOBAL_CONTEXT_INDEX_VERSION = 1;
+    /** Workflow State Index Name */
+    public static final String WORKFLOW_STATE_INDEX = ".plugins-workflow-state";
+    /** Workflow State index mapping file path */
+    public static final String WORKFLOW_STATE_INDEX_MAPPING = "mappings/workflow-state.json";
+    /** Workflow State index mapping version */
+    public static final Integer WORKFLOW_STATE_INDEX_VERSION = 1;
 
     /** The transport action name prefix */
     public static final String TRANSPORT_ACION_NAME_PREFIX = "cluster:admin/opensearch/flow_framework/";

src/main/java/org/opensearch/flowframework/indices/FlowFrameworkIndex.java

@@ -14,6 +14,8 @@
 
 import static org.opensearch.flowframework.common.CommonValue.GLOBAL_CONTEXT_INDEX;
 import static org.opensearch.flowframework.common.CommonValue.GLOBAL_CONTEXT_INDEX_VERSION;
+import static org.opensearch.flowframework.common.CommonValue.WORKFLOW_STATE_INDEX;
+import static org.opensearch.flowframework.common.CommonValue.WORKFLOW_STATE_INDEX_VERSION;
 
 /**
  * An enumeration of Flow Framework indices
@@ -24,8 +26,13 @@ public enum FlowFrameworkIndex {
     */
     GLOBAL_CONTEXT(
         GLOBAL_CONTEXT_INDEX,
-        ThrowingSupplierWrapper.throwingSupplierWrapper(GlobalContextHandler::getGlobalContextMappings),
+        ThrowingSupplierWrapper.throwingSupplierWrapper(FlowFrameworkIndicesHandler::getGlobalContextMappings),
         GLOBAL_CONTEXT_INDEX_VERSION
+    ),
+    WORKFLOW_STATE(
+        WORKFLOW_STATE_INDEX,
+        ThrowingSupplierWrapper.throwingSupplierWrapper(FlowFrameworkIndicesHandler::getGlobalContextMappings),
+        WORKFLOW_STATE_INDEX_VERSION
     );
 
     private final String indexName;

src/main/java/org/opensearch/flowframework/indices/FlowFrameworkIndicesHandler.java

@@ -17,8 +17,8 @@
 import java.util.Map;
 
 import static org.opensearch.core.xcontent.XContentParserUtils.ensureExpectedToken;
-import static org.opensearch.flowframework.common.TemplateUtil.buildStringToStringMap;
-import static org.opensearch.flowframework.common.TemplateUtil.parseStringToStringMap;
+import static org.opensearch.flowframework.util.ParseUtils.buildStringToStringMap;
+import static org.opensearch.flowframework.util.ParseUtils.parseStringToStringMap;
 
 /**
  * This represents a processor associated with search and ingest pipelines in the {@link Template}.

src/main/java/org/opensearch/flowframework/indices/GlobalContextHandler.java

@@ -0,0 +1,15 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+package org.opensearch.flowframework.model;
+
+public enum ProvisioningProgress {
+    IN_PROGRESS,
+    DONE,
+    NOT_STARTED
+}

src/main/java/org/opensearch/flowframework/model/PipelineProcessor.java

@@ -0,0 +1,16 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+package org.opensearch.flowframework.model;
+
+public enum State {
+    NOT_STARTED,
+    PROVISIONING,
+    FAILED,
+    READY
+}

src/main/java/org/opensearch/flowframework/model/ProvisioningProgress.java

@@ -25,8 +25,8 @@
 import java.util.Map.Entry;
 
 import static org.opensearch.core.xcontent.XContentParserUtils.ensureExpectedToken;
-import static org.opensearch.flowframework.common.TemplateUtil.jsonToParser;
-import static org.opensearch.flowframework.common.TemplateUtil.parseStringToStringMap;
+import static org.opensearch.flowframework.util.ParseUtils.jsonToParser;
+import static org.opensearch.flowframework.util.ParseUtils.parseStringToStringMap;
 
 /**
  * The Template is the central data structure which configures workflows. This object is used to parse JSON communicated via REST API.

src/main/java/org/opensearch/flowframework/model/State.java

@@ -24,8 +24,8 @@
 import java.util.Objects;
 
 import static org.opensearch.core.xcontent.XContentParserUtils.ensureExpectedToken;
-import static org.opensearch.flowframework.common.TemplateUtil.buildStringToStringMap;
-import static org.opensearch.flowframework.common.TemplateUtil.parseStringToStringMap;
+import static org.opensearch.flowframework.util.ParseUtils.buildStringToStringMap;
+import static org.opensearch.flowframework.util.ParseUtils.parseStringToStringMap;
 
 /**
  * This represents a process node (step) in a workflow graph in the {@link Template}.