Revert machine migration changes (#176)

go.mod

@@ -25,7 +25,7 @@ require (
 	k8s.io/client-go v0.25.4
 	k8s.io/utils v0.0.0-20221108210102-8e77b1f39fe2
 	knative.dev/pkg v0.0.0-20221123154742-05b694ec4d3a
-	sigs.k8s.io/controller-runtime v0.13.0
+	sigs.k8s.io/controller-runtime v0.13.1
 )
 
 require (

go.sum

@@ -848,8 +848,8 @@ knative.dev/pkg v0.0.0-20221123154742-05b694ec4d3a/go.mod h1:fckNBPf9bu5/p1RbnOh
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
 rsc.io/quote/v3 v3.1.0/go.mod h1:yEA65RcK8LyAZtP9Kv3t0HmxON59tX3rD+tICJqUlj0=
 rsc.io/sampler v1.3.0/go.mod h1:T1hPZKmBbMNahiBKFy5HrXp6adAjACjK9JXDnKaTXpA=
-sigs.k8s.io/controller-runtime v0.13.0 h1:iqa5RNciy7ADWnIc8QxCbOX5FEKVR3uxVxKHRMc2WIQ=
-sigs.k8s.io/controller-runtime v0.13.0/go.mod h1:Zbz+el8Yg31jubvAEyglRZGdLAjplZl+PgtYNI6WNTI=
+sigs.k8s.io/controller-runtime v0.13.1 h1:tUsRCSJVM1QQOOeViGeX3GMT3dQF1eePPw6sEE3xSlg=
+sigs.k8s.io/controller-runtime v0.13.1/go.mod h1:Zbz+el8Yg31jubvAEyglRZGdLAjplZl+PgtYNI6WNTI=
 sigs.k8s.io/json v0.0.0-20220713155537-f223a00ba0e2 h1:iXTIw73aPyC+oRdyqqvVJuloN1p0AC/kzH07hu3NE+k=
 sigs.k8s.io/json v0.0.0-20220713155537-f223a00ba0e2/go.mod h1:B8JuhiUyNFVKdsE8h686QcCxMaH6HrOAZj4vswFpcB0=
 sigs.k8s.io/structured-merge-diff/v4 v4.2.3 h1:PRbqxJClWWYMNV1dhaG4NsibJbArud9kFxnAMREiWFE=

pkg/cloudprovider/fake/cloudprovider.go

@@ -111,9 +111,10 @@ func (c *CloudProvider) Create(ctx context.Context, machine *v1alpha5.Machine) (
 			break
 		}
 	}
+	name := test.RandomName()
 	created := &v1alpha5.Machine{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:        machine.Name,
+			Name:        name,
 			Labels:      lo.Assign(labels, machine.Labels),
 			Annotations: machine.Annotations,
 		},

pkg/controllers/controllers.go

@@ -25,7 +25,6 @@ import (
 	"github.com/aws/karpenter-core/pkg/controllers/counter"
 	"github.com/aws/karpenter-core/pkg/controllers/deprovisioning"
 	"github.com/aws/karpenter-core/pkg/controllers/inflightchecks"
-	"github.com/aws/karpenter-core/pkg/controllers/machine"
 	"github.com/aws/karpenter-core/pkg/controllers/machine/terminator"
 	metricspod "github.com/aws/karpenter-core/pkg/controllers/metrics/pod"
 	metricsprovisioner "github.com/aws/karpenter-core/pkg/controllers/metrics/provisioner"
@@ -65,13 +64,11 @@ func NewControllers(
 		informer.NewNodeController(kubeClient, cluster),
 		informer.NewPodController(kubeClient, cluster),
 		informer.NewProvisionerController(kubeClient, cluster),
-		informer.NewMachineController(kubeClient, cluster),
 		node.NewController(clock, kubeClient, cloudProvider, cluster),
 		termination.NewController(kubeClient, cloudProvider, terminator, recorder),
 		metricspod.NewController(kubeClient),
 		metricsprovisioner.NewController(kubeClient),
 		counter.NewController(kubeClient, cluster),
 		inflightchecks.NewController(clock, kubeClient, recorder, cloudProvider),
-		machine.NewController(clock, kubeClient, cloudProvider, terminator, recorder),
 	}
 }

pkg/controllers/deprovisioning/consolidation.go

@@ -80,16 +80,16 @@ func (c *consolidation) sortAndFilterCandidates(ctx context.Context, nodes []*Ca
 
 	// filter out nodes that can't be terminated
 	nodes = lo.Filter(nodes, func(cn *Candidate, _ int) bool {
-		if !cn.Machine.DeletionTimestamp.IsZero() {
-			c.recorder.Publish(deprovisioningevents.Unconsolidatable(cn.Node, cn.Machine, "in the process of deletion")...)
+		if !cn.Node.DeletionTimestamp.IsZero() {
+			c.recorder.Publish(deprovisioningevents.Unconsolidatable(cn.Node, "in the process of deletion")...)
 			return false
 		}
 		if pdb, ok := pdbs.CanEvictPods(cn.pods); !ok {
-			c.recorder.Publish(deprovisioningevents.Unconsolidatable(cn.Node, cn.Machine, fmt.Sprintf("pdb %s prevents pod evictions", pdb))...)
+			c.recorder.Publish(deprovisioningevents.Unconsolidatable(cn.Node, fmt.Sprintf("pdb %s prevents pod evictions", pdb))...)
 			return false
 		}
 		if p, ok := hasDoNotEvictPod(cn); ok {
-			c.recorder.Publish(deprovisioningevents.Unconsolidatable(cn.Node, cn.Machine, fmt.Sprintf("pod %s/%s has do not evict annotation", p.Namespace, p.Name))...)
+			c.recorder.Publish(deprovisioningevents.Unconsolidatable(cn.Node, fmt.Sprintf("pod %s/%s has do not evict annotation", p.Namespace, p.Name))...)
 			return false
 		}
 		return true
@@ -104,15 +104,15 @@ func (c *consolidation) sortAndFilterCandidates(ctx context.Context, nodes []*Ca
 // ShouldDeprovision is a predicate used to filter deprovisionable nodes
 func (c *consolidation) ShouldDeprovision(_ context.Context, cn *Candidate) bool {
 	if val, ok := cn.Annotations()[v1alpha5.DoNotConsolidateNodeAnnotationKey]; ok {
-		c.recorder.Publish(deprovisioningevents.Unconsolidatable(cn.Node, cn.Machine, fmt.Sprintf("%s annotation exists", v1alpha5.DoNotConsolidateNodeAnnotationKey))...)
+		c.recorder.Publish(deprovisioningevents.Unconsolidatable(cn.Node, fmt.Sprintf("%s annotation exists", v1alpha5.DoNotConsolidateNodeAnnotationKey))...)
 		return val != "true"
 	}
 	if cn.provisioner == nil {
-		c.recorder.Publish(deprovisioningevents.Unconsolidatable(cn.Node, cn.Machine, "provisioner is unknown")...)
+		c.recorder.Publish(deprovisioningevents.Unconsolidatable(cn.Node, "provisioner is unknown")...)
 		return false
 	}
 	if cn.provisioner.Spec.Consolidation == nil || !ptr.BoolValue(cn.provisioner.Spec.Consolidation.Enabled) {
-		c.recorder.Publish(deprovisioningevents.Unconsolidatable(cn.Node, cn.Machine, fmt.Sprintf("provisioner %s has consolidation disabled", cn.provisioner.Name))...)
+		c.recorder.Publish(deprovisioningevents.Unconsolidatable(cn.Node, fmt.Sprintf("provisioner %s has consolidation disabled", cn.provisioner.Name))...)
 		return false
 	}
 	return true
@@ -201,7 +201,7 @@ func (c *consolidation) computeConsolidation(ctx context.Context, candidates ...
 	if !allPodsScheduled {
 		// This method is used by multi-node consolidation as well, so we'll only report in the single node case
 		if len(candidates) == 1 {
-			c.recorder.Publish(deprovisioningevents.Unconsolidatable(candidates[0].Node, candidates[0].Machine, "not all pods would schedule")...)
+			c.recorder.Publish(deprovisioningevents.Unconsolidatable(candidates[0].Node, "not all pods would schedule")...)
 		}
 		return Command{action: actionDoNothing}, nil
 	}
@@ -217,7 +217,7 @@ func (c *consolidation) computeConsolidation(ctx context.Context, candidates ...
 	// we're not going to turn a single node into multiple candidates
 	if len(newMachines) != 1 {
 		if len(candidates) == 1 {
-			c.recorder.Publish(deprovisioningevents.Unconsolidatable(candidates[0].Node, candidates[0].Machine, fmt.Sprintf("can't remove without creating %d candidates", len(newMachines)))...)
+			c.recorder.Publish(deprovisioningevents.Unconsolidatable(candidates[0].Node, fmt.Sprintf("can't remove without creating %d candidates", len(newMachines)))...)
 		}
 		return Command{action: actionDoNothing}, nil
 	}
@@ -231,7 +231,7 @@ func (c *consolidation) computeConsolidation(ctx context.Context, candidates ...
 	newMachines[0].InstanceTypeOptions = filterByPrice(newMachines[0].InstanceTypeOptions, newMachines[0].Requirements, nodesPrice)
 	if len(newMachines[0].InstanceTypeOptions) == 0 {
 		if len(candidates) == 1 {
-			c.recorder.Publish(deprovisioningevents.Unconsolidatable(candidates[0].Node, candidates[0].Machine, "can't replace with a cheaper node")...)
+			c.recorder.Publish(deprovisioningevents.Unconsolidatable(candidates[0].Node, "can't replace with a cheaper node")...)
 		}
 		// no instance types remain after filtering by price
 		return Command{action: actionDoNothing}, nil
@@ -250,7 +250,7 @@ func (c *consolidation) computeConsolidation(ctx context.Context, candidates ...
 	if allExistingAreSpot &&
 		newMachines[0].Requirements.Get(v1alpha5.LabelCapacityType).Has(v1alpha5.CapacityTypeSpot) {
 		if len(candidates) == 1 {
-			c.recorder.Publish(deprovisioningevents.Unconsolidatable(candidates[0].Node, candidates[0].Machine, "can't replace a spot node with a spot node")...)
+			c.recorder.Publish(deprovisioningevents.Unconsolidatable(candidates[0].Node, "can't replace a spot node with a spot node")...)
 		}
 		return Command{action: actionDoNothing}, nil
 	}

pkg/controllers/deprovisioning/controller.go

@@ -114,7 +114,7 @@ func (c *Controller) Reconcile(ctx context.Context, _ reconcile.Request) (reconc
 	for _, d := range c.deprovisioners {
 		candidates, err := GetCandidates(ctx, c.cluster, c.kubeClient, c.clock, c.cloudProvider, d.ShouldDeprovision)
 		if err != nil {
-			return reconcile.Result{}, fmt.Errorf("determining candidate nodes, %w", err)
+			return reconcile.Result{}, fmt.Errorf("determining candidates, %w", err)
 		}
 		// If there are no candidate nodes, move to the next deprovisioner
 		if len(candidates) == 0 {
@@ -135,7 +135,7 @@ func (c *Controller) Reconcile(ctx context.Context, _ reconcile.Request) (reconc
 
 		// Attempt to deprovision
 		if err := c.executeCommand(ctx, d, cmd); err != nil {
-			return reconcile.Result{}, fmt.Errorf("deprovisioning nodes, %w", err)
+			return reconcile.Result{}, fmt.Errorf("deprovisioning candidates, %w", err)
 		}
 		return reconcile.Result{Requeue: true}, nil
 	}
@@ -154,23 +154,23 @@ func (c *Controller) executeCommand(ctx context.Context, d Deprovisioner, comman
 
 	reason := fmt.Sprintf("%s/%s", d, command.action)
 	if command.action == actionReplace {
-		if err := c.launchReplacementMachines(ctx, command, reason); err != nil {
+		if err := c.launchReplacementNodes(ctx, command, reason); err != nil {
 			// If we failed to launch the replacement, don't deprovision.  If this is some permanent failure,
 			// we don't want to disrupt workloads with no way to provision new nodes for them.
 			return fmt.Errorf("launching replacement machine, %w", err)
 		}
 	}
 
 	for _, candidate := range command.candidates {
-		c.recorder.Publish(deprovisioningevents.Terminating(candidate.Node, candidate.Machine, command.String())...)
+		c.recorder.Publish(deprovisioningevents.Terminating(candidate.Node, command.String())...)
 
-		if err := c.kubeClient.Delete(ctx, candidate.Machine); err != nil {
+		if err := c.kubeClient.Delete(ctx, candidate.Node); err != nil {
 			if errors.IsNotFound(err) {
 				continue
 			}
 			logging.FromContext(ctx).Errorf("terminating machine, %s", err)
 		} else {
-			metrics.MachinesTerminatedCounter.With(prometheus.Labels{
+			metrics.NodesTerminatedCounter.With(prometheus.Labels{
 				metrics.ReasonLabel:      reason,
 				metrics.ProvisionerLabel: candidate.provisioner.Name,
 			}).Inc()
@@ -180,14 +180,14 @@ func (c *Controller) executeCommand(ctx context.Context, d Deprovisioner, comman
 	// We wait for nodes to delete to ensure we don't start another round of deprovisioning until this node is fully
 	// deleted.
 	for _, oldCandidate := range command.candidates {
-		c.waitForDeletion(ctx, oldCandidate.Machine)
+		c.waitForDeletion(ctx, oldCandidate.Node)
 	}
 	return nil
 }
 
-// launchReplacementMachines launches replacement machines and blocks until it is ready
+// launchReplacementNodes launches replacement machines and blocks until it is ready
 // nolint:gocyclo
-func (c *Controller) launchReplacementMachines(ctx context.Context, action Command, reason string) error {
+func (c *Controller) launchReplacementNodes(ctx context.Context, action Command, reason string) error {
 	defer metrics.Measure(deprovisioningReplacementNodeInitializedHistogram)()
 	candidateNodeNames := lo.Map(action.candidates, func(c *Candidate, _ int) string { return c.Node.Name })
 
@@ -196,25 +196,25 @@ func (c *Controller) launchReplacementMachines(ctx context.Context, action Comma
 		return fmt.Errorf("cordoning nodes, %w", err)
 	}
 
-	machineNames, err := c.provisioner.LaunchMachines(ctx, action.replacements, provisioning.WithReason(reason))
+	nodeNames, err := c.provisioner.LaunchMachines(ctx, action.replacements, provisioning.WithReason(reason))
 	if err != nil {
 		// uncordon the nodes as the launch may fail (e.g. ICE)
 		err = multierr.Append(err, c.setNodesUnschedulable(ctx, false, candidateNodeNames...))
 		return err
 	}
-	if len(machineNames) != len(action.replacements) {
+	if len(nodeNames) != len(action.replacements) {
 		// shouldn't ever occur since a partially failed LaunchMachines should return an error
-		return fmt.Errorf("expected %d machines, got %d", len(action.replacements), len(machineNames))
+		return fmt.Errorf("expected %d machines, got %d", len(action.replacements), len(nodeNames))
 	}
 
 	// We have the new machines created at the API server so mark the old machines for deletion
 	c.cluster.MarkForDeletion(candidateNodeNames...)
 
-	errs := make([]error, len(machineNames))
-	workqueue.ParallelizeUntil(ctx, len(machineNames), len(machineNames), func(i int) {
+	errs := make([]error, len(nodeNames))
+	workqueue.ParallelizeUntil(ctx, len(nodeNames), len(nodeNames), func(i int) {
 		// machine never became ready or the machines that we tried to launch got Insufficient Capacity or some
 		// other transient error
-		errs[i] = c.waitForReadiness(ctx, action, machineNames[i])
+		errs[i] = c.waitForReadiness(ctx, action, nodeNames[i])
 	})
 	if err = multierr.Combine(errs...); err != nil {
 		c.cluster.UnmarkForDeletion(candidateNodeNames...)
@@ -230,8 +230,8 @@ func (c *Controller) waitForReadiness(ctx context.Context, action Command, name
 	var once sync.Once
 	pollStart := time.Now()
 	return retry.Do(func() error {
-		machine := &v1alpha5.Machine{}
-		if err := c.kubeClient.Get(ctx, types.NamespacedName{Name: name}, machine); err != nil {
+		node := &v1.Node{}
+		if err := c.kubeClient.Get(ctx, types.NamespacedName{Name: name}, node); err != nil {
 			// If the machine was deleted after a few seconds (to give the cache time to update), then we assume
 			// that the machine was deleted due to an Insufficient Capacity error
 			if errors.IsNotFound(err) && c.clock.Since(pollStart) > time.Second*5 {
@@ -240,12 +240,12 @@ func (c *Controller) waitForReadiness(ctx context.Context, action Command, name
 			return fmt.Errorf("getting machine, %w", err)
 		}
 		once.Do(func() {
-			c.recorder.Publish(deprovisioningevents.Launching(machine, action.String()))
+			c.recorder.Publish(deprovisioningevents.Launching(node, action.String()))
 		})
-		if !machine.StatusConditions().GetCondition(v1alpha5.MachineInitialized).IsTrue() {
+		if _, ok := node.Labels[v1alpha5.LabelNodeInitialized]; !ok {
 			// make the user aware of why deprovisioning is paused
-			c.recorder.Publish(deprovisioningevents.WaitingOnReadiness(machine))
-			return fmt.Errorf("machine is not initialized")
+			c.recorder.Publish(deprovisioningevents.WaitingOnReadiness(node))
+			return fmt.Errorf("node is not initialized")
 		}
 		return nil
 	}, waitRetryOptions...)
@@ -254,21 +254,21 @@ func (c *Controller) waitForReadiness(ctx context.Context, action Command, name
 // waitForDeletion waits for the specified machine to be removed from the API server. This deletion can take some period
 // of time if there are PDBs that govern pods on the machine as we need to wait until the node drains before
 // it's actually deleted.
-func (c *Controller) waitForDeletion(ctx context.Context, machine *v1alpha5.Machine) {
+func (c *Controller) waitForDeletion(ctx context.Context, node *v1.Node) {
 	if err := retry.Do(func() error {
-		m := &v1alpha5.Machine{}
-		nerr := c.kubeClient.Get(ctx, client.ObjectKeyFromObject(machine), m)
+		m := &v1.Node{}
+		nerr := c.kubeClient.Get(ctx, client.ObjectKeyFromObject(node), m)
 		// We expect the not machine found error, at which point we know the machine is deleted.
 		if errors.IsNotFound(nerr) {
 			return nil
 		}
 		// make the user aware of why deprovisioning is paused
-		c.recorder.Publish(deprovisioningevents.WaitingOnDeletion(machine))
+		c.recorder.Publish(deprovisioningevents.WaitingOnDeletion(node))
 		if nerr != nil {
-			return fmt.Errorf("expected machine to be not found, %w", nerr)
+			return fmt.Errorf("expected node to be not found, %w", nerr)
 		}
 		// the machine still exists
-		return fmt.Errorf("expected machine to be not found")
+		return fmt.Errorf("expected node to be not found")
 	}, waitRetryOptions...,
 	); err != nil {
 		logging.FromContext(ctx).Errorf("Waiting on machine deletion, %s", err)

pkg/controllers/deprovisioning/drift.go

@@ -66,15 +66,15 @@ func (d *Drift) ComputeCommand(ctx context.Context, candidates ...*Candidate) (C
 	}
 	// filter out machines that can't be terminated
 	candidates = lo.Filter(candidates, func(cn *Candidate, _ int) bool {
-		if !cn.Machine.DeletionTimestamp.IsZero() {
+		if !cn.Node.DeletionTimestamp.IsZero() {
 			return false
 		}
 		if pdb, ok := pdbs.CanEvictPods(cn.pods); !ok {
-			d.recorder.Publish(deprovisioningevents.Blocked(cn.Node, cn.Machine, fmt.Sprintf("pdb %s prevents pod evictions", pdb))...)
+			d.recorder.Publish(deprovisioningevents.Blocked(cn.Node, fmt.Sprintf("pdb %s prevents pod evictions", pdb))...)
 			return false
 		}
 		if p, ok := hasDoNotEvictPod(cn); ok {
-			d.recorder.Publish(deprovisioningevents.Blocked(cn.Node, cn.Machine,
+			d.recorder.Publish(deprovisioningevents.Blocked(cn.Node,
 				fmt.Sprintf("pod %s/%s has do not evict annotation", p.Namespace, p.Name))...)
 			return false
 		}
@@ -93,7 +93,7 @@ func (d *Drift) ComputeCommand(ctx context.Context, candidates ...*Candidate) (C
 		}
 		// Log when all pods can't schedule, as the command will get executed immediately.
 		if !allPodsScheduled {
-			logging.FromContext(ctx).With("machine", candidate.Machine.Name, "node", candidate.Node.Name).Debug("Continuing to terminate drifted machine after scheduling simulation failed to schedule all pods")
+			logging.FromContext(ctx).With("node", candidate.Node.Name).Debug("Continuing to terminate drifted machine after scheduling simulation failed to schedule all pods")
 		}
 		if len(newMachines) == 0 {
 			return Command{