diff --git a/cluster-autoscaler/cloudprovider/azure/azure_client.go b/cluster-autoscaler/cloudprovider/azure/azure_client.go index 4845ae72188c..3192db392e10 100644 --- a/cluster-autoscaler/cloudprovider/azure/azure_client.go +++ b/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. diff --git a/cluster-autoscaler/cloudprovider/azure/clients/diskclient/azure_diskclient.go b/cluster-autoscaler/cloudprovider/azure/clients/diskclient/azure_diskclient.go new file mode 100644 index 000000000000..d61e91abc17f --- /dev/null +++ b/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, "") +} diff --git a/cluster-autoscaler/cloudprovider/azure/clients/diskclient/interface.go b/cluster-autoscaler/cloudprovider/azure/clients/diskclient/interface.go new file mode 100644 index 000000000000..004fe1e09df0 --- /dev/null +++ b/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 +} diff --git a/cluster-autoscaler/cloudprovider/azure/clients/interfaceclient/azure_interfaceclient.go b/cluster-autoscaler/cloudprovider/azure/clients/interfaceclient/azure_interfaceclient.go new file mode 100644 index 000000000000..69e68b3c961f --- /dev/null +++ b/cluster-autoscaler/cloudprovider/azure/clients/interfaceclient/azure_interfaceclient.go @@ -0,0 +1,319 @@ +// +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 interfaceclient + +import ( + "context" + "fmt" + "net/http" + "time" + + "github.com/Azure/azure-sdk-for-go/services/network/mgmt/2019-06-01/network" + "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 network interface client. +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 network interface 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 InterfacesClient (read ops) using rate limit config: QPS=%g, bucket=%d", + config.RateLimitConfig.CloudProviderRateLimitQPS, + config.RateLimitConfig.CloudProviderRateLimitBucket) + klog.V(2).Infof("Azure InterfacesClient (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 network.Interface. +func (c *Client) Get(ctx context.Context, resourceGroupName string, networkInterfaceName string, expand string) (network.Interface, *retry.Error) { + mc := metrics.NewMetricContext("interfaces", "get", resourceGroupName, c.subscriptionID, "") + + // Report errors if the client is rate limited. + if !c.rateLimiterReader.TryAccept() { + mc.RateLimitedCount() + return network.Interface{}, retry.GetRateLimitError(false, "NicGet") + } + + // Report errors if the client is throttled. + if c.RetryAfterReader.After(time.Now()) { + mc.ThrottledCount() + rerr := retry.GetThrottlingError("NicGet", "client throttled", c.RetryAfterReader) + return network.Interface{}, rerr + } + + result, rerr := c.getNetworkInterface(ctx, resourceGroupName, networkInterfaceName, expand) + 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 +} + +// getNetworkInterface gets a network.Interface. +func (c *Client) getNetworkInterface(ctx context.Context, resourceGroupName string, networkInterfaceName string, expand string) (network.Interface, *retry.Error) { + resourceID := armclient.GetResourceID( + c.subscriptionID, + resourceGroupName, + "Microsoft.Network/networkInterfaces", + networkInterfaceName, + ) + result := network.Interface{} + + 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", "nic.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", "nic.get.respond", resourceID, err) + return result, retry.GetError(response, err) + } + + result.Response = autorest.Response{Response: response} + return result, nil +} + +// GetVirtualMachineScaleSetNetworkInterface gets a network.Interface of VMSS VM. +func (c *Client) GetVirtualMachineScaleSetNetworkInterface(ctx context.Context, resourceGroupName string, virtualMachineScaleSetName string, virtualmachineIndex string, networkInterfaceName string, expand string) (network.Interface, *retry.Error) { + mc := metrics.NewMetricContext("interfaces", "get_vmss_nic", resourceGroupName, c.subscriptionID, "") + + // Report errors if the client is rate limited. + if !c.rateLimiterReader.TryAccept() { + mc.RateLimitedCount() + return network.Interface{}, retry.GetRateLimitError(false, "NicGetVirtualMachineScaleSetNetworkInterface") + } + + // Report errors if the client is throttled. + if c.RetryAfterReader.After(time.Now()) { + mc.ThrottledCount() + rerr := retry.GetThrottlingError("NicGetVirtualMachineScaleSetNetworkInterface", "client throttled", c.RetryAfterReader) + return network.Interface{}, rerr + } + + result, rerr := c.getVMSSNetworkInterface(ctx, resourceGroupName, virtualMachineScaleSetName, virtualmachineIndex, networkInterfaceName, expand) + 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 +} + +// getVMSSNetworkInterface gets a network.Interface of VMSS VM. +func (c *Client) getVMSSNetworkInterface(ctx context.Context, resourceGroupName string, virtualMachineScaleSetName string, virtualmachineIndex string, networkInterfaceName string, expand string) (network.Interface, *retry.Error) { + resourceID := fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Compute/virtualMachineScaleSets/%s/virtualMachines/%s/networkInterfaces/%s", + autorest.Encode("path", c.subscriptionID), + autorest.Encode("path", resourceGroupName), + autorest.Encode("path", virtualMachineScaleSetName), + autorest.Encode("path", virtualmachineIndex), + autorest.Encode("path", networkInterfaceName), + ) + + result := network.Interface{} + queryParameters := map[string]interface{}{ + "api-version": ComputeAPIVersion, + } + if len(expand) > 0 { + queryParameters["$expand"] = autorest.Encode("query", expand) + } + decorators := []autorest.PrepareDecorator{ + autorest.WithQueryParameters(queryParameters), + } + response, rerr := c.armClient.GetResourceWithDecorators(ctx, resourceID, decorators) + defer c.armClient.CloseResponse(ctx, response) + if rerr != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vmssnic.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", "vmssnic.get.respond", resourceID, err) + return result, retry.GetError(response, err) + } + + result.Response = autorest.Response{Response: response} + return result, nil +} + +// CreateOrUpdate creates or updates a network.Interface. +func (c *Client) CreateOrUpdate(ctx context.Context, resourceGroupName string, networkInterfaceName string, parameters network.Interface) *retry.Error { + mc := metrics.NewMetricContext("interfaces", "create_or_update", resourceGroupName, c.subscriptionID, "") + + // Report errors if the client is rate limited. + if !c.rateLimiterWriter.TryAccept() { + mc.RateLimitedCount() + return retry.GetRateLimitError(true, "NicCreateOrUpdate") + } + + // Report errors if the client is throttled. + if c.RetryAfterWriter.After(time.Now()) { + mc.ThrottledCount() + rerr := retry.GetThrottlingError("NicCreateOrUpdate", "client throttled", c.RetryAfterWriter) + return rerr + } + + rerr := c.createOrUpdateInterface(ctx, resourceGroupName, networkInterfaceName, parameters) + 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 +} + +// createOrUpdateInterface creates or updates a network.Interface. +func (c *Client) createOrUpdateInterface(ctx context.Context, resourceGroupName string, networkInterfaceName string, parameters network.Interface) *retry.Error { + resourceID := armclient.GetResourceID( + c.subscriptionID, + resourceGroupName, + "Microsoft.Network/networkInterfaces", + networkInterfaceName, + ) + response, rerr := c.armClient.PutResource(ctx, resourceID, parameters) + defer c.armClient.CloseResponse(ctx, response) + if rerr != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "nic.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", "nic.put.respond", resourceID, rerr.Error()) + return rerr + } + } + + return nil +} + +func (c *Client) createOrUpdateResponder(resp *http.Response) (*network.Interface, *retry.Error) { + result := &network.Interface{} + 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 network interface by name. +func (c *Client) Delete(ctx context.Context, resourceGroupName string, networkInterfaceName string) *retry.Error { + mc := metrics.NewMetricContext("interfaces", "delete", resourceGroupName, c.subscriptionID, "") + + // Report errors if the client is rate limited. + if !c.rateLimiterWriter.TryAccept() { + mc.RateLimitedCount() + return retry.GetRateLimitError(true, "NicDelete") + } + + // Report errors if the client is throttled. + if c.RetryAfterWriter.After(time.Now()) { + mc.ThrottledCount() + rerr := retry.GetThrottlingError("NicDelete", "client throttled", c.RetryAfterWriter) + return rerr + } + + rerr := c.deleteInterface(ctx, resourceGroupName, networkInterfaceName) + 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 +} + +// deleteInterface deletes a network interface by name. +func (c *Client) deleteInterface(ctx context.Context, resourceGroupName string, networkInterfaceName string) *retry.Error { + resourceID := armclient.GetResourceID( + c.subscriptionID, + resourceGroupName, + "Microsoft.Network/networkInterfaces", + networkInterfaceName, + ) + + return c.armClient.DeleteResource(ctx, resourceID, "") +} diff --git a/cluster-autoscaler/cloudprovider/azure/clients/interfaceclient/interface.go b/cluster-autoscaler/cloudprovider/azure/clients/interfaceclient/interface.go new file mode 100644 index 000000000000..0ea631481bfc --- /dev/null +++ b/cluster-autoscaler/cloudprovider/azure/clients/interfaceclient/interface.go @@ -0,0 +1,51 @@ +// +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 interfaceclient + +import ( + "context" + + "github.com/Azure/azure-sdk-for-go/services/network/mgmt/2019-06-01/network" + "k8s.io/legacy-cloud-providers/azure/retry" +) + +const ( + // APIVersion is the API version for network. + APIVersion = "2019-06-01" + + // ComputeAPIVersion is the API version for compute. It is required to get VMSS network interface. + ComputeAPIVersion = "2017-03-30" +) + +// Interface is the client interface for NetworkInterface. +// 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/interfaceclient/interface.go -package=mockinterfaceclient Interface > $GOPATH/src/k8s.io/kubernetes/staging/src/k8s.io/legacy-cloud-providers/azure/clients/interfaceclient/mockinterfaceclient/interface.go +type Interface interface { + // Get gets a network.Interface. + Get(ctx context.Context, resourceGroupName string, networkInterfaceName string, expand string) (result network.Interface, rerr *retry.Error) + + // GetVirtualMachineScaleSetNetworkInterface gets a network.Interface of VMSS VM. + GetVirtualMachineScaleSetNetworkInterface(ctx context.Context, resourceGroupName string, virtualMachineScaleSetName string, virtualmachineIndex string, networkInterfaceName string, expand string) (result network.Interface, rerr *retry.Error) + + // CreateOrUpdate creates or updates a network.Interface. + CreateOrUpdate(ctx context.Context, resourceGroupName string, networkInterfaceName string, parameters network.Interface) *retry.Error + + // Delete deletes a network interface by name. + Delete(ctx context.Context, resourceGroupName string, networkInterfaceName string) *retry.Error +} diff --git a/cluster-autoscaler/cloudprovider/azure/clients/storageaccountclient/azure_storageaccountclient.go b/cluster-autoscaler/cloudprovider/azure/clients/storageaccountclient/azure_storageaccountclient.go new file mode 100644 index 000000000000..ee7820afd4a0 --- /dev/null +++ b/cluster-autoscaler/cloudprovider/azure/clients/storageaccountclient/azure_storageaccountclient.go @@ -0,0 +1,466 @@ +// +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 storageaccountclient + +import ( + "context" + "fmt" + "net/http" + "time" + + "github.com/Azure/azure-sdk-for-go/services/storage/mgmt/2019-06-01/storage" + "github.com/Azure/go-autorest/autorest" + "github.com/Azure/go-autorest/autorest/azure" + "github.com/Azure/go-autorest/autorest/to" + + "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 StorageAccount 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 StorageAccount 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 StorageAccountClient (read ops) using rate limit config: QPS=%g, bucket=%d", + config.RateLimitConfig.CloudProviderRateLimitQPS, + config.RateLimitConfig.CloudProviderRateLimitBucket) + klog.V(2).Infof("Azure StorageAccountClient (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 +} + +// GetProperties gets properties of the StorageAccount. +func (c *Client) GetProperties(ctx context.Context, resourceGroupName string, accountName string) (storage.Account, *retry.Error) { + mc := metrics.NewMetricContext("storage_account", "get", resourceGroupName, c.subscriptionID, "") + + // Report errors if the client is rate limited. + if !c.rateLimiterReader.TryAccept() { + mc.RateLimitedCount() + return storage.Account{}, retry.GetRateLimitError(false, "StorageAccountGet") + } + + // Report errors if the client is throttled. + if c.RetryAfterReader.After(time.Now()) { + mc.ThrottledCount() + rerr := retry.GetThrottlingError("StorageAccountGet", "client throttled", c.RetryAfterReader) + return storage.Account{}, rerr + } + + result, rerr := c.getStorageAccount(ctx, resourceGroupName, accountName) + 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 +} + +// getStorageAccount gets properties of the StorageAccount. +func (c *Client) getStorageAccount(ctx context.Context, resourceGroupName string, accountName string) (storage.Account, *retry.Error) { + resourceID := armclient.GetResourceID( + c.subscriptionID, + resourceGroupName, + "Microsoft.Storage/storageAccounts", + accountName, + ) + result := storage.Account{} + + 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", "storageaccount.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", "storageaccount.get.respond", resourceID, err) + return result, retry.GetError(response, err) + } + + result.Response = autorest.Response{Response: response} + return result, nil +} + +// ListKeys get a list of storage account keys. +func (c *Client) ListKeys(ctx context.Context, resourceGroupName string, accountName string) (storage.AccountListKeysResult, *retry.Error) { + mc := metrics.NewMetricContext("storage_account", "list_keys", resourceGroupName, c.subscriptionID, "") + + // Report errors if the client is rate limited. + if !c.rateLimiterReader.TryAccept() { + mc.RateLimitedCount() + return storage.AccountListKeysResult{}, retry.GetRateLimitError(false, "StorageAccountListKeys") + } + + // Report errors if the client is throttled. + if c.RetryAfterReader.After(time.Now()) { + mc.ThrottledCount() + rerr := retry.GetThrottlingError("StorageAccountListKeys", "client throttled", c.RetryAfterReader) + return storage.AccountListKeysResult{}, rerr + } + + result, rerr := c.listStorageAccountKeys(ctx, resourceGroupName, accountName) + 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 +} + +// listStorageAccountKeys get a list of storage account keys. +func (c *Client) listStorageAccountKeys(ctx context.Context, resourceGroupName string, accountName string) (storage.AccountListKeysResult, *retry.Error) { + resourceID := armclient.GetResourceID( + c.subscriptionID, + resourceGroupName, + "Microsoft.Storage/storageAccounts", + accountName, + ) + + result := storage.AccountListKeysResult{} + response, rerr := c.armClient.PostResource(ctx, resourceID, "listKeys", struct{}{}) + defer c.armClient.CloseResponse(ctx, response) + if rerr != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "storageaccount.listkeys.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", "storageaccount.listkeys.respond", resourceID, err) + return result, retry.GetError(response, err) + } + + result.Response = autorest.Response{Response: response} + return result, nil +} + +// Create creates a StorageAccount. +func (c *Client) Create(ctx context.Context, resourceGroupName string, accountName string, parameters storage.AccountCreateParameters) *retry.Error { + mc := metrics.NewMetricContext("storage_account", "create", resourceGroupName, c.subscriptionID, "") + + // Report errors if the client is rate limited. + if !c.rateLimiterWriter.TryAccept() { + mc.RateLimitedCount() + return retry.GetRateLimitError(true, "StorageAccountCreate") + } + + // Report errors if the client is throttled. + if c.RetryAfterWriter.After(time.Now()) { + mc.ThrottledCount() + rerr := retry.GetThrottlingError("StorageAccountCreate", "client throttled", c.RetryAfterWriter) + return rerr + } + + rerr := c.createStorageAccount(ctx, resourceGroupName, accountName, parameters) + 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 +} + +// createStorageAccount creates or updates a StorageAccount. +func (c *Client) createStorageAccount(ctx context.Context, resourceGroupName string, accountName string, parameters storage.AccountCreateParameters) *retry.Error { + resourceID := armclient.GetResourceID( + c.subscriptionID, + resourceGroupName, + "Microsoft.Storage/storageAccounts", + accountName, + ) + + response, rerr := c.armClient.PutResource(ctx, resourceID, parameters) + defer c.armClient.CloseResponse(ctx, response) + if rerr != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "storageAccount.put.request", resourceID, rerr.Error()) + return rerr + } + + if response != nil && response.StatusCode != http.StatusNoContent { + _, rerr = c.createResponder(response) + if rerr != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "storageAccount.put.respond", resourceID, rerr.Error()) + return rerr + } + } + + return nil +} + +func (c *Client) createResponder(resp *http.Response) (*storage.Account, *retry.Error) { + result := &storage.Account{} + err := autorest.Respond( + resp, + azure.WithErrorUnlessStatusCode(http.StatusOK, http.StatusAccepted), + autorest.ByUnmarshallingJSON(&result)) + result.Response = autorest.Response{Response: resp} + return result, retry.GetError(resp, err) +} + +// Delete deletes a StorageAccount by name. +func (c *Client) Delete(ctx context.Context, resourceGroupName string, accountName string) *retry.Error { + mc := metrics.NewMetricContext("storage_account", "delete", resourceGroupName, c.subscriptionID, "") + + // Report errors if the client is rate limited. + if !c.rateLimiterWriter.TryAccept() { + mc.RateLimitedCount() + return retry.GetRateLimitError(true, "StorageAccountDelete") + } + + // Report errors if the client is throttled. + if c.RetryAfterWriter.After(time.Now()) { + mc.ThrottledCount() + rerr := retry.GetThrottlingError("StorageAccountDelete", "client throttled", c.RetryAfterWriter) + return rerr + } + + rerr := c.deleteStorageAccount(ctx, resourceGroupName, accountName) + 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 +} + +// deleteStorageAccount deletes a PublicIPAddress by name. +func (c *Client) deleteStorageAccount(ctx context.Context, resourceGroupName string, accountName string) *retry.Error { + resourceID := armclient.GetResourceID( + c.subscriptionID, + resourceGroupName, + "Microsoft.Storage/storageAccounts", + accountName, + ) + + return c.armClient.DeleteResource(ctx, resourceID, "") +} + +// ListByResourceGroup get a list storage accounts by resourceGroup. +func (c *Client) ListByResourceGroup(ctx context.Context, resourceGroupName string) ([]storage.Account, *retry.Error) { + mc := metrics.NewMetricContext("storage_account", "list_by_resource_group", resourceGroupName, c.subscriptionID, "") + + // Report errors if the client is rate limited. + if !c.rateLimiterReader.TryAccept() { + mc.RateLimitedCount() + return nil, retry.GetRateLimitError(false, "StorageAccountListByResourceGroup") + } + + // Report errors if the client is throttled. + if c.RetryAfterReader.After(time.Now()) { + mc.ThrottledCount() + rerr := retry.GetThrottlingError("StorageAccountListByResourceGroup", "client throttled", c.RetryAfterReader) + return nil, rerr + } + + result, rerr := c.ListStorageAccountByResourceGroup(ctx, resourceGroupName) + 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 +} + +// ListStorageAccountByResourceGroup get a list storage accounts by resourceGroup. +func (c *Client) ListStorageAccountByResourceGroup(ctx context.Context, resourceGroupName string) ([]storage.Account, *retry.Error) { + resourceID := fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Storage/storageAccounts", + autorest.Encode("path", c.subscriptionID), + autorest.Encode("path", resourceGroupName)) + result := make([]storage.Account, 0) + page := &AccountListResultPage{} + page.fn = c.listNextResults + + resp, rerr := c.armClient.GetResource(ctx, resourceID, "") + defer c.armClient.CloseResponse(ctx, resp) + if rerr != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "storageAccount.list.request", resourceID, rerr.Error()) + return result, rerr + } + + var err error + page.alr, err = c.listResponder(resp) + if err != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "storageAccount.list.respond", resourceID, err) + return result, retry.GetError(resp, err) + } + + for page.NotDone() { + result = append(result, *page.Response().Value...) + if err = page.NextWithContext(ctx); err != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "storageAccount.list.next", resourceID, err) + return result, retry.GetError(page.Response().Response.Response, err) + } + } + + return result, nil +} + +func (c *Client) listResponder(resp *http.Response) (result storage.AccountListResult, err error) { + err = autorest.Respond( + resp, + autorest.ByIgnoring(), + azure.WithErrorUnlessStatusCode(http.StatusOK), + autorest.ByUnmarshallingJSON(&result)) + result.Response = autorest.Response{Response: resp} + return +} + +// StorageAccountResultPreparer prepares a request to retrieve the next set of results. +// It returns nil if no more results exist. +func (c *Client) StorageAccountResultPreparer(ctx context.Context, lr storage.AccountListResult) (*http.Request, error) { + if lr.NextLink == nil || len(to.String(lr.NextLink)) < 1 { + return nil, nil + } + + decorators := []autorest.PrepareDecorator{ + autorest.WithBaseURL(to.String(lr.NextLink)), + } + return c.armClient.PrepareGetRequest(ctx, decorators...) +} + +// listNextResults retrieves the next set of results, if any. +func (c *Client) listNextResults(ctx context.Context, lastResults storage.AccountListResult) (result storage.AccountListResult, err error) { + req, err := c.StorageAccountResultPreparer(ctx, lastResults) + if err != nil { + return result, autorest.NewErrorWithError(err, "storageaccount", "listNextResults", nil, "Failure preparing next results request") + } + if req == nil { + return + } + + resp, rerr := c.armClient.Send(ctx, req) + defer c.armClient.CloseResponse(ctx, resp) + if rerr != nil { + result.Response = autorest.Response{Response: resp} + return result, autorest.NewErrorWithError(rerr.Error(), "storageaccount", "listNextResults", resp, "Failure sending next results request") + } + + result, err = c.listResponder(resp) + if err != nil { + err = autorest.NewErrorWithError(err, "storageaccount", "listNextResults", resp, "Failure responding to next results request") + } + + return +} + +// AccountListResultPage contains a page of Account values. +type AccountListResultPage struct { + fn func(context.Context, storage.AccountListResult) (storage.AccountListResult, error) + alr storage.AccountListResult +} + +// NextWithContext advances to the next page of values. If there was an error making +// the request the page does not advance and the error is returned. +func (page *AccountListResultPage) NextWithContext(ctx context.Context) (err error) { + next, err := page.fn(ctx, page.alr) + if err != nil { + return err + } + page.alr = next + return nil +} + +// Next advances to the next page of values. If there was an error making +// the request the page does not advance and the error is returned. +// Deprecated: Use NextWithContext() instead. +func (page *AccountListResultPage) Next() error { + return page.NextWithContext(context.Background()) +} + +// NotDone returns true if the page enumeration should be started or is not yet complete. +func (page AccountListResultPage) NotDone() bool { + return !page.alr.IsEmpty() +} + +// Response returns the raw server response from the last page request. +func (page AccountListResultPage) Response() storage.AccountListResult { + return page.alr +} + +// Values returns the slice of values for the current page or nil if there are no values. +func (page AccountListResultPage) Values() []storage.Account { + if page.alr.IsEmpty() { + return nil + } + return *page.alr.Value +} diff --git a/cluster-autoscaler/cloudprovider/azure/clients/storageaccountclient/interface.go b/cluster-autoscaler/cloudprovider/azure/clients/storageaccountclient/interface.go new file mode 100644 index 000000000000..5a0112c23975 --- /dev/null +++ b/cluster-autoscaler/cloudprovider/azure/clients/storageaccountclient/interface.go @@ -0,0 +1,51 @@ +// +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 storageaccountclient + +import ( + "context" + + "github.com/Azure/azure-sdk-for-go/services/storage/mgmt/2019-06-01/storage" + "k8s.io/legacy-cloud-providers/azure/retry" +) + +const ( + // APIVersion is the API version for network. + APIVersion = "2019-06-01" +) + +// Interface is the client interface for StorageAccounts. +// 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/storageaccountclient/interface.go -package=mockstorageaccountclient Interface > $GOPATH/src/k8s.io/kubernetes/staging/src/k8s.io/legacy-cloud-providers/azure/clients/storageaccountclient/mockstorageaccountclient/interface.go +type Interface interface { + // Create creates a StorageAccount. + Create(ctx context.Context, resourceGroupName string, accountName string, parameters storage.AccountCreateParameters) *retry.Error + + // Delete deletes a StorageAccount by name. + Delete(ctx context.Context, resourceGroupName string, accountName string) *retry.Error + + // ListKeys get a list of storage account keys. + ListKeys(ctx context.Context, resourceGroupName string, accountName string) (storage.AccountListKeysResult, *retry.Error) + + // ListByResourceGroup get a list storage accounts by resourceGroup. + ListByResourceGroup(ctx context.Context, resourceGroupName string) ([]storage.Account, *retry.Error) + + // GetProperties gets properties of the StorageAccount. + GetProperties(ctx context.Context, resourceGroupName string, accountName string) (result storage.Account, rerr *retry.Error) +} diff --git a/cluster-autoscaler/cloudprovider/azure/clients/vmclient/azure_vmclient.go b/cluster-autoscaler/cloudprovider/azure/clients/vmclient/azure_vmclient.go new file mode 100644 index 000000000000..3ed35e8f56d7 --- /dev/null +++ b/cluster-autoscaler/cloudprovider/azure/clients/vmclient/azure_vmclient.go @@ -0,0 +1,479 @@ +// +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 vmclient + +import ( + "context" + "fmt" + "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" + "github.com/Azure/go-autorest/autorest/to" + + "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 VirtualMachine 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 VirtualMachine 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 VirtualMachine client (read ops) using rate limit config: QPS=%g, bucket=%d", + config.RateLimitConfig.CloudProviderRateLimitQPS, + config.RateLimitConfig.CloudProviderRateLimitBucket) + klog.V(2).Infof("Azure VirtualMachine client (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 VirtualMachine. +func (c *Client) Get(ctx context.Context, resourceGroupName string, VMName string, expand compute.InstanceViewTypes) (compute.VirtualMachine, *retry.Error) { + mc := metrics.NewMetricContext("vm", "get", resourceGroupName, c.subscriptionID, "") + + // Report errors if the client is rate limited. + if !c.rateLimiterReader.TryAccept() { + mc.RateLimitedCount() + return compute.VirtualMachine{}, retry.GetRateLimitError(false, "VMGet") + } + + // Report errors if the client is throttled. + if c.RetryAfterReader.After(time.Now()) { + mc.ThrottledCount() + rerr := retry.GetThrottlingError("VMGet", "client throttled", c.RetryAfterReader) + return compute.VirtualMachine{}, rerr + } + + result, rerr := c.getVM(ctx, resourceGroupName, VMName, expand) + 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 +} + +// getVM gets a VirtualMachine. +func (c *Client) getVM(ctx context.Context, resourceGroupName string, VMName string, expand compute.InstanceViewTypes) (compute.VirtualMachine, *retry.Error) { + resourceID := armclient.GetResourceID( + c.subscriptionID, + resourceGroupName, + "Microsoft.Compute/virtualMachines", + VMName, + ) + result := compute.VirtualMachine{} + + response, rerr := c.armClient.GetResource(ctx, resourceID, string(expand)) + defer c.armClient.CloseResponse(ctx, response) + if rerr != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vm.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", "vm.get.respond", resourceID, err) + return result, retry.GetError(response, err) + } + + result.Response = autorest.Response{Response: response} + return result, nil +} + +// List gets a list of VirtualMachine in the resourceGroupName. +func (c *Client) List(ctx context.Context, resourceGroupName string) ([]compute.VirtualMachine, *retry.Error) { + mc := metrics.NewMetricContext("vm", "list", resourceGroupName, c.subscriptionID, "") + + // Report errors if the client is rate limited. + if !c.rateLimiterReader.TryAccept() { + mc.RateLimitedCount() + return nil, retry.GetRateLimitError(false, "VMList") + } + + // Report errors if the client is throttled. + if c.RetryAfterReader.After(time.Now()) { + mc.ThrottledCount() + rerr := retry.GetThrottlingError("VMList", "client throttled", c.RetryAfterReader) + return nil, rerr + } + + result, rerr := c.listVM(ctx, resourceGroupName) + 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 +} + +// listVM gets a list of VirtualMachines in the resourceGroupName. +func (c *Client) listVM(ctx context.Context, resourceGroupName string) ([]compute.VirtualMachine, *retry.Error) { + resourceID := fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Compute/virtualMachines", + autorest.Encode("path", c.subscriptionID), + autorest.Encode("path", resourceGroupName), + ) + + result := make([]compute.VirtualMachine, 0) + page := &VirtualMachineListResultPage{} + page.fn = c.listNextResults + + resp, rerr := c.armClient.GetResource(ctx, resourceID, "") + defer c.armClient.CloseResponse(ctx, resp) + if rerr != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vm.list.request", resourceID, rerr.Error()) + return result, rerr + } + + var err error + page.vmlr, err = c.listResponder(resp) + if err != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vm.list.respond", resourceID, err) + return result, retry.GetError(resp, err) + } + + for page.NotDone() { + result = append(result, *page.Response().Value...) + if err = page.NextWithContext(ctx); err != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vm.list.next", resourceID, err) + return result, retry.GetError(page.Response().Response.Response, err) + } + } + + return result, nil +} + +// Update updates a VirtualMachine. +func (c *Client) Update(ctx context.Context, resourceGroupName string, VMName string, parameters compute.VirtualMachineUpdate, source string) *retry.Error { + mc := metrics.NewMetricContext("vm", "update", resourceGroupName, c.subscriptionID, source) + + // Report errors if the client is rate limited. + if !c.rateLimiterWriter.TryAccept() { + mc.RateLimitedCount() + return retry.GetRateLimitError(true, "VMUpdate") + } + + // Report errors if the client is throttled. + if c.RetryAfterWriter.After(time.Now()) { + mc.ThrottledCount() + rerr := retry.GetThrottlingError("VMUpdate", "client throttled", c.RetryAfterWriter) + return rerr + } + + rerr := c.updateVM(ctx, resourceGroupName, VMName, parameters, source) + 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 +} + +// updateVM updates a VirtualMachine. +func (c *Client) updateVM(ctx context.Context, resourceGroupName string, VMName string, parameters compute.VirtualMachineUpdate, source string) *retry.Error { + resourceID := armclient.GetResourceID( + c.subscriptionID, + resourceGroupName, + "Microsoft.Compute/virtualMachines", + VMName, + ) + + response, rerr := c.armClient.PatchResource(ctx, resourceID, parameters) + defer c.armClient.CloseResponse(ctx, response) + if rerr != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vm.put.request", resourceID, rerr.Error()) + return rerr + } + + if response != nil && response.StatusCode != http.StatusNoContent { + _, rerr = c.updateResponder(response) + if rerr != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vm.put.respond", resourceID, rerr.Error()) + return rerr + } + } + + return nil +} + +func (c *Client) updateResponder(resp *http.Response) (*compute.VirtualMachine, *retry.Error) { + result := &compute.VirtualMachine{} + err := autorest.Respond( + resp, + azure.WithErrorUnlessStatusCode(http.StatusOK, http.StatusCreated), + autorest.ByUnmarshallingJSON(&result), + autorest.ByClosing()) + result.Response = autorest.Response{Response: resp} + return result, retry.GetError(resp, err) +} + +func (c *Client) listResponder(resp *http.Response) (result compute.VirtualMachineListResult, err error) { + err = autorest.Respond( + resp, + autorest.ByIgnoring(), + azure.WithErrorUnlessStatusCode(http.StatusOK), + autorest.ByUnmarshallingJSON(&result), + autorest.ByClosing(), + ) + result.Response = autorest.Response{Response: resp} + return +} + +// vmListResultPreparer prepares a request to retrieve the next set of results. +// It returns nil if no more results exist. +func (c *Client) vmListResultPreparer(ctx context.Context, vmlr compute.VirtualMachineListResult) (*http.Request, error) { + if vmlr.NextLink == nil || len(to.String(vmlr.NextLink)) < 1 { + return nil, nil + } + + decorators := []autorest.PrepareDecorator{ + autorest.WithBaseURL(to.String(vmlr.NextLink)), + } + return c.armClient.PrepareGetRequest(ctx, decorators...) +} + +// listNextResults retrieves the next set of results, if any. +func (c *Client) listNextResults(ctx context.Context, lastResults compute.VirtualMachineListResult) (result compute.VirtualMachineListResult, err error) { + req, err := c.vmListResultPreparer(ctx, lastResults) + if err != nil { + return result, autorest.NewErrorWithError(err, "vmclient", "listNextResults", nil, "Failure preparing next results request") + } + if req == nil { + return + } + + resp, rerr := c.armClient.Send(ctx, req) + defer c.armClient.CloseResponse(ctx, resp) + if rerr != nil { + result.Response = autorest.Response{Response: resp} + return result, autorest.NewErrorWithError(rerr.Error(), "vmclient", "listNextResults", resp, "Failure sending next results request") + } + + result, err = c.listResponder(resp) + if err != nil { + err = autorest.NewErrorWithError(err, "vmclient", "listNextResults", resp, "Failure responding to next results request") + } + + return +} + +// VirtualMachineListResultPage contains a page of VirtualMachine values. +type VirtualMachineListResultPage struct { + fn func(context.Context, compute.VirtualMachineListResult) (compute.VirtualMachineListResult, error) + vmlr compute.VirtualMachineListResult +} + +// NextWithContext advances to the next page of values. If there was an error making +// the request the page does not advance and the error is returned. +func (page *VirtualMachineListResultPage) NextWithContext(ctx context.Context) (err error) { + next, err := page.fn(ctx, page.vmlr) + if err != nil { + return err + } + page.vmlr = next + return nil +} + +// Next advances to the next page of values. If there was an error making +// the request the page does not advance and the error is returned. +// Deprecated: Use NextWithContext() instead. +func (page *VirtualMachineListResultPage) Next() error { + return page.NextWithContext(context.Background()) +} + +// NotDone returns true if the page enumeration should be started or is not yet complete. +func (page VirtualMachineListResultPage) NotDone() bool { + return !page.vmlr.IsEmpty() +} + +// Response returns the raw server response from the last page request. +func (page VirtualMachineListResultPage) Response() compute.VirtualMachineListResult { + return page.vmlr +} + +// Values returns the slice of values for the current page or nil if there are no values. +func (page VirtualMachineListResultPage) Values() []compute.VirtualMachine { + if page.vmlr.IsEmpty() { + return nil + } + return *page.vmlr.Value +} + +// CreateOrUpdate creates or updates a VirtualMachine. +func (c *Client) CreateOrUpdate(ctx context.Context, resourceGroupName string, VMName string, parameters compute.VirtualMachine, source string) *retry.Error { + mc := metrics.NewMetricContext("vm", "create_or_update", resourceGroupName, c.subscriptionID, source) + + // Report errors if the client is rate limited. + if !c.rateLimiterWriter.TryAccept() { + mc.RateLimitedCount() + return retry.GetRateLimitError(true, "VMCreateOrUpdate") + } + + // Report errors if the client is throttled. + if c.RetryAfterWriter.After(time.Now()) { + mc.ThrottledCount() + rerr := retry.GetThrottlingError("VMCreateOrUpdate", "client throttled", c.RetryAfterWriter) + return rerr + } + + rerr := c.createOrUpdateVM(ctx, resourceGroupName, VMName, parameters, source) + 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 +} + +// createOrUpdateVM creates or updates a VirtualMachine. +func (c *Client) createOrUpdateVM(ctx context.Context, resourceGroupName string, VMName string, parameters compute.VirtualMachine, source string) *retry.Error { + resourceID := armclient.GetResourceID( + c.subscriptionID, + resourceGroupName, + "Microsoft.Compute/virtualMachines", + VMName, + ) + + response, rerr := c.armClient.PutResource(ctx, resourceID, parameters) + defer c.armClient.CloseResponse(ctx, response) + if rerr != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vm.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", "vm.put.respond", resourceID, rerr.Error()) + return rerr + } + } + + return nil +} + +func (c *Client) createOrUpdateResponder(resp *http.Response) (*compute.VirtualMachine, *retry.Error) { + result := &compute.VirtualMachine{} + err := autorest.Respond( + resp, + azure.WithErrorUnlessStatusCode(http.StatusOK, http.StatusCreated), + autorest.ByUnmarshallingJSON(&result), + autorest.ByClosing()) + result.Response = autorest.Response{Response: resp} + return result, retry.GetError(resp, err) +} + +// Delete deletes a VirtualMachine. +func (c *Client) Delete(ctx context.Context, resourceGroupName string, VMName string) *retry.Error { + mc := metrics.NewMetricContext("vm", "delete", resourceGroupName, c.subscriptionID, "") + + // Report errors if the client is rate limited. + if !c.rateLimiterWriter.TryAccept() { + mc.RateLimitedCount() + return retry.GetRateLimitError(true, "VMDelete") + } + + // Report errors if the client is throttled. + if c.RetryAfterWriter.After(time.Now()) { + mc.ThrottledCount() + rerr := retry.GetThrottlingError("VMDelete", "client throttled", c.RetryAfterWriter) + return rerr + } + + rerr := c.deleteVM(ctx, resourceGroupName, VMName) + 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 +} + +// deleteVM deletes a VirtualMachine. +func (c *Client) deleteVM(ctx context.Context, resourceGroupName string, VMName string) *retry.Error { + resourceID := armclient.GetResourceID( + c.subscriptionID, + resourceGroupName, + "Microsoft.Compute/virtualMachines", + VMName, + ) + + return c.armClient.DeleteResource(ctx, resourceID, "") +} diff --git a/cluster-autoscaler/cloudprovider/azure/clients/vmclient/interface.go b/cluster-autoscaler/cloudprovider/azure/clients/vmclient/interface.go new file mode 100644 index 000000000000..ad63e8b515ef --- /dev/null +++ b/cluster-autoscaler/cloudprovider/azure/clients/vmclient/interface.go @@ -0,0 +1,51 @@ +// +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 vmclient + +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 VirtualMachine. + APIVersion = "2019-07-01" +) + +// Interface is the client interface for VirtualMachines. +// 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/vmclient/interface.go -package=mockvmclient Interface > $GOPATH/src/k8s.io/kubernetes/staging/src/k8s.io/legacy-cloud-providers/azure/clients/vmclient/mockvmclient/interface.go +type Interface interface { + // Get gets a VirtualMachine. + Get(ctx context.Context, resourceGroupName string, VMName string, expand compute.InstanceViewTypes) (compute.VirtualMachine, *retry.Error) + + // List gets a list of VirtualMachines in the resourceGroupName. + List(ctx context.Context, resourceGroupName string) ([]compute.VirtualMachine, *retry.Error) + + // CreateOrUpdate creates or updates a VirtualMachine. + CreateOrUpdate(ctx context.Context, resourceGroupName string, VMName string, parameters compute.VirtualMachine, source string) *retry.Error + + // Update updates a VirtualMachine. + Update(ctx context.Context, resourceGroupName string, VMName string, parameters compute.VirtualMachineUpdate, source string) *retry.Error + + // Delete deletes a VirtualMachine. + Delete(ctx context.Context, resourceGroupName string, VMName string) *retry.Error +} diff --git a/cluster-autoscaler/cloudprovider/azure/clients/vmssvmclient/azure_vmssvmclient.go b/cluster-autoscaler/cloudprovider/azure/clients/vmssvmclient/azure_vmssvmclient.go new file mode 100644 index 000000000000..a087493c7375 --- /dev/null +++ b/cluster-autoscaler/cloudprovider/azure/clients/vmssvmclient/azure_vmssvmclient.go @@ -0,0 +1,456 @@ +// +build !providerless + +/* +Copyright 2019 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 vmssvmclient + +import ( + "context" + "fmt" + "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" + "github.com/Azure/go-autorest/autorest/to" + + utilerrors "k8s.io/apimachinery/pkg/util/errors" + "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 VMSS 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 vmssVM 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 vmssVM client (read ops) using rate limit config: QPS=%g, bucket=%d", + config.RateLimitConfig.CloudProviderRateLimitQPS, + config.RateLimitConfig.CloudProviderRateLimitBucket) + klog.V(2).Infof("Azure vmssVM client (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 VirtualMachineScaleSetVM. +func (c *Client) Get(ctx context.Context, resourceGroupName string, VMScaleSetName string, instanceID string, expand compute.InstanceViewTypes) (compute.VirtualMachineScaleSetVM, *retry.Error) { + mc := metrics.NewMetricContext("vmssvm", "get", resourceGroupName, c.subscriptionID, "") + + // Report errors if the client is rate limited. + if !c.rateLimiterReader.TryAccept() { + mc.RateLimitedCount() + return compute.VirtualMachineScaleSetVM{}, retry.GetRateLimitError(false, "VMSSVMGet") + } + + // Report errors if the client is throttled. + if c.RetryAfterReader.After(time.Now()) { + mc.ThrottledCount() + rerr := retry.GetThrottlingError("VMSSVMGet", "client throttled", c.RetryAfterReader) + return compute.VirtualMachineScaleSetVM{}, rerr + } + + result, rerr := c.getVMSSVM(ctx, resourceGroupName, VMScaleSetName, instanceID, expand) + 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 +} + +// getVMSSVM gets a VirtualMachineScaleSetVM. +func (c *Client) getVMSSVM(ctx context.Context, resourceGroupName string, VMScaleSetName string, instanceID string, expand compute.InstanceViewTypes) (compute.VirtualMachineScaleSetVM, *retry.Error) { + resourceID := armclient.GetChildResourceID( + c.subscriptionID, + resourceGroupName, + "Microsoft.Compute/virtualMachineScaleSets", + VMScaleSetName, + "virtualMachines", + instanceID, + ) + result := compute.VirtualMachineScaleSetVM{} + + response, rerr := c.armClient.GetResource(ctx, resourceID, string(expand)) + defer c.armClient.CloseResponse(ctx, response) + if rerr != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vmssvm.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", "vmssvm.get.respond", resourceID, err) + return result, retry.GetError(response, err) + } + + result.Response = autorest.Response{Response: response} + return result, nil +} + +// List gets a list of VirtualMachineScaleSetVMs in the virtualMachineScaleSet. +func (c *Client) List(ctx context.Context, resourceGroupName string, virtualMachineScaleSetName string, expand string) ([]compute.VirtualMachineScaleSetVM, *retry.Error) { + mc := metrics.NewMetricContext("vmssvm", "list", resourceGroupName, c.subscriptionID, "") + + // Report errors if the client is rate limited. + if !c.rateLimiterReader.TryAccept() { + mc.RateLimitedCount() + return nil, retry.GetRateLimitError(false, "VMSSVMList") + } + + // Report errors if the client is throttled. + if c.RetryAfterReader.After(time.Now()) { + mc.ThrottledCount() + rerr := retry.GetThrottlingError("VMSSVMList", "client throttled", c.RetryAfterReader) + return nil, rerr + } + + result, rerr := c.listVMSSVM(ctx, resourceGroupName, virtualMachineScaleSetName, expand) + 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 +} + +// listVMSSVM gets a list of VirtualMachineScaleSetVMs in the virtualMachineScaleSet. +func (c *Client) listVMSSVM(ctx context.Context, resourceGroupName string, virtualMachineScaleSetName string, expand string) ([]compute.VirtualMachineScaleSetVM, *retry.Error) { + resourceID := fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Compute/virtualMachineScaleSets/%s/virtualMachines", + autorest.Encode("path", c.subscriptionID), + autorest.Encode("path", resourceGroupName), + autorest.Encode("path", virtualMachineScaleSetName), + ) + + result := make([]compute.VirtualMachineScaleSetVM, 0) + page := &VirtualMachineScaleSetVMListResultPage{} + page.fn = c.listNextResults + + resp, rerr := c.armClient.GetResource(ctx, resourceID, expand) + defer c.armClient.CloseResponse(ctx, resp) + if rerr != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vmssvm.list.request", resourceID, rerr.Error()) + return result, rerr + } + + var err error + page.vmssvlr, err = c.listResponder(resp) + if err != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vmssvm.list.respond", resourceID, err) + return result, retry.GetError(resp, err) + } + + for page.NotDone() { + result = append(result, *page.Response().Value...) + if err = page.NextWithContext(ctx); err != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vmssvm.list.next", resourceID, err) + return result, retry.GetError(page.Response().Response.Response, err) + } + } + + return result, nil +} + +// Update updates a VirtualMachineScaleSetVM. +func (c *Client) Update(ctx context.Context, resourceGroupName string, VMScaleSetName string, instanceID string, parameters compute.VirtualMachineScaleSetVM, source string) *retry.Error { + mc := metrics.NewMetricContext("vmssvm", "update", resourceGroupName, c.subscriptionID, source) + + // Report errors if the client is rate limited. + if !c.rateLimiterWriter.TryAccept() { + mc.RateLimitedCount() + return retry.GetRateLimitError(true, "VMSSVMUpdate") + } + + // Report errors if the client is throttled. + if c.RetryAfterWriter.After(time.Now()) { + mc.ThrottledCount() + rerr := retry.GetThrottlingError("VMSSVMUpdate", "client throttled", c.RetryAfterWriter) + return rerr + } + + rerr := c.updateVMSSVM(ctx, resourceGroupName, VMScaleSetName, instanceID, parameters) + 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 +} + +// updateVMSSVM updates a VirtualMachineScaleSetVM. +func (c *Client) updateVMSSVM(ctx context.Context, resourceGroupName string, VMScaleSetName string, instanceID string, parameters compute.VirtualMachineScaleSetVM) *retry.Error { + resourceID := armclient.GetChildResourceID( + c.subscriptionID, + resourceGroupName, + "Microsoft.Compute/virtualMachineScaleSets", + VMScaleSetName, + "virtualMachines", + instanceID, + ) + + response, rerr := c.armClient.PutResource(ctx, resourceID, parameters) + defer c.armClient.CloseResponse(ctx, response) + if rerr != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vmssvm.put.request", resourceID, rerr.Error()) + return rerr + } + + if response != nil && response.StatusCode != http.StatusNoContent { + _, rerr = c.updateResponder(response) + if rerr != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vmssvm.put.respond", resourceID, rerr.Error()) + return rerr + } + } + + return nil +} + +func (c *Client) updateResponder(resp *http.Response) (*compute.VirtualMachineScaleSetVM, *retry.Error) { + result := &compute.VirtualMachineScaleSetVM{} + 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) +} + +func (c *Client) listResponder(resp *http.Response) (result compute.VirtualMachineScaleSetVMListResult, err error) { + err = autorest.Respond( + resp, + autorest.ByIgnoring(), + azure.WithErrorUnlessStatusCode(http.StatusOK), + autorest.ByUnmarshallingJSON(&result)) + result.Response = autorest.Response{Response: resp} + return +} + +// virtualMachineScaleSetListResultPreparer prepares a request to retrieve the next set of results. +// It returns nil if no more results exist. +func (c *Client) virtualMachineScaleSetVMListResultPreparer(ctx context.Context, vmssvmlr compute.VirtualMachineScaleSetVMListResult) (*http.Request, error) { + if vmssvmlr.NextLink == nil || len(to.String(vmssvmlr.NextLink)) < 1 { + return nil, nil + } + + decorators := []autorest.PrepareDecorator{ + autorest.WithBaseURL(to.String(vmssvmlr.NextLink)), + } + return c.armClient.PrepareGetRequest(ctx, decorators...) +} + +// listNextResults retrieves the next set of results, if any. +func (c *Client) listNextResults(ctx context.Context, lastResults compute.VirtualMachineScaleSetVMListResult) (result compute.VirtualMachineScaleSetVMListResult, err error) { + req, err := c.virtualMachineScaleSetVMListResultPreparer(ctx, lastResults) + if err != nil { + return result, autorest.NewErrorWithError(err, "vmssvmclient", "listNextResults", nil, "Failure preparing next results request") + } + if req == nil { + return + } + + resp, rerr := c.armClient.Send(ctx, req) + defer c.armClient.CloseResponse(ctx, resp) + if rerr != nil { + result.Response = autorest.Response{Response: resp} + return result, autorest.NewErrorWithError(rerr.Error(), "vmssvmclient", "listNextResults", resp, "Failure sending next results request") + } + + result, err = c.listResponder(resp) + if err != nil { + err = autorest.NewErrorWithError(err, "vmssvmclient", "listNextResults", resp, "Failure responding to next results request") + } + + return +} + +// VirtualMachineScaleSetVMListResultPage contains a page of VirtualMachineScaleSetVM values. +type VirtualMachineScaleSetVMListResultPage struct { + fn func(context.Context, compute.VirtualMachineScaleSetVMListResult) (compute.VirtualMachineScaleSetVMListResult, error) + vmssvlr compute.VirtualMachineScaleSetVMListResult +} + +// NextWithContext advances to the next page of values. If there was an error making +// the request the page does not advance and the error is returned. +func (page *VirtualMachineScaleSetVMListResultPage) NextWithContext(ctx context.Context) (err error) { + next, err := page.fn(ctx, page.vmssvlr) + if err != nil { + return err + } + page.vmssvlr = next + return nil +} + +// Next advances to the next page of values. If there was an error making +// the request the page does not advance and the error is returned. +// Deprecated: Use NextWithContext() instead. +func (page *VirtualMachineScaleSetVMListResultPage) Next() error { + return page.NextWithContext(context.Background()) +} + +// NotDone returns true if the page enumeration should be started or is not yet complete. +func (page VirtualMachineScaleSetVMListResultPage) NotDone() bool { + return !page.vmssvlr.IsEmpty() +} + +// Response returns the raw server response from the last page request. +func (page VirtualMachineScaleSetVMListResultPage) Response() compute.VirtualMachineScaleSetVMListResult { + return page.vmssvlr +} + +// Values returns the slice of values for the current page or nil if there are no values. +func (page VirtualMachineScaleSetVMListResultPage) Values() []compute.VirtualMachineScaleSetVM { + if page.vmssvlr.IsEmpty() { + return nil + } + return *page.vmssvlr.Value +} + +// UpdateVMs updates a list of VirtualMachineScaleSetVM from map[instanceID]compute.VirtualMachineScaleSetVM. +func (c *Client) UpdateVMs(ctx context.Context, resourceGroupName string, VMScaleSetName string, instances map[string]compute.VirtualMachineScaleSetVM, source string) *retry.Error { + mc := metrics.NewMetricContext("vmssvm", "update_vms", resourceGroupName, c.subscriptionID, source) + + // Report errors if the client is rate limited. + if !c.rateLimiterWriter.TryAccept() { + mc.RateLimitedCount() + return retry.GetRateLimitError(true, "VMSSVMUpdateVMs") + } + + // Report errors if the client is throttled. + if c.RetryAfterWriter.After(time.Now()) { + mc.ThrottledCount() + rerr := retry.GetThrottlingError("VMSSVMUpdateVMs", "client throttled", c.RetryAfterWriter) + return rerr + } + + rerr := c.updateVMSSVMs(ctx, resourceGroupName, VMScaleSetName, instances) + 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 +} + +// updateVMSSVMs updates a list of VirtualMachineScaleSetVM from map[instanceID]compute.VirtualMachineScaleSetVM. +func (c *Client) updateVMSSVMs(ctx context.Context, resourceGroupName string, VMScaleSetName string, instances map[string]compute.VirtualMachineScaleSetVM) *retry.Error { + resources := make(map[string]interface{}) + for instanceID, parameter := range instances { + resourceID := armclient.GetChildResourceID( + c.subscriptionID, + resourceGroupName, + "Microsoft.Compute/virtualMachineScaleSets", + VMScaleSetName, + "virtualMachines", + instanceID, + ) + resources[resourceID] = parameter + } + + responses := c.armClient.PutResources(ctx, resources) + errors := make([]*retry.Error, 0) + for resourceID, resp := range responses { + if resp == nil { + continue + } + + defer c.armClient.CloseResponse(ctx, resp.Response) + if resp.Error != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vmssvm.put.request", resourceID, resp.Error.Error()) + errors = append(errors, resp.Error) + continue + } + + if resp.Response != nil && resp.Response.StatusCode != http.StatusNoContent { + _, rerr := c.updateResponder(resp.Response) + if rerr != nil { + klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vmssvm.put.respond", resourceID, rerr.Error()) + errors = append(errors, rerr) + } + } + } + + // Aggregate errors. + if len(errors) > 0 { + rerr := &retry.Error{} + errs := make([]error, 0) + for _, err := range errors { + if err.IsThrottled() && err.RetryAfter.After(err.RetryAfter) { + rerr.RetryAfter = err.RetryAfter + } + errs = append(errs, err.Error()) + } + rerr.RawError = utilerrors.Flatten(utilerrors.NewAggregate(errs)) + return rerr + } + + return nil +} diff --git a/cluster-autoscaler/cloudprovider/azure/clients/vmssvmclient/interface.go b/cluster-autoscaler/cloudprovider/azure/clients/vmssvmclient/interface.go new file mode 100644 index 000000000000..19d84c35f107 --- /dev/null +++ b/cluster-autoscaler/cloudprovider/azure/clients/vmssvmclient/interface.go @@ -0,0 +1,48 @@ +// +build !providerless + +/* +Copyright 2019 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 vmssvmclient + +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 VMSS. + APIVersion = "2019-07-01" +) + +// Interface is the client interface for VirtualMachineScaleSetVM. +// 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/vmssvmclient/interface.go -package=mockvmssvmclient Interface > $GOPATH/src/k8s.io/kubernetes/staging/src/k8s.io/legacy-cloud-providers/azure/clients/vmssvmclient/mockvmssvmclient/interface.go +type Interface interface { + // Get gets a VirtualMachineScaleSetVM. + Get(ctx context.Context, resourceGroupName string, VMScaleSetName string, instanceID string, expand compute.InstanceViewTypes) (compute.VirtualMachineScaleSetVM, *retry.Error) + + // List gets a list of VirtualMachineScaleSetVMs in the virtualMachineScaleSet. + List(ctx context.Context, resourceGroupName string, virtualMachineScaleSetName string, expand string) ([]compute.VirtualMachineScaleSetVM, *retry.Error) + + // Update updates a VirtualMachineScaleSetVM. + Update(ctx context.Context, resourceGroupName string, VMScaleSetName string, instanceID string, parameters compute.VirtualMachineScaleSetVM, source string) *retry.Error + + // UpdateVMs updates a list of VirtualMachineScaleSetVM from map[instanceID]compute.VirtualMachineScaleSetVM. + UpdateVMs(ctx context.Context, resourceGroupName string, VMScaleSetName string, instances map[string]compute.VirtualMachineScaleSetVM, source string) *retry.Error +}