Avoid unwanted VMSS VMs caches invalidations

`fetchAutoAsgs()` is called at regular intervals, fetches a list of VMSS,
then call `Register()` to cache each of those. That registration function
will tell the caller wether that vmss' cache is outdated (when the provided
VMSS, supposedly fresh, is different than the one held in cache) and will
replace existing cache entry by the provided VMSS (which in effect will
require a forced refresh since that ScaleSet struct is passed by
fetchAutoAsgs with a nil lastRefresh time and an empty instanceCache).

To detect changes, `Register()` uses an `reflect.DeepEqual()` between the
provided and the cached VMSS. Which does always find them different: cached
VMSS were enriched with instances lists (while the provided one is blank,
fresh from a simple vmss.list call). That DeepEqual is also fragile due to
the compared structs containing mutexes (that may be held or not) and
refresh timestamps, attributes that shoudln't be relevant to the comparison.

As a consequence, all Register() calls causes indirect cache invalidations
and a costly refresh (VMSS VMS List). The number of Register() calls is
directly proportional to the number of VMSS attached to the cluster, and
can easily triggers ARM API throttling.

With a large number of VMSS, that throttling prevents `fetchAutoAsgs` to
ever succeed (and cluster-autoscaler to start). ie.:

```
I0807 16:55:25.875907     153 azure_scale_set.go:344] GetScaleSetVms: starts
I0807 16:55:25.875915     153 azure_scale_set.go:350] GetScaleSetVms: scaleSet.Name: a-testvmss-10, vmList: []
E0807 16:55:25.875919     153 azure_scale_set.go:352] VirtualMachineScaleSetVMsClient.List failed for a-testvmss-10: &{true 0 2020-08-07 17:10:25.875447854 +0000 UTC m=+913.985215807 azure cloud provider throttled for operation VMSSVMList with reason "client throttled"}
E0807 16:55:25.875928     153 azure_manager.go:538] Failed to regenerate ASG cache: Retriable: true, RetryAfter: 899s, HTTPStatusCode: 0, RawError: azure cloud provider throttled for operation VMSSVMList with reason "client throttled"
F0807 16:55:25.875934     153 azure_cloud_provider.go:167] Failed to create Azure Manager: Retriable: true, RetryAfter: 899s, HTTPStatusCode: 0, RawError: azure cloud provider throttled for operation VMSSVMList with reason "client throttled"
goroutine 28 [running]:
```

From [`ScaleSet` struct attributes](https://github.com/kubernetes/autoscaler/blob/master/cluster-autoscaler/cloudprovider/azure/azure_scale_set.go#L74-L89)
(manager, sizes, mutexes, refreshes timestamps) only sizes are relevant
to that comparison. `curSize` is not strictly necessary, but comparing it
will provide early instance caches refreshs.

Author: bpineau

cluster-autoscaler/cloudprovider/azure/azure_cache.go

@@ -17,7 +17,6 @@ limitations under the License.
 package azure
 
 import (
-	"reflect"
 	"regexp"
 	"strings"
 	"sync"
@@ -62,7 +61,9 @@ func (m *asgCache) Register(asg cloudprovider.NodeGroup) bool {
 
 	for i := range m.registeredAsgs {
 		if existing := m.registeredAsgs[i]; strings.EqualFold(existing.Id(), asg.Id()) {
-			if reflect.DeepEqual(existing, asg) {
+			e := existing.(*ScaleSet)
+			a := asg.(*ScaleSet)
+			if e.minSize == a.minSize && e.maxSize == a.maxSize && e.curSize == a.curSize {
 				return false
 			}
 
@@ -181,7 +182,7 @@ func (m *asgCache) regenerate() error {
 
 	m.instanceToAsg = newCache
 
-	// Incalidating unowned instance cache.
+	// Invalidating unowned instance cache.
 	m.invalidateUnownedInstanceCache()
 
 	return nil

cluster-autoscaler/cloudprovider/azure/azure_manager.go

@@ -513,7 +513,7 @@ func (m *AzureManager) buildAsgFromSpec(spec string) (cloudprovider.NodeGroup, e
 	case vmTypeStandard:
 		return NewAgentPool(s, m)
 	case vmTypeVMSS:
-		return NewScaleSet(s, m)
+		return NewScaleSet(s, m, -1)
 	case vmTypeAKS:
 		return NewAKSAgentPool(s, m)
 	default:
@@ -700,7 +700,12 @@ func (m *AzureManager) listScaleSets(filter []labelAutoDiscoveryConfig) ([]cloud
 			continue
 		}
 
-		asg, err := NewScaleSet(spec, m)
+		curSize := int64(-1)
+		if scaleSet.Sku != nil && scaleSet.Sku.Capacity != nil {
+			curSize = *scaleSet.Sku.Capacity
+		}
+
+		asg, err := NewScaleSet(spec, m, curSize)
 		if err != nil {
 			klog.Warningf("ignoring nodegroup %q %s", *scaleSet.Name, err)
 			continue

cluster-autoscaler/cloudprovider/azure/azure_manager_test.go

@@ -751,7 +751,7 @@ func TestListScalesets(t *testing.T) {
 				minSize:           5,
 				maxSize:           50,
 				manager:           manager,
-				curSize:           -1,
+				curSize:           3,
 				sizeRefreshPeriod: defaultVmssSizeRefreshPeriod,
 			}},
 		},
@@ -857,7 +857,7 @@ func TestGetFilteredAutoscalingGroupsVmss(t *testing.T) {
 		minSize:           minVal,
 		maxSize:           maxVal,
 		manager:           manager,
-		curSize:           -1,
+		curSize:           3,
 		sizeRefreshPeriod: defaultVmssSizeRefreshPeriod,
 	}}
 	assert.True(t, assert.ObjectsAreEqualValues(expectedAsgs, asgs), "expected %#v, but found: %#v", expectedAsgs, asgs)

cluster-autoscaler/cloudprovider/azure/azure_scale_set.go

@@ -89,15 +89,15 @@ type ScaleSet struct {
 }
 
 // NewScaleSet creates a new NewScaleSet.
-func NewScaleSet(spec *dynamic.NodeGroupSpec, az *AzureManager) (*ScaleSet, error) {
+func NewScaleSet(spec *dynamic.NodeGroupSpec, az *AzureManager, curSize int64) (*ScaleSet, error) {
 	scaleSet := &ScaleSet{
 		azureRef: azureRef{
 			Name: spec.Name,
 		},
 		minSize: spec.MinSize,
 		maxSize: spec.MaxSize,
 		manager: az,
-		curSize: -1,
+		curSize: curSize,
 	}
 
 	if az.config.VmssCacheTTL != 0 {