Ai 230

src/Common/Configuration/MessageBrokerConfigurationKeys.cs

@@ -88,5 +88,8 @@ public class MessageBrokerConfigurationKeys
 
         [ConfigurationKeyName("notificationEmailCancelation")]
         public string NotificationEmailCancelation { get; set; } = "aide.notification_email.cancellation";
+
+        [ConfigurationKeyName("artifactrecieved")]
+        public string ArtifactRecieved { get; set; } = "md.workflow.artifactrecieved";
     }
 }

src/Monai.Deploy.WorkflowManager.sln.DotSettings

@@ -1,4 +1,5 @@
 <wpf:ResourceDictionary xml:space="preserve" xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml" xmlns:s="clr-namespace:System;assembly=mscorlib" xmlns:ss="urn:shemas-jetbrains-com:settings-storage-xaml" xmlns:wpf="http://schemas.microsoft.com/winfx/2006/xaml/presentation">
+	<s:Boolean x:Key="/Default/CodeInspection/Highlighting/ConvertIfStatementToSwitchStatement/AssumeOpenTypeHierarchy/@EntryValue">True</s:Boolean>
     <s:String x:Key="/Default/CodeStyle/Naming/CSharpNaming/Abbreviations/=AR/@EntryIndexedValue">AR</s:String>
     <s:String x:Key="/Default/CodeStyle/Naming/CSharpNaming/Abbreviations/=AS/@EntryIndexedValue">AS</s:String>
     <s:String x:Key="/Default/CodeStyle/Naming/CSharpNaming/Abbreviations/=ASMT/@EntryIndexedValue">ASMT</s:String>

src/WorkflowManager/Database/Interfaces/IArtifactsRepository.cs

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2022 MONAI Consortium
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Artifact = Monai.Deploy.Messaging.Common.Artifact;
+
+namespace Monai.Deploy.WorkflowManager.Common.Database.Repositories
+{
+    public interface IArtifactsRepository
+    {
+        /// <summary>
+        /// Gets All ArtifactsReceivedItems by workflowInstance and taskId.
+        /// </summary>
+        /// <param name="workflowInstance"></param>
+        /// <param name="taskId"></param>
+        /// <returns></returns>
+        Task<List<ArtifactReceivedItems>?> GetAllAsync(string workflowInstance, string taskId);
+
+        /// <summary>
+        /// Adds an item to the ArtifactsReceivedItems collection.
+        /// </summary>
+        /// <param name="item"></param>
+        /// <returns></returns>
+        Task AddItemAsync(ArtifactReceivedItems item);
+
+        /// <summary>
+        /// Adds an item to the ArtifactsReceivedItems collection.
+        /// </summary>
+        /// <param name="workflowInstanceId"></param>
+        /// <param name="taskId"></param>
+        /// <param name="artifactsOutputs"></param>
+        /// <returns></returns>
+        Task AddItemAsync(string workflowInstanceId, string taskId, List<Artifact> artifactsOutputs);
+
+        Task AddOrUpdateItemAsync(string workflowInstanceId, string taskId,
+            IEnumerable<Artifact> artifactsOutputs);
+    }
+}

src/WorkflowManager/Database/Repositories/ArtifactsRepository.cs

