Merge pull request #4685 from marwanad/stable-cache

azure vmss cache fixes and improvements

cluster-autoscaler/cloudprovider/azure/azure_manager_test.go

@@ -697,6 +697,7 @@ func TestGetFilteredAutoscalingGroupsVmss(t *testing.T) {
 		maxSize:                maxVal,
 		manager:                manager,
 		curSize:                3,
+		sizeRefreshPeriod:      manager.azureCache.refreshInterval,
 		instancesRefreshPeriod: defaultVmssInstancesRefreshPeriod,
 	}}
 	assert.True(t, assert.ObjectsAreEqualValues(expectedAsgs, asgs), "expected %#v, but found: %#v", expectedAsgs, asgs)

cluster-autoscaler/cloudprovider/azure/azure_scale_set.go

@@ -53,6 +53,9 @@ type ScaleSet struct {
 	sizeMutex sync.Mutex
 	curSize   int64
 
+	lastSizeRefresh   time.Time
+	sizeRefreshPeriod time.Duration
+
 	instancesRefreshPeriod time.Duration
 	instancesRefreshJitter int
 
@@ -71,7 +74,7 @@ func NewScaleSet(spec *dynamic.NodeGroupSpec, az *AzureManager, curSize int64) (
 		maxSize:                spec.MaxSize,
 		manager:                az,
 		curSize:                curSize,
-
+		sizeRefreshPeriod:      az.azureCache.refreshInterval,
 		instancesRefreshJitter: az.config.VmssVmsCacheJitter,
 	}
 
@@ -140,6 +143,11 @@ func (scaleSet *ScaleSet) getCurSize() (int64, error) {
 	scaleSet.sizeMutex.Lock()
 	defer scaleSet.sizeMutex.Unlock()
 
+	if scaleSet.lastSizeRefresh.Add(scaleSet.sizeRefreshPeriod).After(time.Now()) {
+		klog.V(3).Infof("VMSS: %s, returning in-memory size: %d", scaleSet.Name, scaleSet.curSize)
+		return scaleSet.curSize, nil
+	}
+
 	set, err := scaleSet.getVMSSFromCache()
 	if err != nil {
 		klog.Errorf("failed to get information for VMSS: %s, error: %v", scaleSet.Name, err)
@@ -148,7 +156,7 @@ func (scaleSet *ScaleSet) getCurSize() (int64, error) {
 
 	// If VMSS state is updating, return the currentSize which would've been proactively incremented or decremented by CA
 	if set.VirtualMachineScaleSetProperties != nil && strings.EqualFold(to.String(set.VirtualMachineScaleSetProperties.ProvisioningState), string(compute.ProvisioningStateUpdating)) {
-		klog.V(3).Infof("VMSS %q is in updating state, returning cached size: %d", scaleSet.Name, scaleSet.curSize)
+		klog.V(3).Infof("VMSS %q is in updating state, returning in-memory size: %d", scaleSet.Name, scaleSet.curSize)
 		return scaleSet.curSize, nil
 	}
 
@@ -161,9 +169,10 @@ func (scaleSet *ScaleSet) getCurSize() (int64, error) {
 		klog.V(5).Infof("VMSS %q size changed from: %d to %d, invalidating instance cache", scaleSet.Name, scaleSet.curSize, curSize)
 		scaleSet.invalidateInstanceCache()
 	}
-	klog.V(3).Infof("VMSS: %s, previous size: %d, new size: %d", scaleSet.Name, scaleSet.curSize, curSize)
+	klog.V(3).Infof("VMSS: %s, in-memory size: %d, new size: %d", scaleSet.Name, scaleSet.curSize, curSize)
 
 	scaleSet.curSize = curSize
+	scaleSet.lastSizeRefresh = time.Now()
 	return scaleSet.curSize, nil
 }
 
@@ -194,6 +203,7 @@ func (scaleSet *ScaleSet) updateVMSSCapacity(future *azure.Future) {
 		if err != nil {
 			klog.Errorf("Failed to update the capacity for vmss %s with error %v, invalidate the cache so as to get the real size from API", scaleSet.Name, err)
 			// Invalidate the VMSS size cache in order to fetch the size from the API.
+			scaleSet.invalidateLastSizeRefreshWithLock()
 			scaleSet.manager.invalidateCache()
 		}
 	}()
@@ -247,6 +257,7 @@ func (scaleSet *ScaleSet) SetScaleSetSize(size int64) error {
 
 	// Proactively set the VMSS size so autoscaler makes better decisions.
 	scaleSet.curSize = size
+	scaleSet.lastSizeRefresh = time.Now()
 
 	go scaleSet.updateVMSSCapacity(future)
 	return nil
@@ -405,6 +416,7 @@ func (scaleSet *ScaleSet) DeleteInstances(instances []*azureRef, hasUnregistered
 	if !hasUnregisteredNodes {
 		scaleSet.sizeMutex.Lock()
 		scaleSet.curSize -= int64(len(instanceIDs))
+		scaleSet.lastSizeRefresh = time.Now()
 		scaleSet.sizeMutex.Unlock()
 	}
 
@@ -567,6 +579,7 @@ func (scaleSet *ScaleSet) setInstanceStatusByProviderID(providerID string, statu
 			scaleSet.instanceCache[k].Status = &status
 		}
 	}
+	scaleSet.lastInstanceRefresh = time.Now()
 }
 
 // instanceStatusFromVM converts the VM provisioning state to cloudprovider.InstanceStatus
@@ -594,3 +607,9 @@ func (scaleSet *ScaleSet) invalidateInstanceCache() {
 	scaleSet.lastInstanceRefresh = time.Now().Add(-1 * scaleSet.instancesRefreshPeriod)
 	scaleSet.instanceMutex.Unlock()
 }
+
+func (scaleSet *ScaleSet) invalidateLastSizeRefreshWithLock() {
+	scaleSet.sizeMutex.Lock()
+	scaleSet.lastSizeRefresh = time.Now().Add(-1 * scaleSet.sizeRefreshPeriod)
+	scaleSet.sizeMutex.Unlock()
+}