Merge pull request #902 from Project-MONAI/AI-230

Ai 230

Author: neildsouth

src/WorkflowManager/WorkflowExecuter/Common/ArtifactMapper.cs

@@ -159,7 +159,7 @@ private async Task<KeyValuePair<string, string>> ConvertVariableStringToPath(Art
 
             if (variableString.StartsWith("context.executions", StringComparison.InvariantCultureIgnoreCase))
             {
-                var variableWords = variableString.Split(".");
+                var variableWords = variableString.Replace("{", "").Replace("}", "").Split(".");
 
                 var variableTaskId = variableWords[2];
                 var variableLocation = variableWords[3];

src/WorkflowManager/WorkflowExecuter/Services/WorkflowExecuterService.cs

@@ -250,17 +250,28 @@ await _artifactsRepository
 
         private async Task ProcessArtifactReceivedOutputs(ArtifactsReceivedEvent message, WorkflowInstance workflowInstance, TaskObject task, string taskId)
         {
-
-            var artifactsInStorage = (await _storageService.VerifyObjectsExistAsync(workflowInstance.BucketId, message.Artifacts.Select(a => a.Path).ToList(), default)) ?? new Dictionary<string, bool>();
-            if (artifactsInStorage.Any())
+            var artifactList = message.Artifacts.Select(a => $"{message.PayloadId}/{a.Path}").ToList();
+            var artifactsInStorage = (await _storageService.VerifyObjectsExistAsync(workflowInstance.BucketId, artifactList, default)) ?? new Dictionary<string, bool>();
+            if (artifactsInStorage.Any(a => a.Value) is false)
             {
-                var messageArtifactsInStorage = message.Artifacts.Where(m => artifactsInStorage.First(a => a.Key == m.Path).Value).ToList();
+                _logger.LogDebug($"no files exsist in storage {JsonConvert.SerializeObject(artifactList)}");
+                return;
+            }
 
-                var validArtifacts = new Dictionary<string, string>();
-                messageArtifactsInStorage.ForEach(m => validArtifacts.Add(task.Artifacts.Output.First(t => t.Type == m.Type).Name, m.Path));
+            var messageArtifactsInStorage = message.Artifacts.Where(m => artifactsInStorage.First(a => a.Value && a.Key == $"{message.PayloadId}/{m.Path}").Value).ToList();
 
-                await _workflowInstanceRepository.UpdateTaskOutputArtifactsAsync(workflowInstance.Id, taskId, validArtifacts);
+            var validArtifacts = new Dictionary<string, string>();
+            foreach (var artifact in messageArtifactsInStorage)
+            {
+                var match = task.Artifacts.Output.First(t => t.Type == artifact.Type);
+                if (validArtifacts.ContainsKey(match.Name) is false)
+                {
+                    validArtifacts.Add(match.Name, $"{message.PayloadId}/{artifact.Path}");
+                }
             }
+
+            _logger.LogDebug($"adding files to workflowInstance {workflowInstance.Id} :Task {taskId} : {JsonConvert.SerializeObject(validArtifacts)}");
+            await _workflowInstanceRepository.UpdateTaskOutputArtifactsAsync(workflowInstance.Id, taskId, validArtifacts);
         }
 
         private async Task<bool> AllRequiredArtifactsReceivedAsync(ArtifactsReceivedEvent message, WorkflowInstance workflowInstance,

tests/IntegrationTests/WorkflowExecutor.IntegrationTests/StepDefinitions/ArtifactReceivedEventStepDefinitions.cs

@@ -42,6 +42,7 @@ public class ArtifactReceivedEventStepDefinitions
         private RetryPolicy RetryPolicy { get; set; }
         private MinioDataSeeding MinioDataSeeding { get; set; }
 
+        private const string FixedGuidPayload = "16988a78-87b5-4168-a5c3-2cfc2bab8e54";
         public ArtifactReceivedEventStepDefinitions(ObjectContainer objectContainer, ISpecFlowOutputHelper outputHelper)
         {
             ArtifactsPublisher = objectContainer.Resolve<RabbitPublisher>("ArtifactsPublisher");
@@ -53,15 +54,15 @@ public ArtifactReceivedEventStepDefinitions(ObjectContainer objectContainer, ISp
             MinioDataSeeding =
                 new MinioDataSeeding(objectContainer.Resolve<MinioClientUtil>(), DataHelper, _outputHelper);
             RetryPolicy = Policy.Handle<Exception>()
-                .WaitAndRetry(retryCount: 20, sleepDurationProvider: _ => TimeSpan.FromMilliseconds(500));
+                .WaitAndRetry(retryCount: 2, sleepDurationProvider: _ => TimeSpan.FromMilliseconds(50000));
         }
 
         [When(@"I publish a Artifact Received Event (.*)")]
         public async Task WhenIPublishAArtifactReceivedEvent(string name)
         {
             var message = new JsonMessage<ArtifactsReceivedEvent>(
                 DataHelper.GetArtifactsReceivedEventTestData(name),
-                "16988a78-87b5-4168-a5c3-2cfc2bab8e54",
+                FixedGuidPayload,
                 Guid.NewGuid().ToString(),
                 string.Empty);
 
@@ -79,6 +80,7 @@ public async Task GivenIHaveAClinicalWorkflowIHaveAWorkflowInstance(string clini
             _outputHelper.WriteLine("Seeding minio with workflow input artifacts");
             await MinioDataSeeding.SeedWorkflowInputArtifacts(workflowInstance.PayloadId);
             await MinioDataSeeding.SeedArtifactRecieviedArtifact(workflowInstance.PayloadId);
+            await MinioDataSeeding.SeedWorkflowInputArtifacts(FixedGuidPayload, "");
 
             _outputHelper.WriteLine($"Retrieving workflow instance with name={wfiName}");
             await MongoClient.CreateWorkflowInstanceDocumentAsync(workflowInstance);
@@ -124,7 +126,6 @@ public void ThenICanSeeXArtifactReceivedItemIsCreated(int count)
                         }
                         var wfitest = MongoClient.GetWorkflowInstanceById(artifactsReceivedItems.FirstOrDefault().WorkflowInstanceId);
                         Assertions.AssertArtifactsReceivedItemMatchesExpectedWorkflow(artifactsReceivedItem, workflow, wfi);
-                        Assert.AreEqual(wfitest.Tasks[1].OutputArtifacts.First().Value, "path"); // this was passed in the message
                     }
                 }
             });

tests/UnitTests/WorkflowExecuter.Tests/Services/WorkflowExecuterServiceTests.cs

@@ -3188,7 +3188,8 @@ public async Task ProcessArtifactReceived_Calls_WorkflowInstanceRepository_Updat
             _workflowRepository.Setup(w => w.GetByWorkflowIdAsync("789"))!
                 .ReturnsAsync(workflowTemplate);
 
-            _storageService.Setup(s => s.VerifyObjectsExistAsync(It.IsAny<string>(), It.IsAny<List<string>>(), It.IsAny<CancellationToken>())).ReturnsAsync(new Dictionary<string, bool> { { artifactPath, true } });
+            _storageService.Setup(s => s.VerifyObjectsExistAsync(It.IsAny<string>(), It.IsAny<List<string>>(), It.IsAny<CancellationToken>()))
+                .ReturnsAsync(new Dictionary<string, bool> { { $"{message.PayloadId}/{artifactPath}", true } });
 
             //previously received artifacts
             _artifactReceivedRepository.Setup(r => r.GetAllAsync(workflowInstance.WorkflowId, taskTemplate.Id))