tests pass again and scale reduction is now done via machine delete

Author: devigned

api/v1alpha3/tags.go

@@ -29,12 +29,6 @@ func (t Tags) Equals(other Tags) bool {
 	return reflect.DeepEqual(t, other)
 }
 
-// HasMatchingSpecVersionHash returns true if the resource has been tagged with a matching resource spec hash value.
-func (t Tags) HasMatchingSpecVersionHash(hash string) bool {
-	value, ok := t[SpecVersionHashTagKey()]
-	return ok && value == hash
-}
-
 // HasOwned returns true if the tags contains a tag that marks the resource as owned by the cluster from the perspective of this management tooling.
 func (t Tags) HasOwned(cluster string) bool {
 	value, ok := t[ClusterTagKey(cluster)]
@@ -74,12 +68,6 @@ func (t Tags) Merge(other Tags) {
 	}
 }
 
-// AddSpecVersionHashTag adds a spec version hash to the Azure resource tags to determine if state has changed quickly
-func (t Tags) AddSpecVersionHashTag(hash string) Tags {
-	t[SpecVersionHashTagKey()] = hash
-	return t
-}
-
 // ResourceLifecycle configures the lifecycle of a resource
 type ResourceLifecycle string
 
@@ -138,11 +126,6 @@ const (
 	VMTagsLastAppliedAnnotation = "sigs.k8s.io/cluster-api-provider-azure-last-applied-tags-vm"
 )
 
