[azeventhubs,azservicebus] Fixing bug where connection string tokens weren't renewing (Azure#20469)

When I made the fix for the SAS connection string path I removed the expiryTime, which caused the background renewer to assume the tokens were non-renewable.

I've re-added in the old expiry-time behavior if the expiration time is not zero, and added some tests to sure it works properly.

Author: richardpark-msft

sdk/messaging/azeventhubs/internal/eh/stress/templates/deploy-job.yaml renamed to sdk/messaging/azeventhubs/internal/eh/stress/templates/stress-test-job.yaml

@@ -6,6 +6,7 @@ metadata:
     # This'll make it so the resources aren't deleted on test exit.
     # Skip.RemoveTestResources: "true"
     chaos: "{{ default false .Stress.chaos }}"
+    testInstance: "{{.Stress.Scenario}}-{{ .Release.Name }}-{{ .Release.Revision }}"
 spec:
   containers:
     - name: main
@@ -56,7 +57,8 @@ spec:
       namespaces:
         - "{{ .Release.Namespace }}"
       labelSelectors:
-        scenario: {{ .Stress.Scenario }}
+        testInstance: "{{.Stress.Scenario}}-{{ .Release.Name }}-{{ .Release.Revision }}"
+        chaos: "true"
     mode: all
     action: loss
     duration: 10s
@@ -69,7 +71,8 @@ spec:
         namespaces:
           - {{ .Release.Namespace }}
         labelSelectors:
-          scenario: {{ .Stress.Scenario }}
+          testInstance: "{{.Stress.Scenario}}-{{ .Release.Name }}-{{ .Release.Revision }}"
+          chaos: "true"
       mode: all
     externalTargets:
       - {{ .Stress.BaseName }}.servicebus.windows.net

sdk/messaging/azeventhubs/internal/sas/sas_test.go

@@ -32,7 +32,10 @@ func TestNewSigner(t *testing.T) {
 	keyName, key := "foo", "superSecret"
 	signer := NewSigner(keyName, key)
 	before := time.Now().UTC().Add(-2 * time.Second)
-	sigStr, expiry, err := signer.SignWithDuration("http://microsoft.com", 1*time.Hour)
+
+	// the URL is lowercased and escaped when used as the audience in our signature.
+	sigStr, expiry, err := signer.SignWithDuration("http://MiCrosoft.com", 1*time.Hour)
+
 	require.NoError(t, err)
 	nixExpiry, err := strconv.ParseInt(expiry, 10, 64)
 	require.NoError(t, err)

sdk/messaging/azeventhubs/internal/sbauth/token_provider.go

@@ -117,7 +117,17 @@ func (tpa *TokenProvider) getSASToken(uri string) (*auth.Token, time.Time, error
 		return nil, time.Time{}, err
 	}
 
+	// we can ignore the error here since we did the string-izing of the time
+	// in the first place.
+	var expiryTime time.Time
+
+	if authToken.Expiry != "0" {
+		// TODO: I'd like to just use the actual Expiry time we generated
+		// Filed here https://github.com/Azure/azure-sdk-for-go/issues/20468
+		expiryTime = time.Now().Add(time.Minute * 15)
+	}
+
 	return authToken,
-		time.Time{},
+		expiryTime,
 		nil
 }

sdk/messaging/azeventhubs/producer_client_test.go

@@ -69,9 +69,11 @@ func TestProducerClient_SAS(t *testing.T) {
 	require.NotEmpty(t, events)
 
 	logs := getLogsFn()
-	require.Contains(t, logs, "[azeh.Auth] Token does not have an expiration date, no background renewal needed.")
+	require.Contains(t, logs, backgroundRenewalDisabledMsg)
 }
 
