Adds REST APIs for creating and provisioning a workflow

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

@@ -9,28 +9,53 @@
 package org.opensearch.flowframework;
 
 import com.google.common.collect.ImmutableList;
+import org.opensearch.action.ActionRequest;
 import org.opensearch.client.Client;
 import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
+import org.opensearch.cluster.node.DiscoveryNodes;
 import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.common.settings.ClusterSettings;
+import org.opensearch.common.settings.IndexScopedSettings;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.settings.SettingsFilter;
+import org.opensearch.common.util.concurrent.OpenSearchExecutors;
+import org.opensearch.core.action.ActionResponse;
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 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.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;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.repositories.RepositoriesService;
+import org.opensearch.rest.RestController;
+import org.opensearch.rest.RestHandler;
 import org.opensearch.script.ScriptService;
+import org.opensearch.threadpool.ExecutorBuilder;
+import org.opensearch.threadpool.FixedExecutorBuilder;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.watcher.ResourceWatcherService;
 
 import java.util.Collection;
+import java.util.List;
 import java.util.function.Supplier;
 
+import static org.opensearch.flowframework.common.CommonValue.FLOW_FRAMEWORK_THREAD_POOL_PREFIX;
+import static org.opensearch.flowframework.common.CommonValue.PROVISION_THREAD_POOL;
+
 /**
  * An OpenSearch plugin that enables builders to innovate AI apps on OpenSearch.
  */
