Skip to content

Commit b44839a

Browse files
woodheadiowoodheadio
authored
Merge pull request #395 from Project-MONAI/AC-1298
remove the need for a double copy to minio
2 parents 70bd3da + 556d2c3 commit b44839a

20 files changed

+68
-183
lines changed

‎src/Api/Storage/FileStorageMetadata.cs

+2
Original file line numberDiff line numberDiff line change
@@ -116,5 +116,7 @@ public virtual void SetFailed()
116116
{
117117
File.SetFailed();
118118
}
119+
120+
public string? PayloadId { get; set; }
119121
}
120122
}

‎src/Configuration/ConfigurationValidator.cs

+1-1
Original file line numberDiff line numberDiff line change
@@ -120,7 +120,7 @@ private bool IsValidDirectory(string source, string directory)
120120
private bool IsValidBucketName(string source, string bucketName)
121121
{
122122
var valid = IsNotNullOrWhiteSpace(source, bucketName);
123-
var regex = new Regex("(?=^.{3,63}$)(^[a-z0-9]+[a-z0-9\\-]+[a-z0-9]+$)");
123+
var regex = new Regex("(?=^.{3,63}$)(^[a-z0-9]+[a-z0-9\\-]+[a-z0-9]+$)", new RegexOptions(), TimeSpan.FromSeconds(5));
124124
if (!regex.IsMatch(bucketName))
125125
{
126126
valid = false;

‎src/Database/EntityFramework/Test/SourceApplicationEntityRepositoryTest.cs

+1
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,7 @@ public async Task GivenAAETitleName_WhenFindByAETAsyncIsCalled_ExpectItToReturnM
119119
Assert.Equal("AET1", actual.FirstOrDefault()!.Name);
120120

121121
actual = await store.FindByAETAsync("AET6").ConfigureAwait(false);
122+
Assert.NotNull(actual);
122123
Assert.Empty(actual);
123124
}
124125

‎src/Database/MongoDB/Integration.Test/SourceApplicationEntityRepositoryTest.cs

+1
Original file line numberDiff line numberDiff line change
@@ -124,6 +124,7 @@ public async Task GivenAETitle_WhenFindByAETitleAsyncIsCalled_ExpectItToReturnMa
124124
Assert.Equal("AET1", actual.FirstOrDefault()!.Name);
125125

126126
actual = await store.FindByAETAsync("AET6").ConfigureAwait(false);
127+
Assert.NotNull(actual);
127128
Assert.Empty(actual);
128129
}
129130

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

+5-5
Original file line numberDiff line numberDiff line change
@@ -27,20 +27,20 @@ public static partial class Log
2727
[LoggerMessage(EventId = 4001, Level = LogLevel.Debug, Message = "Upload statistics: {threads} threads, {seconds} seconds.")]
2828
public static partial void UploadStats(this ILogger logger, int threads, double seconds);
2929

30-
[LoggerMessage(EventId = 4002, Level = LogLevel.Debug, Message = "Uploading file to temporary store at {filePath}.")]
31-
public static partial void UploadingFileToTemporaryStore(this ILogger logger, string filePath);
30+
[LoggerMessage(EventId = 4002, Level = LogLevel.Debug, Message = "Uploading file to storeage at {filePath}.")]
31+
public static partial void UploadingFileToStoreage(this ILogger logger, string filePath);
3232

3333
[LoggerMessage(EventId = 4003, Level = LogLevel.Information, Message = "Instance queued for upload {identifier}. Items in queue {count} using memory {memoryUsageKb}KB.")]
3434
public static partial void InstanceAddedToUploadQueue(this ILogger logger, string identifier, int count, double memoryUsageKb);
3535

3636
[LoggerMessage(EventId = 4004, Level = LogLevel.Debug, Message = "Error removing objects that are pending upload during startup.")]
3737
public static partial void ErrorRemovingPendingUploadObjects(this ILogger logger, Exception ex);
3838

39-
[LoggerMessage(EventId = 4005, Level = LogLevel.Error, Message = "Error uploading temporary store. Waiting {timeSpan} before next retry. Retry attempt {retryCount}.")]
39+
[LoggerMessage(EventId = 4005, Level = LogLevel.Error, Message = "Error uploading storeage. Waiting {timeSpan} before next retry. Retry attempt {retryCount}.")]
4040
public static partial void ErrorUploadingFileToTemporaryStore(this ILogger logger, TimeSpan timespan, int retryCount, Exception ex);
4141