-// SpecVersionHashTagKey is the key for the spec version hash used to enable quick spec difference comparison
-func SpecVersionHashTagKey() string {
-	return fmt.Sprintf("%s%s", NameAzureProviderPrefix, "spec-version-hash")
-}
-
 // ClusterTagKey generates the key for resources associated with a cluster.
 func ClusterTagKey(name string) string {
 	return fmt.Sprintf("%s%s", NameAzureProviderOwned, name)

cloud/converters/vmss_test.go

@@ -99,11 +99,12 @@ func Test_SDKToVMSS(t *testing.T) {
 
 				for i := 0; i < 2; i++ {
 					expected.Instances[i] = infrav1exp.VMSSVM{
-						ID:               fmt.Sprintf("vm/%d", i),
-						InstanceID:       fmt.Sprintf("%d", i),
-						Name:             fmt.Sprintf("instance-00000%d", i),
-						AvailabilityZone: fmt.Sprintf("zone%d", i),
-						State:            "Succeeded",
+						ID:                 fmt.Sprintf("vm/%d", i),
+						InstanceID:         fmt.Sprintf("%d", i),
+						Name:               fmt.Sprintf("instance-00000%d", i),
+						AvailabilityZone:   fmt.Sprintf("zone%d", i),
+						State:              "Succeeded",
+						LatestModelApplied: true,
 					}
 				}
 				g.Expect(actual).To(gomega.Equal(&expected))

cloud/scope/machinepool.go

@@ -25,7 +25,6 @@ import (
 	"github.com/go-logr/logr"
 	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
-	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/klog/klogr"
@@ -241,19 +240,85 @@ func (m *MachinePoolScope) applyAzureMachinePoolMachines(ctx context.Context) er
 		}
 	}
 
-	// determine which machines need to be deleted since they are not in Azure
-	for key, val := range existingMachinesByProviderID {
-		val := val
-		if _, ok := latestMachinesByProviderID[key]; !ok {
-			if err := m.client.Delete(ctx, &val); err != nil && !apierrors.IsNotFound(err) {
-				return errors.Wrap(err, "failed deleting machine")
-			}
+	// select machines to delete to lower the replica count
+	toDelete := m.selectMachinesToDelete(existingMachinesByProviderID)
+	for _, machine := range toDelete {
+		machine := machine
+		if err := m.client.Delete(ctx, &machine); err != nil {
+			return errors.Wrap(err, "failed deleting machine to reduce replica count")
 		}
 	}
 
 	return nil
 }
 
+func getProviderIDs(machinesByProviderID map[string]infrav1exp.AzureMachinePoolMachine) []string {
+	ids := make([]string, len(machinesByProviderID))
+	idx := 0
+	for k := range machinesByProviderID {
+		ids[idx] = k
+		idx++
+	}
+	return ids
+}
+
+func (m *MachinePoolScope) selectMachinesToDelete(machinesByProviderID map[string]infrav1exp.AzureMachinePoolMachine) map[string]infrav1exp.AzureMachinePoolMachine {
+	desiredReplicaCount := int(*m.MachinePool.Spec.Replicas)
+	readyMachines := getReadyMachines(machinesByProviderID)
+	if desiredReplicaCount > len(readyMachines) {
+		m.V(4).Info("we don't have enough ready machines, so don't try to delete any", "desired", desiredReplicaCount, "ready", getProviderIDs(readyMachines))
+		return nil
+	}
+
+	readyAndNotMarkedForDelete := make(map[string]infrav1exp.AzureMachinePoolMachine)
+	for providerID, machine := range readyMachines {
+		if machine.ObjectMeta.DeletionTimestamp.IsZero() {
+			// this is not marked for delete
+			readyAndNotMarkedForDelete[providerID] = machine
+		}
+	}
+
+	numberOfOverProvisioned := len(machinesByProviderID) - desiredReplicaCount
+	if numberOfOverProvisioned <= 0 {
+		m.V(4).Info("no over-provisioned machines", "desired", desiredReplicaCount, "overprovisioned", numberOfOverProvisioned)
+		return nil // no over-provisioned machines
+	}
+
+	m.V(4).Info("found over-provisioned machines", "desired", desiredReplicaCount, "overprovisioned", numberOfOverProvisioned)
+	// pick machines that are not
+	numberofMachinesSelectedToBeDeleted := 0
+	toDelete := make(map[string]infrav1exp.AzureMachinePoolMachine)
+	for k, v := range machinesByProviderID {
+		if _, ok := readyAndNotMarkedForDelete[k]; ok {
+			// if not a known good machine move on
+			continue
+		}
+
+		toDelete[k] = v
+		numberofMachinesSelectedToBeDeleted++
+		if numberofMachinesSelectedToBeDeleted >= numberOfOverProvisioned {
+			break
+		}
+	}
+
+	if len(toDelete) >= numberOfOverProvisioned {
+		m.V(4).Info("found enough machines not ready or marked for delete", "toDelete", getProviderIDs(toDelete), "numToDelete", len(toDelete), "overprovisioned", numberOfOverProvisioned)
+		return toDelete
+	}
+
+	// we are still over-provisioned, select ready machines to delete
+	for k, v := range readyAndNotMarkedForDelete {
+		toDelete[k] = v
+		numberofMachinesSelectedToBeDeleted++
+		if numberofMachinesSelectedToBeDeleted >= numberOfOverProvisioned {
+			break
+		}
+	}
+
+	m.V(4).Info("included ready machines to delete", "toDelete", getProviderIDs(toDelete), "numToDelete", len(toDelete), "overprovisioned", numberOfOverProvisioned)
+	return toDelete
+}
+
 func (m *MachinePoolScope) createMachine(ctx context.Context, machine infrav1exp.VMSSVM) error {
 	if machine.InstanceID == "" {
 		return errors.New("machine.InstanceID must not be empty")
@@ -524,3 +589,14 @@ func getAzureMachineTemplate(ctx context.Context, c client.Client, name, namespa
 	}
 	return m, nil
 }
+
+func getReadyMachines(machinesByProviderID map[string]infrav1exp.AzureMachinePoolMachine) map[string]infrav1exp.AzureMachinePoolMachine {
+	readyMachines := make(map[string]infrav1exp.AzureMachinePoolMachine)
+	for k, v := range machinesByProviderID {
+		if v.Status.Ready && v.Status.ProvisioningState != nil && *v.Status.ProvisioningState == infrav1.VMStateSucceeded {
+			readyMachines[k] = v
+		}
+	}
+
+	return readyMachines
+}

cloud/scope/machinepoolmachine.go

@@ -133,18 +133,14 @@ func (s *MachinePoolMachineScope) ProvisioningState() infrav1.VMState {
 	return ""
 }
 
-// SetReady sets the AzureMachinePoolMachine Ready Status to true.
-func (s *MachinePoolMachineScope) SetReady() {
-	s.AzureMachinePool.Status.Ready = true
-}
-
-// SetNotReady sets the AzureMachinePoolMachine Ready Status to false.
-func (s *MachinePoolMachineScope)  SetNotReady() {
-	s.AzureMachinePool.Status.Ready = false
+// IsReady indicates the machine has successfully provisioned and has a node ref associated
+func (s *MachinePoolMachineScope) IsReady() bool {
+	state := s.AzureMachinePoolMachine.Status.ProvisioningState
+	return s.AzureMachinePoolMachine.Status.Ready && state != nil && *state == infrav1.VMStateSucceeded
 }
 
 // SetFailureMessage sets the AzureMachinePoolMachine status failure message.
-func (s *MachinePoolMachineScope)  SetFailureMessage(v error) {
+func (s *MachinePoolMachineScope) SetFailureMessage(v error) {
 	s.AzureMachinePool.Status.FailureMessage = pointer.StringPtr(v.Error())
 }
 
@@ -153,7 +149,6 @@ func (s *MachinePoolMachineScope) SetFailureReason(v capierrors.MachineStatusErr
 	s.AzureMachinePool.Status.FailureReason = &v
 }
 
-
 // ProviderID returns the AzureMachinePool ID by parsing Spec.ProviderID.
 func (s *MachinePoolMachineScope) ProviderID() string {
 	return s.AzureMachinePoolMachine.Spec.ProviderID
@@ -164,10 +159,6 @@ func (s *MachinePoolMachineScope) Close(ctx context.Context) error {
 	ctx, span := tele.Tracer().Start(ctx, "scope.MachinePoolMachineScope.Close")
 	defer span.End()
 
-	if err := s.updateState(ctx); err != nil {
-		return errors.Wrap(err, "failed to update state")
-	}
-
 	return s.patchHelper.Patch(ctx, s.AzureMachinePoolMachine)
 }
 
@@ -199,7 +190,9 @@ func (s *MachinePoolMachineScope) Drain(ctx context.Context) error {
 	return nil
 }
 
-func (s *MachinePoolMachineScope) updateState(ctx context.Context) error {
+// UpdateStatus updates the node reference for the machine and other status fields. This func should be called at the
+// end of a reconcile request and after updating the scope with the most recent Azure data.
+func (s *MachinePoolMachineScope) UpdateStatus(ctx context.Context) error {
 	ctx, span := tele.Tracer().Start(ctx, "scope.MachinePoolMachineScope.Get")
 	defer span.End()
 

cloud/services/scalesets/mock_scalesets/scalesets_mock.go

@@ -27,6 +27,7 @@ import (
 	"github.com/Azure/go-autorest/autorest/to"
 	"github.com/go-logr/logr"
 	"github.com/pkg/errors"
+
 	"sigs.k8s.io/cluster-api-provider-azure/util/generators"
 
 	infrav1 "sigs.k8s.io/cluster-api-provider-azure/api/v1alpha3"
@@ -50,6 +51,7 @@ type (
 		NeedsK8sVersionUpdate() bool
 		SaveK8sVersion()
 		ScaleSetSpec() azure.ScaleSetSpec
+		VMSSExtensionSpecs() []azure.VMSSExtensionSpec
 		SetAnnotation(string, string)
 		SetLongRunningOperationState(*infrav1.Future)
 		SetProviderID(string)
@@ -176,7 +178,6 @@ func (s *Service) createVMSS(ctx context.Context) (*infrav1.Future, error) {
 	}
 
 	vmss := result.VMSSWithoutHash
-	vmss.Tags = converters.TagsToMap(result.Tags.AddSpecVersionHashTag(result.Hash))
 	future, err := s.Client.CreateOrUpdateAsync(ctx, s.Scope.ResourceGroup(), spec.Name, vmss)
 	if err != nil {
 		return future, errors.Wrapf(err, "cannot create VMSS")
@@ -199,31 +200,25 @@ func (s *Service) patchVMSSIfNeeded(ctx context.Context, infraVMSS *infrav1exp.V
 		return nil, errors.Wrapf(err, "failed to generate scale set update parameters for %s", spec.Name)
 	}
 
-	if infraVMSS.Tags.HasMatchingSpecVersionHash(result.Hash) {
-		// The VMSS built from the AzureMachinePool spec matches the hash in the tag of the existing VMSS. This means
-		// the VMSS does not need to be patched since it has not changed.
-		//
-		// hash(AzureMachinePool.Spec)
-		//
-		// Note: if a user were to mutate the VMSS in Azure rather than through CAPZ, this hash match may match, but not
-		// reflect the state of the specification in K8s.
-		return nil, nil
-	}
-
 	vmss := result.VMSSWithoutHash
-	vmss.Tags = converters.TagsToMap(result.Tags.AddSpecVersionHashTag(result.Hash))
 	patch, err := getVMSSUpdateFromVMSS(vmss)
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to generate vmss patch for %s", spec.Name)
 	}
 
-	if s.Scope.MaxSurge() > 0 && hasModelModifyingDifferences(infraVMSS, vmss) {
+	hasModelChanges := hasModelModifyingDifferences(infraVMSS, vmss)
+	if s.Scope.MaxSurge() > 0 && hasModelChanges {
 		// surge capacity with the intention of lowering during instance reconciliation
 		patch.Sku.Capacity = to.Int64Ptr(*patch.Sku.Capacity + int64(s.Scope.MaxSurge()))
 	}
 
-	// wipe out network profile, so updates won't conflict with Cloud Provider updates
-	patch.VirtualMachineProfile.NetworkProfile = nil
+	// If there are no model changes and no increase in the replica count, do not update the VMSS.
+	// Decreases in replica count is handled by deleting AzureMachinePoolMachine instances in the MachinePoolScope
+	if *patch.Sku.Capacity <= infraVMSS.Capacity && !hasModelChanges {
+		s.Scope.V(4).Info("nothing to update on vmss", "scale set", spec.Name, "newReplicas", *patch.Sku.Capacity, "oldReplicas", infraVMSS.Capacity, "hasChanges", hasModelChanges)
+		return nil, nil
+	}
+
 	future, err := s.UpdateAsync(ctx, s.Scope.ResourceGroup(), spec.Name, patch)
 	if err != nil {
 		if azure.ResourceConflict(err) {
@@ -332,6 +327,8 @@ func (s *Service) buildVMSSFromSpec(ctx context.Context, vmssSpec azure.ScaleSet
 		return result, err
 	}
 
+	extensions := s.generateExtensions()
+
 	vmss := compute.VirtualMachineScaleSet{
 		Location: to.StringPtr(s.Scope.Location()),
 		Sku: &compute.Sku{
@@ -381,6 +378,9 @@ func (s *Service) buildVMSSFromSpec(ctx context.Context, vmssSpec azure.ScaleSet
 				Priority:       priority,
 				EvictionPolicy: evictionPolicy,
 				BillingProfile: billingProfile,
+				ExtensionProfile: &compute.VirtualMachineScaleSetExtensionProfile{
+					Extensions: &extensions,
+				},
 			},
 		},
 	}
@@ -476,6 +476,23 @@ func (s *Service) getVirtualMachineScaleSetIfDone(ctx context.Context, future *i
 	return converters.SDKToVMSS(vmss, vmssInstances), nil
 }
 
+func (s *Service) generateExtensions() []compute.VirtualMachineScaleSetExtension {
+	extensions := make([]compute.VirtualMachineScaleSetExtension, len(s.Scope.VMSSExtensionSpecs()))
+	for i, extensionSpec := range s.Scope.VMSSExtensionSpecs() {
+		extensions[i] = compute.VirtualMachineScaleSetExtension{
+			Name: &extensionSpec.Name,
+			VirtualMachineScaleSetExtensionProperties: &compute.VirtualMachineScaleSetExtensionProperties{
+				Publisher:          to.StringPtr(extensionSpec.Publisher),
+				Type:               to.StringPtr(extensionSpec.Name),
+				TypeHandlerVersion: to.StringPtr(extensionSpec.Version),
+				Settings:           nil,
+				ProtectedSettings:  nil,
+			},
+		}
+	}
+	return extensions
+}
+
 // generateStorageProfile generates a pointer to a compute.VirtualMachineScaleSetStorageProfile which can utilized for VM creation.
 func (s *Service) generateStorageProfile(vmssSpec azure.ScaleSetSpec, sku resourceskus.SKU) (*compute.VirtualMachineScaleSetStorageProfile, error) {
 	storageProfile := &compute.VirtualMachineScaleSetStorageProfile{
@@ -581,9 +598,15 @@ func getVMSSUpdateFromVMSS(vmss compute.VirtualMachineScaleSet) (compute.Virtual
 	if err != nil {
 		return compute.VirtualMachineScaleSetUpdate{}, err
 	}
+
 	var update compute.VirtualMachineScaleSetUpdate
-	err = update.UnmarshalJSON(jsonData)
-	return update, err
+	if err := update.UnmarshalJSON(jsonData); err != nil {
+		return update, err
+	}
+
+	// wipe out network profile, so updates won't conflict with Cloud Provider updates
+	update.VirtualMachineProfile.NetworkProfile = nil
+	return update, nil
 }
 
 func getSecurityProfile(vmssSpec azure.ScaleSetSpec, sku resourceskus.SKU) (*compute.SecurityProfile, error) {