[Event Hubs] Update stress and perf tests post "prefetch feature" tes…

…ting (Azure#26095)

## Updates to the tests following Azure#26065 
1. default test duration is set to 2 days for stress tests
2. Added a new "log-median-batch-size" option for the perf test
- `log-median-batch-size` is a boolean that logs more information
related to the batch size, such as median, max, average, etc.
   - It can be useful when relevant code is updated. 
- Being introduced in the perf test when the prefetch feature was added
to Event Hubs.

---------

Co-authored-by: Deyaaeldeen Almahallawi <dealmaha@microsoft.com>

sdk/eventhub/event-hubs/test/stress/app/src/scenarioCheckpointStore.ts

@@ -16,7 +16,7 @@ interface scenarioCheckpointStoreOptions {
 function sanitizeOptions(args: string[]): Required<scenarioCheckpointStoreOptions> {
   const options = parsedArgs<scenarioCheckpointStoreOptions>(args);
   return {
-    testDurationInMs: options.testDurationInMs || 20 * 24 * 60 * 60 * 1000, // Default = 20 days
+    testDurationInMs: options.testDurationInMs || 2 * 24 * 60 * 60 * 1000, // Default = 2 days
   };
 }
 

sdk/eventhub/event-hubs/test/stress/app/src/scenarioGetRuntimeProps.ts

@@ -13,7 +13,7 @@ interface scenarioGetRuntimePropertiesOptions {
 function sanitizeOptions(args: string[]): Required<scenarioGetRuntimePropertiesOptions> {
   const options = parsedArgs<scenarioGetRuntimePropertiesOptions>(args);
   return {
-    testDurationInMs: options.testDurationInMs || 10 * 60 * 60 * 1000, // Default = 10 hrs
+    testDurationInMs: options.testDurationInMs || 2 * 24 * 60 * 60 * 1000, // Default = 2 days
   };
 }
 

sdk/eventhub/event-hubs/test/stress/app/src/scenarioNoActivity.ts

@@ -2,14 +2,10 @@
 // Licensed under the MIT license.
 
 import { delay } from "@azure/core-util";
-import {
-  EventHubsStressTester,
-  createEventHubsConsumerClient,
-  createEventHubsProducerClient,
-  defaultClientAppInsights,
-} from "./eventHubsStressTester";
+import { EventHubsStressTester, defaultClientAppInsights } from "./eventHubsStressTester";
 import parsedArgs from "minimist";
 import { Subscription } from "@azure/event-hubs";
+import { createEventHubsConsumerClient, createEventHubsProducerClient } from "./utils";
 
 interface ScenarioNoActivityOptions {
   testDurationInMs?: number;
@@ -19,7 +15,7 @@ interface ScenarioNoActivityOptions {
 function sanitizeOptions(args: string[]): Required<ScenarioNoActivityOptions> {
   const options = parsedArgs<ScenarioNoActivityOptions>(args);
   return {
-    testDurationInMs: options.testDurationInMs || 10 * 60 * 60 * 1000, // Default = 10 hrs
+    testDurationInMs: options.testDurationInMs || 2 * 24 * 60 * 60 * 1000, // Default = 2 days
     maxBatchSize: options.maxBatchSize || 100,
   };
 }

sdk/eventhub/event-hubs/test/stress/generatedValues.yaml

@@ -0,0 +1,17 @@
+scenarios:
+- testTarget: scenarioNoActivity
+  image: Dockerfile
+  Scenario: node18-noActivity
+  memory: 1.5Gi
+  imageTag: stresspgs7b6dif73rup6.azurecr.io/harshan-eh-prefetch/jseh/dockerfile:harshanalluru
+- testTarget: scenarioCheckpointStore
+  image: Dockerfile
+  Scenario: node18-checkpointStore
+  memory: 1.5Gi
+  imageTag: stresspgs7b6dif73rup6.azurecr.io/harshan-eh-prefetch/jseh/dockerfile:harshanalluru
+- testTarget: scenarioGetRuntimeProps
+  image: Dockerfile
+  Scenario: node18-getRuntimeProps
+  memory: 1.5Gi
+  imageTag: stresspgs7b6dif73rup6.azurecr.io/harshan-eh-prefetch/jseh/dockerfile:harshanalluru
+

sdk/eventhub/perf-tests/event-hubs/test/subscribe.spec.ts

@@ -16,6 +16,12 @@ interface ReceiverOptions {
   "event-size-in-bytes": number;
   partitions: number;
   "max-batch-size": number;
+  /**
+   * Logs more information related to the batch size, such as median, max, average, etc
+   * Useful when relevant code is updated
+   * Introduced when prefetch feature was added to Event Hubs
+   */
+  "log-median-batch-size": boolean
 }
 
 const connectionString = getEnvVar("EVENTHUB_CONNECTION_STRING");
@@ -27,6 +33,8 @@ const consumer = new EventHubConsumerClient(consumerGroup, connectionString, eve
 export class SubscribeTest extends EventPerfTest<ReceiverOptions> {
   receiver: EventHubConsumerClient;
   subscriber: { close: () => Promise<void> } | undefined;
+  callbackCallsCount = 0;
+  messagesPerBatch: Array<number> = [];
 
   options: PerfOptionDictionary<ReceiverOptions> = {
     "number-of-events": {
@@ -57,6 +65,11 @@ export class SubscribeTest extends EventPerfTest<ReceiverOptions> {
       longName: "max-batch-size",
       defaultValue: 100,
     },
+    "log-median-batch-size": {
+      required: false,
+      description: "Logs more information related to the batch size, such as median, max, average, etc",
+      defaultValue: false,
+    }
   };
 
   constructor() {
@@ -83,6 +96,10 @@ export class SubscribeTest extends EventPerfTest<ReceiverOptions> {
           for (const _event of events) {
             this.eventRaised();
           }
+          if (this.parsedOptions["log-median-batch-size"].value) {
+            this.callbackCallsCount++;
+            this.messagesPerBatch.push(events.length);
+          }
         },
         processError: async (error: Error | MessagingError, _context: PartitionContext) => {
           this.errorRaised(error);
@@ -102,6 +119,12 @@ export class SubscribeTest extends EventPerfTest<ReceiverOptions> {
 
   async globalCleanup(): Promise<void> {
     await consumer.close();
+    // The following might just be noise if we don't think there are related changes to the code
+    if (this.parsedOptions["log-median-batch-size"].value) {
+      console.log(`\tBatch count: ${this.callbackCallsCount}, Batch count per sec: ${this.callbackCallsCount / this.parsedOptions.duration.value}`);
+      console.log(`\tmessagesPerBatch: ${this.messagesPerBatch}`);
+      console.log(`\tmessagesPerBatch... median: ${median(this.messagesPerBatch)}, avg: ${this.messagesPerBatch.reduce((a, b) => a + b, 0) / this.messagesPerBatch.length}, max: ${Math.max(...this.messagesPerBatch)}, min: ${Math.min(...this.messagesPerBatch)}`);
+    }
   }
 }
 
@@ -138,3 +161,18 @@ async function sendBatch(
 
   await producer.close();
 }
+
+function median(values: number[]) {
+  if (values.length === 0) throw new Error("No inputs while calculating median");
+
+  values.sort(function (a, b) {
+    return a - b;
+  });
+
+  const half = Math.floor(values.length / 2);
+
+  if (values.length % 2)
+    return values[half];
+
+  return (values[half - 1] + values[half]) / 2.0;
+}