move more clients out of vendor

cluster-autoscaler/cloudprovider/azure/azure_client.go

@@ -29,13 +29,13 @@ import (
 	"github.com/Azure/go-autorest/autorest/adal"
 	"github.com/Azure/go-autorest/autorest/azure"
 
+	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider/azure/clients/diskclient"
+	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider/azure/clients/interfaceclient"
+	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider/azure/clients/storageaccountclient"
+	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider/azure/clients/vmclient"
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider/azure/clients/vmssclient"
+	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider/azure/clients/vmssvmclient"
 	"k8s.io/klog"
-	"k8s.io/legacy-cloud-providers/azure/clients/diskclient"
-	"k8s.io/legacy-cloud-providers/azure/clients/interfaceclient"
-	"k8s.io/legacy-cloud-providers/azure/clients/storageaccountclient"
-	"k8s.io/legacy-cloud-providers/azure/clients/vmclient"
-	"k8s.io/legacy-cloud-providers/azure/clients/vmssvmclient"
 )
 
 // DeploymentsClient defines needed functions for azure network.DeploymentsClient.

cluster-autoscaler/cloudprovider/azure/clients/diskclient/azure_diskclient.go

@@ -0,0 +1,248 @@
+// +build !providerless
+
+/*
+Copyright 2020 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package diskclient
+
+import (
+	"context"
+	"net/http"
+	"time"
+
+	"github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2019-07-01/compute"
+	"github.com/Azure/go-autorest/autorest"
+	"github.com/Azure/go-autorest/autorest/azure"
+
+	"k8s.io/client-go/util/flowcontrol"
+	"k8s.io/klog"
+	azclients "k8s.io/legacy-cloud-providers/azure/clients"
+	"k8s.io/legacy-cloud-providers/azure/clients/armclient"
+	"k8s.io/legacy-cloud-providers/azure/metrics"
+	"k8s.io/legacy-cloud-providers/azure/retry"
+)
+
+var _ Interface = &Client{}
+
+// Client implements Disk client Interface.
+type Client struct {
+	armClient      armclient.Interface
+	subscriptionID string
+
+	// Rate limiting configures.
+	rateLimiterReader flowcontrol.RateLimiter
+	rateLimiterWriter flowcontrol.RateLimiter
+
+	// ARM throttling configures.
+	RetryAfterReader time.Time
+	RetryAfterWriter time.Time
+}
+
+// New creates a new Disk client with ratelimiting.
+func New(config *azclients.ClientConfig) *Client {
+	baseURI := config.ResourceManagerEndpoint
+	authorizer := config.Authorizer
+	armClient := armclient.New(authorizer, baseURI, "", APIVersion, config.Location, config.Backoff)
+	rateLimiterReader, rateLimiterWriter := azclients.NewRateLimiter(config.RateLimitConfig)
+
+	klog.V(2).Infof("Azure DisksClient (read ops) using rate limit config: QPS=%g, bucket=%d",
+		config.RateLimitConfig.CloudProviderRateLimitQPS,
+		config.RateLimitConfig.CloudProviderRateLimitBucket)
+	klog.V(2).Infof("Azure DisksClient (write ops) using rate limit config: QPS=%g, bucket=%d",
+		config.RateLimitConfig.CloudProviderRateLimitQPSWrite,
+		config.RateLimitConfig.CloudProviderRateLimitBucketWrite)
+
+	client := &Client{
+		armClient:         armClient,
+		rateLimiterReader: rateLimiterReader,
+		rateLimiterWriter: rateLimiterWriter,
+		subscriptionID:    config.SubscriptionID,
+	}
+
+	return client
+}
+
+// Get gets a Disk.
+func (c *Client) Get(ctx context.Context, resourceGroupName string, diskName string) (compute.Disk, *retry.Error) {
+	mc := metrics.NewMetricContext("disks", "get", resourceGroupName, c.subscriptionID, "")
+
+	// Report errors if the client is rate limited.
+	if !c.rateLimiterReader.TryAccept() {
+		mc.RateLimitedCount()
+		return compute.Disk{}, retry.GetRateLimitError(false, "GetDisk")
+	}
+
+	// Report errors if the client is throttled.
+	if c.RetryAfterReader.After(time.Now()) {
+		mc.ThrottledCount()
+		rerr := retry.GetThrottlingError("GetDisk", "client throttled", c.RetryAfterReader)
+		return compute.Disk{}, rerr
+	}
+
+	result, rerr := c.getDisk(ctx, resourceGroupName, diskName)
+	mc.Observe(rerr.Error())
+	if rerr != nil {
+		if rerr.IsThrottled() {
+			// Update RetryAfterReader so that no more requests would be sent until RetryAfter expires.
+			c.RetryAfterReader = rerr.RetryAfter
+		}
+
+		return result, rerr
+	}
+
+	return result, nil
+}
+
+// getDisk gets a Disk.
+func (c *Client) getDisk(ctx context.Context, resourceGroupName string, diskName string) (compute.Disk, *retry.Error) {
+	resourceID := armclient.GetResourceID(
+		c.subscriptionID,
+		resourceGroupName,
+		"Microsoft.Compute/disks",
+		diskName,
+	)
+	result := compute.Disk{}
+
+	response, rerr := c.armClient.GetResource(ctx, resourceID, "")
+	defer c.armClient.CloseResponse(ctx, response)
+	if rerr != nil {
+		klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "disk.get.request", resourceID, rerr.Error())
+		return result, rerr
+	}
+
+	err := autorest.Respond(
+		response,
+		azure.WithErrorUnlessStatusCode(http.StatusOK),
+		autorest.ByUnmarshallingJSON(&result))
+	if err != nil {
+		klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "disk.get.respond", resourceID, err)
+		return result, retry.GetError(response, err)
+	}
+
+	result.Response = autorest.Response{Response: response}
+	return result, nil
+}
+
+// CreateOrUpdate creates or updates a Disk.
+func (c *Client) CreateOrUpdate(ctx context.Context, resourceGroupName string, diskName string, diskParameter compute.Disk) *retry.Error {
+	mc := metrics.NewMetricContext("disks", "create_or_update", resourceGroupName, c.subscriptionID, "")
+
+	// Report errors if the client is rate limited.
+	if !c.rateLimiterWriter.TryAccept() {
+		mc.RateLimitedCount()
+		return retry.GetRateLimitError(true, "DiskCreateOrUpdate")
+	}
+
+	// Report errors if the client is throttled.
+	if c.RetryAfterWriter.After(time.Now()) {
+		mc.ThrottledCount()
+		rerr := retry.GetThrottlingError("DiskCreateOrUpdate", "client throttled", c.RetryAfterWriter)
+		return rerr
+	}
+
+	rerr := c.createOrUpdateDisk(ctx, resourceGroupName, diskName, diskParameter)
+	mc.Observe(rerr.Error())
+	if rerr != nil {
+		if rerr.IsThrottled() {
+			// Update RetryAfterReader so that no more requests would be sent until RetryAfter expires.
+			c.RetryAfterWriter = rerr.RetryAfter
+		}
+
+		return rerr
+	}
+
+	return nil
+}
+
+// createOrUpdateDisk creates or updates a Disk.
+func (c *Client) createOrUpdateDisk(ctx context.Context, resourceGroupName string, diskName string, diskParameter compute.Disk) *retry.Error {
+	resourceID := armclient.GetResourceID(
+		c.subscriptionID,
+		resourceGroupName,
+		"Microsoft.Compute/disks",
+		diskName,
+	)
+
+	response, rerr := c.armClient.PutResource(ctx, resourceID, diskParameter)
+	defer c.armClient.CloseResponse(ctx, response)
+	if rerr != nil {
+		klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "disk.put.request", resourceID, rerr.Error())
+		return rerr
+	}
+
+	if response != nil && response.StatusCode != http.StatusNoContent {
+		_, rerr = c.createOrUpdateResponder(response)
+		if rerr != nil {
+			klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "disk.put.respond", resourceID, rerr.Error())
+			return rerr
+		}
+	}
+
+	return nil
+}
+
+func (c *Client) createOrUpdateResponder(resp *http.Response) (*compute.Disk, *retry.Error) {
+	result := &compute.Disk{}
+	err := autorest.Respond(
+		resp,
+		azure.WithErrorUnlessStatusCode(http.StatusOK, http.StatusCreated),
+		autorest.ByUnmarshallingJSON(&result))
+	result.Response = autorest.Response{Response: resp}
+	return result, retry.GetError(resp, err)
+}
+
+// Delete deletes a Disk by name.
+func (c *Client) Delete(ctx context.Context, resourceGroupName string, diskName string) *retry.Error {
+	mc := metrics.NewMetricContext("disks", "delete", resourceGroupName, c.subscriptionID, "")
+
+	// Report errors if the client is rate limited.
+	if !c.rateLimiterWriter.TryAccept() {
+		mc.RateLimitedCount()
+		return retry.GetRateLimitError(true, "DiskDelete")
+	}
+
+	// Report errors if the client is throttled.
+	if c.RetryAfterWriter.After(time.Now()) {
+		mc.ThrottledCount()
+		rerr := retry.GetThrottlingError("DiskDelete", "client throttled", c.RetryAfterWriter)
+		return rerr
+	}
+
+	rerr := c.deleteDisk(ctx, resourceGroupName, diskName)
+	mc.Observe(rerr.Error())
+	if rerr != nil {
+		if rerr.IsThrottled() {
+			// Update RetryAfterReader so that no more requests would be sent until RetryAfter expires.
+			c.RetryAfterWriter = rerr.RetryAfter
+		}
+
+		return rerr
+	}
+
+	return nil
+}
+
+// deleteDisk deletes a PublicIPAddress by name.
+func (c *Client) deleteDisk(ctx context.Context, resourceGroupName string, diskName string) *retry.Error {
+	resourceID := armclient.GetResourceID(
+		c.subscriptionID,
+		resourceGroupName,
+		"Microsoft.Compute/disks",
+		diskName,
+	)
+
+	return c.armClient.DeleteResource(ctx, resourceID, "")
+}

cluster-autoscaler/cloudprovider/azure/clients/diskclient/interface.go

@@ -0,0 +1,45 @@
+// +build !providerless
+
+/*
+Copyright 2020 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package diskclient
+
+import (
+	"context"
+
+	"github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2019-07-01/compute"
+	"k8s.io/legacy-cloud-providers/azure/retry"
+)
+
+const (
+	// APIVersion is the API version for compute.
+	APIVersion = "2019-07-01"
+)
+
+// Interface is the client interface for Disks.
+// Don't forget to run the following command to generate the mock client:
+// mockgen -source=$GOPATH/src/k8s.io/kubernetes/staging/src/k8s.io/legacy-cloud-providers/azure/clients/diskclient/interface.go -package=mockdiskclient Interface > $GOPATH/src/k8s.io/kubernetes/staging/src/k8s.io/legacy-cloud-providers/azure/clients/diskclient/mockdiskclient/interface.go
+type Interface interface {
+	// Get gets a Disk.
+	Get(ctx context.Context, resourceGroupName string, diskName string) (result compute.Disk, rerr *retry.Error)
+
+	// CreateOrUpdate creates or updates a Disk.
+	CreateOrUpdate(ctx context.Context, resourceGroupName string, diskName string, diskParameter compute.Disk) *retry.Error
+
+	// Delete deletes a Disk by name.
+	Delete(ctx context.Context, resourceGroupName string, diskName string) *retry.Error
+}