Refactor azcore (#15349)

* Refactor azcore

See the CHANGELOG for details.

* add additional test coverage

move early header check back to Pipeline.Do

* add nil body check

* added more test coverage

removed spurious error return from NewPoller() sig.
fixed bug in FinalResponse() to perform the final GET if the URL is
specified but the terminal response contains no content.

* fix tests that don't return a final payload

* remove bad examples

* move constants to make version info easier to find

* moved Poller constructors out of root packages

for core, moved to azcore/runtime
for arm, moved to azcore/arm/runtime
also moved ARM RP registration policy to runtime

In order to do this, it was necessary to move all pipeline related
content to its own internal package azcore/internal/pipeline.  This was
to avoid problems with circular dependencies.  It also groups things
that are logically related and cleaned up the code.

sdk/azcore/CHANGELOG.md

@@ -1,13 +1,36 @@
 # Release History
 
+## v0.19.0
+
+### Breaking Changes
+* Split content out of `azcore` into various packages.  The intent is to separate content based on its usage (common, uncommon, SDK authors).
+  * `azcore` has all core functionality.
+  * `log` contains facilities for configuring in-box logging.
+  * `policy` is used for configuring pipeline options and creating custom pipeline policies.
+  * `runtime` contains various helpers used by SDK authors and generated content.
+  * `streaming` has helpers for streaming IO operations.
+* `NewTelemetryPolicy()` now requires module and version parameters and the `Value` option has been removed.
+  * As a result, the `Request.Telemetry()` method has been removed.
+* The telemetry policy now includes the SDK prefix `azsdk-go-` so callers no longer need to provide it.
+* The `*http.Request` in `runtime.Request` is no longer anonymously embedded.  Use the `Raw()` method to access it.
+* The `UserAgent` and `Version` constants have been made internal, `Module` and `Version` respectively.
+
+### Bug Fixes
+* Fixed an issue in the retry policy where the request body could be overwritten after a rewind.
+
+### Other Changes
+* Moved modules `armcore` and `to` content into `arm` and `to` packages respectively.
+  * The `Pipeline()` method on `armcore.Connection` has been replaced by `NewPipeline()` in `arm.Connection`.  It takes module and version parameters used by the telemetry policy.
+* Poller logic has been consolidated across ARM and core implementations.
+  * This required some changes to the internal interfaces for core pollers.
+* The core poller types have been improved, including more logging and test coverage.
+
 ## v0.18.1
 
 ### Features Added
 * Adds an `ETag` type for comparing etags and handling etags on requests
 * Simplifies the `requestBodyProgess` and `responseBodyProgress` into a single `progress` object
 
-### Breaking Changes
-
 ### Bugs Fixed
 * `JoinPaths` will preserve query parameters encoded in the `root` url.
 

sdk/azcore/arm/connection.go

@@ -0,0 +1,120 @@
+//go:build go1.16
+// +build go1.16
+
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package arm
+
+import (
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+	armruntime "github.com/Azure/azure-sdk-for-go/sdk/azcore/arm/runtime"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/internal/pipeline"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/internal/shared"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
+	azruntime "github.com/Azure/azure-sdk-for-go/sdk/azcore/runtime"
+)
+
+const (
+	// AzureChina is the Azure Resource Manager China cloud endpoint.
+	AzureChina = "https://management.chinacloudapi.cn/"
+	// AzureGermany is the Azure Resource Manager Germany cloud endpoint.
+	AzureGermany = "https://management.microsoftazure.de/"
+	// AzureGovernment is the Azure Resource Manager US government cloud endpoint.
+	AzureGovernment = "https://management.usgovcloudapi.net/"
+	// AzurePublicCloud is the Azure Resource Manager public cloud endpoint.
+	AzurePublicCloud = "https://management.azure.com/"
+)
+
+// ConnectionOptions contains configuration settings for the connection's pipeline.
+// All zero-value fields will be initialized with their default values.
+type ConnectionOptions struct {
+	// AuxiliaryTenants contains a list of additional tenants to be used to authenticate
+	// across multiple tenants.
+	AuxiliaryTenants []string
+
+	// HTTPClient sets the transport for making HTTP requests.
+	HTTPClient policy.Transporter
+
+	// Retry configures the built-in retry policy behavior.
+	Retry policy.RetryOptions
+
+	// Telemetry configures the built-in telemetry policy behavior.
+	Telemetry policy.TelemetryOptions
+
+	// Logging configures the built-in logging policy behavior.
+	Logging policy.LogOptions
+
+	// DisableRPRegistration disables the auto-RP registration policy.
+	// The default value is false.
+	DisableRPRegistration bool
+
+	// PerCallPolicies contains custom policies to inject into the pipeline.
+	// Each policy is executed once per request.
+	PerCallPolicies []policy.Policy
+
+	// PerRetryPolicies contains custom policies to inject into the pipeline.
+	// Each policy is executed once per request, and for each retry request.
+	PerRetryPolicies []policy.Policy
+}
+
+// Connection is a connection to an Azure Resource Manager endpoint.
+// It contains the base ARM endpoint and a pipeline for making requests.
+type Connection struct {
+	ep   string
+	cred azcore.TokenCredential
+	opt  ConnectionOptions
+}
+
+// NewDefaultConnection creates an instance of the Connection type using the AzurePublicCloud.
+// Pass nil to accept the default options; this is the same as passing a zero-value options.
+func NewDefaultConnection(cred azcore.TokenCredential, options *ConnectionOptions) *Connection {
+	return NewConnection(AzurePublicCloud, cred, options)
+}
+
+// NewConnection creates an instance of the Connection type with the specified endpoint.
+// Use this when connecting to clouds other than the Azure public cloud (stack/sovereign clouds).
+// Pass nil to accept the default options; this is the same as passing a zero-value options.
+func NewConnection(endpoint string, cred azcore.TokenCredential, options *ConnectionOptions) *Connection {
+	if options == nil {
+		options = &ConnectionOptions{}
+	}
+	return &Connection{ep: endpoint, cred: cred, opt: *options}
+}
+
+// Endpoint returns the connection's ARM endpoint.
+func (con *Connection) Endpoint() string {
+	return con.ep
+}
+
+// NewPipeline creates a pipeline from the connection's options.
+// The telemetry policy, when enabled, will use the specified module and version info.
+func (con *Connection) NewPipeline(module, version string) pipeline.Pipeline {
+	policies := []policy.Policy{}
+	if !con.opt.Telemetry.Disabled {
+		policies = append(policies, azruntime.NewTelemetryPolicy(module, version, &con.opt.Telemetry))
+	}
+	if !con.opt.DisableRPRegistration {
+		regRPOpts := armruntime.RegistrationOptions{
+			HTTPClient: con.opt.HTTPClient,
+			Logging:    con.opt.Logging,
+			Retry:      con.opt.Retry,
+			Telemetry:  con.opt.Telemetry,
+		}
+		policies = append(policies, armruntime.NewRPRegistrationPolicy(con.ep, con.cred, &regRPOpts))
+	}
+	policies = append(policies, con.opt.PerCallPolicies...)
+	policies = append(policies, azruntime.NewRetryPolicy(&con.opt.Retry))
+	policies = append(policies, con.opt.PerRetryPolicies...)
+	policies = append(policies,
+		con.cred.NewAuthenticationPolicy(
+			azruntime.AuthenticationOptions{
+				TokenRequest: policy.TokenRequestOptions{
+					Scopes: []string{shared.EndpointToScope(con.ep)},
+				},
+				AuxiliaryTenants: con.opt.AuxiliaryTenants,
+			},
+		),
+		azruntime.NewLogPolicy(&con.opt.Logging))
+	return azruntime.NewPipeline(con.opt.HTTPClient, policies...)
+}

sdk/azcore/arm/connection_test.go

@@ -0,0 +1,202 @@
+//go:build go1.16
+// +build go1.16
+
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package arm
+
+import (
+	"context"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+	armruntime "github.com/Azure/azure-sdk-for-go/sdk/azcore/arm/runtime"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/internal/pipeline"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/log"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
+	azruntime "github.com/Azure/azure-sdk-for-go/sdk/azcore/runtime"
+	"github.com/Azure/azure-sdk-for-go/sdk/internal/mock"
+)
+
+type mockTokenCred struct{}
+
+func (mockTokenCred) NewAuthenticationPolicy(azruntime.AuthenticationOptions) policy.Policy {
+	return pipeline.PolicyFunc(func(req *policy.Request) (*http.Response, error) {
+		return req.Next()
+	})
+}
+
+func (mockTokenCred) GetToken(context.Context, policy.TokenRequestOptions) (*azcore.AccessToken, error) {
+	return &azcore.AccessToken{
+		Token:     "abc123",
+		ExpiresOn: time.Now().Add(1 * time.Hour),
+	}, nil
+}
+
+const rpUnregisteredResp = `{
+	"error":{
+		"code":"MissingSubscriptionRegistration",
+		"message":"The subscription registration is in 'Unregistered' state. The subscription must be registered to use namespace 'Microsoft.Storage'. See https://aka.ms/rps-not-found for how to register subscriptions.",
+		"details":[{
+				"code":"MissingSubscriptionRegistration",
+				"target":"Microsoft.Storage",
+				"message":"The subscription registration is in 'Unregistered' state. The subscription must be registered to use namespace 'Microsoft.Storage'. See https://aka.ms/rps-not-found for how to register subscriptions."
+			}
+		]
+	}
+}`
+
+func TestNewDefaultConnection(t *testing.T) {
+	opt := ConnectionOptions{}
+	con := NewDefaultConnection(mockTokenCred{}, &opt)
+	if ep := con.Endpoint(); ep != AzurePublicCloud {
+		t.Fatalf("unexpected endpoint %s", ep)
+	}
+}
+
+func TestNewConnection(t *testing.T) {
+	const customEndpoint = "https://contoso.com/fake/endpoint"
+	con := NewConnection(customEndpoint, mockTokenCred{}, nil)
+	if ep := con.Endpoint(); ep != customEndpoint {
+		t.Fatalf("unexpected endpoint %s", ep)
+	}
+}
+
+func TestNewConnectionWithOptions(t *testing.T) {
+	srv, close := mock.NewServer()
+	defer close()
+	srv.AppendResponse()
+	opt := ConnectionOptions{}
+	opt.HTTPClient = srv
+	con := NewConnection(srv.URL(), mockTokenCred{}, &opt)
+	if ep := con.Endpoint(); ep != srv.URL() {
+		t.Fatalf("unexpected endpoint %s", ep)
+	}
+	req, err := azruntime.NewRequest(context.Background(), http.MethodGet, srv.URL())
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	resp, err := con.NewPipeline("armtest", "v1.2.3").Do(req)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("unexpected status code: %d", resp.StatusCode)
+	}
+	if ua := resp.Request.Header.Get("User-Agent"); !strings.HasPrefix(ua, "azsdk-go-armtest/v1.2.3") {
+		t.Fatalf("unexpected User-Agent %s", ua)
+	}
+}
+
+func TestNewConnectionWithCustomTelemetry(t *testing.T) {
+	const myTelemetry = "something"
+	srv, close := mock.NewServer()
+	defer close()
+	srv.AppendResponse()
+	opt := ConnectionOptions{}
+	opt.HTTPClient = srv
+	opt.Telemetry.ApplicationID = myTelemetry
+	con := NewConnection(srv.URL(), mockTokenCred{}, &opt)
+	if ep := con.Endpoint(); ep != srv.URL() {
+		t.Fatalf("unexpected endpoint %s", ep)
+	}
+	if opt.Telemetry.ApplicationID != myTelemetry {
+		t.Fatalf("telemetry was modified: %s", opt.Telemetry.ApplicationID)
+	}
+	req, err := azruntime.NewRequest(context.Background(), http.MethodGet, srv.URL())
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	resp, err := con.NewPipeline("armtest", "v1.2.3").Do(req)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("unexpected status code: %d", resp.StatusCode)
+	}
+	if ua := resp.Request.Header.Get("User-Agent"); !strings.HasPrefix(ua, myTelemetry+" "+"azsdk-go-armtest/v1.2.3") {
+		t.Fatalf("unexpected User-Agent %s", ua)
+	}
+}
+
+func TestDisableAutoRPRegistration(t *testing.T) {
+	srv, close := mock.NewServer()
+	defer close()
+	// initial response that RP is unregistered
+	srv.SetResponse(mock.WithStatusCode(http.StatusConflict), mock.WithBody([]byte(rpUnregisteredResp)))
+	con := NewConnection(srv.URL(), mockTokenCred{}, &ConnectionOptions{DisableRPRegistration: true})
+	if ep := con.Endpoint(); ep != srv.URL() {
+		t.Fatalf("unexpected endpoint %s", ep)
+	}
+	req, err := azruntime.NewRequest(context.Background(), http.MethodGet, srv.URL())
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	// log only RP registration
+	log.SetClassifications(armruntime.LogRPRegistration)
+	defer func() {
+		// reset logging
+		log.SetClassifications()
+	}()
+	logEntries := 0
+	log.SetListener(func(cls log.Classification, msg string) {
+		logEntries++
+	})
+	resp, err := con.NewPipeline("armtest", "v1.2.3").Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if resp.StatusCode != http.StatusConflict {
+		t.Fatalf("unexpected status code %d:", resp.StatusCode)
+	}
+	// shouldn't be any log entries
+	if logEntries != 0 {
+		t.Fatalf("expected 0 log entries, got %d", logEntries)
+	}
+}
+
+// policy that tracks the number of times it was invoked
+type countingPolicy struct {
+	count int
+}
+
+func (p *countingPolicy) Do(req *policy.Request) (*http.Response, error) {
+	p.count++
+	return req.Next()
+}
+
+func TestConnectionWithCustomPolicies(t *testing.T) {
+	srv, close := mock.NewServer()
+	defer close()
+	// initial response is a failure to trigger retry
+	srv.AppendResponse(mock.WithStatusCode(http.StatusInternalServerError))
+	srv.AppendResponse(mock.WithStatusCode(http.StatusOK))
+	perCallPolicy := countingPolicy{}
+	perRetryPolicy := countingPolicy{}
+	con := NewConnection(srv.URL(), mockTokenCred{}, &ConnectionOptions{
+		DisableRPRegistration: true,
+		PerCallPolicies:       []policy.Policy{&perCallPolicy},
+		PerRetryPolicies:      []policy.Policy{&perRetryPolicy},
+	})
+	req, err := azruntime.NewRequest(context.Background(), http.MethodGet, srv.URL())
+	if err != nil {
+		t.Fatal(err)
+	}
+	resp, err := con.NewPipeline("armtest", "v1.2.3").Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("unexpected status code %d", resp.StatusCode)
+	}
+	if perCallPolicy.count != 1 {
+		t.Fatalf("unexpected per call policy count %d", perCallPolicy.count)
+	}
+	if perRetryPolicy.count != 2 {
+		t.Fatalf("unexpected per retry policy count %d", perRetryPolicy.count)
+	}
+}