Patch/0.3.17

.github/.gitversion.yml

@@ -29,4 +29,4 @@ branches:
 ignore:
  sha: []
 merge-message-formats: {}
-next-version: 0.3.15
+next-version: 0.3.16

.github/workflows/ci.yml

@@ -425,7 +425,7 @@ jobs:
 
  - name: Anchore container scan
  id: anchore-scan
- uses: anchore/scan-action@v3.3.4
+ uses: anchore/scan-action@v3.3.5
  if: ${{ (matrix.os == 'ubuntu-latest') }}
  with:
  image: ${{ fromJSON(steps.meta.outputs.json).tags[0] }}

src/Api/Storage/Payload.cs

@@ -76,6 +76,7 @@ public TimeSpan Elapsed
  public string? CalledAeTitle { get => Files.OfType<DicomFileStorageMetadata>().Select(p => p.CalledAeTitle).FirstOrDefault(); }
 
  public int FilesUploaded { get => Files.Count(p => p.IsUploaded); }
+
  public int FilesFailedToUpload { get => Files.Count(p => p.IsUploadFailed); }
 
  public Payload(string key, string correlationId, uint timeout)

src/Common/ExtensionMethods.cs

@@ -76,5 +76,15 @@ public static async Task<bool> Post<TInput>(this ActionBlock<TInput> actionBlock
  await Task.Delay(delay).ConfigureAwait(false);
  return actionBlock.Post(input);
  }
+
+ /// <summary>
+ /// Checks if a given task is faulted or cancelled.
+ /// </summary>
+ /// <param name="task">The task object</param>
+ /// <returns>True if canceled or faulted. False otherwise.</returns>
+ public static bool IsCanceledOrFaulted(this Task task)
+ {
+ return task.IsCanceled || task.IsFaulted;
+ }
  }
 }

src/Configuration/DicomWebConfiguration.cs

@@ -39,7 +39,7 @@ public class DicomWebConfiguration
  /// Gets or sets the maximum number of simultaneous DICOMweb connections.
  /// </summary>
  [ConfigurationKeyName("maximumNumberOfConnections")]
- public int MaximumNumberOfConnection { get; set; } = 2;
+ public ushort MaximumNumberOfConnection { get; set; } = 2;
 
  /// <summary>
  /// Gets or set the maximum allowed file size in bytes with default to 2GiB.

src/Configuration/ScuConfiguration.cs

@@ -54,7 +54,7 @@ public class ScuConfiguration
  /// Gets or sets the maximum number of simultaneous DICOM associations for the SCU service.
  /// </summary>
  [ConfigurationKeyName("maximumNumberOfAssociations")]
