Eventhub Stress Test Onboarding

eng/Packages.Data.props

@@ -188,6 +188,7 @@
     <PackageReference Update="FluentAssertions" Version="5.10.3" />
     <PackageReference Update="FsCheck.Xunit" Version="2.14.0" />
     <PackageReference Update="Microsoft.Azure.ApplicationInsights.Query" Version="1.0.0" />
+    <PackageReference Update="Microsoft.ApplicationInsights" Version="2.20.0" />
     <PackageReference Update="Microsoft.AspNetCore" Version="2.2.0" />
     <PackageReference Update="Microsoft.AspNetCore.Mvc.Testing" Version="2.2.0" />
     <PackageReference Update="Microsoft.AspNetCore.Server.Kestrel" Version="2.1.3" />

sdk/eventhub/Azure.Messaging.EventHubs.Shared/src/Azure.Messaging.EventHubs.Shared.Stress.projitems

@@ -0,0 +1,15 @@
+<?xml version="1.0" encoding="utf-8"?>
+<Project xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
+  <PropertyGroup>
+    <MSBuildAllProjects>$(MSBuildAllProjects);$(MSBuildThisFileFullPath)</MSBuildAllProjects>
+    <HasSharedItems>true</HasSharedItems>
+  </PropertyGroup>
+
+  <PropertyGroup>
+    <Import_RootNamespace>Azure.Messaging.EventHubs.Tests</Import_RootNamespace>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <Compile Include="$(MSBuildThisFileDirectory)Testing\EventGenerator.cs" Link="SharedSource\Stress\%(Filename)%(Extension)" />
+  </ItemGroup>
+</Project>

sdk/eventhub/Azure.Messaging.EventHubs/stress/Chart.lock

@@ -0,0 +1,6 @@
+dependencies:
+- name: stress-test-addons
+  repository: https://stresstestcharts.blob.core.windows.net/helm/
+  version: 0.1.13
+digest: sha256:007ec9983233e0f8c8ad8aa7f1df5f57b1b4c127d223d59d233b3c5366bd5173
+generated: "2022-04-12T11:20:37.5250017-07:00"

sdk/eventhub/Azure.Messaging.EventHubs/stress/Chart.yaml

@@ -0,0 +1,15 @@
+apiVersion: v2
+name: eventhub-net-stress-test
+version: 0.1.1
+description: Stress tests for Event Hubs for .NET
+
+dependencies:
+- name: stress-test-addons
+  version: 0.1.13
+  repository: https://stresstestcharts.blob.core.windows.net/helm/
+
+annotations:
+  stressTest: 'true'
+  namespace: 'net'
+  dockerbuilddir: '../../../..'
+  dockerfile: './Dockerfile'

sdk/eventhub/Azure.Messaging.EventHubs/stress/Dockerfile

