[azservicebus] Fixing the stress tests to take care of some debt (Azure#21719)

- Upgrade to mariner:2.0
- Remove "custom" stats from the tests and use the names that Liudmila's standardizing.
- Some tests weren't properly ending their runs, which meant they didn't report an End() event (mostly affected reporting)
- Start events contain some more context about the test (ie, number of rounds, etc..)

Author: richardpark-msft

sdk/messaging/azservicebus/internal/stress/Dockerfile

@@ -1,4 +1,4 @@
-FROM mcr.microsoft.com/oss/go/microsoft/golang:1.18-fips-cbl-mariner2.0 as build
+FROM mcr.microsoft.com/oss/go/microsoft/golang:1.21-cbl-mariner2.0 as build
 # you'll need to run this build from the root of the azservicebus module
 ENV GOOS=linux 
 ENV GOARCH=amd64 
@@ -12,7 +12,8 @@ RUN go build -o stress .
 #
 # This image pulls from the other container (using COPY --from=build). The resulting image
 # will be pushed to our container registry.
-FROM mcr.microsoft.com/cbl-mariner/base/core:1.0
+FROM mcr.microsoft.com/cbl-mariner/base/core:2.0
 WORKDIR /app
 COPY --from=build /src/internal/stress/stress /app/stress
+RUN yum update && yum install -y ca-certificates
 ENTRYPOINT ["/bin/bash"]

sdk/messaging/azservicebus/internal/stress/scenarios-matrix.yaml

@@ -28,6 +28,10 @@ matrix:
     finiteSessions:
       testTarget: finiteSessions
       memory: "4Gi"
+    finiteSessionsWithChaos:
+      testTarget: finiteSessions
+      memory: "4Gi"
+      chaos: "true"
     idleFastReconnect:
       testTarget: idleFastReconnect
       memory: "0.5Gi"

sdk/messaging/azservicebus/internal/stress/shared/metric_names.go

@@ -0,0 +1,54 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package shared
+
+type Metric string
+
+// These names are modeled off of the metrics from Java
+// https://github.com/Azure/azure-sdk-for-java/tree/main/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/instrumentation/ServiceBusMeter.java
+//
+// and from our standard for attributes:
+// https://gist.github.com/lmolkova/e4215c0f44a49ef824983382762e6b92
+const (
+	MetricConnectionLost Metric = "messaging.servicebus.connectionlost"
+	MetricMessagesSent   Metric = "messaging.servicebus.messages.sent"
+
+	// metrics related to Service Bus sessions (NOT amqp sessions)
+	MetricSessionAccept    Metric = "messaging.servicebus.session.accept"
+	MetricSessionTimeoutMS Metric = "messaging.servicebus.session.timeout"
+
+	MetricSettlementRequestDuration Metric = "messaging.servicebus.settlement.request.duration"
+	MetricReceiveLag                Metric = "messaging.servicebus.receiver.lag"
+
+	MetricAMQPSendDuration Metric = "messaging.az.amqp.producer.send.duration"
+	// - amqp.delivery_state (ex: 'accepted')
+
+	MetricAMQPMgmtRequestDuration Metric = "messaging.az.amqp.management.request.duration"
+	// - amqp.status_code (ex: 'accepted')
+
+	MetricAMQPSettlementRequestDuration Metric = "messaging.servicebus.settlement.request.duration"
+	MetricAMQPSettlementSequenceNum     Metric = "messaging.servicebus.settlement.sequence_number"
+
+	// OTel attributes (should come in naturally when we swap over to using OTel throughout)
+	// otel.status_code
+
+	// TODO: I've made these up entirely.
+	MetricMessageReceived Metric = "messaging.servicebus.messages.received"
+	MetricMessagePeeked   Metric = "messaging.servicebus.messages.peeked"
+	MetricCloseDuration   Metric = "messaging.servicebus.close.duration"
+	MetricLockRenew       Metric = "messaging.servicebus.lockrenew.duration" // TODO: separate for session vs message lock?
+)
+
+const (
+	AttrAMQPDeliveryState string = "amqp.delivery_state"
+	AttrAMQPStatusCode    string = "amqp.status_code"
+
+	// TODO: I made these up entirely
+	AttrMessageCount string = "amqp.message_count"
+)
+
+// these metrics are specific to stress tests and wouldn't be in customer code.
+const (
+	MetricStressSuccessfulCancels = "stress.cancels"
+)

sdk/messaging/azservicebus/internal/stress/shared/stats.go

@@ -23,7 +23,7 @@ type internalBatch interface {
 }
 
 type senderWrapper struct {
-	inner *azservicebus.Sender
+	inner *TrackingSender
 }
 
 func (sw *senderWrapper) SendMessageBatch(ctx context.Context, batch internalBatch) error {
@@ -34,7 +34,7 @@ func (sw *senderWrapper) NewMessageBatch(ctx context.Context, options *azservice
 	return sw.inner.NewMessageBatch(ctx, options)
 }
 
-func NewStreamingMessageBatch(ctx context.Context, sender internalBatchSender, stats *Stats) (*StreamingMessageBatch, error) {
+func NewStreamingMessageBatch(ctx context.Context, sender internalBatchSender) (*StreamingMessageBatch, error) {
 	batch, err := sender.NewMessageBatch(ctx, nil)
 
 	if err != nil {
@@ -43,14 +43,12 @@ func NewStreamingMessageBatch(ctx context.Context, sender internalBatchSender, s
 
 	return &StreamingMessageBatch{
 		sender:       sender,
-		stats:        stats,
 		currentBatch: batch,
 	}, nil
 }
 
 type StreamingMessageBatch struct {
 	sender       internalBatchSender
-	stats        *Stats
 	currentBatch internalBatch
 }
 
@@ -72,10 +70,6 @@ func (sb *StreamingMessageBatch) Add(ctx context.Context, msg *azservicebus.Mess
 		return err
 	}
 
-	if sb.stats != nil {
-		sb.stats.AddSent(sb.currentBatch.NumMessages())
-	}
-
 	// throttle a teeny bit.
 	time.Sleep(time.Second)
 
@@ -106,8 +100,5 @@ func (sb *StreamingMessageBatch) Close(ctx context.Context) error {
 		return err
 	}
 
-	if sb.stats != nil {
-		sb.stats.AddSent(sb.currentBatch.NumMessages())
-	}
 	return nil
 }