42-
[LoggerMessage(EventId = 4006, Level = LogLevel.Information, Message = "File uploaded to temporary store at {filePath}.")]
43-
public static partial void UploadedFileToTemporaryStore(this ILogger logger, string filePath);
42+
[LoggerMessage(EventId = 4006, Level = LogLevel.Information, Message = "File uploaded to storeage at {filePath}.")]
43+
public static partial void UploadedFileToStoreage(this ILogger logger, string filePath);
4444

4545
[LoggerMessage(EventId = 4007, Level = LogLevel.Debug, Message = "Items in queue {count}.")]
4646
public static partial void InstanceInUploadQueue(this ILogger logger, int count);

‎src/InformaticsGateway/Services/Connectors/DataRetrievalService.cs

+4-1
Original file line numberDiff line numberDiff line change
@@ -198,8 +198,11 @@ private async Task NotifyNewInstance(InferenceRequest inferenceRequest, Dictiona
198198
{
199199
retrievedFiles[key].SetWorkflows(inferenceRequest.Application.Id);
200200
}
201+
var FileMeta = retrievedFiles[key];
202+
203+
var payloadId = await _payloadAssembler.Queue(inferenceRequest.TransactionId, retrievedFiles[key]).ConfigureAwait(false);
204+
retrievedFiles[key].PayloadId = payloadId.ToString();
201205
_uploadQueue.Queue(retrievedFiles[key]);
202-
await _payloadAssembler.Queue(inferenceRequest.TransactionId, retrievedFiles[key]).ConfigureAwait(false);
203206
}
204207
}
205208

‎src/InformaticsGateway/Services/Connectors/IPayloadAssembler.cs

+3-2
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
* limitations under the License.
1515
*/
1616

17+
using System;
1718
using System.Threading;
1819
using System.Threading.Tasks;
1920
using Monai.Deploy.InformaticsGateway.Api.Storage;
@@ -30,15 +31,15 @@ internal interface IPayloadAssembler
3031
/// </summary>
3132
/// <param name="bucket">The bucket group the file belongs to.</param>
3233
/// <param name="file">Path to the file to be added to the payload bucket.</param>
33-
Task Queue(string bucket, FileStorageMetadata file);
34+
Task<Guid> Queue(string bucket, FileStorageMetadata file);
3435

3536
/// <summary>
3637
/// Queue a new file for the spcified payload bucket.
3738
/// </summary>
3839
/// <param name="bucket">The bucket group the file belongs to.</param>
3940
/// <param name="file">Path to the file to be added to the payload bucket.</param>
4041
/// <param name="timeout">Number of seconds to wait for additional files.</param>
41-
Task Queue(string bucket, FileStorageMetadata file, uint timeout);
42+
Task<Guid> Queue(string bucket, FileStorageMetadata file, uint timeout);
4243

4344
/// <summary>
4445
/// Dequeue a payload from the queue for the message broker to notify subscribers.

‎src/InformaticsGateway/Services/Connectors/PayloadAssembler.cs

+3-2
Original file line numberDiff line numberDiff line change
@@ -87,15 +87,15 @@ private async Task RemovePendingPayloads()
8787
/// </summary>
8888
/// <param name="bucket">Name of the bucket where the file would be added to</param>
8989
/// <param name="file">Instance to be queued</param>
90-
public async Task Queue(string bucket, FileStorageMetadata file) => await Queue(bucket, file, DEFAULT_TIMEOUT).ConfigureAwait(false);
90+
public async Task<Guid> Queue(string bucket, FileStorageMetadata file) => await Queue(bucket, file, DEFAULT_TIMEOUT).ConfigureAwait(false);
9191

9292
/// <summary>
9393
/// Queues a new instance of <see cref="FileStorageMetadata"/>.
9494
/// </summary>
9595
/// <param name="bucket">Name of the bucket where the file would be added to</param>
9696
/// <param name="file">Instance to be queued</param>
9797
/// <param name="timeout">Number of seconds the bucket shall wait before sending the payload to be processed. Note: timeout cannot be modified once the bucket is created.</param>
98-
public async Task Queue(string bucket, FileStorageMetadata file, uint timeout)
98+
public async Task<Guid> Queue(string bucket, FileStorageMetadata file, uint timeout)
9999
{
100100
Guard.Against.Null(file);
101101

@@ -106,6 +106,7 @@ public async Task Queue(string bucket, FileStorageMetadata file, uint timeout)
106106
var payload = await CreateOrGetPayload(bucket, file.CorrelationId, timeout).ConfigureAwait(false);
107107
payload.Add(file);
108108
_logger.FileAddedToBucket(payload.Key, payload.Count);
109+
return payload.PayloadId;
109110
}
110111