- public int MaximumNumberOfAssociations { get; set; } = 8;
+ public ushort MaximumNumberOfAssociations { get; set; } = 8;
 
  public ScuConfiguration()
  {

src/InformaticsGateway/Common/PostPayloadException.cs

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2023 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.Runtime.Serialization;
+using Monai.Deploy.InformaticsGateway.Api.Storage;
+
+namespace Monai.Deploy.InformaticsGateway.Common
+{
+ internal class PostPayloadException : Exception
+ {
+ public Payload.PayloadState TargetQueue { get; }
+ public Payload Payload { get; }
+
+ public PostPayloadException()
+ {
+ }
+
+ public PostPayloadException(Api.Storage.Payload.PayloadState targetState, Payload payload)
+ {
+ TargetQueue = targetState;
+ Payload = payload;
+ }
+
+ public PostPayloadException(string message) : base(message)
+ {
+ }
+
+ public PostPayloadException(string message, Exception innerException) : base(message, innerException)
+ {
+ }
+
+ protected PostPayloadException(SerializationInfo info, StreamingContext context) : base(info, context)
+ {
+ }
+ }
+}

src/InformaticsGateway/Logging/Log.3000.PayloadAssembler.cs

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 MONAI Consortium
+ * Copyright 2022-2023 MONAI Consortium
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -33,8 +33,8 @@ public static partial class Log
  [LoggerMessage(EventId = 3004, Level = LogLevel.Trace, Message = "Number of incomplete payloads waiting for processing: {count}.")]
  public static partial void BucketsActive(this ILogger logger, int count);
 
- [LoggerMessage(EventId = 3005, Level = LogLevel.Trace, Message = "Checking elapsed time for bucket: {key} with timeout set to {timeout}s. Elapsed {elapsed}s with {failedFiles} failures out of {totalNumberOfFiles}.")]
- public static partial void BucketElapsedTime(this ILogger logger, string key, uint timeout, double elapsed, int totalNumberOfFiles, int failedFiles);
+ [LoggerMessage(EventId = 3005, Level = LogLevel.Trace, Message = "Checking elapsed time for bucket: {key} with timeout set to {timeout}s. Elapsed {elapsed}s with {succeededFiles} uplaoded and {failedFiles} failures  out of {totalNumberOfFiles}.")]
+ public static partial void BucketElapsedTime(this ILogger logger, string key, uint timeout, double elapsed, int totalNumberOfFiles, int succeededFiles, int failedFiles);
 
  [LoggerMessage(EventId = 3007, Level = LogLevel.Information, Message = "Bucket {key} sent to processing queue with {count} files.")]
  public static partial void BucketReady(this ILogger logger, string key, int count);

src/InformaticsGateway/Logging/Log.4000.ObjectUploadService.cs

@@ -54,7 +54,7 @@ public static partial class Log
  [LoggerMessage(EventId = 4010, Level = LogLevel.Debug, Message = "File {path} exists={exists}.")]
  public static partial void VerifyFileExists(this ILogger logger, string path, bool exists);
 
- [LoggerMessage(EventId = 4011, Level = LogLevel.Information, Message = "Initializing Object Uploader service with {threads} workers.")]
+ [LoggerMessage(EventId = 4011, Level = LogLevel.Information, Message = "Initializing Object Uploser service with {threads} workers.")]
  public static partial void InitializeThreads(this ILogger logger, int threads);
  }
 }

src/InformaticsGateway/Logging/Log.500.ExportService.cs

@@ -123,5 +123,11 @@ public static partial class Log
 
  [LoggerMessage(EventId = 533, Level = LogLevel.Error, Message = "Recovering messaging service connection due to {reason}.")]
  public static partial void MessagingServiceErrorRecover(this ILogger logger, string reason);
+
+ [LoggerMessage(EventId = 534, Level = LogLevel.Error, Message = "Error posting export job for processing correlation ID {correlationId}, export task ID {exportTaskId}.")]
+ public static partial void ErrorPostingExportJobToQueue(this ILogger logger, string correlationId, string exportTaskId);
+
+ [LoggerMessage(EventId = 535, Level = LogLevel.Warning, Message = "Exceeded maximum number of worker in {serviceName}: {count}.")]
+ public static partial void ExceededMaxmimumNumberOfWorkers(this ILogger logger, string serviceName, ulong count);
  }
 }

src/InformaticsGateway/Logging/Log.700.PayloadService.cs

@@ -136,5 +136,17 @@ public static partial class Log
 
  [LoggerMessage(EventId = 743, Level = LogLevel.Error, Message = "Exception moving payload.")]
  public static partial void PayloadMoveException(this ILogger logger, Exception ex);
+
+ [LoggerMessage(EventId = 744, Level = LogLevel.Warning, Message = "PayloadNotification move payload queue: faulted: {isFauled}, cancelled: {isCancelled}.")]
+ public static partial void MoveQueueFaulted(this ILogger logger, bool isFauled, bool isCancelled);
+
+ [LoggerMessage(EventId = 745, Level = LogLevel.Warning, Message = "PayloadNotification publishing payload queue: faulted: {isFauled}, cancelled: {isCancelled}.")]
+ public static partial void PublishQueueFaulted(this ILogger logger, bool isFauled, bool isCancelled);
+
+ [LoggerMessage(EventId = 746, Level = LogLevel.Error, Message = "Error posting payload to move queue.")]
+ public static partial void ErrorPostingJobToMovePayloadsQueue(this ILogger logger);
+
+ [LoggerMessage(EventId = 747, Level = LogLevel.Error, Message = "Error posting payload to publish queue.")]
+ public static partial void ErrorPostingJobToPublishPayloadsQueue(this ILogger logger);
  }
 }

