Implements Update index Step to support updating index settings, modi…

…fies updating resource created script to remove error if any Signed-off-by: Joshua Palis <[email protected]>
joshpalis · Jul 24, 2024 · 8530df0 · 8530df0
1 parent 202072c
commit 8530df0
Show file tree

Hide file tree

Showing 4 changed files with 61 additions and 6 deletions.
diff --git a/src/main/java/org/opensearch/flowframework/common/WorkflowResources.java b/src/main/java/org/opensearch/flowframework/common/WorkflowResources.java
@@ -29,6 +29,7 @@
 import org.opensearch.flowframework.workflow.RegisterLocalSparseEncodingModelStep;
 import org.opensearch.flowframework.workflow.RegisterModelGroupStep;
 import org.opensearch.flowframework.workflow.RegisterRemoteModelStep;
+import org.opensearch.flowframework.workflow.ReindexStep;
 import org.opensearch.flowframework.workflow.UndeployModelStep;
 import org.opensearch.flowframework.workflow.UpdateIndexStep;
 import org.opensearch.flowframework.workflow.UpdateIngestPipelineStep;

diff --git a/src/main/java/org/opensearch/flowframework/indices/FlowFrameworkIndicesHandler.java b/src/main/java/org/opensearch/flowframework/indices/FlowFrameworkIndicesHandler.java
@@ -689,7 +689,7 @@ public void updateResourceInStateIndex(
         Script script = new Script(
             ScriptType.INLINE,
             "painless",
-            "ctx._source.resources_created.add(params.newResource)",
+            "ctx._source.resources_created.add(params.newResource); if(ctx._source.containsKey('error')){ctx._source.remove('error')}",
             Collections.singletonMap("newResource", newResource.resourceMap())
         );
 

diff --git a/src/main/java/org/opensearch/flowframework/transport/ReprovisionWorkflowTransportAction.java b/src/main/java/org/opensearch/flowframework/transport/ReprovisionWorkflowTransportAction.java
@@ -141,9 +141,9 @@ protected void doExecute(Task task, ReprovisionWorkflowRequest request, ActionLi
                         logger.info("Updated workflow {} state to {}", request.getWorkflowId(), State.PROVISIONING);
 
                         // Attach last provisioned time to updated template and execute reprovisioning
-                        Template updatedTemplateWithProvisionedTime = new Template.Builder(updatedTemplate).lastProvisionedTime(
-                            Instant.now()
-                        ).build();
+                        Template updatedTemplateWithProvisionedTime = Template.builder(updatedTemplate)
+                            .lastProvisionedTime(Instant.now())
+                            .build();
                         executeWorkflowAsync(workflowId, updatedTemplateWithProvisionedTime, reprovisionProcessSequence, listener);
 
                         listener.onResponse(new WorkflowResponse(workflowId));

diff --git a/src/main/java/org/opensearch/flowframework/workflow/UpdateIndexStep.java b/src/main/java/org/opensearch/flowframework/workflow/UpdateIndexStep.java
@@ -11,9 +11,11 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.opensearch.ExceptionsHelper;
+import org.opensearch.action.admin.indices.settings.get.GetSettingsRequest;
 import org.opensearch.action.admin.indices.settings.put.UpdateSettingsRequest;
 import org.opensearch.action.support.PlainActionFuture;
 import org.opensearch.client.Client;
+import org.opensearch.common.settings.Settings;
 import org.opensearch.common.xcontent.XContentHelper;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.common.bytes.BytesArray;
@@ -26,6 +28,7 @@
 
 import java.nio.charset.StandardCharsets;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
 
@@ -99,9 +102,42 @@ public PlainActionFuture<WorkflowData> execute(
                         + ", settings are not found in the index configuration";
                     throw new FlowFrameworkException(errorMessage, RestStatus.BAD_REQUEST);
                 } else {
+
                     @SuppressWarnings("unchecked")
-                    Map<String, Object> indexSettings = (Map<String, Object>) sourceAsMap.get("settings");
-                    updateSettingsRequest.settings(indexSettings);
+                    Map<String, Object> updatedSettings = (Map<String, Object>) sourceAsMap.get("settings");
+
+                    // check if settings are flattened or expanded
+                    Map<String, Object> flattenedSettings = new HashMap<>();
+                    if (updatedSettings.containsKey("index")) {
+                        flattenSettings("", updatedSettings, flattenedSettings);
+                    } else {
+                        flattenedSettings.putAll(updatedSettings);
+                    }
+
+                    Map<String, Object> filteredSettings = new HashMap<>();
+
+                    // Retrieve current Index Settings
+                    GetSettingsRequest getSettingsRequest = new GetSettingsRequest();
+                    getSettingsRequest.indices(indexName);
+                    getSettingsRequest.includeDefaults(true);
+                    client.admin().indices().getSettings(getSettingsRequest, ActionListener.wrap(response -> {
+                        Map<String, Settings> indexToSettings = new HashMap<String, Settings>(response.getIndexToSettings());
+
+                        // Include in the update request only settings with updated values
+                        Settings currentIndexSettings = indexToSettings.get(indexName);
+                        for (Map.Entry<String, Object> e : flattenedSettings.entrySet()) {
+                            String val = e.getValue().toString();
+                            if (!val.equals(currentIndexSettings.get(e.getKey()))) {
+                                filteredSettings.put(e.getKey(), e.getValue());
+                            }
+                        }
+                    }, ex -> {
+                        Exception e = getSafeException(ex);
+                        String errorMessage = (e == null ? "Failed to update the index settings for index " + indexName : e.getMessage());
+                        updateIndexFuture.onFailure(new WorkflowStepException(errorMessage, ExceptionsHelper.status(e)));
+                    }));
+
+                    updateSettingsRequest.settings(filteredSettings);
                 }
             }
 
@@ -126,6 +162,24 @@ public PlainActionFuture<WorkflowData> execute(
 
     }
 
+    /**
+     * Flattens a nested map of settings, delimitted by a period
+     * @param prefix the setting prefix
+     * @param settings the nested setting map
+     * @param flattenedSettings the final flattend map of settings
+     */
+    public static void flattenSettings(String prefix, Map<String, Object> settings, Map<String, Object> flattenedSettings) {
+        for (Map.Entry<String, Object> entry : settings.entrySet()) {
+            String key = prefix.isEmpty() ? entry.getKey() : prefix + "." + entry.getKey();
+            Object value = entry.getValue();
+            if (value instanceof Map) {
+                flattenSettings(key, (Map<String, Object>) value, flattenedSettings);
+            } else {
+                flattenedSettings.put(key, value.toString());
+            }
+        }
+    }
+
     @Override
     public String getName() {
         return NAME;