Update to use new UpdateStatusEvent

Signed-off-by: Jack Schofield <jack.schofield@answerdigital.com>
Project-MONAI · Jun 9, 2022 · b647d41 · b647d41
1 parent 271f282
commit b647d41
Show file tree

Hide file tree

Showing 9 changed files with 22 additions and 134 deletions.
diff --git a/src/Monai.Deploy.WorkflowManager.sln b/src/Monai.Deploy.WorkflowManager.sln
@@ -55,9 +55,7 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Monai.Deploy.WorkflowManage
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Monai.Deploy.WorkflowManager.ConditionsResolver.Tests", "..\tests\UnitTests\ConditionsResolver.Tests\Monai.Deploy.WorkflowManager.ConditionsResolver.Tests.csproj", "{918E4DE3-A7BF-4B7F-9B5A-5C36FEFA3C30}"
 EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Monai.Deploy.WorkflowManager.TaskManager.API.Tests", "..\tests\UnitTests\TaskManager.API.Tests\Monai.Deploy.WorkflowManager.TaskManager.API.Tests.csproj", "{7AF31246-C97C-4AF5-90C3-422199CF7B4A}"
-EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Monai.Deploy.WorkflowManager.Common.Tests", "..\tests\UnitTests\WorkflowManager.Common.Tests\Monai.Deploy.WorkflowManager.Common.Tests.csproj", "{A44F975E-70CA-49D6-8513-78F2D5210EAF}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Monai.Deploy.WorkflowManager.Common.Tests", "..\tests\UnitTests\WorkflowManager.Common.Tests\Monai.Deploy.WorkflowManager.Common.Tests.csproj", "{A44F975E-70CA-49D6-8513-78F2D5210EAF}"
 EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -161,10 +159,6 @@ Global
 		{918E4DE3-A7BF-4B7F-9B5A-5C36FEFA3C30}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{918E4DE3-A7BF-4B7F-9B5A-5C36FEFA3C30}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{918E4DE3-A7BF-4B7F-9B5A-5C36FEFA3C30}.Release|Any CPU.Build.0 = Release|Any CPU
-		{7AF31246-C97C-4AF5-90C3-422199CF7B4A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
-		{7AF31246-C97C-4AF5-90C3-422199CF7B4A}.Debug|Any CPU.Build.0 = Debug|Any CPU
-		{7AF31246-C97C-4AF5-90C3-422199CF7B4A}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{7AF31246-C97C-4AF5-90C3-422199CF7B4A}.Release|Any CPU.Build.0 = Release|Any CPU
 		{A44F975E-70CA-49D6-8513-78F2D5210EAF}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
 		{A44F975E-70CA-49D6-8513-78F2D5210EAF}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{A44F975E-70CA-49D6-8513-78F2D5210EAF}.Release|Any CPU.ActiveCfg = Release|Any CPU
@@ -189,7 +183,6 @@ Global
 		{89D3D817-CCFE-4933-9089-D1283F2EA1B5} = {71DDEE7B-E213-4E39-A7F4-4646783A27F7}
 		{918E4DE3-A7BF-4B7F-9B5A-5C36FEFA3C30} = {71DDEE7B-E213-4E39-A7F4-4646783A27F7}
 		{A44F975E-70CA-49D6-8513-78F2D5210EAF} = {71DDEE7B-E213-4E39-A7F4-4646783A27F7}
-		{7AF31246-C97C-4AF5-90C3-422199CF7B4A} = {71DDEE7B-E213-4E39-A7F4-4646783A27F7}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {DC0D56C8-D8CB-45CE-B528-F3DCF86D63ED}

diff --git a/src/TaskManager/API/Extensions/StorageListExtensions.cs b/src/TaskManager/API/Extensions/StorageListExtensions.cs
diff --git a/src/TaskManager/Plug-ins/Argo/ExitHookTemplate.cs b/src/TaskManager/Plug-ins/Argo/ExitHookTemplate.cs
@@ -3,7 +3,6 @@
 
 using Argo;
 using Monai.Deploy.Messaging.Events;
-using Monai.Deploy.WorkflowManager.TaskManager.API.Extensions;
 using Newtonsoft.Json;
 
 namespace Monai.Deploy.WorkflowManager.TaskManager.Argo
@@ -82,7 +81,7 @@ public Template2 GenerateMessageTemplate(S3Artifact2 artifact)
                 ExecutionId = _taskDispatchEvent.ExecutionId,
                 CorrelationId = _taskDispatchEvent.CorrelationId,
                 Identity = "{{workflow.name}}",
-                OutputArtifacts = _taskDispatchEvent.Outputs?.ToArtifactDictionary() ?? new Dictionary<string, string>()
+                Outputs = _taskDispatchEvent.Outputs ?? new List<Messaging.Common.Storage>()
             };
 
         private object GenerateTaskCallbackMessage() =>

