[Service Bus] StressTests scenarios

eng/.docsettings.yml

@@ -13,6 +13,7 @@ omitted_paths:
   - sdk/eventhub/*/test/README.md
   - sdk/search/*/test/README.md
   - sdk/servicebus/*/test/README.md
+  - sdk/servicebus/service-bus/test/stress-test-track-2/*
   - sdk/schemaregistry/README.md
   - sdk/servicebus/service-bus/samples-v1/*
   - sdk/storage/*/test/README.md

sdk/servicebus/service-bus/test/stress-test-track-2/SetupAndGuide.md

@@ -0,0 +1,91 @@
+### Setup
+
+Before running the code,
+
+- Setup the `SERVICEBUS_CONNECTION_STRING` environment variable. (Queue resource will be created in the beginning for the test period and teared down at the end)
+- Run `rushx pack` at `sdk/servicebus/service-bus` folder to generate the current version of service-bus package. (Or skip this step and install the version that you prefer)
+- Run `npm install` in this folder to install the Service Bus package that should be stress tested.
+
+### Scenarios
+
+Here are the scenarios being covered and how you can run them.
+
+1. Single sender that keeps sending messages in series.
+   `ts-node scenarioSend.ts` (runs with defaults)
+   `ts-node scenarioSend.ts --numberOfMessagesPerSend=100 delayBetweenSendsInMs=50`(with flags)
+   Flags that can be set
+
+   - testDurationInMs (default = 3600000) // 60 minutes
+   - numberOfMessagesPerSend (default = 1)
+   - delayBetweenSendsInMs (default = 0)
+   - totalNumberOfMessagesToSend (default = Infinity)
+   - useScheduleApi (default = false)
+
+2. Batch receive with a streaming sender
+   `ts-node scenarioBatchReceive.ts` (runs with defaults)
+   `ts-node scenarioBatchReceive.ts --delayBetweenReceivesInMs=200 --totalNumberOfMessagesToSend=5000` (With flags)
+   Flags that can be set
+
+   - testDurationInMs (default = 3600000) // 1hr
+   - receiveMode (default = "peekLock")
+   - receiveBatchMaxMessageCount (default = 10)
+   - receiveBatchMaxWaitTimeInMs (default = 10000)
+   - delayBetweenReceivesInMs (default = 0)
+   - numberOfMessagesPerSend (default = 1)
+   - delayBetweenSendsInMs (default = 0)
+   - totalNumberOfMessagesToSend (default = Infinity)
+
+3. Renew message locks for a duration
+   `ts-node scenarioRenewMessageLock.ts` (runs with defaults)
+   `ts-node scenarioRenewMessageLock.ts --delayBetweenReceivesInMs=200 --totalNumberOfMessagesToSend=5000` (With flags)
+   Flags that can be set
+
+   - testDurationInMs (default = 3600000) // 1hr
+   - receiveBatchMaxMessageCount (default = 10)
+   - receiveBatchMaxWaitTimeInMs (default = 10000)
+   - delayBetweenReceivesInMs (default = 0)
+   - numberOfMessagesPerSend (default = 1)
+   - delayBetweenSendsInMs (default = 0)
+   - totalNumberOfMessagesToSend (default = Infinity)
+
+4. Renew session locks for a duration
+   `ts-node scenarioRenewSessionLock.ts` (runs with defaults)
+   `ts-node scenarioRenewSessionLock.ts --delayBetweenReceivesInMs=200 --totalNumberOfMessagesToSend=5000` (With flags)
+   Flags that can be set
+
+   - testDurationInMs (default = 3600000) // 1hr
+   - receiveBatchMaxMessageCount (default = 10)
+   - receiveBatchMaxWaitTimeInMs (default = 10000)
+   - delayBetweenReceivesInMs (default = 0)
+   - numberOfMessagesPerSend (default = 100)
+   - delayBetweenSendsInMs (default = 0)
+   - totalNumberOfMessagesToSend (default = Infinity)
+   - autoLockRenewal (default = false)
+
+5. Streaming Receive
+   `ts-node scenarioStreamingReceive.ts` (runs with defaults)
+   `ts-node scenarioStreamingReceive.ts --manualLockRenewal=false --totalNumberOfMessagesToSend=5000` (With flags)
+   Flags that can be set
+
+   - testDurationInMs (default = 3600000) // 1hr
+   - receiveMode (default = "peekLock")
+   - autoComplete (default = true)
+   - maxConcurrentCalls (default = 100)
+   - maxAutoRenewLockDurationInMs (default = 0)
+   - manualLockRenewal (default = true)
+   - numberOfMessagesPerSend (default = 1)
+   - delayBetweenSendsInMs (default = 0)
+   - totalNumberOfMessagesToSend (default = Infinity)
+   - completeMessageAfterDuration (default = true)
+
+6. Management client - peek messages
+   `ts-node scenarioPeekMessages.ts` (runs with defaults)
+   `ts-node scenarioPeekMessages.ts --delayBetweenPeeksInMs=200 --totalNumberOfMessagesToSend=5000` (With flags)
+   Flags that can be set
+
+   - testDurationInMs (default = 3600000) // 1hr
+   - peekMaxMessageCount (default = 10)
+   - delayBetweenPeeksInMs (default = 0)
+   - numberOfMessagesPerSend (default = 1)
+   - delayBetweenSendsInMs (default = 0)
+   - totalNumberOfMessagesToSend (default = Infinity)

