Adds REST APIs for creating and provisioning a workflow

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

@@ -9,28 +9,54 @@
 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.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.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.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.ThreadPool;
 import org.opensearch.watcher.ResourceWatcherService;
 
 import java.util.Collection;
+import java.util.List;
 import java.util.function.Supplier;
 
 /**
  * An OpenSearch plugin that enables builders to innovate AI apps on OpenSearch.
  */
-public class FlowFrameworkPlugin extends Plugin {
+public class FlowFrameworkPlugin extends Plugin implements ActionPlugin {
+
+    /**
+     * The base URI for this plugin's rest actions
+     */
+    public static final String AI_FLOW_FRAMEWORK_BASE_URI = "/_plugins/_ai_flow";
+    /**
+     * The URI for this plugin's workflow rest actions
+     */
+    public static final String WORKFLOWS_URI = AI_FLOW_FRAMEWORK_BASE_URI + "/workflows";
 
     @Override
     public Collection<Object> createComponents(
@@ -51,4 +77,26 @@ public Collection<Object> createComponents(
 
         return ImmutableList.of(workflowStepFactory, workflowProcessSorter);
     }
+
+    @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)
+        );
+    }
+
 }

src/main/java/org/opensearch/flowframework/rest/RestCreateWorkflowAction.java

@@ -0,0 +1,69 @@
+/*
+ * 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.rest;
+
+import com.google.common.collect.ImmutableList;
+import org.opensearch.client.node.NodeClient;
+import org.opensearch.flowframework.FlowFrameworkPlugin;
+import org.opensearch.flowframework.model.Template;
+import org.opensearch.flowframework.transport.CreateWorkflowAction;
+import org.opensearch.flowframework.transport.WorkflowRequest;
+import org.opensearch.rest.BaseRestHandler;
+import org.opensearch.rest.RestRequest;
+import org.opensearch.rest.action.RestToXContentListener;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+
+/**
+ * Rest Action to facilitate requests to create and update a use case template
+ */
+public class RestCreateWorkflowAction extends BaseRestHandler {
+
+    private static final String CREATE_WORKFLOW_ACTION = "create_workflow_action";
+
+    // TODO : move to common values class, pending implementation
+    /**
+     * Field name for workflow Id, the document Id of the indexed use case template
+     */
+    public static final String WORKFLOW_ID = "workflowId";
+
+    /**
+     * Intantiates a new RestCreateWorkflowAction
+     */
+    public RestCreateWorkflowAction() {
+        // TODO : Pass settings and cluster service to constructor and add settings update consumer for request timeout value
+    }
+
+    @Override
+    public String getName() {
+        return CREATE_WORKFLOW_ACTION;
+    }
+
+    @Override
+    public List<Route> routes() {
+        return ImmutableList.of(
+            // Create new workflow
+            new Route(RestRequest.Method.POST, String.format(Locale.ROOT, "%s", FlowFrameworkPlugin.WORKFLOWS_URI)),
+            // Update workflow
+            new Route(RestRequest.Method.PUT, String.format(Locale.ROOT, "%s/{%s}", FlowFrameworkPlugin.WORKFLOWS_URI, WORKFLOW_ID))
+        );
+    }
+
+    @Override
+    protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
+
+        String workflowId = request.param(WORKFLOW_ID, null);
+        Template template = Template.parse(request.content().utf8ToString());
+        WorkflowRequest workflowRequest = new WorkflowRequest(workflowId, template);
+        return channel -> client.execute(CreateWorkflowAction.INSTANCE, workflowRequest, new RestToXContentListener<>(channel));
+    }
+
+}

src/main/java/org/opensearch/flowframework/rest/RestProvisionWorkflowAction.java

