[Azure.Monitor.Ingestion] Add tests and update Readme (Azure#34021)

Author: nisha-bhatia

sdk/monitor/Azure.Monitor.Ingestion/README.md

@@ -88,6 +88,8 @@ We guarantee that all client instance methods are thread-safe and independent of
 
 - [Register the client with dependency injection](#register-the-client-with-dependency-injection)
 - [Upload custom logs](#upload-custom-logs)
+- [Upload custom logs as IEnumerable](#upload-custom-logs-ienumerable)
+- [Upload custom logs as IEnumerable with EventHandler](#upload-custom-logs-ienumerable-eventhandler)
 - [Verify logs](#verify-logs)
 
 You can familiarize yourself with different APIs using [samples](https://github.com/Azure/azure-sdk-for-net/tree/main/sdk/monitor/Azure.Monitor.Ingestion/samples).
@@ -98,7 +100,7 @@ To register `LogsIngestionClient` with the dependency injection (DI) container,
 
 ### Upload custom logs
 
-You can upload logs using either the `LogsIngestionClient.Upload` or the `LogsIngestionClient.UploadAsync` method. Note the data ingestion [limits](https://learn.microsoft.com/azure/azure-monitor/service-limits#custom-logs).
+You can upload logs using either the `LogsIngestionClient.Upload` or the `LogsIngestionClient.UploadAsync` method. Note the data ingestion [limits](https://learn.microsoft.com/azure/azure-monitor/service-limits#custom-logs). This method has an optional parameter: string contentEncoding. This refers to the encoding of the RequestContent that is being passed in. If you are passing in content that is already manipulated, set the contentEncoding parameter. For example if your content is gzipped, set contentEncoding to be "gzip". The default behavior if this parameter is not set is to `gzip` all input. 
 
 ```C# Snippet:UploadCustomLogsAsync
 var endpoint = new Uri("<data_collection_endpoint>");
@@ -147,6 +149,81 @@ Response response = await client.UploadAsync(
     RequestContent.Create(data)).ConfigureAwait(false);
 ```
 
+### Upload custom logs as IEnumerable
+
+You can also upload logs using either the `LogsIngestionClient.Upload` or the `LogsIngestionClient.UploadAsync` method in which  logs are passed in a generic `IEnumerable` type along with an optional `LogsUploadOptions` parameter. The `LogsUploadOptions` parameter includes a serializer, concurrency, and an EventHandler.
+
+```C# Snippet:UploadLogDataIEnumerableAsync
+var endpoint = new Uri("<data_collection_endpoint_uri>");
+var ruleId = "<data_collection_rule_id>";
+var streamName = "<stream_name>";
+
+var credential = new DefaultAzureCredential();
+LogsIngestionClient client = new(endpoint, credential);
+
+DateTimeOffset currentTime = DateTimeOffset.UtcNow;
+
+var entries = new List<Object>();
+for (int i = 0; i < 100; i++)
+{
+    entries.Add(
+        new {
+            Time = currentTime,
+            Computer = "Computer" + i.ToString(),
+            AdditionalContext = i
+        }
+    );
+}
+
+// Upload our logs
+Response response = await client.UploadAsync(ruleId, streamName, entries).ConfigureAwait(false);
+```
+
+### Upload custom logs as IEnumerable with EventHandler
+
+You can upload logs using either the `LogsIngestionClient.Upload` or the `LogsIngestionClient.UploadAsync` method. In these two methods, logs are passed in a generic `IEnumerable` type. Additionally, there's an `LogsUploadOptions`-typed parameter in which a serializer, concurrency, and EventHandler can be set. The default serializer is set to `System.Text.Json`, but you can pass in the serializer you would like used. The `MaxConcurrency` property sets the number of threads that will be used in the `UploadAsync` method. The default value is 5, and this parameter is unused in the `Upload` method. The EventHandler is used for error handling. It gives the user the option to abort the upload if a batch fails and access the failed logs and corresponding exception. Without the EventHandler, if an upload fails, an `AggregateException` will be thrown.
+
+```C# Snippet:LogDataIEnumerableEventHandlerAsync
+var endpoint = new Uri("<data_collection_endpoint_uri>");
+var ruleId = "<data_collection_rule_id>";
+var streamName = "<stream_name>";
+
+var credential = new DefaultAzureCredential();
+LogsIngestionClient client = new(endpoint, credential);
+
+DateTimeOffset currentTime = DateTimeOffset.UtcNow;
+
+var entries = new List<Object>();
+for (int i = 0; i < 100; i++)
+{
+    entries.Add(
+        new {
+            Time = currentTime,
+            Computer = "Computer" + i.ToString(),
+            AdditionalContext = i
+        }
+    );
+}
+// Set concurrency and EventHandler in LogsUploadOptions
+LogsUploadOptions options = new LogsUploadOptions();
+options.MaxConcurrency = 10;
+options.UploadFailed += Options_UploadFailed;
+
+// Upload our logs
+Response response = await client.UploadAsync(ruleId, streamName, entries, options).ConfigureAwait(false);
+
+Task Options_UploadFailed(UploadFailedEventArgs e)
+{
+    // Throw exception from EventHandler to stop Upload if there is a failure
+    IReadOnlyList<object> failedLogs = e.FailedLogs;
+    // 413 status is RequestTooLarge - don't throw here because other batches can successfully upload
+    if ((e.Exception is RequestFailedException) && (((RequestFailedException)e.Exception).Status != 413))
+        throw e.Exception;
+    else
+        return Task.CompletedTask;
+}
+```
+
 ### Verify logs
 
 You can verify that your data has been uploaded correctly by using the [Azure Monitor Query](https://github.com/Azure/azure-sdk-for-net/blob/main/sdk/monitor/Azure.Monitor.Query/README.md#install-the-package) library. Run the [Upload custom logs](#upload-custom-logs) sample first before verifying the logs.

sdk/monitor/Azure.Monitor.Ingestion/src/LogsIngestionClient.cs

@@ -219,6 +219,8 @@ public virtual Response Upload<T>(string ruleId, string streamName, IEnumerable<
                     break;
                 try
                 {
+                    // Cancel all future Uploads if user triggers CancellationToken
+                    cancellationToken.ThrowIfCancellationRequested();
                     // Because we are uploading in sequence, wait for each batch to upload before starting the next batch
                     response = UploadBatchListSyncOrAsync(
                         batch,
@@ -232,18 +234,23 @@ public virtual Response Upload<T>(string ruleId, string streamName, IEnumerable<
                         // if there is no Handler on options, throw exception otherwise raise Handler
                         if (!options.HasHandler)
                         {
+                            // throw exception here that is caught in catch and we increment LogsFailed
                             throw new RequestFailedException(response);
                         }
                         else
                         {
+                            logsFailed += batch.Logs.Count;
                             var eventArgs = new UploadFailedEventArgs(batch.Logs, new RequestFailedException(response), isRunningSynchronously: true, ClientDiagnostics, cancellationToken);
 #pragma warning disable AZC0106 // Non-public asynchronous method needs 'async' parameter.
                             // sync/async parameter in eventArgs
-                            var ex = options.OnUploadFailedAsync(eventArgs).EnsureCompleted();
+                            var userThrownException = options.OnUploadFailedAsync(eventArgs).EnsureCompleted();
 #pragma warning restore AZC0106 // Non-public asynchronous method needs 'async' parameter.
-                            shouldAbort = ex != null;
-                            if (shouldAbort)
-                                AddException(ref exceptions, ex);
+                            // if exception is thrown stop processing future batches
+                            if (userThrownException != null)
+                            {
+                                shouldAbort = true;
+                                AddException(ref exceptions, userThrownException);
+                            }
                         }
                     }
                 }
@@ -259,13 +266,24 @@ public virtual Response Upload<T>(string ruleId, string streamName, IEnumerable<
                     }
                     else
                     {
+                        logsFailed += batch.Logs.Count;
                         var eventArgs = new UploadFailedEventArgs(batch.Logs, new RequestFailedException(response), isRunningSynchronously: true, ClientDiagnostics, cancellationToken);
 #pragma warning disable AZC0106 // Non-public asynchronous method needs 'async' parameter.
                         var exceptionOnUpload = options.OnUploadFailedAsync(eventArgs).EnsureCompleted();
 #pragma warning restore AZC0106 // Non-public asynchronous method needs 'async' parameter.
-                        shouldAbort = exceptionOnUpload != null;
-                        if (shouldAbort)
+                        // if exception is thrown stop processing future batches
+                        if (exceptionOnUpload != null)
+                        {
+                            shouldAbort = true;
                             AddException(ref exceptions, exceptionOnUpload);
+                        }
+                    }
+
+                    // Cancel all future Uploads if user triggers CancellationToken
+                    if (ex is OperationCanceledException && cancellationToken.IsCancellationRequested)
+                    {
+                        shouldAbort = true;
+                        AddException(ref exceptions, ex);
                     }
                 }
             }
@@ -332,6 +350,8 @@ public virtual async Task<Response> UploadAsync<T>(string ruleId, string streamN
                     break;
                 try
                 {
+                    // Cancel all future Uploads if user triggers CancellationToken
+                    cancellationToken.ThrowIfCancellationRequested();
                     // Start staging the next batch (but don't await the Task!)
                     Task<Response> task = UploadBatchListSyncOrAsync(
                         batch,
@@ -363,11 +383,16 @@ public virtual async Task<Response> UploadAsync<T>(string ruleId, string streamN
                             }
                             else
                             {
-                                Exception exceptionEventHandler = await ProcessCompletedTaskEventHandlerAsync(runningTask, batch.Logs, options, cancellationToken).ConfigureAwait(false);
-                                shouldAbort = exceptionEventHandler != null;
-                                if (shouldAbort)
-                                    AddException(ref exceptions, exceptionEventHandler);
+                                var processCompletedTask = await ProcessCompletedTaskEventHandlerAsync(runningTask, batch.Logs, options, cancellationToken).ConfigureAwait(false);
+                                logsFailed += processCompletedTask.FailedLogsCount;
+                                // if exception is thrown stop processing future batches
+                                if (processCompletedTask.Exception != null)
+                                {
+                                    shouldAbort = true;
+                                    AddException(ref exceptions, processCompletedTask.Exception);
+                                }
                             }
+
                             // Remove completed task from task list
                             runningTasks.RemoveAt(i);
                             i--;
@@ -377,9 +402,15 @@ public virtual async Task<Response> UploadAsync<T>(string ruleId, string streamN
                     // Wait for all the remaining blocks to finish uploading
                     await Task.WhenAll(runningTasks.Select(_ => _.CurrentTask)).ConfigureAwait(false);
                 }
-                catch (Exception)
+                catch (Exception ex)
                 {
                     // We do not want to log exceptions here as we will loop through all the tasks later
+                    // Cancel all future Uploads if user triggers CancellationToken
+                    if (ex is OperationCanceledException && cancellationToken.IsCancellationRequested)
+                    {
+                        shouldAbort = true;
+                        AddException(ref exceptions, ex);
+                    }
                 }
             }
 
@@ -393,10 +424,14 @@ public virtual async Task<Response> UploadAsync<T>(string ruleId, string streamN
                 }
                 else
                 {
-                    Exception exceptionEventHandler = await ProcessCompletedTaskEventHandlerAsync(task.CurrentTask, task.Logs, options, cancellationToken).ConfigureAwait(false);
-                    shouldAbort = exceptionEventHandler != null;
-                    if (shouldAbort)
-                        AddException(ref exceptions, exceptionEventHandler);
+                    var processTaskResult = await ProcessCompletedTaskEventHandlerAsync(task.CurrentTask, task.Logs, options, cancellationToken).ConfigureAwait(false);
+                    logsFailed += processTaskResult.FailedLogsCount;
+                    // if exception is thrown stop processing future batches
+                    if (processTaskResult.Exception != null)
+                    {
+                        shouldAbort = true;
+                        AddException(ref exceptions, processTaskResult.Exception);
+                    }
                 }
             }
             if (exceptions?.Count > 0)
@@ -431,22 +466,24 @@ private static void ProcessCompletedTask((Task<Response> CurrentTask, List<objec
             }
         }
 
-        internal async Task<Exception> ProcessCompletedTaskEventHandlerAsync(Task<Response> completedTask, List<object> logs, LogsUploadOptions options, CancellationToken cancellationToken)
+        internal async Task<(Exception Exception, int FailedLogsCount)> ProcessCompletedTaskEventHandlerAsync(Task<Response> completedTask, List<object> logs, LogsUploadOptions options, CancellationToken cancellationToken)
         {
             UploadFailedEventArgs eventArgs;
             if (completedTask.Exception != null)
             {
                 eventArgs = new UploadFailedEventArgs(logs, completedTask.Exception, isRunningSynchronously: false, ClientDiagnostics, cancellationToken);
-                return await options.OnUploadFailedAsync(eventArgs).ConfigureAwait(false);
+                var exception = await options.OnUploadFailedAsync(eventArgs).ConfigureAwait(false);
+                return (exception, logs.Count);
             }
             else if (completedTask.Result.Status != 204)
             {
                 eventArgs = new UploadFailedEventArgs(logs, new RequestFailedException(completedTask.Result), isRunningSynchronously: false, ClientDiagnostics, cancellationToken);
-                return await options.OnUploadFailedAsync(eventArgs).ConfigureAwait(false);
+                var exception = await options.OnUploadFailedAsync(eventArgs).ConfigureAwait(false);
+                return (exception, logs.Count);
             }
             else
             {
-                return null;
+                return (null, 0);
             }
         }
 

sdk/monitor/Azure.Monitor.Ingestion/tests/ErrorTest.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using Azure.Core;
@@ -131,7 +132,6 @@ public async Task OneFailureWithEventHandler()
             var cts = new CancellationTokenSource();
             bool isTriggered = false;
             options.UploadFailed += Options_UploadFailed;
-            //await client.UploadAsync(TestEnvironment.DCRImmutableId, TestEnvironment.StreamName, entries, options, cts.Token).ConfigureAwait(false);
             await client.UploadAsync(TestEnvironment.DCRImmutableId, TestEnvironment.StreamName, entries, options).ConfigureAwait(false);
             Assert.IsTrue(isTriggered);
             Task Options_UploadFailed(UploadFailedEventArgs e)
@@ -183,5 +183,87 @@ Task Options_UploadFailed(UploadFailedEventArgs e)
                 return Task.CompletedTask;
             }
         }
+
+        [Test]
+        public void TwoFailuresWithEventHandlerCancellationToken()
+        {
+            LogsIngestionClient client = CreateClient();
+            // set compression to gzip so SDK does not gzip data (assumes already gzipped)
+            LogsIngestionClient.Compression = "gzip";
+            var entries = GenerateEntries(800, Recording.Now.DateTime);
+            entries.Add(new object[] {
+                    new {
+                        Time = Recording.Now.DateTime,
+                        Computer = "Computer" + new string('*', Mb),
+                        AdditionalContext = 1
+                    }
+                });
+            entries.Add(new object[] {
+                    new {
+                        Time = Recording.Now.DateTime,
+                        Computer = "Computer" + new string('!', Mb),
+                        AdditionalContext = 1
+                    }
+                });
+            entries.Add(new object[] {
+                    new {
+                        Time = Recording.Now.DateTime,
+                        Computer = "Computer" + new string(';', Mb),
+                        AdditionalContext = 1
+                    }
+                });
+
+            // Make the request
+            LogsUploadOptions options = new LogsUploadOptions();
+            options.MaxConcurrency = 2;
+            bool isTriggered = false;
+            var cts = new CancellationTokenSource();
+            options.UploadFailed += Options_UploadFailed;
+            AggregateException exceptions = Assert.ThrowsAsync<AggregateException>(async () => { await client.UploadAsync(TestEnvironment.DCRImmutableId, TestEnvironment.StreamName, entries, options, cts.Token).ConfigureAwait(false); });
+            Assert.IsTrue(isTriggered);
+            Assert.IsTrue(cts.IsCancellationRequested);
+            // check if OperationCanceledException is in the Exception list
+            // may not be first one in async case
+            Assert.IsTrue(exceptions.InnerExceptions.Any(exception => exception is OperationCanceledException));
+            Task Options_UploadFailed(UploadFailedEventArgs e)
+            {
+                cts.Cancel();
+                isTriggered = true;
+                Assert.IsInstanceOf<RequestFailedException>(e.Exception);
+                Assert.AreEqual("ContentLengthLimitExceeded", ((RequestFailedException)(e.Exception)).ErrorCode);
+                Assert.IsNull(((RequestFailedException)(e.Exception)).InnerException);
+                Assert.AreEqual(413, ((RequestFailedException)(e.Exception)).Status);
+                return Task.CompletedTask;
+            }
+        }
+
+        [Test]
+        public void OneFailureWithEventHandlerThrowException()
+        {
+            LogsIngestionClient client = CreateClient();
+            // set compression to gzip so SDK does not gzip data (assumes already gzipped)
+            LogsIngestionClient.Compression = "gzip";
+            var entries = GenerateEntries(800, Recording.Now.DateTime);
+            entries.Add(new object[] {
+                    new {
+                        Time = Recording.Now.DateTime,
+                        Computer = "Computer" + new string('*', Mb),
+                        AdditionalContext = 1
+                    }
+                });
+
+            // Make the request
+            LogsUploadOptions options = new LogsUploadOptions();
+            options.UploadFailed += Options_UploadFailed;
+            var exceptions = Assert.ThrowsAsync<AggregateException>(async () => { await client.UploadAsync(TestEnvironment.DCRImmutableId, TestEnvironment.StreamName, entries, options).ConfigureAwait(false); });
+            Task Options_UploadFailed(UploadFailedEventArgs e)
+            {
+                Assert.IsInstanceOf<RequestFailedException>(e.Exception);
+                Assert.AreEqual("ContentLengthLimitExceeded", ((RequestFailedException)(e.Exception)).ErrorCode);
+                Assert.IsNull(((RequestFailedException)(e.Exception)).InnerException);
+                Assert.AreEqual(413, ((RequestFailedException)(e.Exception)).Status);
+                throw e.Exception;
+            }
+        }
     }
 }