Change last provisioned timestamp when provisioning workflow

Signed-off-by: Daniel Widdis <[email protected]>
opensearch-project · Mar 5, 2024 · 1fa040e · 1fa040e
1 parent ae36573
commit 1fa040e
Show file tree

Hide file tree

Showing 3 changed files with 43 additions and 3 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -21,6 +21,8 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.1.0/)
 - Added an optional workflow_step param to the get workflow steps API ([#538](https://github.com/opensearch-project/flow-framework/pull/538))
 
 ### Enhancements
+- Add created, updated, and provisioned timestamps to saved template ([#551](https://github.com/opensearch-project/flow-framework/pull/551))
+
 ### Bug Fixes
 ### Infrastructure
 ### Documentation

diff --git a/src/main/java/org/opensearch/flowframework/transport/ProvisionWorkflowTransportAction.java b/src/main/java/org/opensearch/flowframework/transport/ProvisionWorkflowTransportAction.java
@@ -118,10 +118,10 @@ protected void doExecute(Task task, WorkflowRequest request, ActionListener<Work
                 }
 
                 // Parse template from document source
-                Template template = Template.parse(response.getSourceAsString());
+                Template parsedTemplate = Template.parse(response.getSourceAsString());
 
                 // Decrypt template
-                template = encryptorUtils.decryptTemplateCredentials(template);
+                final Template template = encryptorUtils.decryptTemplateCredentials(parsedTemplate);
 
                 // Sort and validate graph
                 Workflow provisionWorkflow = template.workflows().get(PROVISION_WORKFLOW);
@@ -134,6 +134,7 @@ protected void doExecute(Task task, WorkflowRequest request, ActionListener<Work
 
                 flowFrameworkIndicesHandler.isWorkflowNotStarted(workflowId, workflowIsNotStarted -> {
                     if (TRUE.equals(workflowIsNotStarted)) {
+                        // update state index
                         flowFrameworkIndicesHandler.updateFlowFrameworkSystemIndexDoc(
                             workflowId,
                             Map.ofEntries(
@@ -145,7 +146,36 @@ protected void doExecute(Task task, WorkflowRequest request, ActionListener<Work
                             ActionListener.wrap(updateResponse -> {
                                 logger.info("updated workflow {} state to {}", request.getWorkflowId(), State.PROVISIONING);
                                 executeWorkflowAsync(workflowId, provisionProcessSequence, listener);
-                                listener.onResponse(new WorkflowResponse(workflowId));
+                                // update last provisioned field in template
+                                Template newTemplate = new Template.Builder().name(template.name())
+                                    .description(template.description())
+                                    .useCase(template.useCase())
+                                    .templateVersion(template.templateVersion())
+                                    .compatibilityVersion(template.compatibilityVersion())
+                                    .workflows(template.workflows())
+                                    .uiMetadata(template.getUiMetadata())
+                                    .user(template.getUser()) // Should we care about old user here?
+                                    .createdTime(template.createdTime())
+                                    .lastUpdatedTime(template.lastUpdatedTime())
+                                    .lastProvisionedTime(Instant.now().toEpochMilli())
+                                    .build();
+                                flowFrameworkIndicesHandler.updateTemplateInGlobalContext(
+                                    request.getWorkflowId(),
+                                    newTemplate,
+                                    ActionListener.wrap(templateResponse -> {
+                                        listener.onResponse(new WorkflowResponse(request.getWorkflowId()));
+                                    }, exception -> {
+                                        String errorMessage = "Failed to update use case template " + request.getWorkflowId();
+                                        logger.error(errorMessage, exception);
+                                        if (exception instanceof FlowFrameworkException) {
+                                            listener.onFailure(exception);
+                                        } else {
+                                            listener.onFailure(
+                                                new FlowFrameworkException(errorMessage, ExceptionsHelper.status(exception))
+                                            );
+                                        }
+                                    })
+                                );
                             }, exception -> {
                                 String errorMessage = "Failed to update workflow state: " + workflowId;
                                 logger.error(errorMessage, exception);

diff --git a/...st/java/org/opensearch/flowframework/transport/ProvisionWorkflowTransportActionTests.java b/...st/java/org/opensearch/flowframework/transport/ProvisionWorkflowTransportActionTests.java
@@ -11,6 +11,7 @@
 import org.opensearch.Version;
 import org.opensearch.action.get.GetRequest;
 import org.opensearch.action.get.GetResponse;
+import org.opensearch.action.index.IndexResponse;
 import org.opensearch.action.support.ActionFilters;
 import org.opensearch.action.update.UpdateResponse;
 import org.opensearch.client.Client;
@@ -19,6 +20,7 @@
 import org.opensearch.common.xcontent.XContentFactory;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.common.bytes.BytesReference;
+import org.opensearch.core.index.shard.ShardId;
 import org.opensearch.core.xcontent.XContentBuilder;
 import org.opensearch.flowframework.TestHelpers;
 import org.opensearch.flowframework.indices.FlowFrameworkIndicesHandler;
@@ -147,6 +149,12 @@ public void testProvisionWorkflow() {
             return null;
         }).when(flowFrameworkIndicesHandler).updateFlowFrameworkSystemIndexDoc(any(), any(), any());
 
+        doAnswer(invocation -> {
+            ActionListener<IndexResponse> responseListener = invocation.getArgument(2);
+            responseListener.onResponse(new IndexResponse(new ShardId(GLOBAL_CONTEXT_INDEX, "", 1), "1", 1L, 1L, 1L, true));
+            return null;
+        }).when(flowFrameworkIndicesHandler).updateTemplateInGlobalContext(any(), any(Template.class), any());
+
         provisionWorkflowTransportAction.doExecute(mock(Task.class), workflowRequest, listener);
         ArgumentCaptor<WorkflowResponse> responseCaptor = ArgumentCaptor.forClass(WorkflowResponse.class);
         verify(listener, times(1)).onResponse(responseCaptor.capture());