Skip to content

Commit b804bad

Browse files
berndverstberndverst
authored
Merge branch 'master' into huawei-obs-bindings
2 parents df661fd + e1b6b01 commit b804bad

10 files changed

Lines changed: 239 additions & 107 deletions

File tree

‎bindings/azure/servicebusqueues/servicebusqueues.go‎

Lines changed: 80 additions & 57 deletions
Original file line numberDiff line numberDiff line change
@@ -19,13 +19,12 @@ import (
1919
"errors"
2020
"strings"
2121
"sync"
22-
"sync/atomic"
2322
"time"
2423

2524
"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
2625
servicebus "github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus"
2726
sbadmin "github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/admin"
28-
"github.com/cenkalti/backoff/v4"
27+
"go.uber.org/ratelimit"
2928

3029
azauth "github.com/dapr/components-contrib/authentication/azure"
3130
"github.com/dapr/components-contrib/bindings"
@@ -39,33 +38,37 @@ const (
3938
label = "label"
4039
id = "id"
4140

42-
// AzureServiceBusDefaultMessageTimeToLive defines the default time to live for queues, which is 14 days. The same way Azure Portal does.
43-
AzureServiceBusDefaultMessageTimeToLive = time.Hour * 24 * 14
41+
// azureServiceBusDefaultMessageTimeToLive defines the default time to live for queues, which is 14 days. The same way Azure Portal does.
42+
azureServiceBusDefaultMessageTimeToLive = time.Hour * 24 * 14
4443

4544
// Default timeout in seconds
46-
DefaultTimeoutInSec = 60
45+
defaultTimeoutInSec = 60
46+
47+
// Default rate of retriable errors per second
48+
defaultMaxRetriableErrorsPerSec = 10
4749
)
4850

4951
// AzureServiceBusQueues is an input/output binding reading from and sending events to Azure Service Bus queues.
5052
type AzureServiceBusQueues struct {
51-
metadata *serviceBusQueuesMetadata
52-
client *servicebus.Client
53-
adminClient *sbadmin.Client
54-
shutdownSignal int32
55-
timeout time.Duration
56-
sender *servicebus.Sender
57-
senderLock sync.RWMutex
58-
logger logger.Logger
59-
ctx context.Context
60-
cancel context.CancelFunc
53+
metadata *serviceBusQueuesMetadata
54+
client *servicebus.Client
55+
adminClient *sbadmin.Client
56+
timeout time.Duration
57+
sender *servicebus.Sender
58+
senderLock sync.RWMutex
59+
retriableErrLimit ratelimit.Limiter
60+
logger logger.Logger
61+
ctx context.Context
62+
cancel context.CancelFunc
6163
}
6264

6365
type serviceBusQueuesMetadata struct {
64-
ConnectionString string `json:"connectionString"`
65-
NamespaceName string `json:"namespaceName,omitempty"`
66-
QueueName string `json:"queueName"`
67-
TimeoutInSec int `json:"timeoutInSec"`
68-
ttl time.Duration
66+
ConnectionString string `json:"connectionString"`
67+
NamespaceName string `json:"namespaceName,omitempty"`
68+
QueueName string `json:"queueName"`
69+
TimeoutInSec int `json:"timeoutInSec"`
70+
MaxRetriableErrorsPerSec *int `json:"maxRetriableErrorsPerSec"`
71+
ttl time.Duration
6972
}
7073

7174
// NewAzureServiceBusQueues returns a new AzureServiceBusQueues instance.
@@ -83,6 +86,11 @@ func (a *AzureServiceBusQueues) Init(metadata bindings.Metadata) (err error) {
8386
return err
8487
}
8588
a.timeout = time.Duration(a.metadata.TimeoutInSec) * time.Second
89+
if *a.metadata.MaxRetriableErrorsPerSec > 0 {
90+
a.retriableErrLimit = ratelimit.New(*a.metadata.MaxRetriableErrorsPerSec)
91+
} else {
92+
a.retriableErrLimit = ratelimit.NewUnlimited()
93+
}
8694

8795
userAgent := "dapr-" + logger.DaprVersion
8896
if a.metadata.ConnectionString != "" {
@@ -144,8 +152,6 @@ func (a *AzureServiceBusQueues) Init(metadata bindings.Metadata) (err error) {
144152
}
145153
}
146154

147-
a.clearShutdown()
148-
149155
a.ctx, a.cancel = context.WithCancel(context.Background())
150156

151157
return nil
@@ -173,15 +179,22 @@ func (a *AzureServiceBusQueues) parseMetadata(metadata bindings.Metadata) (*serv
173179
}
174180
if !ok {
175181
// set the same default message time to live as suggested in Azure Portal to 14 days (otherwise it will be 10675199 days)
176-
ttl = AzureServiceBusDefaultMessageTimeToLive
182+
ttl = azureServiceBusDefaultMessageTimeToLive
177183
}
178184
m.ttl = ttl
179185

180186
// Queue names are case-insensitive and are forced to lowercase. This mimics the Azure portal's behavior.
181187
m.QueueName = strings.ToLower(m.QueueName)
182188

183189
if m.TimeoutInSec < 1 {
184-
m.TimeoutInSec = DefaultTimeoutInSec
190+
m.TimeoutInSec = defaultTimeoutInSec
191+
}
192+
193+
if m.MaxRetriableErrorsPerSec == nil {
194+
m.MaxRetriableErrorsPerSec = to.Ptr(defaultMaxRetriableErrorsPerSec)
195+
}
196+
if *m.MaxRetriableErrorsPerSec < 0 {
197+
return nil, errors.New("maxRetriableErrorsPerSec must be non-negative")
185198
}
186199

187200
return &m, nil
@@ -232,26 +245,34 @@ func (a *AzureServiceBusQueues) Invoke(ctx context.Context, req *bindings.Invoke
232245
}
233246

234247
func (a *AzureServiceBusQueues) Read(handler func(context.Context, *bindings.ReadResponse) ([]byte, error)) error {
235-
// Connections need to retry forever with a maximum backoff of 5 minutes and exponential scaling.
236-
connConfig := retry.DefaultConfig()
237-
connConfig.Policy = retry.PolicyExponential
238-
connConfig.MaxInterval = 5 * time.Minute
239-
connBackoff := connConfig.NewBackOffWithContext(a.ctx)
240-
241-
for !a.isShutdown() {
242-
receiver := a.attemptConnectionForever(connBackoff)
248+
for a.ctx.Err() == nil {
249+
receiver := a.attemptConnectionForever()
243250
if receiver == nil {
244251
a.logger.Errorf("Failed to connect to Azure Service Bus Queue.")
245252
continue
246253
}
247254

248-
msgs, err := receiver.ReceiveMessages(a.ctx, 10, nil)
249-
if err != nil {
250-
a.logger.Warnf("Error reading from Azure Service Bus Queue binding: %s", err.Error())
251-
}
255+
// Receive messages loop
256+
// This continues until the context is canceled
257+
for a.ctx.Err() == nil {
258+
// Blocks until the connection is closed or the context is canceled
259+
msgs, err := receiver.ReceiveMessages(a.ctx, 1, nil)
260+
if err != nil {
261+
a.logger.Warnf("Error reading from Azure Service Bus Queue binding: %s", err.Error())
262+
}
263+
264+
l := len(msgs)
265+
if l == 0 {
266+
// We got no message, which is unusual too
267+
a.logger.Warn("Received 0 messages from Service Bus")
268+
continue
269+
} else if l > 1 {
270+
// We are requesting one message only; this should never happen
271+
a.logger.Errorf("Expected one message from Service Bus, but received %d", l)
272+
}
273+
274+
msg := msgs[0]
252275

253-
// Blocks until the connection is closed
254-
for _, msg := range msgs {
255276
body, err := msg.Body()
256277
if err != nil {
257278
a.logger.Warnf("Error reading message body: %s", err.Error())
@@ -277,7 +298,10 @@ func (a *AzureServiceBusQueues) Read(handler func(context.Context, *bindings.Rea
277298
continue
278299
}
279300

280-
err = receiver.CompleteMessage(a.ctx, msg, nil)
301+
// Use a background context in case a.ctx has been canceled already
302+
ctx, cancel := context.WithTimeout(context.Background(), a.timeout)
303+
err = receiver.CompleteMessage(ctx, msg, nil)
304+
cancel()
281305
if err != nil {
282306
a.logger.Warnf("Error completing message: %s", err.Error())
283307
continue
@@ -297,17 +321,30 @@ func (a *AzureServiceBusQueues) Read(handler func(context.Context, *bindings.Rea
297321
}
298322

299323
func (a *AzureServiceBusQueues) abandonMessage(receiver *servicebus.Receiver, msg *servicebus.ReceivedMessage) {
300-
ctx, cancel := context.WithTimeout(a.ctx, a.timeout)
324+
// Use a background context in case a.ctx has been canceled already
325+
ctx, cancel := context.WithTimeout(context.Background(), a.timeout)
301326
err := receiver.AbandonMessage(ctx, msg, nil)
327+
cancel()
302328
if err != nil {
303329
// Log only
304330
a.logger.Warnf("Error abandoning message: %s", err.Error())
305331
}
306-
cancel()
332+
333+
// If we're here, it means we got a retriable error, so we need to consume a retriable error token before this (synchronous) method returns
334+
// If there have been too many retriable errors per second, this method slows the consuming down
335+
a.logger.Debugf("Taking a retriable error token")
336+
before := time.Now()
337+
_ = a.retriableErrLimit.Take()
338+
a.logger.Debugf("Resumed after pausing for %v", time.Now().Sub(before))
307339
}
308340

309-
func (a *AzureServiceBusQueues) attemptConnectionForever(backoff backoff.BackOff) *servicebus.Receiver {
310-
var receiver *servicebus.Receiver
341+
func (a *AzureServiceBusQueues) attemptConnectionForever() (receiver *servicebus.Receiver) {
342+
// Connections need to retry forever with a maximum backoff of 5 minutes and exponential scaling.
343+
config := retry.DefaultConfig()
344+
config.Policy = retry.PolicyExponential
345+
config.MaxInterval = 5 * time.Minute
346+
backoff := config.NewBackOffWithContext(a.ctx)
347+
311348
retry.NotifyRecover(
312349
func() error {
313350
clientAttempt, err := a.client.NewReceiverForQueue(a.metadata.QueueName, nil)
@@ -345,19 +382,5 @@ func (a *AzureServiceBusQueues) Close() (err error) {
345382
return err
346383
}
347384
}
348-
a.setShutdown()
349385
return nil
350386
}
351-
352-
func (a *AzureServiceBusQueues) setShutdown() {
353-
atomic.CompareAndSwapInt32(&a.shutdownSignal, 0, 1)
354-
}
355-
356-
func (a *AzureServiceBusQueues) clearShutdown() {
357-
atomic.CompareAndSwapInt32(&a.shutdownSignal, 1, 0)
358-
}
359-
360-
func (a *AzureServiceBusQueues) isShutdown() bool {
361-
val := atomic.LoadInt32(&a.shutdownSignal)
362-
return val == 1
363-
}

‎bindings/azure/servicebusqueues/servicebusqueues_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -39,14 +39,14 @@ func TestParseMetadata(t *testing.T) {
3939
properties: map[string]string{"connectionString": "connString", "queueName": "queue1"},
4040
expectedConnectionString: "connString",
4141
expectedQueueName: "queue1",
42-
expectedTTL: AzureServiceBusDefaultMessageTimeToLive,
42+
expectedTTL: azureServiceBusDefaultMessageTimeToLive,
4343
},
4444
{
4545
name: "Empty TTL",
4646
properties: map[string]string{"connectionString": "connString", "queueName": "queue1", metadata.TTLMetadataKey: ""},
4747
expectedConnectionString: "connString",
4848
expectedQueueName: "queue1",
49-
expectedTTL: AzureServiceBusDefaultMessageTimeToLive,
49+
expectedTTL: azureServiceBusDefaultMessageTimeToLive,
5050
},
5151
{
5252
name: "With TTL",

‎go.mod‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -163,6 +163,7 @@ require (
163163
github.com/huaweicloud/huaweicloud-sdk-go-v3 v0.0.87
164164
github.com/labd/commercetools-go-sdk v0.3.2
165165
github.com/nacos-group/nacos-sdk-go/v2 v2.0.1
166+
go.uber.org/ratelimit v0.2.0
166167
gopkg.in/couchbase/gocb.v1 v1.6.4
167168
)
168169

@@ -172,6 +173,7 @@ require (
172173
github.com/99designs/go-keychain v0.0.0-20191008050251-8e49817e8af4 // indirect
173174
github.com/Azure/azure-sdk-for-go/sdk/messaging/internal v0.1.0 // indirect
174175
github.com/AzureAD/microsoft-authentication-library-for-go v0.4.0 // indirect
176+
github.com/andres-erbsen/clock v0.0.0-20160526145045-9e14626cd129 // indirect
175177
github.com/appscode/go-querystring v0.0.0-20170504095604-0126cfb3f1dc // indirect
176178
github.com/fsnotify/fsnotify v1.5.1 // indirect
177179
github.com/hashicorp/go-hclog v0.14.1 // indirect

‎go.sum‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -236,6 +236,8 @@ github.com/aliyun/credentials-go v1.1.2 h1:qU1vwGIBb3UJ8BwunHDRFtAhS6jnQLnde/yk0
236236
github.com/aliyun/credentials-go v1.1.2/go.mod h1:ozcZaMR5kLM7pwtCMEpVmQ242suV6qTJya2bDq4X1Tw=
237237
github.com/aliyunmq/mq-http-go-sdk v1.0.3 h1:/uhH7DUoaw9XTtsPgDp7zdPUyG5FBKj2GmJJph9z+6o=
238238
github.com/aliyunmq/mq-http-go-sdk v1.0.3/go.mod h1:JYfRMQoPexERvnNNBcal0ZQ2TVQ5ialDiW9ScjaadEM=
239+
github.com/andres-erbsen/clock v0.0.0-20160526145045-9e14626cd129 h1:MzBOUgng9orim59UnfUTLRjMpd09C5uEVQ6RPGeCaVI=
240+
github.com/andres-erbsen/clock v0.0.0-20160526145045-9e14626cd129/go.mod h1:rFgpPQZYZ8vdbc+48xibu8ALc3yeyd64IhHS+PU6Yyg=
239241
github.com/andybalholm/brotli v1.0.0/go.mod h1:loMXtMfwqflxFJPmdbJO0a3KNoPuLBgiu3qAvBg8x/Y=
240242
github.com/andybalholm/brotli v1.0.2 h1:JKnhI/XQ75uFBTiuzXpzFrUriDPiZjlOSzh6wXogP0E=
241243
github.com/andybalholm/brotli v1.0.2/go.mod h1:loMXtMfwqflxFJPmdbJO0a3KNoPuLBgiu3qAvBg8x/Y=
@@ -1404,6 +1406,8 @@ go.uber.org/multierr v1.5.0/go.mod h1:FeouvMocqHpRaaGuG9EjoKcStLC43Zu/fmqdUMPcKY
14041406
go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
14051407
go.uber.org/multierr v1.7.0 h1:zaiO/rmgFjbmCXdSYJWQcdvOCsthmdaHfr3Gm2Kx4Ec=
14061408
go.uber.org/multierr v1.7.0/go.mod h1:7EAYxJLBy9rStEaz58O2t4Uvip6FSURkq8/ppBp95ak=
1409+
go.uber.org/ratelimit v0.2.0 h1:UQE2Bgi7p2B85uP5dC2bbRtig0C+OeNRnNEafLjsLPA=
1410+
go.uber.org/ratelimit v0.2.0/go.mod h1:YYBV4e4naJvhpitQrWJu1vCpgB7CboMe0qhltKt6mUg=
14071411
go.uber.org/tools v0.0.0-20190618225709-2cfd321de3ee/go.mod h1:vJERXedbb3MVM5f9Ejo0C68/HhF8uaILCdgjnY+goOA=
14081412
go.uber.org/zap v1.9.1/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
14091413
go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=

‎pubsub/azure/servicebus/metadata.go‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ type metadata struct {
2525
MaxReconnectionAttempts int `json:"maxReconnectionAttempts"`
2626
ConnectionRecoveryInSec int `json:"connectionRecoveryInSec"`
2727
DisableEntityManagement bool `json:"disableEntityManagement"`
28+
MaxRetriableErrorsPerSec int `json:"MaxRetriableErrorsPerSec"`
2829
MaxDeliveryCount *int `json:"maxDeliveryCount"`
2930
LockDurationInSec *int `json:"lockDurationInSec"`
3031
DefaultMessageTimeToLiveInSec *int `json:"defaultMessageTimeToLiveInSec"`
@@ -34,3 +35,39 @@ type metadata struct {
3435
PublishInitialRetryIntervalInMs int `json:"publishInitialRetryInternalInMs"`
3536
NamespaceName string `json:"namespaceName,omitempty"`
3637
}
38+
39+
const (
40+
// Keys.
41+
connectionString = "connectionString"
42+
consumerID = "consumerID"
43+
timeoutInSec = "timeoutInSec"
44+
handlerTimeoutInSec = "handlerTimeoutInSec"
45+
lockRenewalInSec = "lockRenewalInSec"
46+
maxActiveMessages = "maxActiveMessages"
47+
maxReconnectionAttempts = "maxReconnectionAttempts"
48+
connectionRecoveryInSec = "connectionRecoveryInSec"
49+
disableEntityManagement = "disableEntityManagement"
50+
maxRetriableErrorsPerSec = "maxRetriableErrorsPerSec"
51+
maxDeliveryCount = "maxDeliveryCount"
52+
lockDurationInSec = "lockDurationInSec"
53+
defaultMessageTimeToLiveInSec = "defaultMessageTimeToLiveInSec"
54+
autoDeleteOnIdleInSec = "autoDeleteOnIdleInSec"
55+
maxConcurrentHandlers = "maxConcurrentHandlers"
56+
publishMaxRetries = "publishMaxRetries"
57+
publishInitialRetryInternalInMs = "publishInitialRetryInternalInMs"
58+
namespaceName = "namespaceName"
59+
60+
// Defaults.
61+
defaultTimeoutInSec = 60
62+
defaultHandlerTimeoutInSec = 60
63+
defaultLockRenewalInSec = 20
64+
defaultMaxRetriableErrorsPerSec = 10
65+
// ASB Messages can be up to 256Kb. 10000 messages at this size would roughly use 2.56Gb.
66+
// We should change this if performance testing suggests a more sensible default.
67+
defaultMaxActiveMessages = 10000
68+
defaultDisableEntityManagement = false
69+
defaultMaxReconnectionAttempts = 30
70+
defaultConnectionRecoveryInSec = 2
71+
defaultPublishMaxRetries = 5
72+
defaultPublishInitialRetryInternalInMs = 500
73+
)

0 commit comments

Comments
 (0)