src/InformaticsGateway/Services/Connectors/PayloadAssembler.cs

@@ -135,7 +135,7 @@ private async void OnTimedEvent(Object source, System.Timers.ElapsedEventArgs e)
  var payload = await _payloads[key].Task.ConfigureAwait(false);
  using var loggerScope = _logger.BeginScope(new LoggingDataDictionary<string, object> { { "CorrelationId", payload.CorrelationId } });
 
- _logger.BucketElapsedTime(key, payload.Timeout, payload.ElapsedTime().TotalSeconds, payload.Files.Count, payload.FilesFailedToUpload);
+ _logger.BucketElapsedTime(key, payload.Timeout, payload.ElapsedTime().TotalSeconds, payload.Files.Count, payload.FilesUploaded, payload.FilesFailedToUpload);
  // Wait for timer window closes before sending payload for processing
  if (payload.HasTimedOut)
  {

src/InformaticsGateway/Services/Connectors/PayloadMoveActionHandler.cs

@@ -86,7 +86,10 @@ public async Task MoveFilesAsync(Payload payload, ActionBlock<Payload> moveQueue
  var action = await UpdatePayloadState(payload, ex, cancellationToken).ConfigureAwait(false);
  if (action == PayloadAction.Updated)
  {
- await moveQueue.Post(payload, _options.Value.Storage.Retries.RetryDelays.ElementAt(payload.RetryCount - 1)).ConfigureAwait(false);
+ if (!await moveQueue.Post(payload, _options.Value.Storage.Retries.RetryDelays.ElementAt(payload.RetryCount - 1)).ConfigureAwait(false))
+ {
+ throw new PostPayloadException(Payload.PayloadState.Move, payload);
+ }
  }
  }
  finally
@@ -111,7 +114,11 @@ private async Task NotifyIfCompleted(Payload payload, ActionBlock<Payload> notif
  await repository.UpdateAsync(payload, cancellationToken).ConfigureAwait(false);
  _logger.PayloadSaved(payload.PayloadId);
 
- notificationQueue.Post(payload);
+ if (!notificationQueue.Post(payload))
+ {
+ throw new PostPayloadException(Payload.PayloadState.Notify, payload);
+ }
+
  _logger.PayloadReadyToBePublished(payload.PayloadId);
  }
  else // we should never hit this else block.
@@ -185,7 +192,7 @@ await _storageService.CopyObjectAsync(
 
  await VerifyFileExists(payloadId, file, cancellationToken).ConfigureAwait(false);
  }
- catch (StorageServiceException ex) when (ex.Message.Contains("Not found", StringComparison.OrdinalIgnoreCase)) // TODO: StorageLib shall not throw any errors from MINIO
+ catch (StorageObjectNotFoundException ex) when (ex.Message.Contains("Not found", StringComparison.OrdinalIgnoreCase)) // TODO: StorageLib shall not throw any errors from MINIO
  {
  // when file cannot be found on the Storage Service, we assume file has been moved previously by verifying the file exists on destination.
  _logger.FileMissingInPayload(payloadId, file.GetTempStoragPath(_options.Value.Storage.RemoteTemporaryStoragePath), ex);

src/InformaticsGateway/Services/Connectors/PayloadNotificationActionHandler.cs

@@ -80,8 +80,11 @@ public async Task NotifyAsync(Payload payload, ActionBlock<Payload> notification
  var action = await UpdatePayloadState(payload, cancellationToken).ConfigureAwait(false);
  if (action == PayloadAction.Updated)
  {
- await notificationQueue.Post(payload, _options.Value.Messaging.Retries.RetryDelays.ElementAt(payload.RetryCount - 1)).ConfigureAwait(false);
  _logger.FailedToPublishWorkflowRequest(payload.PayloadId, ex);
+ if (!await notificationQueue.Post(payload, _options.Value.Messaging.Retries.RetryDelays.ElementAt(payload.RetryCount - 1)).ConfigureAwait(false))
+ {
+ throw new PostPayloadException(Payload.PayloadState.Notify, payload);
+ }
  }
  }
  }