@@ -0,0 +1,152 @@
+/*
+ * Copyright 2022 MONAI Consortium
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using Monai.Deploy.WorkflowManager.Common.Database.Options;
+using MongoDB.Driver;
+using Artifact = Monai.Deploy.Messaging.Common.Artifact;
+
+namespace Monai.Deploy.WorkflowManager.Common.Database.Repositories
+{
+    public class ArtifactReceivedDetails : Artifact
+    {
+        /// <summary>
+        /// Gets or Sets LastReceived.
+        /// </summary>
+        public DateTime? Received { get; set; } = null;
+
+        public static ArtifactReceivedDetails FromArtifact(Artifact artifact) =>
+            new()
+            {
+                Received = DateTime.UtcNow,
+                Type = artifact.Type,
+                Path = artifact.Path
+            };
+    }
+
+    public class ArtifactReceivedItems
+    {
+        /// <summary>
+        /// Gets or Sets the Id.
+        /// </summary>
+        public double Id { get; set; }
+
+        /// <summary>
+        /// Gets or Sets WorkflowInstanceId.
+        /// </summary>
+        public string WorkflowInstanceId { get; set; } = string.Empty;
+
+        /// <summary>
+        /// Gets or Sets TaskId.
+        /// </summary>
+        public string TaskId { get; set; } = string.Empty;
+
+        /// <summary>
+        /// Gets or Sets Artifacts.
+        /// </summary>
+        public List<ArtifactReceivedDetails> Artifacts { get; set; } = new();
+    }
+
+    public class ArtifactsRepository : IArtifactsRepository
+    {
+        private readonly ILogger<ArtifactsRepository> _logger;
+        private readonly IMongoCollection<ArtifactReceivedItems> _artifactReceivedItemsCollection;
+
+        public ArtifactsRepository(
+            IMongoClient client,
+            IOptions<WorkloadManagerDatabaseSettings> bookStoreDatabaseSettings,
+            ILogger<ArtifactsRepository> logger)
+        {
+            if (client == null)
+            {
+                throw new ArgumentNullException(nameof(client));
+            }
+
+            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+            var mongoDatabase = client.GetDatabase(bookStoreDatabaseSettings.Value.DatabaseName);
+            _artifactReceivedItemsCollection = mongoDatabase.GetCollection<ArtifactReceivedItems>("ArtifactReceivedItems");
+        }
+
+        public async Task<List<ArtifactReceivedItems>> GetAllAsync(string workflowInstance, string taskId)
+        {
+            var result = await _artifactReceivedItemsCollection.FindAsync(a => a.WorkflowInstanceId == workflowInstance && a.TaskId == taskId).ConfigureAwait(false);
+            return await result.ToListAsync().ConfigureAwait(false);
+        }
+
+        public Task AddItemAsync(ArtifactReceivedItems item)
+        {
+            return _artifactReceivedItemsCollection.InsertOneAsync(item);
+        }
+
+        public Task AddItemAsync(string workflowInstanceId, string taskId, List<Artifact> artifactsOutputs)
+        {
+            var artifacts = artifactsOutputs.Select(a => new ArtifactReceivedDetails()
+            {
+                Path = a.Path,
+                Type = a.Type,
+                Received = DateTime.UtcNow
+            });
+
+            var item = new ArtifactReceivedItems()
+            {
+                WorkflowInstanceId = workflowInstanceId,
+                TaskId = taskId,
+                Artifacts = artifacts.ToList()
+            };
+
+            return _artifactReceivedItemsCollection.InsertOneAsync(item);
+        }
+
+        public async Task AddOrUpdateItemAsync(string workflowInstanceId, string taskId,
+            IEnumerable<Artifact> artifactsOutputs)
+        {
+            var artifacts = artifactsOutputs.Select(a => new ArtifactReceivedDetails()
+            {
+                Path = a.Path,
+                Type = a.Type,
+                Received = DateTime.UtcNow
+            });
+
+            var item = new ArtifactReceivedItems()
+            {
+                WorkflowInstanceId = workflowInstanceId,
+                TaskId = taskId,
+                Artifacts = artifacts.ToList()
+            };
+
+            var result = await _artifactReceivedItemsCollection
+                .FindAsync(a => a.WorkflowInstanceId == workflowInstanceId && a.TaskId == taskId).ConfigureAwait(false);
+            var existing = await result.FirstOrDefaultAsync().ConfigureAwait(false);
+
+            if (existing == null)
+            {
+                await _artifactReceivedItemsCollection.InsertOneAsync(item).ConfigureAwait(false);
+            }
+            else
+            {
+                var update = Builders<ArtifactReceivedItems>.Update.Set(a => a.Artifacts, item.Artifacts);
+                await _artifactReceivedItemsCollection
+                    .UpdateOneAsync(a => a.WorkflowInstanceId == workflowInstanceId && a.TaskId == taskId, update)
+                    .ConfigureAwait(false);
+            }
+        }
+    }
+}

src/WorkflowManager/Database/Repositories/RepositoryBase.cs

@@ -24,8 +24,8 @@ namespace Monai.Deploy.WorkflowManager.Common.Database.Repositories
 {
     public abstract class RepositoryBase
     {
-        public static async Task<long> CountAsync<T>(IMongoCollection<T> collection, FilterDefinition<T>? filter)
-            => await collection.CountDocumentsAsync(filter ?? Builders<T>.Filter.Empty);
+        public static Task<long> CountAsync<T>(IMongoCollection<T> collection, FilterDefinition<T>? filter)
+            => collection.CountDocumentsAsync(filter ?? Builders<T>.Filter.Empty);
 
         /// <summary>
         /// Get All T that match filters provided.
@@ -44,7 +44,7 @@ public static async Task<IList<T>> GetAllAsync<T>(IMongoCollection<T> collection
                 .Skip(skip)
                 .Limit(limit)
                 .Sort(sortFunction)
-                .ToListAsync();
+                .ToListAsync().ConfigureAwait(false);
         }
 
         public static async Task<IList<T>> GetAllAsync<T>(IMongoCollection<T> collection, FilterDefinition<T> filterFunction, SortDefinition<T> sortFunction, int? skip = null, int? limit = null)
@@ -54,7 +54,7 @@ public static async Task<IList<T>> GetAllAsync<T>(IMongoCollection<T> collection
                 .Skip(skip)
                 .Limit(limit)
                 .Sort(sortFunction)
-                .ToListAsync();
+                .ToListAsync().ConfigureAwait(false);
         }
     }
 }

src/WorkflowManager/Logging/Log.200000.Workflow.cs

@@ -81,6 +81,9 @@ public static partial class Log
         [LoggerMessage(EventId = 200019, Level = LogLevel.Debug, Message = "Task destination condition for task {taskId} with resolved condition: {resolvedConditional} resolved to false. initial conditional: {conditions}")]
         public static partial void TaskDestinationConditionFalse(this ILogger logger, string resolvedConditional, string conditions, string taskId);
 
+        [LoggerMessage(EventId = 200020, Level = LogLevel.Warning, Message = "Use new ArtifactReceived Queue for continuation messages.")]
+        public static partial void DontUseWorkflowReceivedForPayload(this ILogger logger);
+
         // Conditions Resolver
         [LoggerMessage(EventId = 210000, Level = LogLevel.Warning, Message = "Failed to parse condition: {condition}. resolvedConditional: {resolvedConditional}")]
         public static partial void FailedToParseCondition(this ILogger logger, string resolvedConditional, string condition, Exception ex);

src/WorkflowManager/Logging/Log.500000.Messaging.cs

@@ -71,7 +71,20 @@ public static partial class Log
         [LoggerMessage(EventId = 500016, Level = LogLevel.Debug, Message = "Export complete message received.")]
         public static partial void ExportCompleteReceived(this ILogger logger);
 
-        [LoggerMessage(EventId = 200017, Level = LogLevel.Debug, Message = "Workflow continuation event so not creating payload.")]
+        [LoggerMessage(EventId = 500017, Level = LogLevel.Debug, Message = "ArtifactReceived message so not creating payload.")]
         public static partial void WorkflowContinuation(this ILogger logger);
+
+        [LoggerMessage(EventId = 500018, Level = LogLevel.Debug, Message = "ArtifactReceived message received.")]
+        public static partial void ArtifactReceivedReceived(this ILogger logger);
+
+        [LoggerMessage(EventId = 500019, Level = LogLevel.Error, Message = "ArtifactReceived message {messageId} failed unexpectedly (no workflowId or TaskId ?) and has been requeued.")]
+        public static partial void ArtifactReceivedRequeuePayloadCreateError(this ILogger logger, string messageId);
+
+        [LoggerMessage(EventId = 500020, Level = LogLevel.Error, Message = "ArtifactReveived message {messageId} failed unexpectedly and has been requeued.")]
+        public static partial void ArtifactReceivedRequeueUnknownError(this ILogger logger, string messageId, Exception ex);
+
+        [LoggerMessage(EventId = 500021, Level = LogLevel.Error, Message = "ArtifactReveived message {messageId} is invalid and has been rejected without being requeued.")]
+        public static partial void ArtifactReceivedRejectValidationError(this ILogger logger, string messageId);
+
     }
 }

src/WorkflowManager/Logging/Log.700000.Artifact.cs

@@ -44,5 +44,12 @@ public static partial class Log
 
         [LoggerMessage(EventId = 700007, Level = LogLevel.Information, Message = "Task Dispatch resolved successfully output artifacts: PayloadId: {payloadId}, TaskId: {taskId}, WorkflowInstanceId: {workflowInstanceId}, WorkflowRevisionId: {workflowRevisionId}, output artifact object: {pathOutputArtifacts}")]
         public static partial void LogGeneralTaskDispatchInformation(this ILogger logger, string payloadId, string taskId, string workflowInstanceId, string workflowRevisionId, string pathOutputArtifacts);
+
+        [LoggerMessage(EventId = 700008, Level = LogLevel.Warning, Message = "Unexpected Artifacts output artifacts: TaskId: {taskId}, WorkflowInstanceId: {workflowInstanceId}, output artifact object: {unexpectedArtifacts}")]
+        public static partial void UnexpectedArtifactsReceived(this ILogger logger, string taskId, string workflowInstanceId, string unexpectedArtifacts);
+
+        [LoggerMessage(EventId = 700009, Level = LogLevel.Debug, Message = "Mandatory output artifacts for task {taskId} are missing. waiting for remaining artifacts... {missingArtifacts}")]
+        public static partial void MandatoryOutputArtifactsMissingForTask(this ILogger logger, string taskId, string missingArtifacts);
+
     }
 }

src/WorkflowManager/PayloadListener/Extensions/ValidationExtensions.cs

@@ -15,6 +15,7 @@
  */
 
 using Ardalis.GuardClauses;