111112
/// <summary>

‎src/InformaticsGateway/Services/Connectors/PayloadMoveActionHandler.cs

-144
Original file line numberDiff line numberDiff line change
@@ -77,7 +77,6 @@ public async Task MoveFilesAsync(Payload payload, ActionBlock<Payload> moveQueue
7777
var stopwatch = Stopwatch.StartNew();
7878
try
7979
{
80-
await Move(payload, cancellationToken).ConfigureAwait(false);
8180
await NotifyIfCompleted(payload, notificationQueue, cancellationToken).ConfigureAwait(false);
8281
}
8382
catch (Exception ex)
@@ -127,149 +126,6 @@ private async Task NotifyIfCompleted(Payload payload, ActionBlock<Payload> notif
127126
}
128127
}
129128

130-
private async Task Move(Payload payload, CancellationToken cancellationToken)
131-
{
132-
Guard.Against.Null(payload);
133-
134-
_logger.MovingFIlesInPayload(payload.PayloadId, _options.Value.Storage.StorageServiceBucketName);
135-
136-
var options = new ParallelOptions
137-
{
138-
CancellationToken = cancellationToken,
139-
MaxDegreeOfParallelism = _options.Value.Storage.ConcurrentUploads
140-
};
141-
142-
var exceptions = new List<Exception>();
143-
await Parallel.ForEachAsync(payload.Files, options, async (file, cancellationToke) =>
144-
{
145-
try
146-
{
147-
switch (file)
148-
{
149-
case DicomFileStorageMetadata dicom:
150-
if (!string.IsNullOrWhiteSpace(dicom.JsonFile.TemporaryPath))
151-
{
152-
await MoveFile(payload.PayloadId, dicom.Id, dicom.JsonFile, cancellationToken).ConfigureAwait(false);
153-
}
154-
break;
155-
}
156-
157-
await MoveFile(payload.PayloadId, file.Id, file.File, cancellationToken).ConfigureAwait(false);
158-
}
159-
catch (Exception ex)
160-
{
161-
exceptions.Add(ex);
162-
}
163-
}).ConfigureAwait(false);
164-
165-
if (exceptions.Any())
166-
{
167-
throw new AggregateException(exceptions);
168-
}
169-
}
170-
171-
private async Task MoveFile(Guid payloadId, string identity, StorageObjectMetadata file, CancellationToken cancellationToken)
172-
{
173-
Guard.Against.NullOrWhiteSpace(identity);
174-
Guard.Against.Null(file);
175-
176-
if (file.IsMoveCompleted)
177-
{
178-
_logger.AlreadyMoved(payloadId, file.UploadPath);
179-
return;
180-
}
181-
182-
_logger.MovingFileToPayloadDirectory(payloadId, file.UploadPath);
183-
184-
try
185-
{
186-
await _storageService.CopyObjectAsync(
187-
file.TemporaryBucketName,
188-
file.GetTempStoragPath(_options.Value.Storage.RemoteTemporaryStoragePath),
189-
_options.Value.Storage.StorageServiceBucketName,
190-
file.GetPayloadPath(payloadId),
191-
cancellationToken).ConfigureAwait(false);
192-
193-
await VerifyFileExists(payloadId, file, cancellationToken).ConfigureAwait(false);
194-
}
195-
catch (StorageObjectNotFoundException ex) when (ex.Message.Contains("Not found", StringComparison.OrdinalIgnoreCase)) // TODO: StorageLib shall not throw any errors from MINIO
196-
{
197-
// when file cannot be found on the Storage Service, we assume file has been moved previously by verifying the file exists on destination.
198-
_logger.FileMissingInPayload(payloadId, file.GetTempStoragPath(_options.Value.Storage.RemoteTemporaryStoragePath), ex);
199-
await VerifyFileExists(payloadId, file, cancellationToken).ConfigureAwait(false);
200-
}
201-
catch (StorageConnectionException ex)
202-
{
203-
_logger.StorageServiceConnectionError(ex);
204-
throw new PayloadNotifyException(PayloadNotifyException.FailureReason.ServiceUnavailable);
205-
}
206-
catch (Exception ex)
207-
{
208-
_logger.PayloadMoveException(ex);
209-
await LogFilesInMinIo(file.TemporaryBucketName, cancellationToken).ConfigureAwait(false);
210-
throw new FileMoveException(file.GetTempStoragPath(_options.Value.Storage.RemoteTemporaryStoragePath), file.UploadPath, ex);
211-
}
212-
213-
try
214-
{
215-
_logger.DeletingFileFromTemporaryBbucket(file.TemporaryBucketName, identity, file.TemporaryPath);
216-
await _storageService.RemoveObjectAsync(file.TemporaryBucketName, file.GetTempStoragPath(_options.Value.Storage.RemoteTemporaryStoragePath), cancellationToken).ConfigureAwait(false);
217-
}
218-
catch (Exception)
219-
{
220-
_logger.ErrorDeletingFileAfterMoveComplete(file.TemporaryBucketName, identity, file.TemporaryPath);
221-
}
222-
finally
223-
{
224-
file.SetMoved(_options.Value.Storage.StorageServiceBucketName);
225-
}
226-
}
227-
228-
private async Task VerifyFileExists(Guid payloadId, StorageObjectMetadata file, CancellationToken cancellationToken)
229-
{
230-
await Policy
231-
.Handle<VerifyObjectsException>()
232-
.WaitAndRetryAsync(
233-
_options.Value.Storage.Retries.RetryDelays,
234-
(exception, timeSpan, retryCount, context) =>
235-
{
236-
_logger.ErrorUploadingFileToTemporaryStore(timeSpan, retryCount, exception);
237-
})
238-
.ExecuteAsync(async () =>
239-
{
240-
var internalCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
241-
internalCancellationTokenSource.CancelAfter(_options.Value.Storage.StorageServiceListTimeout);
242-
var exists = await _storageService.VerifyObjectExistsAsync(_options.Value.Storage.StorageServiceBucketName, file.GetPayloadPath(payloadId), cancellationToken).ConfigureAwait(false);
243-
if (!exists)
244-
{
245-
_logger.FileMovedVerificationFailure(payloadId, file.UploadPath);
246-
throw new PayloadNotifyException(PayloadNotifyException.FailureReason.MoveFailure, false);
247-
}
248-
})
249-
.ConfigureAwait(false);
250-
}
251-
252-
private async Task LogFilesInMinIo(string bucketName, CancellationToken cancellationToken)
253-
{
254-
try
255-
{
256-
var internalCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
257-
internalCancellationTokenSource.CancelAfter(_options.Value.Storage.StorageServiceListTimeout);
258-
var listingResults = await _storageService.ListObjectsAsync(bucketName, recursive: true, cancellationToken: internalCancellationTokenSource.Token).ConfigureAwait(false);
259-
_logger.FilesFounddOnStorageService(bucketName, listingResults.Count);
260-
var files = new List<string>();
261-
foreach (var item in listingResults)
262-
{
263-
files.Add(item.FilePath);
264-
}
265-
_logger.FileFounddOnStorageService(bucketName, string.Join(Environment.NewLine, files));
266-
}
267-
catch (Exception ex)
268-
{
269-
_logger.ErrorListingFilesOnStorageService(ex);
270-
}
271-
}
272-
273129
private async Task<PayloadAction> UpdatePayloadState(Payload payload, Exception ex, CancellationToken cancellationToken = default)
274130
{
275131
Guard.Against.Null(payload);

‎src/InformaticsGateway/Services/DicomWeb/IStreamsWriter.cs

+4-2
Original file line numberDiff line numberDiff line change
@@ -168,12 +168,14 @@ private async Task SaveInstance(Stream stream, string studyInstanceUid, string w
168168
{
169169
dicomInfo.SetWorkflows(workflowName);
170170
}
171+
// for DICOMweb, use correlation ID as the grouping key
172+
var payloadId = await _payloadAssembler.Queue(correlationId, dicomInfo, _configuration.Value.DicomWeb.Timeout).ConfigureAwait(false);
173+
dicomInfo.PayloadId = payloadId.ToString();
171174

172175
await dicomInfo.SetDataStreams(dicomFile, dicomFile.ToJson(_configuration.Value.Dicom.WriteDicomJson, _configuration.Value.Dicom.ValidateDicomOnSerialization), _configuration.Value.Storage.TemporaryDataStorage, _fileSystem, _configuration.Value.Storage.LocalTemporaryStoragePath).ConfigureAwait(false);
173176
_uploadQueue.Queue(dicomInfo);
174177

175-
// for DICOMweb, use correlation ID as the grouping key
176-
await _payloadAssembler.Queue(correlationId, dicomInfo, _configuration.Value.DicomWeb.Timeout).ConfigureAwait(false);
178+
177179
_logger.QueuedStowInstance();
178180

179181
AddSuccess(null, uids);

‎src/InformaticsGateway/Services/Fhir/FhirService.cs

+2-1
Original file line numberDiff line numberDiff line change
@@ -87,8 +87,9 @@ public async Task<FhirStoreResult> StoreAsync(HttpRequest request, string correl
8787
throw new FhirStoreException(correlationId, $"Provided resource is of type '{content.InternalResourceType}' but request targeted type '{resourceType}'.", IssueType.Invalid);
8888
}
8989

90+
var payloadId = await _payloadAssembler.Queue(correlationId, content.Metadata, Resources.PayloadAssemblerTimeout).ConfigureAwait(false);
91+
content.Metadata.PayloadId = payloadId.ToString();
9092
_uploadQueue.Queue(content.Metadata);
91-
await _payloadAssembler.Queue(correlationId, content.Metadata, Resources.PayloadAssemblerTimeout).ConfigureAwait(false);
9293
_logger.QueuedStowInstance();
9394

9495
content.StatusCode = StatusCodes.Status201Created;

‎src/InformaticsGateway/Services/HealthLevel7/MllpService.cs

+2-1
Original file line numberDiff line numberDiff line change
@@ -170,8 +170,9 @@ private async Task OnDisconnect(IMllpClient client, MllpClientResult result)
170170
{
171171
var hl7Fileetadata = new Hl7FileStorageMetadata(client.ClientId.ToString());
172172
await hl7Fileetadata.SetDataStream(message.HL7Message, _configuration.Value.Storage.TemporaryDataStorage, _fileSystem, _configuration.Value.Storage.LocalTemporaryStoragePath).ConfigureAwait(false);
173+
var payloadId = await _payloadAssembler.Queue(client.ClientId.ToString(), hl7Fileetadata).ConfigureAwait(false);
174+
hl7Fileetadata.PayloadId = payloadId.ToString();
173175
_uploadQueue.Queue(hl7Fileetadata);
174-
await _payloadAssembler.Queue(client.ClientId.ToString(), hl7Fileetadata).ConfigureAwait(false);
175176
}
176177
}
177178
catch (Exception ex)

‎src/InformaticsGateway/Services/Scp/ApplicationEntityHandler.cs

+4-2
Original file line numberDiff line numberDiff line change
@@ -113,12 +113,14 @@ public async Task HandleInstanceAsync(DicomCStoreRequest request, string calledA
113113
}
114114

115115
await dicomInfo.SetDataStreams(request.File, request.File.ToJson(_dicomJsonOptions, _validateDicomValueOnJsonSerialization), _options.Value.Storage.TemporaryDataStorage, _fileSystem, _options.Value.Storage.LocalTemporaryStoragePath).ConfigureAwait(false);
116-
_uploadQueue.Queue(dicomInfo);
117116

118117
var dicomTag = FellowOakDicom.DicomTag.Parse(_configuration.Grouping);
119118
_logger.QueueInstanceUsingDicomTag(dicomTag);
120119
var key = request.Dataset.GetSingleValue<string>(dicomTag);
121-
await _payloadAssembler.Queue(key, dicomInfo, _configuration.Timeout).ConfigureAwait(false);
120+
121+
var payloadid = await _payloadAssembler.Queue(key, dicomInfo, _configuration.Timeout).ConfigureAwait(false);
122+
dicomInfo.PayloadId = payloadid.ToString();
123+
_uploadQueue.Queue(dicomInfo);
122124
}
123125

124126
private bool AcceptsSopClass(string sopClassUid)

0 commit comments

Comments
 (0)