[Egress] Clean-Up For CopyBufferSize

documentation/schema.json

@@ -2579,7 +2579,9 @@
             "null"
           ],
           "description": "Buffer size used when copying data from an egress callback returning a stream to the egress callback that is provided a stream to which data is written.",
-          "format": "int32"
+          "format": "int32",
+          "maximum": 2147483647.0,
+          "minimum": 1.0
         },
         "QueueName": {
           "type": [

src/Extensions/AzureBlobStorage/AzureBlobEgressProvider.cs

@@ -23,7 +23,7 @@ namespace Microsoft.Diagnostics.Monitoring.AzureBlobStorage
     /// </remarks>
     internal partial class AzureBlobEgressProvider : EgressProvider<AzureBlobEgressProviderOptions>
     {
-        private int BlobStorageBufferSize = 4 * 1024 * 1024;
+        private int DefaultBlobStorageBufferSize = 4 * 1024 * 1024;
 
         public AzureBlobEgressProvider(ILogger logger) : base(logger)
         {
@@ -49,10 +49,10 @@ public override async Task<string> EgressAsync(
 
                 var bloboptions = new BlockBlobOpenWriteOptions
                 {
-                    BufferSize = BlobStorageBufferSize,
+                    BufferSize = options.CopyBufferSize.GetValueOrDefault(DefaultBlobStorageBufferSize),
                 };
                 using (Stream blobStream = await blobClient.OpenWriteAsync(overwrite: true, options: bloboptions, cancellationToken: token))
-                using (AutoFlushStream flushStream = new AutoFlushStream(blobStream, BlobStorageBufferSize))
+                using (AutoFlushStream flushStream = new AutoFlushStream(blobStream, bloboptions.BufferSize.Value))
                 {
                     //Azure's stream from OpenWriteAsync will do the following
                     //1. Write the data to a local buffer

src/Extensions/AzureBlobStorage/AzureBlobEgressProviderOptions.cs

@@ -55,6 +55,7 @@ internal sealed partial class AzureBlobEgressProviderOptions
         [Display(
             ResourceType = typeof(OptionsDisplayStrings),
             Description = nameof(OptionsDisplayStrings.DisplayAttributeDescription_CommonEgressProviderOptions_CopyBufferSize))]
+        [Range(1, int.MaxValue)]
         public int? CopyBufferSize { get; set; }
 
         [Display(

src/Extensions/S3Storage/MultiPartUploadStream.cs

@@ -27,9 +27,9 @@ internal sealed class MultiPartUploadStream : Stream
     public const int MinimumSize = 5 * 1024 * 1024; // the minimum size of an upload part (except for the last part)
     private readonly int _bufferSize;
 
-    public MultiPartUploadStream(IS3Storage client, string bucketName, string objectKey, string uploadId, int bufferSize)
+    public MultiPartUploadStream(IS3Storage client, string bucketName, string objectKey, string uploadId, int? bufferSize)
     {
-        _bufferSize = Math.Max(bufferSize, MinimumSize); // has to be at least the minimum
+        _bufferSize = Math.Max(bufferSize ?? 0, MinimumSize);
         _buffer = ArrayPool<byte>.Shared.Rent(_bufferSize);
         _offset = 0;
         _client = client;

src/Extensions/S3Storage/S3StorageEgressProvider.cs

@@ -43,8 +43,7 @@ public override async Task<string> EgressAsync(
             {
                 client = await ClientFactory.CreateAsync(options, artifactSettings, token);
                 uploadId = await client.InitMultiPartUploadAsync(artifactSettings.Metadata, token);
-                int copyBufferSize = options.CopyBufferSize.GetValueOrDefault(0x100000);
-                await using var stream = new MultiPartUploadStream(client, options.BucketName, artifactSettings.Name, uploadId, copyBufferSize);
+                await using var stream = new MultiPartUploadStream(client, options.BucketName, artifactSettings.Name, uploadId, options.CopyBufferSize);
                 _logger.EgressProviderInvokeStreamAction(Constants.S3StorageProviderName);
                 await action(stream, token);
                 await stream.FinalizeAsync(token); // force to push the last part

src/Microsoft.Diagnostics.Monitoring.WebApi/IEgressService.cs

@@ -12,15 +12,6 @@ internal interface IEgressService
     {
         void ValidateProvider(string providerName);
 
-        Task<EgressResult> EgressAsync(
-            string providerName,
-            Func<CancellationToken, Task<Stream>> action,
-            string fileName,
-            string contentType,
-            IEndpointInfo source,
-            CollectionRuleMetadata collectionRuleMetadata,
-            CancellationToken token);
-
         Task<EgressResult> EgressAsync(
             string providerName,
             Func<Stream, CancellationToken, Task> action,

src/Microsoft.Diagnostics.Monitoring.WebApi/Operation/EgressOperation.cs

@@ -22,17 +22,6 @@ internal class EgressOperation : IEgressOperation
 
         private readonly IArtifactOperation _operation;
 
-
-        public EgressOperation(Func<CancellationToken, Task<Stream>> action, string endpointName, string artifactName, IProcessInfo processInfo, string contentType, KeyValueLogScope scope, string tags, CollectionRuleMetadata collectionRuleMetadata = null)
-        {
-            _egress = (service, token) => service.EgressAsync(endpointName, action, artifactName, contentType, processInfo.EndpointInfo, collectionRuleMetadata, token);
-            _scope = scope;
-
-            EgressProviderName = endpointName;
-            ProcessInfo = new EgressProcessInfo(processInfo.ProcessName, processInfo.EndpointInfo.ProcessId, processInfo.EndpointInfo.RuntimeInstanceCookie);
-            Tags = Utilities.SplitTags(tags);
-        }
-
         public EgressOperation(Func<Stream, CancellationToken, Task> action, string endpointName, string artifactName, IProcessInfo processInfo, string contentType, KeyValueLogScope scope, string tags, CollectionRuleMetadata collectionRuleMetadata = null)
         {
             _egress = (service, token) => service.EgressAsync(endpointName, action, artifactName, contentType, processInfo.EndpointInfo, collectionRuleMetadata, token);
@@ -57,13 +46,6 @@ public EgressOperation(Func<Stream, CancellationToken, Task> action, string endp
             _scope = scope;
         }
 
-        public EgressOperation(Func<CancellationToken, Task<Stream>> action, string endpointName, string artifactName, IEndpointInfo source, string contentType, KeyValueLogScope scope, CollectionRuleMetadata collectionRuleMetadata)
-        {
-            _egress = (service, token) => service.EgressAsync(endpointName, action, artifactName, contentType, source, collectionRuleMetadata, token);
-            EgressProviderName = endpointName;
-            _scope = scope;
-        }
-
         public async Task<ExecutionResult<EgressResult>> ExecuteAsync(IServiceProvider serviceProvider, CancellationToken token)
         {
             ILogger<EgressOperation> logger = serviceProvider

src/Tests/CollectionRuleActions.UnitTests/Egress/Pipe/PipeEgressProvider.cs

@@ -4,7 +4,6 @@
 using Microsoft.Diagnostics.Tools.Monitor.Egress;
 using System;
 using System.IO;
-using System.IO.Pipelines;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -14,15 +13,6 @@ internal sealed class PipeEgressProvider : IEgressProvider<PipeEgressOptions>
     {
         public const string Name = "Pipe";
 
-        public async Task<string> EgressAsync(string providerType, string providerName, PipeEgressOptions options, Func<CancellationToken, Task<Stream>> action, EgressArtifactSettings artifactSettings, CancellationToken token)
-        {
-            using Stream stream = await action(token);
-
-            await stream.CopyToAsync(options.Writer, token);
-
-            return null;
-        }
-
         public async Task<string> EgressAsync(string providerType, string providerName, PipeEgressOptions options, Func<Stream, CancellationToken, Task> action, EgressArtifactSettings artifactSettings, CancellationToken token)
         {
             using Stream stream = options.Writer.AsStream();

src/Tools/dotnet-monitor/Egress/EgressProvider.cs

@@ -44,47 +44,6 @@ protected EgressProvider(ILogger logger)
             Logger = logger;
         }
 
-        /// <summary>
-        /// Egress a stream via a callback by returning the stream from the callback.
-        /// </summary>
-        /// <param name="options">Described to where stream data should be egressed.</param>
-        /// <param name="action">Callback that is invoked in order to get the stream to be egressed.</param>
-        /// <param name="artifactSettings">Describes data about the artifact, such as file name and content type.</param>
-        /// <param name="token">The token to monitor for cancellation requests.</param>
-        /// <returns>A task that completes with a value of the identifier of the egress result. Typically,
-        /// this is a path to access the stream without any information indicating whether any particular
-        /// user has access to it (e.g. no file system permissions or SAS tokens).</returns>
-        public virtual Task<string> EgressAsync(
-            string providerType,
-            string providerName,
-            TOptions options,
-            Func<CancellationToken, Task<Stream>> action,
-            EgressArtifactSettings artifactSettings,
-            CancellationToken token)
-        {
-            Func<Stream, CancellationToken, Task> wrappingAction = async (targetStream, token) =>
-            {
-                using var sourceStream = await action(token);
-
-                int copyBufferSize = options.CopyBufferSize.GetValueOrDefault(0x100000);
-
-                Logger?.EgressCopyActionStreamToEgressStream(copyBufferSize);
-
-                await sourceStream.CopyToAsync(
-                    targetStream,
-                    copyBufferSize,
-                    token);
-            };
-
-            return EgressAsync(
-                providerType,
-                providerName,
-                options,
-                wrappingAction,
-                artifactSettings,
-                token);
-        }
-
         /// <summary>
         /// Egress a stream via a callback by writing to the provided stream.
         /// </summary>

src/Tools/dotnet-monitor/Egress/EgressProviderInternal.cs

@@ -34,23 +34,6 @@ public EgressProviderInternal(
             _monitor = monitor;
         }
 
-        /// <inheritdoc/>
-        public Task<string> EgressAsync(
-            string providerType,
-            string providerName,
-            Func<CancellationToken, Task<Stream>> action,
-            EgressArtifactSettings artifactSettings,
-            CancellationToken token)
-        {
-            return _provider.EgressAsync(
-                providerType,
-                providerName,
-                GetOptions(providerName),
-                action,
-                artifactSettings,
-                token);
-        }
-
         /// <inheritdoc/>
         public Task<string> EgressAsync(
             string providerType,

src/Tools/dotnet-monitor/Egress/EgressService.cs

@@ -62,21 +62,6 @@ public void ValidateProvider(string providerName)
             string providerType = GetProviderType(providerName);
         }
 
-        public async Task<EgressResult> EgressAsync(string providerName, Func<CancellationToken, Task<Stream>> action, string fileName, string contentType, IEndpointInfo source, CollectionRuleMetadata collectionRuleMetadata, CancellationToken token)
-        {
-            string providerType = GetProviderType(providerName);
-            IEgressProviderInternal provider = GetProvider(providerType);
-
-            string value = await provider.EgressAsync(
-                providerType,
-                providerName,
-                action,
-                await CreateSettings(source, fileName, contentType, collectionRuleMetadata, token),
-                token);
-
-            return new EgressResult(value);
-        }
-
         public async Task<EgressResult> EgressAsync(string providerName, Func<Stream, CancellationToken, Task> action, string fileName, string contentType, IEndpointInfo source, CollectionRuleMetadata collectionRuleMetadata, CancellationToken token)
         {
             string providerType = GetProviderType(providerName);

src/Tools/dotnet-monitor/Egress/IEgressProvider.cs

@@ -10,14 +10,6 @@ namespace Microsoft.Diagnostics.Tools.Monitor.Egress
 {
     internal interface IEgressProvider<TOptions>
     {
-        Task<string> EgressAsync(
-            string providerType,
-            string providerName,
-            TOptions options,
-            Func<CancellationToken, Task<Stream>> action,
-            EgressArtifactSettings artifactSettings,
-            CancellationToken token);
-
         Task<string> EgressAsync(
             string providerType,
             string providerName,

src/Tools/dotnet-monitor/Egress/IEgressProviderInternal.cs

@@ -10,13 +10,6 @@ namespace Microsoft.Diagnostics.Tools.Monitor.Egress
 {
     internal interface IEgressProviderInternal
     {
-        Task<string> EgressAsync(
-            string providerType,
-            string providerName,
-            Func<CancellationToken, Task<Stream>> action,
-            EgressArtifactSettings artifactSettings,
-            CancellationToken token);
-
         Task<string> EgressAsync(
             string providerType,
             string providerName,