Skip to content

[azservicebus] Fixing several reliability/recovery bugs #16831

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
richardpark-msft merged 58 commits into from
Jan 25, 2022
Merged

[azservicebus] Fixing several reliability/recovery bugs #16831

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
58 commits
Select commit Hold shift + click to select a range
a9a8404
Remove unneeded error types and tests
Jan 13, 2022
85018de
Creating a Retry() implmentation for SB. Based on what the azcore imp…
Jan 13, 2022
b605c1d
ATOM
Jan 13, 2022
7aced8a
Adding in a simple test helper that we can use in internal to create …
Jan 13, 2022
a79e5b5
NIT: fixing an extra blank line
Jan 13, 2022
86ce75a
Changing everything over to using the ServiceBusError info (soon to b…
Jan 13, 2022
9b15f0f
Renaming ServiceBusError to SBErrInfo to make it more clear that it's…
Jan 13, 2022
32c5724
Remove probably invalid isPermanentNetError classifier.
Jan 13, 2022
1df31b6
Trimming more unneeded stuff from errors
Jan 13, 2022
fb81dc3
The processor type isn't going to ship immediately and it's starting …
Jan 13, 2022
ca9cf84
Add a safety net where test queues are just auto-deleted after being …
Jan 13, 2022
749e3a7
Adding in retries to the session receiver's operations.
Jan 13, 2022
dcb802b
Adding in some more tests for retries.
Jan 14, 2022
bf0e335
Handling different mgmt link errors
Jan 14, 2022
b77360e
Removed the management client - it was just yet another lock around a…
Jan 14, 2022
fb90bea
A 1/2 measure, but making it so the admin_client at least returns the…
Jan 14, 2022
5932ea9
We no longer use the older retry, preferring our newer one that looks…
Jan 14, 2022
d1405bc
Bigger changes here for amqpLinks:
Jan 14, 2022
ea9f48f
Lots of fixes for namespace:
Jan 14, 2022
779d2b0
Converted the one tab.Log into an azure logger and we can stop passin…
Jan 14, 2022
43845c1
Expose the retry options publicly, allowing customers to configure it…
Jan 14, 2022
84b8afb
Going all in on the Azure logger
Jan 14, 2022
c01be62
- Reducing our dependency on the `messaging/internal` package by movi…
Jan 14, 2022
a92e656
Removed dependency on 3rd party backoff package
Jan 14, 2022
5957c99
Change over to the new amqpLinks.Retry() function, which properly han…
Jan 14, 2022
b323281
Switch over to using amqpLinks.Retry().
Jan 14, 2022
bf07c54
Use amqpLinks.Retry(), use azure logging.
Jan 14, 2022
32348d8
Merged isMgmtError into the sberrorinfo code.
Jan 14, 2022
eccf04f
update mocks, remove dead code.
Jan 14, 2022
7701bdd
Updating stress tests:
Jan 14, 2022
12d2751
Fixing copyright header in finite_peeks.go
Jan 14, 2022
19db935
Remove unused switch
Jan 14, 2022
c9af647
Updating, error code doesn't look the same between separate 404 style…
Jan 14, 2022
ff74b68
Fixing broken retry tests.
Jan 15, 2022
305cf26
glitched, missed a file
Jan 15, 2022
3429053
Don't parse the body, it's not actually conforming to anything that I…
Jan 15, 2022
e6e2ba5
Fixiung some more of the HTTP based tests
Jan 15, 2022
825dbb6
session link recovery means my bogus way of testing this doesn't work…
Jan 15, 2022
858ea89
There was overlap between the two interfaces and it turns out one of …
Jan 15, 2022
9f2f472
lint
Jan 15, 2022
d4620b0
ErrorCode isn't going to get filled out. We'll rely on accurate http …
Jan 15, 2022
888df18
Merge remote-tracking branch 'upstream/main' into export-retry-policy…
Jan 18, 2022
5472cd4
- Make sure we're 64-bit aligned for our atomic counter values.
Jan 18, 2022
640ea5f
go mod tidy
Jan 19, 2022
e6fc70b
Doc comment for retry options in admin_client.go
Jan 19, 2022
d9a7e91
Comments and rename from 'rev' to ID.
Jan 20, 2022
283e5e9
Checking in the working stress tests that are testing this changeset,…
Jan 20, 2022
b99f513
Test list.
Jan 20, 2022
fd251c1
chart.yaml update to a shorter namespace name, for benbp
Jan 20, 2022
98d7a1d
Merge remote-tracking branch 'upstream/main' into export-retry-policy…
Jan 20, 2022
4216dd5
revert change to the deploy script
Jan 20, 2022
1a459bf
Adding a little bit of documentation for which deps are for the produ…
Jan 20, 2022
1c9594d
Unneeded break
Jan 22, 2022
0a49382
Mean tot be a type alias.
Jan 22, 2022
05e8ba3
Remove unneeded retryOptions (it's actually just in args)
Jan 22, 2022
2d0efa6
retryOptions wasn't supposed to be in args since it's optional.
Jan 22, 2022
5f8db60
Fallthrough instead of else.
Jan 22, 2022
4fc0933
Don't need to zero initialize a struct.
Jan 25, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 14 additions & 2 deletions sdk/messaging/azservicebus/admin/admin_client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ import (
"github.com/Azure/azure-sdk-for-go/sdk/azcore"
"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal"
"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/atom"
"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/utils"
)

// Client allows you to administer resources in a Service Bus Namespace.
Expand All @@ -21,8 +22,13 @@ type Client struct {
em atom.EntityManager
}

// RetryOptions represent the options for retries.
type RetryOptions = utils.RetryOptions
Comment thread
richardpark-msft marked this conversation as resolved.

// ClientOptions allows you to set optional configuration for `Client`.
type ClientOptions struct {
// for future expansion
// RetryOptions controls how often operations are retried from this client.
RetryOptions *RetryOptions
}

// NewClientFromConnectionString creates a Client authenticating using a connection string.
Expand All @@ -38,7 +44,13 @@ func NewClientFromConnectionString(connectionString string, options *ClientOptio

// NewClient creates a Client authenticating using a TokenCredential.
func NewClient(fullyQualifiedNamespace string, tokenCredential azcore.TokenCredential, options *ClientOptions) (*Client, error) {
em, err := atom.NewEntityManager(fullyQualifiedNamespace, tokenCredential, internal.Version)
var retryOptions utils.RetryOptions

if options != nil && options.RetryOptions != nil {
retryOptions = *options.RetryOptions
}

em, err := atom.NewEntityManager(fullyQualifiedNamespace, tokenCredential, internal.Version, retryOptions)
Comment thread
jhendrixMSFT marked this conversation as resolved.

if err != nil {
return nil, err
Expand Down
8 changes: 4 additions & 4 deletions sdk/messaging/azservicebus/admin/admin_client_queue.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -196,7 +196,7 @@ func (ac *Client) GetQueue(ctx context.Context, queueName string, options *GetQu
props, err := newQueueProperties(&atomResp.Content.QueueDescription)

if err != nil {
return nil, atom.NewResponseError(err, resp)
return nil, err
}

return &GetQueueResponse{
Expand Down Expand Up @@ -234,7 +234,7 @@ func (ac *Client) GetQueueRuntimeProperties(ctx context.Context, queueName strin
props, err := newQueueRuntimeProperties(&atomResp.Content.QueueDescription)

if err != nil {
return nil, atom.NewResponseError(err, resp)
return nil, err
}

return &GetQueueRuntimePropertiesResponse{
Expand Down Expand Up @@ -349,7 +349,7 @@ func (ac *Client) createOrUpdateQueueImpl(ctx context.Context, queueName string,
newProps, err := newQueueProperties(&atomResp.Content.QueueDescription)

if err != nil {
return nil, nil, atom.NewResponseError(err, resp)
return nil, nil, err
}

return newProps, resp, nil
Expand Down Expand Up @@ -395,7 +395,7 @@ func (p *QueuePager) getNextPage(ctx context.Context) (*ListQueuesResponse, erro
props, err := newQueueProperties(&env.Content.QueueDescription)

if err != nil {
return nil, atom.NewResponseError(err, resp)
return nil, err
}

all = append(all, &QueueItem{
Expand Down
10 changes: 5 additions & 5 deletions sdk/messaging/azservicebus/admin/admin_client_subscription.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,7 @@ func (ac *Client) GetSubscription(ctx context.Context, topicName string, subscri
props, err := newSubscriptionProperties(&atomResp.Content.SubscriptionDescription)

if err != nil {
return nil, atom.NewResponseError(err, resp)
return nil, err
}

return &GetSubscriptionResponse{
Expand Down Expand Up @@ -186,7 +186,7 @@ func (ac *Client) GetSubscriptionRuntimeProperties(ctx context.Context, topicNam
props, err := newSubscriptionRuntimeProperties(&atomResp.Content.SubscriptionDescription)

if err != nil {
return nil, atom.NewResponseError(err, resp)
return nil, err
}

return &GetSubscriptionRuntimePropertiesResponse{
Expand Down Expand Up @@ -257,7 +257,7 @@ func (p *SubscriptionPager) getNext(ctx context.Context) (*ListSubscriptionsResp
props, err := newSubscriptionProperties(&env.Content.SubscriptionDescription)

if err != nil {
return nil, atom.NewResponseError(err, resp)
return nil, err
}

all = append(all, &SubscriptionPropertiesItem{
Expand Down Expand Up @@ -346,7 +346,7 @@ func (p *SubscriptionRuntimePropertiesPager) getNextPage(ctx context.Context) (*
props, err := newSubscriptionRuntimeProperties(&entry.Content.SubscriptionDescription)

if err != nil {
return nil, atom.NewResponseError(err, resp)
return nil, err
}

all = append(all, &SubscriptionRuntimePropertiesItem{
Expand Down Expand Up @@ -452,7 +452,7 @@ func (ac *Client) createOrUpdateSubscriptionImpl(ctx context.Context, topicName
newProps, err := newSubscriptionProperties(&atomResp.Content.SubscriptionDescription)

if err != nil {
return nil, nil, atom.NewResponseError(err, resp)
return nil, nil, err
}

return newProps, resp, nil
Expand Down
67 changes: 53 additions & 14 deletions sdk/messaging/azservicebus/admin/admin_client_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import (
"testing"
"time"

"github.com/Azure/azure-sdk-for-go/sdk/azcore"
"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/atom"
Expand Down Expand Up @@ -204,7 +205,12 @@ func TestAdminClient_UpdateQueue(t *testing.T) {

updatedProps, err = adminClient.UpdateQueue(context.Background(), "non-existent-queue", createdProps.QueueProperties, nil)
// a little awkward, we'll make these programatically inspectable as we add in better error handling.
require.Contains(t, err.Error(), "error code: 404")
require.Contains(t, err.Error(), "404 Not Found")

var asResponseErr *azcore.ResponseError
require.ErrorAs(t, err, &asResponseErr)
require.EqualValues(t, 404, asResponseErr.StatusCode)

require.Nil(t, updatedProps)
}

Expand Down Expand Up @@ -475,7 +481,12 @@ func TestAdminClient_UpdateTopic(t *testing.T) {

updateResp, err = adminClient.UpdateTopic(context.Background(), "non-existent-topic", addResp.TopicProperties, nil)
// a little awkward, we'll make these programatically inspectable as we add in better error handling.
require.Contains(t, err.Error(), "error code: 404")
require.Contains(t, err.Error(), "404 Not Found")

var asResponseErr *azcore.ResponseError
require.ErrorAs(t, err, &asResponseErr)
require.EqualValues(t, 404, asResponseErr.StatusCode)

require.Nil(t, updateResp)
}

Expand Down Expand Up @@ -738,8 +749,12 @@ func TestAdminClient_UpdateSubscription(t *testing.T) {
require.Nil(t, updateResp)

updateResp, err = adminClient.UpdateSubscription(context.Background(), topicName, "non-existent-subscription", addResp.CreateSubscriptionResult.SubscriptionProperties, nil)
// a little awkward, we'll make these programatically inspectable as we add in better error handling.
require.Contains(t, err.Error(), "error code: 404")
require.Contains(t, err.Error(), "404 Not Found")

var asResponseErr *azcore.ResponseError
require.ErrorAs(t, err, &asResponseErr)
require.EqualValues(t, 404, asResponseErr.StatusCode)

require.Nil(t, updateResp)
}

Expand All @@ -754,21 +769,33 @@ func TestAdminClient_LackPermissions_Queue(t *testing.T) {
require.True(t, notFound)
require.NotNil(t, resp)

var re *azcore.ResponseError

_, err = testData.Client.GetQueue(ctx, testData.QueueName, nil)
require.Contains(t, err.Error(), "error code: 401, Details: Manage,EntityRead claims")
require.Contains(t, err.Error(), "Manage,EntityRead claims required for this operation")
require.ErrorAs(t, err, &re)
require.EqualValues(t, 401, re.StatusCode)

pager := testData.Client.ListQueues(nil)
require.False(t, pager.NextPage(context.Background()))
require.Contains(t, pager.Err().Error(), "error code: 401, Details: Manage,EntityRead claims required for this operation")
require.Contains(t, pager.Err().Error(), "Manage,EntityRead claims required for this operation")
require.ErrorAs(t, err, &re)
require.EqualValues(t, 401, re.StatusCode)

_, err = testData.Client.CreateQueue(ctx, "canneverbecreated", nil, nil)
require.Contains(t, err.Error(), "error code: 401, Details: Authorization failed for specified action: Manage,EntityWrite")
require.Contains(t, err.Error(), "Authorization failed for specified action: Manage,EntityWrite")
require.ErrorAs(t, err, &re)
require.EqualValues(t, 401, re.StatusCode)

_, err = testData.Client.UpdateQueue(ctx, "canneverbecreated", QueueProperties{}, nil)
require.Contains(t, err.Error(), "error code: 401, Details: Authorization failed for specified action: Manage,EntityWrite")
require.Contains(t, err.Error(), "Authorization failed for specified action: Manage,EntityWrite")
require.ErrorAs(t, err, &re)
require.EqualValues(t, 401, re.StatusCode)

_, err = testData.Client.DeleteQueue(ctx, testData.QueueName, nil)
require.Contains(t, err.Error(), "error code: 401, Details: Authorization failed for specified action: Manage,EntityDelete.")
require.Contains(t, err.Error(), "Authorization failed for specified action: Manage,EntityDelete.")
require.ErrorAs(t, err, &re)
require.EqualValues(t, 401, re.StatusCode)
}

func TestAdminClient_LackPermissions_Topic(t *testing.T) {
Expand All @@ -782,21 +809,33 @@ func TestAdminClient_LackPermissions_Topic(t *testing.T) {
require.True(t, notFound)
require.NotNil(t, resp)

var asResponseErr *azcore.ResponseError

_, err = testData.Client.GetTopic(ctx, testData.TopicName, nil)
require.Contains(t, err.Error(), "error code: 401, Details: Manage,EntityRead claims")
require.Contains(t, err.Error(), ">Manage,EntityRead claims required for this operation")
require.ErrorAs(t, err, &asResponseErr)
require.EqualValues(t, 401, asResponseErr.StatusCode)

pager := testData.Client.ListTopics(nil)
require.False(t, pager.NextPage(context.Background()))
require.Contains(t, pager.Err().Error(), "error code: 401, Details: Manage,EntityRead claims required for this operation")
require.Contains(t, pager.Err().Error(), ">Manage,EntityRead claims required for this operation")
require.ErrorAs(t, err, &asResponseErr)
require.EqualValues(t, 401, asResponseErr.StatusCode)

_, err = testData.Client.CreateTopic(ctx, "canneverbecreated", nil, nil)
require.Contains(t, err.Error(), "error code: 401, Details: Authorization failed for specified action")
require.Contains(t, err.Error(), "Authorization failed for specified action")
require.ErrorAs(t, err, &asResponseErr)
require.EqualValues(t, 401, asResponseErr.StatusCode)

_, err = testData.Client.UpdateTopic(ctx, "canneverbecreated", TopicProperties{}, nil)
require.Contains(t, err.Error(), "error code: 401, Details: Authorization failed for specified action")
require.Contains(t, err.Error(), "Authorization failed for specified action")
require.ErrorAs(t, err, &asResponseErr)
require.EqualValues(t, 401, asResponseErr.StatusCode)

_, err = testData.Client.DeleteTopic(ctx, testData.TopicName, nil)
require.Contains(t, err.Error(), "error code: 401, Details: Authorization failed for specified action: Manage,EntityDelete.")
require.Contains(t, err.Error(), "Authorization failed for specified action: Manage,EntityDelete.")
require.ErrorAs(t, err, &asResponseErr)
require.EqualValues(t, 401, asResponseErr.StatusCode)
}

func TestAdminClient_LackPermissions_Subscription(t *testing.T) {
Expand Down
8 changes: 4 additions & 4 deletions sdk/messaging/azservicebus/admin/admin_client_topic.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,7 @@ func (ac *Client) GetTopicRuntimeProperties(ctx context.Context, topicName strin
props, err := newTopicRuntimeProperties(&atomResp.Content.TopicDescription)

if err != nil {
return nil, atom.NewResponseError(err, resp)
return nil, err
}

return &GetTopicRuntimePropertiesResponse{
Expand Down Expand Up @@ -239,7 +239,7 @@ func (p *TopicsPager) getNextPage(ctx context.Context) (*ListTopicsResponse, err
props, err := newTopicProperties(&env.Content.TopicDescription)

if err != nil {
return nil, atom.NewResponseError(err, resp)
return nil, err
}

all = append(all, &TopicItem{
Expand Down Expand Up @@ -326,7 +326,7 @@ func (p *TopicRuntimePropertiesPager) getNextPage(ctx context.Context) (*ListTop
props, err := newTopicRuntimeProperties(&entry.Content.TopicDescription)

if err != nil {
return nil, atom.NewResponseError(err, resp)
return nil, err
}

all = append(all, &TopicRuntimePropertiesItem{
Expand Down Expand Up @@ -436,7 +436,7 @@ func (ac *Client) createOrUpdateTopicImpl(ctx context.Context, topicName string,
topicProps, err := newTopicProperties(&atomResp.Content.TopicDescription)

if err != nil {
return nil, nil, atom.NewResponseError(err, resp)
return nil, nil, err
}

return topicProps, resp, nil
Expand Down
Loading