Adds CreateIndex building block

Signed-off-by: Owais Kazi <[email protected]>

build.gradle

@@ -105,6 +105,7 @@ repositories {
 dependencies {
     implementation "org.opensearch:opensearch:${opensearch_version}"
     implementation 'org.junit.jupiter:junit-jupiter:5.10.0'
+    implementation "org.apache.logging.log4j:log4j-slf4j-impl:2.19.0"
     compileOnly "com.google.guava:guava:32.1.2-jre"
     api group: 'org.opensearch', name:'opensearch-ml-client', version: "${opensearch_build}"
 

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

@@ -8,11 +8,47 @@
  */
 package org.opensearch.flowframework;
 
+import com.google.common.collect.ImmutableList;
+import org.opensearch.client.Client;
+import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
+import org.opensearch.cluster.service.ClusterService;
+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.workflow.CreateIndexStep;
 import org.opensearch.plugins.Plugin;
+import org.opensearch.repositories.RepositoriesService;
+import org.opensearch.script.ScriptService;
+import org.opensearch.threadpool.ThreadPool;
+import org.opensearch.watcher.ResourceWatcherService;
+
+import java.util.Collection;
+import java.util.function.Supplier;
 
 /**
  * An OpenSearch plugin that enables builders to innovate AI apps on OpenSearch.
  */
 public class FlowFrameworkPlugin extends Plugin {
-    // Implement the relevant Plugin Interfaces here
+
+    private Client client;
+
+    @Override
+    public Collection<Object> createComponents(
+        Client client,
+        ClusterService clusterService,
+        ThreadPool threadPool,
+        ResourceWatcherService resourceWatcherService,
+        ScriptService scriptService,
+        NamedXContentRegistry xContentRegistry,
+        Environment environment,
+        NodeEnvironment nodeEnvironment,
+        NamedWriteableRegistry namedWriteableRegistry,
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        Supplier<RepositoriesService> repositoriesServiceSupplier
+    ) {
+        this.client = client;
+        CreateIndexStep createIndexStep = new CreateIndexStep(client);
+        return ImmutableList.of(createIndexStep);
+    }
 }

src/main/java/org/opensearch/flowframework/workflow/CreateIndexStep.java

@@ -10,26 +10,54 @@
 
 import com.google.common.base.Charsets;
 import com.google.common.io.Resources;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.opensearch.action.ActionListener;
 import org.opensearch.action.admin.indices.create.CreateIndexRequest;
-import org.opensearch.client.AdminClient;
+import org.opensearch.action.admin.indices.create.CreateIndexResponse;
+import org.opensearch.client.Client;
 import org.opensearch.common.xcontent.XContentType;
 
 import java.io.IOException;
 import java.net.URL;
 import java.util.concurrent.CompletableFuture;
 
-public class CreateIndexStep implements Workflow {
+public class CreateIndexStep implements WorkflowStep {
 
-    AdminClient adminClient;
+    private static final Logger logger = LogManager.getLogger(CreateIndexStep.class);
+    private Client client;
+    private final String CREATE_INDEX_STEP = "create_index_step";
+
+    public CreateIndexStep(Client client) {
+        this.client = client;
+    }
 
     @Override
-    public CompletableFuture<Workflow> execute() throws Exception {
+    public CompletableFuture<WorkflowData> execute(WorkflowData data) {
+
+        ActionListener<CreateIndexResponse> actionListener = new ActionListener<>() {
+
+            @Override
+            public void onResponse(CreateIndexResponse createIndexResponse) {
+                logger.info("created index:{}");
+            }
 
-        // ActionListener<CreateIndexResponse> actionListener
+            @Override
+            public void onFailure(Exception e) {
+                logger.error("Index creation failed", e);
+            }
+        };
+
+        // Fetch indexName, fileName and settings from WorkflowData
         CreateIndexRequest request = new CreateIndexRequest(indexName).mapping(getIndexMappings(fileName), XContentType.JSON)
             .settings(settings);
-        adminClient.indices().create(request, actionListener);
+        client.admin().indices().create(request, actionListener);
+        return null;
+    }
 
+    @Override
+    public String getName() {
+        return CREATE_INDEX_STEP;
     }
 
     /**

src/main/resources/mappings/knn-index-mapping.json

@@ -1 +1,16 @@
-
+{
+  "properties": {
+    "desc_v": {
+      "type": "keyword"
+    },
+    "name_v": {
+      "type": "keyword"
+    },
+    "description": {
+      "type": "keyword"
+    },
+    "name": {
+      "type": "keyword"
+    }
+  }
+}