diff --git a/src/TaskManager/TaskManager.cs b/src/TaskManager/TaskManager.cs
@@ -312,7 +312,7 @@ private void AcknowledgeMessage<T>(JsonMessage<T> message)
             }
         }
 
-        private static JsonMessage<TaskUpdateEvent> GenerateUpdateEventMessage<T>(JsonMessage<T> message, string executionId, string WorkflowInstanceId, string taskId, ExecutionStatus executionStatus, Dictionary<string, string> outputArtifacts = null)
+        private static JsonMessage<TaskUpdateEvent> GenerateUpdateEventMessage<T>(JsonMessage<T> message, string executionId, string WorkflowInstanceId, string taskId, ExecutionStatus executionStatus, List<Messaging.Common.Storage> outputs = null)
         {
             Guard.Against.Null(message, nameof(message));
             Guard.Against.Null(executionStatus, nameof(executionStatus));
@@ -326,7 +326,7 @@ private static JsonMessage<TaskUpdateEvent> GenerateUpdateEventMessage<T>(JsonMe
                 WorkflowInstanceId = WorkflowInstanceId,
                 TaskId = taskId,
                 Message = executionStatus.Errors,
-                OutputArtifacts = outputArtifacts ?? new Dictionary<string, string>(),
+                Outputs = outputs ?? new List<Messaging.Common.Storage>(),
             }, TaskManagerApplicationId, message.CorrelationId);
         }
 

diff --git a/src/WorkflowExecuter/Common/EventMapper.cs b/src/WorkflowExecuter/Common/EventMapper.cs
@@ -44,7 +44,15 @@ public static TaskDispatchEvent ToTaskDispatchEvent(TaskExecution task, string w
                 TaskPluginArguments = task.TaskPluginArguments,
                 Inputs = inputs,
                 TaskPluginType = task.TaskType,
-                Metadata = task.Metadata
+                Metadata = task.Metadata,
+                IntermediateStorage = new Messaging.Common.Storage
+                {
+                    Bucket = configuration.Settings["bucket"],
+                    RelativeRootPath = $"{task.OutputDirectory}/tmp",
+                    Endpoint = configuration.Settings["endpoint"],
+                    Name = task.TaskId,
+                    SecuredConnection = bool.Parse(configuration.Settings["securedConnection"])
+                }
             };
         }
     }

diff --git a/tests/UnitTests/TaskManager.API.Tests/Extensions/StorageListExtensionsTests.cs b/tests/UnitTests/TaskManager.API.Tests/Extensions/StorageListExtensionsTests.cs
diff --git a/...UnitTests/TaskManager.API.Tests/Monai.Deploy.WorkflowManager.TaskManager.API.Tests.csproj b/...UnitTests/TaskManager.API.Tests/Monai.Deploy.WorkflowManager.TaskManager.API.Tests.csproj
diff --git a/tests/UnitTests/WorkflowExecuter.Tests/Common/EventMapperTests.cs b/tests/UnitTests/WorkflowExecuter.Tests/Common/EventMapperTests.cs
@@ -71,6 +71,14 @@ public void ToTaskDispatchEvent_ValidAeTitleWorkflowRequest_ReturnesTrue()
                 TaskPluginArguments = new Dictionary<string, string>
                 {
                     { "key", "value" }
+                },
+                IntermediateStorage = new Messaging.Common.Storage
+                {
+                    Bucket = configuration.Settings["bucket"],
+                    Endpoint = configuration.Settings["endpoint"],
+                    Name = task.TaskId,
+                    RelativeRootPath = "minio/workflowid/taskid/tmp",
+                    SecuredConnection = bool.Parse(configuration.Settings["securedConnection"])
                 }
             };
 

diff --git a/tests/UnitTests/WorkflowExecuter.Tests/Services/WorkflowExecuterServiceTests.cs b/tests/UnitTests/WorkflowExecuter.Tests/Services/WorkflowExecuterServiceTests.cs
@@ -47,7 +47,7 @@ public WorkflowExecuterServiceTests()
             _messageBrokerPublisherService = new Mock<IMessageBrokerPublisherService>();
             _storageService = new Mock<IStorageService>();
             _configuration = Options.Create(new WorkflowManagerOptions() { Messaging = new MessageBrokerConfiguration { Topics = new MessageBrokerConfigurationKeys { TaskDispatchRequest = "md.task.dispatch" } } });
-            _storageConfiguration = Options.Create(new StorageServiceConfiguration());
+            _storageConfiguration = Options.Create(new StorageServiceConfiguration() { Settings = new Dictionary<string, string> { { "bucket", "testbucket" }, { "endpoint", "localhost" }, { "securedConnection", "False" } } });
 
             WorkflowExecuterService = new WorkflowExecuterService(_logger.Object, _configuration, _storageConfiguration, _workflowRepository.Object, _workflowInstanceRepository.Object, _messageBrokerPublisherService.Object, _artifactMapper.Object, _storageService.Object);
         }