+using Monai.Deploy.Messaging.Common;
 using Monai.Deploy.Messaging.Events;
 using Monai.Deploy.WorkflowManager.Common.Contracts.Models;
 
@@ -39,6 +40,39 @@ public static bool IsValid(this WorkflowRequestEvent workflowRequestMessage, out
             return valid;
         }
 
+        public static bool IsValid(this ArtifactsReceivedEvent artifactReceivedMessage, out IList<string> validationErrors)
+        {
+            Guard.Against.Null(artifactReceivedMessage, nameof(artifactReceivedMessage));
+
+            validationErrors = new List<string>();
+
+            var valid = true;
+
+            valid &= IsAeTitleValid(artifactReceivedMessage.GetType().Name, artifactReceivedMessage.DataTrigger.Source, validationErrors);
+            valid &= IsAeTitleValid(artifactReceivedMessage.GetType().Name, artifactReceivedMessage.DataTrigger.Destination, validationErrors);
+            valid &= IsBucketValid(artifactReceivedMessage.GetType().Name, artifactReceivedMessage.Bucket, validationErrors);
+            valid &= IsCorrelationIdValid(artifactReceivedMessage.GetType().Name, artifactReceivedMessage.CorrelationId, validationErrors);
+            valid &= IsPayloadIdValid(artifactReceivedMessage.GetType().Name, artifactReceivedMessage.PayloadId.ToString(), validationErrors);
+            valid &= string.IsNullOrEmpty(artifactReceivedMessage.WorkflowInstanceId) is false && string.IsNullOrEmpty(artifactReceivedMessage.TaskId) is false;
+            valid &= AllArtifactsAreValid(artifactReceivedMessage, validationErrors);
+            return valid;
+        }
+
+        private static bool AllArtifactsAreValid(this ArtifactsReceivedEvent artifactReceivedMessage, IList<string> validationErrors)
+        {
+            Guard.Against.Null(artifactReceivedMessage, nameof(artifactReceivedMessage));
+
+            var valid = artifactReceivedMessage.Artifacts.All(a => a.Type != ArtifactType.Unset);
+
+            if (valid is false)
+            {
+                var unsetArtifacts = string.Join(',', artifactReceivedMessage.Artifacts.Where(a => a.Type == ArtifactType.Unset).Select(a => a.Path));
+                validationErrors.Add($"The following artifacts are have unset artifact types: {unsetArtifacts}");
+            }
+
+            return valid;
+        }
+
         public static bool IsInformaticsGatewayNotNull(string source, InformaticsGateway informaticsGateway, IList<string> validationErrors)
         {
             Guard.Against.NullOrWhiteSpace(source, nameof(source));

src/WorkflowManager/PayloadListener/Services/EventPayloadRecieverService.cs

@@ -178,5 +178,47 @@ public async Task ExportCompletePayload(MessageReceivedEventArgs message)
                 await _messageSubscriber.RequeueWithDelay(message.Message);
             }
         }