@@ -0,0 +1,83 @@
+/*
+ * 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.rest;
+
+import com.google.common.collect.ImmutableList;
+import org.opensearch.client.node.NodeClient;
+import org.opensearch.flowframework.FlowFrameworkPlugin;
+import org.opensearch.flowframework.model.Template;
+import org.opensearch.flowframework.transport.ProvisionWorkflowAction;
+import org.opensearch.flowframework.transport.WorkflowRequest;
+import org.opensearch.rest.BaseRestHandler;
+import org.opensearch.rest.RestRequest;
+import org.opensearch.rest.action.RestToXContentListener;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+
+/**
+ * Rest action to facilitate requests to provision a workflow from an inline defined or stored use case template
+ */
+public class RestProvisionWorkflowAction extends BaseRestHandler {
+
+    private static final String PROVISION_WORKFLOW_ACTION = "provision_workflow_action";
+
+    // TODO : move to common values class, pending implementation
+    /**
+     * Field name for workflow Id, the document Id of the indexed use case template
+     */
+    public static final String WORKFLOW_ID = "workflowId";
+
+    /**
+     * Instantiates a new RestProvisionWorkflowAction
+     */
+    public RestProvisionWorkflowAction() {
+        // TODO : Pass settings and cluster service to constructor and add settings update consumer for request timeout value
+    }
+
+    @Override
+    public String getName() {
+        return PROVISION_WORKFLOW_ACTION;
+    }
+
+    @Override
+    public List<Route> routes() {
+        return ImmutableList.of(
+            // Provision workflow from inline use case template
+            new Route(RestRequest.Method.POST, String.format(Locale.ROOT, "%s/%s", FlowFrameworkPlugin.WORKFLOWS_URI, "_provision")),
+            // Provision workflow from indexed use case template
+            new Route(
+                RestRequest.Method.POST,
+                String.format(Locale.ROOT, "%s/{%s}/%s", FlowFrameworkPlugin.WORKFLOWS_URI, WORKFLOW_ID, "_provision")
+            )
+        );
+    }
+
+    @Override
+    protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
+
+        String workflowId = request.param(WORKFLOW_ID, null);
+        Template template = null;
+
+        if (request.hasContent()) {
+            template = Template.parse(request.content().utf8ToString());
+        }
+
+        // Validate workflow request inputs
+        if (workflowId == null && template == null) {
+            throw new IOException("WorkflowId and template cannot be both null");
+        }
+
+        // Create request and provision
+        WorkflowRequest workflowRequest = new WorkflowRequest(workflowId, template);
+        return channel -> client.execute(ProvisionWorkflowAction.INSTANCE, workflowRequest, new RestToXContentListener<>(channel));
+    }
+
+}

src/main/java/org/opensearch/flowframework/transport/CreateWorkflowAction.java

@@ -0,0 +1,27 @@
+/*
+ * 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.transport;
+
+import org.opensearch.action.ActionType;
+
+/**
+ * External Action for public facing RestCreateWorkflowActiom
+ */
+public class CreateWorkflowAction extends ActionType<WorkflowResponse> {
+
+    // TODO : Determine external action prefix for plugin
+    /** The name of this action */
+    public static final String NAME = "workflows/create";
+    /** An instance of this action */
+    public static final CreateWorkflowAction INSTANCE = new CreateWorkflowAction();
+
+    private CreateWorkflowAction() {
+        super(NAME, WorkflowResponse::new);
+    }
+}

src/main/java/org/opensearch/flowframework/transport/CreateWorkflowTransportAction.java

@@ -0,0 +1,65 @@
+/*
+ * 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.transport;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.opensearch.action.support.ActionFilters;
+import org.opensearch.action.support.HandledTransportAction;
+import org.opensearch.client.Client;
+import org.opensearch.common.inject.Inject;
+import org.opensearch.core.action.ActionListener;
+import org.opensearch.tasks.Task;
+import org.opensearch.transport.TransportService;
+
+/**
+ * Transport Action to index or update a use case template within the Global Context
+ */
+public class CreateWorkflowTransportAction extends HandledTransportAction<WorkflowRequest, WorkflowResponse> {
+
+    private final Logger logger = LogManager.getLogger(CreateWorkflowTransportAction.class);
+
+    private final Client client;
+
+    /**
+     * Intantiates a new CreateWorkflowTransportAction
+     * @param transportService the TransportService
+     * @param actionFilters action filters
+     * @param client the node client to interact with an index
+     */
+    @Inject
+    public CreateWorkflowTransportAction(TransportService transportService, ActionFilters actionFilters, Client client) {
+        super(CreateWorkflowAction.NAME, transportService, actionFilters, WorkflowRequest::new);
+        this.client = client;
+    }
+
+    @Override
+    protected void doExecute(Task task, WorkflowRequest request, ActionListener<WorkflowResponse> listener) {
+
+        String workflowId;
+        // TODO : Check if global context index exists, and if it does not then create
+
+        if (request.getWorkflowId() == null) {
+            // TODO : Create new entry
+            // TODO : Insert doc
+
+            // TODO : get document ID
+            workflowId = "";
+            // TODO : check if state index exists, and if it does not, then create
+            // TODO : insert state index doc, mapped with documentId, defaulted to NOT_STARTED
+        } else {
+            // TODO : Update existing entry
+            workflowId = request.getWorkflowId();
+            // TODO : Update state index entry, default back to NOT_STARTED
+        }
+
+        listener.onResponse(new WorkflowResponse(workflowId));
+    }
+
+}

src/main/java/org/opensearch/flowframework/transport/ProvisionWorkflowAction.java

@@ -0,0 +1,27 @@
+/*
+ * 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.transport;
+
+import org.opensearch.action.ActionType;
+
+/**
+ * External Action for public facing RestProvisionWorkflowAction
+ */
+public class ProvisionWorkflowAction extends ActionType<WorkflowResponse> {
+
+    // TODO : Determine external action prefix for plugin
+    /** The name of this action */
+    public static final String NAME = "workflows/provision";
+    /** An instance of this action */
+    public static final ProvisionWorkflowAction INSTANCE = new ProvisionWorkflowAction();
+
+    private ProvisionWorkflowAction() {
+        super(NAME, WorkflowResponse::new);
+    }
+}