sdk/servicebus/service-bus/test/stress-test-track-2/package.json

@@ -0,0 +1,20 @@
+{
+  "name": "stress-test-track-2",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "ts-node scenarioBatchReceive.ts --testDurationInMs=10000 --recieveMode=receiveAndDelete"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "@azure/service-bus": "file:../../azure-service-bus-7.0.0-preview.8.tgz",
+    "minimist": "^1.2.5",
+    "uuid": "^8.3.0"
+  },
+  "devDependencies": {
+    "@types/uuid": "^8.3.0"
+  }
+}

sdk/servicebus/service-bus/test/stress-test-track-2/scenarioBatchReceive.ts

@@ -0,0 +1,120 @@
+import {
+  ReceiveMode,
+  ServiceBusClient,
+  ServiceBusReceivedMessage,
+  ServiceBusReceiver
+} from "@azure/service-bus";
+import { SBStressTestsBase } from "./stressTestsBase";
+import { delay } from "rhea-promise";
+import parsedArgs from "minimist";
+
+// Load the .env file if it exists
+import * as dotenv from "dotenv";
+dotenv.config();
+
+// Define connection string and related Service Bus entity names here
+const connectionString = process.env.SERVICEBUS_CONNECTION_STRING || "<connection string>";
+
+interface ScenarioReceiveBatchOptions {
+  testDurationInMs?: number;
+  receiveMode?: ReceiveMode;
+  receiveBatchMaxMessageCount?: number;
+  receiveBatchMaxWaitTimeInMs?: number;
+  delayBetweenReceivesInMs?: number;
+  numberOfMessagesPerSend?: number;
+  delayBetweenSendsInMs?: number;
+  totalNumberOfMessagesToSend?: number;
+  maxAutoLockRenewalDurationInMs?: number;
+  settleMessageOnReceive: boolean;
+}
+
+function sanitizeOptions(
+  options: ScenarioReceiveBatchOptions
+): Required<ScenarioReceiveBatchOptions> {
+  return {
+    testDurationInMs: options.testDurationInMs || 60 * 60 * 1000, // Default = 60 minutes
+    receiveMode: (options.receiveMode as ReceiveMode) || "peekLock",
+    receiveBatchMaxMessageCount: options.receiveBatchMaxMessageCount || 10,
+    receiveBatchMaxWaitTimeInMs: options.receiveBatchMaxWaitTimeInMs || 10000,
+    delayBetweenReceivesInMs: options.delayBetweenReceivesInMs || 0,
+    numberOfMessagesPerSend: options.numberOfMessagesPerSend || 1,
+    delayBetweenSendsInMs: options.delayBetweenSendsInMs || 0,
+    totalNumberOfMessagesToSend: options.totalNumberOfMessagesToSend || Infinity,
+    maxAutoLockRenewalDurationInMs: options.maxAutoLockRenewalDurationInMs || 0, // 0 = disabled
+    settleMessageOnReceive: options.settleMessageOnReceive || false
+  };
+}
+
+export async function scenarioReceiveBatch() {
+  const testOptions = sanitizeOptions(parsedArgs<ScenarioReceiveBatchOptions>(process.argv));
+  const {
+    testDurationInMs,
+    receiveMode,
+    receiveBatchMaxMessageCount,
+    receiveBatchMaxWaitTimeInMs,
+    delayBetweenReceivesInMs,
+    numberOfMessagesPerSend,
+    delayBetweenSendsInMs,
+    totalNumberOfMessagesToSend,
+    maxAutoLockRenewalDurationInMs,
+    settleMessageOnReceive
+  } = testOptions;
+
+  // Sending stops after 70% of total duration to give the receiver a chance to clean up and receive all the messages
+  const testDurationForSendInMs = testDurationInMs * 0.7;
+
+  const startedAt = new Date();
+
+  const stressBase = new SBStressTestsBase({
+    snapshotFocus: ["send-info", "receive-info"]
+  });
+  const sbClient = new ServiceBusClient(connectionString);
+
+  await stressBase.init(undefined, undefined, testOptions);
+  const sender = sbClient.createSender(stressBase.queueName);
+  let receiver: ServiceBusReceiver<ServiceBusReceivedMessage>;
+
+  if (receiveMode === "receiveAndDelete") {
+    receiver = sbClient.createReceiver(stressBase.queueName, {
+      receiveMode: "receiveAndDelete",
+      maxAutoLockRenewalDurationInMs
+    });
+  } else {
+    receiver = sbClient.createReceiver(stressBase.queueName, { maxAutoLockRenewalDurationInMs });
+  }
+
+  async function sendMessages() {
+    let elapsedTime = new Date().valueOf() - startedAt.valueOf();
+    while (
+      elapsedTime < testDurationForSendInMs &&
+      stressBase.messagesSent.length < totalNumberOfMessagesToSend
+    ) {
+      await stressBase.sendMessages([sender], numberOfMessagesPerSend);
+      elapsedTime = new Date().valueOf() - startedAt.valueOf();
+      await delay(delayBetweenSendsInMs);
+    }
+  }
+
+  async function receiveMessages() {
+    let elapsedTime = new Date().valueOf() - startedAt.valueOf();
+    while (elapsedTime < testDurationInMs) {
+      await stressBase.receiveMessages(
+        receiver,
+        receiveBatchMaxMessageCount,
+        receiveBatchMaxWaitTimeInMs,
+        settleMessageOnReceive
+      );
+      elapsedTime = new Date().valueOf() - startedAt.valueOf();
+      await delay(delayBetweenReceivesInMs);
+    }
+  }
+
+  await Promise.all([sendMessages(), receiveMessages()]);
+  await sbClient.close();
+
+  await stressBase.end();
+}
+
+scenarioReceiveBatch().catch((err) => {
+  console.log("Error occurred: ", err);
+});