+
+        public async Task ArtifactReceivePayload(MessageReceivedEventArgs message)
+        {
+            try
+            {
+                var requestEvent = message.Message.ConvertTo<ArtifactsReceivedEvent>();
+
+                using var loggingScope = Logger.BeginScope(new LoggingDataDictionary<string, object>
+                {
+                    ["correlationId"] = requestEvent.CorrelationId,
+                    ["workflowId"] = requestEvent.Workflows.FirstOrDefault(),
+                    ["workflowInstanceId"] = requestEvent.WorkflowInstanceId,
+                    ["taskId"] = requestEvent.TaskId
+                });
+
+                Logger.WorkflowContinuation();
+
+                var validation = PayloadValidator.ValidateArtifactReceived(requestEvent);
+
+                if (!validation)
+                {
+                    Logger.ArtifactReceivedRejectValidationError(message.Message.MessageId);
+                    _messageSubscriber.Reject(message.Message, false);
+
+                    return;
+                }
+
+                if (!await WorkflowExecuterService.ProcessArtifactReceivedAsync(requestEvent))
+                {
+                    Logger.ArtifactReceivedRequeuePayloadCreateError(message.Message.MessageId);
+                    await _messageSubscriber.RequeueWithDelay(message.Message);
+
+                    return;
+                }
+                _messageSubscriber.Acknowledge(message.Message);
+            }
+            catch (Exception e)
+            {
+                Logger.ArtifactReceivedRequeueUnknownError(message.Message.MessageId, e);
+                await _messageSubscriber.RequeueWithDelay(message.Message);
+            }
+        }
     }
 }