@@ -0,0 +1,28 @@
+# right version of dotnet?
+FROM mcr.microsoft.com/dotnet/sdk:6.0 AS build-env 
+
+# Copy in engineering system needed to build
+COPY /sdk/eventhub/Azure.Messaging.EventHubs/ /app/sdk/eventhub/Azure.Messaging.EventHubs/
+COPY /eng /app/eng
+COPY /tools /app/tools
+COPY ./*.proj /app
+COPY ./Directory.Build.props /app
+COPY ./Directory.Build.targets /app
+COPY ./NuGet.Config /app
+COPY ./sdk/eventhub/Azure.Messaging.EventHubs.Shared /app/sdk/eventhub/Azure.Messaging.EventHubs.Shared
+COPY ./sdk/core /app/sdk/core
+
+# Run restore and build, copy the .dll files over so they can used in the running container
+WORKDIR /app
+RUN dotnet restore './sdk/eventhub/Azure.Messaging.EventHubs/stress/src/'
+RUN dotnet build './sdk/eventhub/Azure.Messaging.EventHubs/stress/src/'
+COPY /artifacts/bin/Azure.Messaging.EventHubs.Stress/Debug /app/bin/Debug
+
+# Copy in the dll files to be ready to run
+FROM build-env as publish
+WORKDIR /app
+COPY --from=build-env /app/artifacts/bin/Azure.Messaging.EventHubs.Stress/Debug /app/artifacts/bin/Azure.Messaging.EventHubs.Stress/Debug
+COPY --from=build-env /app/sdk/eventhub/Azure.Messaging.EventHubs.Shared /app/sdk/evethub/Azure.Messaging.EventHubs.Shared
+COPY --from=build-env /app/sdk/eventhub/Azure.Messaging.EventHubs/stress /app/sdk/eventhub/Azure.Messaging.EventHubs/stress
+
+WORKDIR /app/artifacts/bin/Azure.Messaging.EventHubs.Stress/Debug/net6.0

sdk/eventhub/Azure.Messaging.EventHubs/stress/README.md

@@ -0,0 +1,63 @@
+# Event Hub Stress Tests
+This is a preliminary README. It demonstrates how to use the stress tests in the current state.
+
+In order to run the stress tests locally, the necessary resource connections need to be input through the command line interface. Test runs can call any of the following tests:
+- EventProducerTest : "EventProd"
+- EventBufferedProducerTest : "EventBuffProd"
+- BurstBufferedProducerTest : "BurstBuffProd"
+- ConcurrentBufferedProducerTest : "ConcurBuffProd"
+
+## Local Stress Test Runs
+### Setting up resources and packages
+```cmd
+(env) ~/stress/src> dotnet clean
+(env) ~/stress/src> dotnet build
+```
+
+### Running Tests
+When tests are run locally, Azure resources need to be created prior to running the test. This can be done through the Azure CLI, an ARM template or bicep file, or the Azure Portal. The user is required to input the connection strings upon request on the command line when the test is being run. For more information about what resources are needed for each test, see the "Scenario Information" section below. 
+
+The recommended approach is to run tests one at a time when running locally.
+To run any one test, run the following:
+```cmd
+(env) ~/stress/src> dotnet run -f netcoreapp3.1 BasicPublishReadTest local
+```
+
+## Deploy a stress test
+In order to deploy stress tests to be run in kubernetes clusters, run:
+```cmd 
+(env) ~/azure-sdk-for-net/eng/common/scripts/stress-testing/deploy-stress-tests.ps1 `
+>>     -Login `
+>>     -PushImages
+```
+This command requires Azure login credentials.
+
+## Scenario Information
+### Event Producer Test
+This test requires an event hub namespace, an event hub, and an application insights resource. Note that an event hub may experience throttling if too few partitions are used. This test creates 2 producers and has 5 concurrent processes per producer sending batches of events. This is a long-running, consistent volume test.
+
+### Event Buffered Producer Test
+This test requires an event hub namespace, an event hub, and an application insights resource. High CPU usage may be experienced if too few partitions are used. This test creates 2 producers and continuously sends to each producer separately. This is a long-running, consistent volume test.
+
+### Concurrent Buffered Producer Test
+This test requires an event hub namespace, an event hub, and an application insights resource. High CPU usage may be experienced if too few partitions are used. This test creates 2 producers and has 5 concurrent processes per producer continuously sending events. This is a long-running, consistent volume test.
+
+### Burst Buffered Producer Test
+This test requires an event hub namespace, an event hub, and an application insights resource. Note that this test may have a high CPU usage. This test creates 2 producers and sends sets of events to each producer separately every 15 minutes. This is a long-running, variable volume test.
+
+## Seeing Metrics and Logging in App Insights
+All metrics and logging are sent to App Insights via the Instrumentation Key provided during the initialization of the test. A brief explanation of the metrics collection approach is described below.
+- Any exceptions that occur are tracked by the telemetry client as exception telemetry. They can be accessed through the logs by filtering for exceptions, or through the application insights portal in the "Exceptions" blade. If an exception occured during send, the exception telemetry will include a "process" property containing "send"
+- Successful enqueues and sends are tracked through Metrics. For the buffered producer, the total number of enqueues is tracked, and the actual sends include the ability to separate counts into partition Ids.
+
+## Contributing  
+
+This project welcomes contributions and suggestions.  Most contributions require you to agree to a Contributor License Agreement (CLA) declaring that you have the right to, and actually do, grant us the rights to use your contribution. For details, visit https://cla.microsoft.com.
+
+When you submit a pull request, a CLA-bot will automatically determine whether you need to provide a CLA and decorate the PR appropriately (e.g., label, comment). Simply follow the instructions provided by the bot. You will only need to do this once across all repos using our CLA.
+
+This project has adopted the [Microsoft Open Source Code of Conduct](https://opensource.microsoft.com/codeofconduct/). For more information see the [Code of Conduct FAQ](https://opensource.microsoft.com/codeofconduct/faq/) or contact [[email protected]](mailto:[email protected]) with any additional questions or comments.
+
+Please see our [contributing guide](https://github.com/Azure/azure-sdk-for-net/blob/main/sdk/eventhub/Azure.Messaging.EventHubs/CONTRIBUTING.md) for more information.
+
+![Impressions](https://azure-sdk-impressions.azurewebsites.net/api/impressions/azure-sdk-for-net%2Fsdk%2Feventhub%2FAzure.Messaging.EventHubs%2stress%2FREADME.png)

sdk/eventhub/Azure.Messaging.EventHubs/stress/src/Azure.Messaging.EventHubs.Stress.csproj

@@ -0,0 +1,28 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <Version>1.0.0</Version>
+    <OutputType>Exe</OutputType>
+    <LangVersion>latest</LangVersion>
+    <IsTestSupportProject>true</IsTestSupportProject>
+    <IsStressProject>false</IsStressProject>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="System.Memory.Data" />
+    <PackageReference Include="Azure.Storage.Blobs" />
+    <PackageReference Include="Microsoft.ApplicationInsights" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\src\Azure.Messaging.EventHubs.csproj" />
+  </ItemGroup>
+
+  <!-- Import Azure.Core shared source -->
+  <ItemGroup>
+    <Compile Include="$(AzureCoreSharedSources)Argument.cs" LinkBase="SharedSource\Azure.Core" />
+    <Compile Include="$(AzureCoreSharedSources)TaskExtensions.cs" LinkBase="SharedSource\Azure.Core" />
+  </ItemGroup>
+
+  <!-- Import Event Hubs shared source -->
+  <Import Project="$(MSBuildThisFileDirectory)..\..\..\Azure.Messaging.EventHubs.Shared\src\Azure.Messaging.EventHubs.Shared.Stress.projitems" Label="Stress" />
+</Project>

sdk/eventhub/Azure.Messaging.EventHubs/stress/src/Infrastructure/BufferedPublisher.cs

@@ -0,0 +1,155 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.IO;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Azure.Messaging.EventHubs.Consumer;
+using Azure.Messaging.EventHubs.Producer;
+using Azure.Messaging.EventHubs.Tests;
+
+namespace Azure.Messaging.EventHubs.Stress
+{
+    internal class BufferedPublisher
+    {
+        private string connectionString;
+        private string eventHubName;
+        private Metrics metrics;
+        private ProducerConfiguration testConfiguration;
+
+        public BufferedPublisher(ProducerConfiguration testConfigurationIn, Metrics metricsIn)
+        {
+            connectionString = testConfigurationIn.EventHubsConnectionString;
+            eventHubName = testConfigurationIn.EventHub;
+            testConfiguration = testConfigurationIn;
+            metrics = metricsIn;
+        }
+
+        public async Task Start(CancellationToken cancellationToken)
+        {
+            var enqueueTasks = new List<Task>();
+
+            while (!cancellationToken.IsCancellationRequested)
+            {
+                using var backgroundCancellationSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+
+                try
+                {
+                    var producer = new EventHubBufferedProducerClient(connectionString, eventHubName);
+                    producer.SendEventBatchSucceededAsync += args =>
+                    {
+                        var numEvents = args.EventBatch.ToList().Count;
+                        var eventProperties = new Dictionary<String, String>();
+
+                        metrics.Client.GetMetric(metrics.SuccessfullySentFromQueue, args.PartitionId).TrackValue(numEvents);
+
+                        return Task.CompletedTask;
+                    };
+
+                    producer.SendEventBatchFailedAsync += args =>
+                    {
+                        var numEvents = args.EventBatch.ToList().Count;
+
+                        var eventProperties = new Dictionary<String, String>();
+
+                        metrics.Client.GetMetric(metrics.EventsNotSentAfterEnqueue, args.PartitionId).TrackValue(numEvents);
+                        metrics.Client.TrackException(args.Exception);
+
+                        return Task.CompletedTask;
+                    };
+
+                    await using (producer.ConfigureAwait(false))
+                    {
+                        // Create a set of background tasks to handle all but one of the concurrent sends.  The final
+                        // send will be performed directly.
+
+                        if (testConfiguration.ConcurrentSends > 1)
+                        {
+                            for (var index = 0; index < testConfiguration.ConcurrentSends - 1; ++index)
+                            {
+                                enqueueTasks.Add(Task.Run(async () =>
+                                {
+                                    while (!cancellationToken.IsCancellationRequested)
+                                    {
+                                        await PerformSend(producer, cancellationToken).ConfigureAwait(false);
+
+                                        if ((testConfiguration.ProducerPublishingDelay.HasValue) && (testConfiguration.ProducerPublishingDelay.Value > TimeSpan.Zero))
+                                        {
+                                            await Task.Delay(testConfiguration.ProducerPublishingDelay.Value, backgroundCancellationSource.Token).ConfigureAwait(false);
+                                        }
+                                    }
+                                }));
+                            }
+                        }
+
+                        // Perform one of the sends in the foreground, which will allow easier detection of a
+                        // processor-level issue.
+
+                        while (!cancellationToken.IsCancellationRequested)
+                        {
+                            try
+                            {
+                                await PerformSend(producer, cancellationToken).ConfigureAwait(false);
+
+                                if ((testConfiguration.ProducerPublishingDelay.HasValue) && (testConfiguration.ProducerPublishingDelay.Value > TimeSpan.Zero))
+                                {
+                                    await Task.Delay(testConfiguration.ProducerPublishingDelay.Value, cancellationToken).ConfigureAwait(false);
+                                }
+                            }
+                            catch (TaskCanceledException)
+                            {
+                                backgroundCancellationSource.Cancel();
+                                await Task.WhenAll(enqueueTasks).ConfigureAwait(false);
+                            }
+                        }
+                    }
+                }
+                catch (TaskCanceledException)
+                {
+                    // No action needed.
+                }
+                catch (Exception ex) when
+                    (ex is OutOfMemoryException
+                    || ex is StackOverflowException
+                    || ex is ThreadAbortException)
+                {
+                    throw;
+                }
+                catch (Exception ex)
+                {
+                    metrics.Client.TrackException(ex);
+                }
+            }
+        }
+
+        private async Task PerformSend(EventHubBufferedProducerClient producer,
+                                       CancellationToken cancellationToken)
+        {
+            var events = EventGenerator.CreateEvents(testConfiguration.EventEnqueueListSize);
+
+            try
+            {
+                await producer.EnqueueEventsAsync(events, cancellationToken).ConfigureAwait(false);
+
+                metrics.Client.GetMetric(metrics.EventsEnqueued).TrackValue(testConfiguration.EventEnqueueListSize);
+            }
+            catch (TaskCanceledException)
+            {
+                // Run is completed.
+            }
+            catch (Exception ex)
+            {
+                var eventProperties = new Dictionary<String, String>();
+                eventProperties.Add("Process", "Send");
+
+                metrics.Client.TrackException(ex, eventProperties);
+            }
+        }
+    }
+}