+const backgroundRenewalDisabledMsg = "[azeh.Auth] Token does not have an expiration date, no background renewal needed."
+
 func TestClientsUnauthorizedCreds(t *testing.T) {
 	testParams := test.GetConnectionParamsForTest(t)
 
@@ -169,6 +171,7 @@ func TestClientsUnauthorizedCreds(t *testing.T) {
 }
 
 func TestProducerClient_GetHubAndPartitionProperties(t *testing.T) {
+	getLogsFn := test.CaptureLogsForTest()
 	testParams := test.GetConnectionParamsForTest(t)
 
 	producer, err := azeventhubs.NewProducerClientFromConnectionString(testParams.ConnectionString, testParams.EventHubName, nil)
@@ -198,6 +201,21 @@ func TestProducerClient_GetHubAndPartitionProperties(t *testing.T) {
 	}
 
 	wg.Wait()
+	logs := getLogsFn()
+	checkForTokenRefresh(t, logs, testParams.EventHubName)
+}
+
+// checkForTokenRefresh just makes sure that background token refresh has been started
+// and that we haven't somehow fallen into the trap of marking all tokens are expired.
+func checkForTokenRefresh(t *testing.T, logs []string, eventHubName string) {
+	require.NotContains(t, logs, backgroundRenewalDisabledMsg)
+
+	for _, log := range logs {
+		if strings.HasPrefix(log, fmt.Sprintf("[azeh.Auth] (%s/$management) next refresh in ", eventHubName)) {
+			return
+		}
+	}
+	require.Failf(t, "No token negotiation log lines", "logs:%s", strings.Join(logs, "\n"))
 }
 
 func TestProducerClient_GetEventHubsProperties(t *testing.T) {

sdk/messaging/azservicebus/client_test.go

@@ -172,9 +172,11 @@ func TestNewClientUsingSharedAccessSignature(t *testing.T) {
 	require.EqualValues(t, "hello world", string(messages[0].Body))
 
 	logs := getLogsFn()
-	require.Contains(t, logs, "[azsb.Auth] Token does not have an expiration date, no background renewal needed.")
+	require.Contains(t, logs, backgroundRenewalDisabledMsg)
 }
 
+const backgroundRenewalDisabledMsg = "[azsb.Auth] Token does not have an expiration date, no background renewal needed."
+
 const fastNotFoundDuration = 10 * time.Second
 
 func TestNewClientNewSenderNotFound(t *testing.T) {

sdk/messaging/azservicebus/internal/sas/sas_test.go

@@ -32,7 +32,9 @@ func TestNewSigner(t *testing.T) {
 	keyName, key := "foo", "superSecret"
 	signer := NewSigner(keyName, key)
 	before := time.Now().UTC().Add(-2 * time.Second)
-	sigStr, expiry, err := signer.SignWithDuration("http://microsoft.com", 1*time.Hour)
+
+	// the URL is lowercased and escaped when used as the audience in our signature.
+	sigStr, expiry, err := signer.SignWithDuration("http://MiCrosoft.com", 1*time.Hour)
 	require.NoError(t, err)
 	nixExpiry, err := strconv.ParseInt(expiry, 10, 64)
 	require.NoError(t, err)

sdk/messaging/azservicebus/internal/sbauth/token_provider.go

@@ -117,7 +117,17 @@ func (tpa *TokenProvider) getSASToken(uri string) (*auth.Token, time.Time, error
 		return nil, time.Time{}, err
 	}
 
+	// we can ignore the error here since we did the string-izing of the time
+	// in the first place.
+	var expiryTime time.Time
+
+	if authToken.Expiry != "0" {
+		// TODO: I'd like to just use the actual Expiry time we generated
+		// Filed here https://github.com/Azure/azure-sdk-for-go/issues/20468
+		expiryTime = time.Now().Add(time.Minute * 15)
+	}
+
 	return authToken,
-		time.Time{},
+		expiryTime,
 		nil
 }

sdk/messaging/azservicebus/internal/stress/templates/deploy-job.yaml renamed to sdk/messaging/azservicebus/internal/stress/templates/stress-test-job.yaml

@@ -3,6 +3,7 @@
 metadata:
   labels:
     chaos: "{{ default false .Stress.chaos }}"
+    testInstance: "{{.Stress.Scenario}}-{{ .Release.Name }}-{{ .Release.Revision }}"
 spec:
   # uncomment to deploy to the southeastasia region.
   # nodeSelector:
@@ -46,7 +47,8 @@ spec:
       namespaces:
         - "{{ .Release.Namespace }}"
       labelSelectors:
-        scenario: {{ .Stress.Scenario }}
+        testInstance: "{{.Stress.Scenario}}-{{ .Release.Name }}-{{ .Release.Revision }}"
+        chaos: "true"
     mode: all
     action: loss
     duration: 10s
@@ -59,7 +61,8 @@ spec:
         namespaces:
           - {{ .Release.Namespace }}
         labelSelectors:
-          scenario: {{ .Stress.Scenario }}
+          testInstance: "{{.Stress.Scenario}}-{{ .Release.Name }}-{{ .Release.Revision }}"
+          chaos: "true"
       mode: all
     externalTargets:
       - {{ .Stress.BaseName }}.servicebus.windows.net

sdk/messaging/azservicebus/receiver_test.go

@@ -37,6 +37,8 @@ func TestReceiverCancel(t *testing.T) {
 }
 
 func TestReceiverSendFiveReceiveFive(t *testing.T) {
+	getLogsFn := test.CaptureLogsForTest()
+
 	serviceBusClient, cleanup, queueName := setupLiveTest(t, nil)
 	defer cleanup()
 
@@ -63,6 +65,21 @@ func TestReceiverSendFiveReceiveFive(t *testing.T) {
 
 		require.NoError(t, receiver.CompleteMessage(context.Background(), messages[i], nil))
 	}
+
+	logs := getLogsFn()
+	checkForTokenRefresh(t, logs, queueName)
+}
+
+// checkForTokenRefresh just makes sure that background token refresh has been started
+// and that we haven't somehow fallen into the trap of marking all tokens are expired.
+func checkForTokenRefresh(t *testing.T, logs []string, queueName string) {
+	require.NotContains(t, logs, backgroundRenewalDisabledMsg)
+	for _, log := range logs {
+		if strings.HasPrefix(log, fmt.Sprintf("[azsb.Auth] (%s) next refresh in ", queueName)) {
+			return
+		}
+	}
+	require.Fail(t, "No token negotiation log lines")
 }
 
 func TestReceiverSendFiveReceiveFive_Subscription(t *testing.T) {