sdk/servicebus/service-bus/test/stress-test-track-2/scenarioCloseOpen.ts

@@ -0,0 +1,80 @@
+import { ServiceBusClient } from "@azure/service-bus";
+import { SBStressTestsBase } from "./stressTestsBase";
+import { delay } from "rhea-promise";
+import parsedArgs from "minimist";
+
+// Load the .env file if it exists
+import * as dotenv from "dotenv";
+dotenv.config();
+
+// Define connection string and related Service Bus entity names here
+const connectionString = process.env.SERVICEBUS_CONNECTION_STRING || "<connection string>";
+
+interface ScenarioCloseOptions {
+  testDurationInMs?: number;
+  receiveBatchMaxMessageCount?: number;
+  receiveBatchMaxWaitTimeInMs?: number;
+  numberOfMessagesPerSend?: number;
+  delayBeforeCallingCloseInMs?: number;
+  shouldCreateNewClientEachTime?: boolean;
+}
+
+function sanitizeOptions(options: ScenarioCloseOptions): Required<ScenarioCloseOptions> {
+  return {
+    testDurationInMs: options.testDurationInMs || 60 * 1000, // Default = 60 minutes
+    receiveBatchMaxMessageCount: options.receiveBatchMaxMessageCount || 10,
+    receiveBatchMaxWaitTimeInMs: options.receiveBatchMaxWaitTimeInMs || 10000,
+    numberOfMessagesPerSend: options.numberOfMessagesPerSend || 1,
+    delayBeforeCallingCloseInMs: options.delayBeforeCallingCloseInMs || 100,
+    shouldCreateNewClientEachTime: options.shouldCreateNewClientEachTime || true
+  };
+}
+
+export async function scenarioClose() {
+  const testOptions = sanitizeOptions(parsedArgs<ScenarioCloseOptions>(process.argv));
+
+  const {
+    testDurationInMs,
+    receiveBatchMaxMessageCount,
+    receiveBatchMaxWaitTimeInMs,
+    numberOfMessagesPerSend,
+    delayBeforeCallingCloseInMs,
+    shouldCreateNewClientEachTime
+  } = testOptions;
+
+  const startedAt = new Date();
+
+  const stressBase = new SBStressTestsBase({
+    snapshotFocus: ["send-info", "receive-info", "close-info"]
+  });
+  let sbClient = new ServiceBusClient(connectionString);
+
+  await stressBase.init(undefined, undefined, testOptions);
+
+  let elapsedTime = new Date().valueOf() - startedAt.valueOf();
+  while (elapsedTime < testDurationInMs) {
+    const sender = sbClient.createSender(stressBase.queueName);
+    const receiver = sbClient.createReceiver(stressBase.queueName);
+    await stressBase.sendMessages([sender], numberOfMessagesPerSend);
+    await stressBase.receiveMessages(
+      receiver,
+      receiveBatchMaxMessageCount,
+      receiveBatchMaxWaitTimeInMs
+    );
+    await delay(delayBeforeCallingCloseInMs);
+    await stressBase.callClose(sender, "sender");
+    await stressBase.callClose(receiver, "receiver");
+    if (shouldCreateNewClientEachTime) {
+      await stressBase.callClose(sbClient, "client");
+      sbClient = new ServiceBusClient(connectionString);
+    }
+    elapsedTime = new Date().valueOf() - startedAt.valueOf();
+  }
+
+  await sbClient.close();
+  await stressBase.end();
+}
+
+scenarioClose().catch((err) => {
+  console.log("Error occurred: ", err);
+});