-public class FlowFrameworkPlugin extends Plugin {
+public class FlowFrameworkPlugin extends Plugin implements ActionPlugin {
 
     /**
      * Instantiate this plugin.
@@ -54,6 +79,45 @@ public Collection<Object> createComponents(
         WorkflowStepFactory workflowStepFactory = new WorkflowStepFactory(clusterService, client);
         WorkflowProcessSorter workflowProcessSorter = new WorkflowProcessSorter(workflowStepFactory, threadPool);
 
-        return ImmutableList.of(workflowStepFactory, workflowProcessSorter);
+        // TODO : Refactor, move system index creation/associated methods outside of the CreateIndexStep
+        GlobalContextHandler globalContextHandler = new GlobalContextHandler(client, new CreateIndexStep(clusterService, client));
+
+        return ImmutableList.of(workflowStepFactory, workflowProcessSorter, globalContextHandler);
+    }
+
+    @Override
+    public List<RestHandler> getRestHandlers(
+        Settings settings,
+        RestController restController,
+        ClusterSettings clusterSettings,
+        IndexScopedSettings indexScopedSettings,
+        SettingsFilter settingsFilter,
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        Supplier<DiscoveryNodes> nodesInCluster
+    ) {
+        return ImmutableList.of(new RestCreateWorkflowAction(), new RestProvisionWorkflowAction());
+    }
+
+    @Override
+    public List<ActionHandler<? extends ActionRequest, ? extends ActionResponse>> getActions() {
+        return ImmutableList.of(
+            new ActionHandler<>(CreateWorkflowAction.INSTANCE, CreateWorkflowTransportAction.class),
+            new ActionHandler<>(ProvisionWorkflowAction.INSTANCE, ProvisionWorkflowTransportAction.class)
+        );
     }
+
+    @Override
+    public List<ExecutorBuilder<?>> getExecutorBuilders(Settings settings) {
+        // TODO : Determine final size/queueSize values for the provision thread pool
+        return ImmutableList.of(
+            new FixedExecutorBuilder(
+                settings,
+                PROVISION_THREAD_POOL,
+                OpenSearchExecutors.allocatedProcessors(settings),
+                10,
+                FLOW_FRAMEWORK_THREAD_POOL_PREFIX + PROVISION_THREAD_POOL
+            )
+        );
+    }
+
 }

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

@@ -13,19 +13,49 @@
  */
 public class CommonValue {
 
+    /** Default value for no schema version */
     public static Integer NO_SCHEMA_VERSION = 0;
+    /** Index mapping meta field name*/
     public static final String META = "_meta";
+    /** Schema Version field name */
     public static final String SCHEMA_VERSION_FIELD = "schema_version";
+    /** Global Context Index Name */
     public static final String GLOBAL_CONTEXT_INDEX = ".plugins-ai-global-context";
+    /** Global Context index mapping file path */
     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;
+
+    /** The base URI for this plugin's rest actions */
+    public static final String AI_FLOW_FRAMEWORK_BASE_URI = "/_plugins/_flow_framework";
+    /** The URI for this plugin's workflow rest actions */
+    public static final String WORKFLOW_URI = AI_FLOW_FRAMEWORK_BASE_URI + "/workflow";
+    /** Field name for workflow Id, the document Id of the indexed use case template */
+    public static final String WORKFLOW_ID = "workflow_id";
+    /** The field name for provision workflow within a use case template*/
+    public static final String PROVISION_WORKFLOW = "provision";
+
+    /** Flow Framework plugin thread pool name prefix */
+    public static final String FLOW_FRAMEWORK_THREAD_POOL_PREFIX = "thread_pool.flow_framework.";
+    /** The provision workflow thread pool name */
+    public static final String PROVISION_THREAD_POOL = "opensearch_workflow_provision";
+
+    /** Model Id field */
     public static final String MODEL_ID = "model_id";
+    /** Function Name field */
     public static final String FUNCTION_NAME = "function_name";
+    /** Model Name field */
     public static final String MODEL_NAME = "name";
+    /** Model Version field */
     public static final String MODEL_VERSION = "model_version";
+    /** Model Group Id field */
     public static final String MODEL_GROUP_ID = "model_group_id";
+    /** Description field */
     public static final String DESCRIPTION = "description";
+    /** Connector Id field */
     public static final String CONNECTOR_ID = "connector_id";
+    /** Model format field */
     public static final String MODEL_FORMAT = "model_format";
+    /** Model config field */
     public static final String MODEL_CONFIG = "model_config";
 }

src/main/java/org/opensearch/flowframework/exception/FlowFrameworkException.java

@@ -17,6 +17,7 @@ public class FlowFrameworkException extends RuntimeException {
 
     private static final long serialVersionUID = 1L;
 
+    /** The rest status code of this exception */
     private final RestStatus restStatus;
 
     /**

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

@@ -19,6 +19,9 @@
  * An enumeration of Flow Framework indices
  */
 public enum FlowFrameworkIndex {
+    /**
+     * Global Context Index
+    */
     GLOBAL_CONTEXT(
         GLOBAL_CONTEXT_INDEX,
         ThrowingSupplierWrapper.throwingSupplierWrapper(GlobalContextHandler::getGlobalContextMappings),
@@ -35,14 +38,26 @@ public enum FlowFrameworkIndex {
         this.version = version;
     }
 
+    /**
+     * Retrieves the index name
+     * @return the index name
+     */
     public String getIndexName() {
         return indexName;
     }
 
+    /**
+     * Retrieves the index mapping
+     * @return the index mapping
+     */
     public String getMapping() {
         return mapping;
     }
 
+    /**
+     * Retrieves the index version
+     * @return the index version
+     */
     public Integer getVersion() {
         return version;
     }

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

@@ -81,7 +81,7 @@
                 XContentBuilder builder = XContentFactory.jsonBuilder();
                 ThreadContext.StoredContext context = client.threadPool().getThreadContext().stashContext()
             ) {
-                request.source(template.toXContent(builder, ToXContent.EMPTY_PARAMS))
+                request.source(template.toDocumentSource(builder, ToXContent.EMPTY_PARAMS))
                     .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
                 client.index(request, ActionListener.runBefore(listener, () -> context.restore()));
             } catch (Exception e) {
@@ -94,6 +94,35 @@
         }));
     }
 
+    /**
+     * Replaces a document in the global context index
+     * @param documentId the document Id
+     * @param template the use-case template
+     * @param listener action listener
+     */
+    public void updateTemplate(String documentId, Template template, ActionListener<IndexResponse> listener) {
+        if (!createIndexStep.doesIndexExist(GLOBAL_CONTEXT_INDEX)) {
+            String exceptionMessage = "Failed to update template for workflow_id : "
+                + documentId
+                + ", global_context index does not exist.";
+            logger.error(exceptionMessage);
+            listener.onFailure(new Exception(exceptionMessage));
+        } else {
+            IndexRequest request = new IndexRequest(GLOBAL_CONTEXT_INDEX).id(documentId);
+            try (
+                XContentBuilder builder = XContentFactory.jsonBuilder();
+                ThreadContext.StoredContext context = client.threadPool().getThreadContext().stashContext()
+            ) {
+                request.source(template.toDocumentSource(builder, ToXContent.EMPTY_PARAMS))
+                    .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+                client.index(request, ActionListener.runBefore(listener, () -> context.restore()));
+            } catch (Exception e) {
+                logger.error("Failed to update global_context entry : {}. {}", documentId, e.getMessage());
+                listener.onFailure(e);
+            }
+        }
+    }
+
     /**
      * Update global context index for specific fields
      * @param documentId global context index document id