forked from Azure/azure-sdk-for-js
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[Perf] [Service Bus] [Event Hubs] Event Based SDKs Perf Framework - p…
…art 2 (Azure#18859) * clean up old scripts * tests -> perfTest * base, event, batch perf test classes * program.ts * incorrect comment * postSetup and preCleanup * ParsedPerfOptions<TOptions = Record<string, unknown>> and related updates * remove ! from options * MockReceiverTest * import "@azure/abort-controller" at perf classes * BatchReceiveTest * remove ! * changelog * receiveBatch test * remove aborter * docs * BatchSendTest * receiveMessages * no need of timer * - Adds `PerfTestBase` and `BatchPerfTest`. * lint fixes * checkpoitn * checkpoint * manage parallels in base class * parallels in base class * MockEventReceiverTest and the framework * fix readme * formatting * unneeded changes * EventPerfTest * mock test * subscribe test for service-bus * lint error * formatting * simplify * update desc * checkpoint * subscribe-test * mock event receiver 2 * formatting * Address Mike's feedback * lint format * remove min-max delay * maxEventsPerSecond logic * Apply suggestions from code review * Make EventPerfTest.eventRaise() and errorRaised() sync instead of async - Methods do not perform any async work, and sync methods add less overhead to perf framework * Cache event arrays to increase MockEventHubConsumerClient throughput * Add MockReceiverTest to test list * Improve throughput of MockReceiverTest - Now matches NoOpTest * AverageBatchSize - Also update options to match .NET * Jeremy's feedback * Clear the duration timer if the test ended sooner * Mike's feedbakc * lint:fix * minor feedback Co-authored-by: Mike Harder <mharder@microsoft.com>
- Loading branch information
1 parent
aaff27a
commit efe321c
Showing
18 changed files
with
751 additions
and
51 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
136 changes: 136 additions & 0 deletions
136
sdk/eventhub/perf-tests/event-hubs/test/subscribe.spec.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,136 @@ | ||
| // Copyright (c) Microsoft Corporation. | ||
| // Licensed under the MIT license. | ||
|
|
||
| import { | ||
| earliestEventPosition, | ||
| EventHubConsumerClient, | ||
| EventHubProducerClient, | ||
| MessagingError, | ||
| PartitionContext, | ||
| ReceivedEventData, | ||
| } from "@azure/event-hubs"; | ||
| import { PerfOptionDictionary, EventPerfTest, getEnvVar } from "@azure/test-utils-perf"; | ||
|
|
||
| interface ReceiverOptions { | ||
| "number-of-events": number; | ||
| "event-size-in-bytes": number; | ||
| partitions: number; | ||
| "max-batch-size": number; | ||
| } | ||
|
|
||
| const connectionString = getEnvVar("EVENTHUB_CONNECTION_STRING"); | ||
| const eventHubName = getEnvVar("EVENTHUB_NAME"); | ||
| const consumerGroup = getEnvVar("CONSUMER_GROUP_NAME"); | ||
|
|
||
| const consumer = new EventHubConsumerClient(consumerGroup, connectionString, eventHubName); | ||
|
|
||
| export class SubscribeTest extends EventPerfTest<ReceiverOptions> { | ||
| receiver: EventHubConsumerClient; | ||
| subscriber: { close: () => Promise<void> } | undefined; | ||
|
|
||
| options: PerfOptionDictionary<ReceiverOptions> = { | ||
| "number-of-events": { | ||
| required: true, | ||
| description: "Total number of events to send", | ||
| shortName: "events", | ||
| longName: "events", | ||
| defaultValue: 10000, | ||
| }, | ||
| "event-size-in-bytes": { | ||
| required: true, | ||
| description: "Size of each event in bytes", | ||
| shortName: "size", | ||
| longName: "size-in-bytes", | ||
| defaultValue: 2000, | ||
| }, | ||
| partitions: { | ||
| required: true, | ||
| description: "number of partitions to publish, -1 publishes to all partitions", | ||
| shortName: "p", | ||
| longName: "partitions", | ||
| defaultValue: 10, | ||
| }, | ||
| "max-batch-size": { | ||
| required: true, | ||
| description: "The number of events to request per batch", | ||
| shortName: "max-count", | ||
| longName: "max-batch-size", | ||
| defaultValue: 100, | ||
| }, | ||
| }; | ||
|
|
||
| constructor() { | ||
| super(); | ||
| this.receiver = consumer; | ||
| } | ||
|
|
||
| /** | ||
| * Sends the messages to be received later. | ||
| */ | ||
| async globalSetup(): Promise<void> { | ||
| const { | ||
| "number-of-events": { value: numberOfEvents }, | ||
| "event-size-in-bytes": { value: eventSize }, | ||
| } = this.parsedOptions; | ||
|
|
||
| await sendBatch(numberOfEvents, eventSize, this.parsedOptions["partitions"].value); | ||
| } | ||
|
|
||
| setup() { | ||
| this.subscriber = this.receiver.subscribe( | ||
| { | ||
| processEvents: async (events: ReceivedEventData[], _context: PartitionContext) => { | ||
| for (const _event of events) { | ||
| this.eventRaised(); | ||
| } | ||
| }, | ||
| processError: async (error: Error | MessagingError, _context: PartitionContext) => { | ||
| this.errorRaised(error); | ||
| }, | ||
| }, | ||
| { | ||
| maxBatchSize: this.parsedOptions["max-batch-size"].value, | ||
| startPosition: earliestEventPosition, | ||
| } | ||
| ); | ||
| } | ||
|
|
||
| async cleanup() { | ||
| await this.subscriber?.close(); | ||
| await this.receiver.close(); | ||
| } | ||
|
|
||
| async globalCleanup(): Promise<void> { | ||
| await consumer.close(); | ||
| } | ||
| } | ||
|
|
||
| async function sendBatch( | ||
| numberOfEvents: number, | ||
| eventBodySize: number, | ||
| partitions: number | ||
| ): Promise<void> { | ||
| const _payload = Buffer.alloc(eventBodySize); | ||
| const producer = new EventHubProducerClient(connectionString, eventHubName); | ||
| let partitionIds = await producer.getPartitionIds(); | ||
| const numberOfPartitions = | ||
| partitionIds.length < partitions || partitions === -1 ? partitionIds.length : partitions; | ||
| partitionIds = partitionIds.slice(0, numberOfPartitions); | ||
| const numberOfEventsPerPartition = Math.ceil(numberOfEvents / numberOfPartitions); | ||
|
|
||
| for (const partitionId of partitionIds) { | ||
| const batch = await producer.createBatch({ partitionId }); | ||
| let numberOfEventsSent = 0; | ||
| // add events to our batch | ||
| while (numberOfEventsSent < numberOfEventsPerPartition) { | ||
| while ( | ||
| batch.tryAdd({ body: _payload }) && | ||
| numberOfEventsSent + batch.count <= numberOfEventsPerPartition | ||
| ); | ||
| await producer.sendBatch(batch); | ||
| numberOfEventsSent = numberOfEventsSent + batch.count; | ||
| } | ||
| } | ||
|
|
||
| await producer.close(); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
93 changes: 93 additions & 0 deletions
93
sdk/servicebus/perf-tests/service-bus/test/subscribe.spec.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,93 @@ | ||
| // Copyright (c) Microsoft Corporation. | ||
| // Licensed under the MIT license. | ||
|
|
||
| import { | ||
| ProcessErrorArgs, | ||
| ServiceBusReceivedMessage, | ||
| ServiceBusReceiver, | ||
| } from "@azure/service-bus"; | ||
| import { PerfOptionDictionary, EventPerfTest } from "@azure/test-utils-perf"; | ||
| import { sendMessages } from "./receiveBatch.spec"; | ||
| import { ServiceBusTest } from "./sbBase.spec"; | ||
|
|
||
| interface ReceiverOptions { | ||
| "number-of-messages": number; | ||
| "message-body-size-in-bytes": number; | ||
| "max-concurrent-calls": number; | ||
| } | ||
|
|
||
| export class SubscribeTest extends EventPerfTest<ReceiverOptions> { | ||
| receiver: ServiceBusReceiver; | ||
| subscriber: { close: () => Promise<void> } | undefined; | ||
|
|
||
| options: PerfOptionDictionary<ReceiverOptions> = { | ||
| "number-of-messages": { | ||
| required: true, | ||
| description: "Total number of messages to send", | ||
| shortName: "messages", | ||
| longName: "messages", | ||
| defaultValue: 100000, | ||
| }, | ||
| "message-body-size-in-bytes": { | ||
| required: true, | ||
| description: "Size of each message body in bytes", | ||
| shortName: "size", | ||
| longName: "size-in-bytes", | ||
| defaultValue: 2000, | ||
| }, | ||
| "max-concurrent-calls": { | ||
| required: true, | ||
| description: "max concurrent calls", | ||
| shortName: "mcc", | ||
| longName: "max-concurrent-calls", | ||
| defaultValue: 10, | ||
| }, | ||
| }; | ||
|
|
||
| constructor() { | ||
| super(); | ||
| this.receiver = ServiceBusTest.sbClient.createReceiver(ServiceBusTest.queueName, { | ||
| receiveMode: "receiveAndDelete", | ||
| }); | ||
| } | ||
|
|
||
| /** | ||
| * Sends the messages to be received later. | ||
| */ | ||
| async globalSetup(): Promise<void> { | ||
| await ServiceBusTest.sbAdminClient.createQueue(ServiceBusTest.queueName); | ||
| const sender = ServiceBusTest.sbClient.createSender(ServiceBusTest.queueName); | ||
|
|
||
| const { | ||
| "number-of-messages": { value: numberOfMessages }, | ||
| "message-body-size-in-bytes": { value: messageBodySize }, | ||
| } = this.parsedOptions; | ||
|
|
||
| await sendMessages(sender, numberOfMessages, messageBodySize); | ||
| } | ||
|
|
||
| setup() { | ||
| this.subscriber = this.receiver.subscribe( | ||
| { | ||
| processMessage: async (_message: ServiceBusReceivedMessage) => { | ||
| // { event: _message } | ||
| this.eventRaised(); | ||
| }, | ||
| processError: async (args: ProcessErrorArgs) => { | ||
| this.errorRaised(args.error); | ||
| }, | ||
| }, | ||
| { maxConcurrentCalls: this.parsedOptions["max-concurrent-calls"].value } | ||
| ); | ||
| } | ||
|
|
||
| async cleanup() { | ||
| this.subscriber && (await this.subscriber.close()); | ||
| await this.receiver.close(); | ||
| } | ||
|
|
||
| async globalCleanup(): Promise<void> { | ||
| await ServiceBusTest.sbClient.close(); | ||
| await ServiceBusTest.sbAdminClient.deleteQueue(ServiceBusTest.queueName); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.