Revert "Revert "fix(platform): delete machine failed & move machine operation to provider (#1435)" (#1451)"

This reverts commit cad879c.

Author: leoryu

pkg/platform/controller/cluster/cluster_controller.go

@@ -152,7 +152,7 @@ func (c *Controller) needsUpdate(old *platformv1.Cluster, new *platformv1.Cluste
 		return true
 	}
 
-	if old.Status.Phase == platformv1.ClusterRunning && new.Status.Phase == platformv1.ClusterTerminating {
+	if old.Status.Phase != platformv1.ClusterTerminating && new.Status.Phase == platformv1.ClusterTerminating {
 		return true
 	}
 
@@ -281,7 +281,7 @@ func (c *Controller) reconcile(ctx context.Context, key string, cluster *platfor
 		log.FromContext(ctx).Info("Cluster has been terminated. Attempting to cleanup resources")
 		err = c.deleter.Delete(ctx, key)
 		if err == nil {
-			log.FromContext(ctx).Info("Machine has been successfully deleted")
+			log.FromContext(ctx).Info("Cluster has been successfully deleted")
 		}
 	default:
 		log.FromContext(ctx).Info("unknown cluster phase", "status.phase", cluster.Status.Phase)

pkg/platform/controller/cluster/deletion/cluster_deleter.go

@@ -24,9 +24,9 @@ import (
 
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/fields"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/sets"
+
 	v1clientset "tkestack.io/tke/api/client/clientset/versioned/typed/platform/v1"
 	platformv1 "tkestack.io/tke/api/platform/v1"
 	clusterprovider "tkestack.io/tke/pkg/platform/provider/cluster"
@@ -244,7 +244,6 @@ var deleteResourceFuncs = []deleteResourceFunc{
 	deleteIPAM,
 	deleteTappControllers,
 	deleteClusterProvider,
-	deleteMachine,
 }
 
 // deleteAllContent will use the client to delete each resource identified in cluster.
@@ -433,29 +432,3 @@ func deleteClusterCredential(ctx context.Context, deleter *clusterDeleter, clust
 	return nil
 }
 */
-
-func deleteMachine(ctx context.Context, deleter *clusterDeleter, cluster *platformv1.Cluster) error {
-	log.FromContext(ctx).Info("deleteMachine doing")
-
-	fieldSelector := fields.OneTermEqualSelector("spec.clusterName", cluster.Name).String()
-	machineList, err := deleter.platformClient.Machines().List(ctx, metav1.ListOptions{FieldSelector: fieldSelector})
-	if err != nil {
-		return err
-	}
-	if len(machineList.Items) == 0 {
-		return nil
-	}
-	background := metav1.DeletePropagationForeground
-	deleteOpt := metav1.DeleteOptions{PropagationPolicy: &background}
-	for _, machine := range machineList.Items {
-		if err := deleter.platformClient.Machines().Delete(ctx, machine.Name, deleteOpt); err != nil {
-			if !errors.IsNotFound(err) {
-				return err
-			}
-		}
-	}
-
-	log.FromContext(ctx).Info("deleteMachine done")
-
-	return nil
-}

pkg/platform/controller/machine/deletion/machine_deleter.go

@@ -26,12 +26,11 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/sets"
+
 	v1clientset "tkestack.io/tke/api/client/clientset/versioned/typed/platform/v1"
-	platformv1 "tkestack.io/tke/api/platform/v1"
 	v1 "tkestack.io/tke/api/platform/v1"
 	clusterprovider "tkestack.io/tke/pkg/platform/provider/cluster"
 	machineprovider "tkestack.io/tke/pkg/platform/provider/machine"
-	"tkestack.io/tke/pkg/util/apiclient"
 	"tkestack.io/tke/pkg/util/log"
 )
 
@@ -101,7 +100,7 @@ func (d *machineDeleter) Delete(ctx context.Context, name string) error {
 
 	// ensure that the status is up to date on the machine
 	// if we get a not found error, we assume the machine is truly gone
-	machine, err = d.retryOnConflictError(machine, d.updateMachineStatusFunc)
+	machine, err = d.retryOnConflictError(ctx, machine, d.updateMachineStatusFunc)
 	if err != nil {
 		if errors.IsNotFound(err) {
 			return nil
@@ -126,7 +125,7 @@ func (d *machineDeleter) Delete(ctx context.Context, name string) error {
 	}
 
 	// we have removed content, so mark it finalized by us
-	machine, err = d.retryOnConflictError(machine, d.finalizeMachine)
+	machine, err = d.retryOnConflictError(ctx, machine, d.finalizeMachine)
 	if err != nil {
 		// in normal practice, this should not be possible, but if a deployment is running
 		// two controllers to do machine deletion that share a common finalizer token it's
@@ -159,15 +158,15 @@ func (d *machineDeleter) deleteMachine(machine *v1.Machine) error {
 }
 
 // updateMachineFunc is a function that makes an update to a namespace
-type updateMachineFunc func(machine *v1.Machine) (*v1.Machine, error)
+type updateMachineFunc func(ctx context.Context, machine *v1.Machine) (*v1.Machine, error)
 
 // retryOnConflictError retries the specified fn if there was a conflict error
 // it will return an error if the UID for an object changes across retry operations.
 // TODO RetryOnConflict should be a generic concept in client code
-func (d *machineDeleter) retryOnConflictError(machine *v1.Machine, fn updateMachineFunc) (result *v1.Machine, err error) {
+func (d *machineDeleter) retryOnConflictError(ctx context.Context, machine *v1.Machine, fn updateMachineFunc) (result *v1.Machine, err error) {
 	latestMachine := machine
 	for {
-		result, err = fn(latestMachine)
+		result, err = fn(ctx, latestMachine)
 		if err == nil {
 			return result, nil
 		}
@@ -186,7 +185,7 @@ func (d *machineDeleter) retryOnConflictError(machine *v1.Machine, fn updateMach
 }
 
 // updateMachineStatusFunc will verify that the status of the machine is correct
-func (d *machineDeleter) updateMachineStatusFunc(machine *v1.Machine) (*v1.Machine, error) {
+func (d *machineDeleter) updateMachineStatusFunc(ctx context.Context, machine *v1.Machine) (*v1.Machine, error) {
 	if machine.DeletionTimestamp.IsZero() || machine.Status.Phase == v1.MachineTerminating {
 		return machine, nil
 	}
@@ -203,7 +202,7 @@ func finalized(machine *v1.Machine) bool {
 }
 
 // finalizeMachine removes the specified finalizerToken and finalizes the machine
-func (d *machineDeleter) finalizeMachine(machine *v1.Machine) (*v1.Machine, error) {
+func (d *machineDeleter) finalizeMachine(ctx context.Context, machine *v1.Machine) (*v1.Machine, error) {
 	machineFinalize := v1.Machine{}
 	machineFinalize.ObjectMeta = machine.ObjectMeta
 	machineFinalize.Spec = machine.Spec
@@ -225,7 +224,7 @@ func (d *machineDeleter) finalizeMachine(machine *v1.Machine) (*v1.Machine, erro
 		Name(machineFinalize.Name).
 		SubResource("finalize").
 		Body(&machineFinalize).
-		Do(context.Background()).
+		Do(ctx).
 		Into(machine)
 
 	if err != nil {
@@ -241,7 +240,6 @@ type deleteResourceFunc func(ctx context.Context, deleter *machineDeleter, machi
 
 var deleteResourceFuncs = []deleteResourceFunc{
 	deleteMachineProvider,
-	deleteNode,
 }
 
 // deleteAllContent will use the client to delete each resource identified in machine.
@@ -287,39 +285,3 @@ func deleteMachineProvider(ctx context.Context, deleter *machineDeleter, machine
 
 	return nil
 }
-
-func deleteNode(ctx context.Context, deleter *machineDeleter, machine *v1.Machine) error {
-	log.FromContext(ctx).Info("deleteNode doing")
-
-	cluster, err := clusterprovider.GetV1ClusterByName(context.Background(), deleter.platformClient, machine.Spec.ClusterName, clusterprovider.AdminUsername)
-	if err != nil {
-		return err
-	}
-	if cluster.Status.Phase == platformv1.ClusterTerminating {
-		return nil
-	}
-	clientset, err := cluster.Clientset()
-	if err != nil {
-		return err
-	}
-
-	node, err := apiclient.GetNodeByMachineIP(ctx, clientset, machine.Spec.IP)
-	if err != nil {
-		if !errors.IsNotFound(err) {
-			return err
-		}
-		log.FromContext(ctx).Info("deleteNode done")
-		return nil
-	}
-
-	err = clientset.CoreV1().Nodes().Delete(context.Background(), node.Name, metav1.DeleteOptions{})
-	if err != nil {
-		if !errors.IsNotFound(err) {
-			return err
-		}
-	}
-
-	log.FromContext(ctx).Info("deleteNode done")
-
-	return nil
-}

pkg/platform/controller/machine/machine_controller.go

@@ -34,6 +34,7 @@ import (
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/retry"
 	"k8s.io/client-go/util/workqueue"
+
 	platformversionedclient "tkestack.io/tke/api/client/clientset/versioned/typed/platform/v1"
 	platformv1informer "tkestack.io/tke/api/client/informers/externalversions/platform/v1"
 	platformv1lister "tkestack.io/tke/api/client/listers/platform/v1"
@@ -123,6 +124,10 @@ func (c *Controller) needsUpdate(oldMachine *platformv1.Machine, newMachine *pla
 		return true
 	}
 
+	if oldMachine.Status.Phase != platformv1.MachineTerminating && newMachine.Status.Phase == platformv1.MachineTerminating {
+		return true
+	}
+
 	// Control the synchronization interval through the health detection interval
 	// to avoid version conflicts caused by concurrent modification
 	healthCondition := newMachine.GetCondition(conditionTypeHealthCheck)

pkg/platform/provider/baremetal/cluster/delete.go

@@ -20,9 +20,14 @@ package cluster
 
 import (
 	"context"
+	"time"
 
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/util/wait"
+
+	platformv1client "tkestack.io/tke/api/client/clientset/versioned/typed/platform/v1"
 	"tkestack.io/tke/pkg/platform/provider/baremetal/phases/kubeadm"
 	"tkestack.io/tke/pkg/platform/provider/util/mark"
 	typesv1 "tkestack.io/tke/pkg/platform/types/v1"
@@ -75,3 +80,44 @@ func (p *Provider) EnsureRemoveNode(ctx context.Context, c *v1.Cluster) error {
 	}
 	return nil
 }
+
+func (p *Provider) EnsureRemoveMachine(ctx context.Context, c *v1.Cluster) error {
+	log.FromContext(ctx).Info("delete machine start")
+	fieldSelector := fields.OneTermEqualSelector("spec.clusterName", c.Name).String()
+	machineList, err := p.platformClient.Machines().List(ctx, metav1.ListOptions{FieldSelector: fieldSelector})
+	if err != nil {
+		return err
+	}
+	if len(machineList.Items) == 0 {
+		return nil
+	}
+	for _, machine := range machineList.Items {
+		if err := p.platformClient.Machines().Delete(ctx, machine.Name, metav1.DeleteOptions{}); err != nil {
+			if errors.IsNotFound(err) {
+				return nil
+			}
+			return err
+		}
+
+		if err = wait.PollImmediate(5*time.Second, 5*time.Minute, waitForMachineDelete(ctx, p.platformClient, machine.Name)); err != nil {
+			return err
+		}
+	}
+
+	log.FromContext(ctx).Info("delete machine done")
+
+	return nil
+}
+
+func waitForMachineDelete(ctx context.Context, c platformv1client.PlatformV1Interface, machineName string) wait.ConditionFunc {
+	return func() (done bool, err error) {
+
+		if _, err := c.Machines().Get(ctx, machineName, metav1.GetOptions{}); err != nil {
+			if errors.IsNotFound(err) {
+				return true, nil
+			}
+		}
+
+		return false, nil
+	}
+}

pkg/platform/provider/baremetal/cluster/provider.go

@@ -154,6 +154,7 @@ func NewProvider() (*Provider, error) {
 			p.EnsureRemoveNode,
 		},
 		DeleteHandlers: []clusterprovider.Handler{
+			p.EnsureRemoveMachine,
 			p.EnsureCleanClusterMark,
 		},
 	}

pkg/platform/provider/baremetal/machine/delete.go

@@ -0,0 +1,59 @@
+package machine
+
+import (
+	"context"
+	"time"
+
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/kubernetes"
+
+	platformv1 "tkestack.io/tke/api/platform/v1"
+	typesv1 "tkestack.io/tke/pkg/platform/types/v1"
+	"tkestack.io/tke/pkg/util/apiclient"
+	"tkestack.io/tke/pkg/util/log"
+)
+
+func (p *Provider) EnsureRemoveNode(ctx context.Context, machine *platformv1.Machine, cluster *typesv1.Cluster) error {
+	log.FromContext(ctx).Info("deleteNode doing")
+
+	if cluster.Status.Phase == platformv1.ClusterTerminating {
+		return nil
+	}
+
+	clientset, err := cluster.Clientset()
+	if err != nil {
+		return err
+	}
+
+	node, err := apiclient.GetNodeByMachineIP(ctx, clientset, machine.Spec.IP)
+	if err != nil {
+		return err
+	}
+	err = clientset.CoreV1().Nodes().Delete(ctx, node.Name, metav1.DeleteOptions{})
+	if err != nil {
+		if errors.IsNotFound(err) {
+			return nil
+		}
+		return err
+	}
+	if err = wait.PollImmediate(5*time.Second, 5*time.Minute, waitForNodeDelete(ctx, clientset, node.Name)); err != nil {
+		return err
+	}
+
+	log.FromContext(ctx).Info("deleteNode done")
+	return nil
+}
+
+func waitForNodeDelete(ctx context.Context, c kubernetes.Interface, nodeName string) wait.ConditionFunc {
+	return func() (done bool, err error) {
+		if _, err := c.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{}); err != nil {
+			if errors.IsNotFound(err) {
+				return true, nil
+			}
+		}
+
+		return false, nil
+	}
+}

pkg/platform/provider/baremetal/machine/provider.go

@@ -94,6 +94,9 @@ func NewProvider() (*Provider, error) {
 			p.EnsureUpgrade,
 			p.EnsurePostUpgradeHook,
 		},
+		DeleteHandlers: []machineprovider.Handler{
+			p.EnsureRemoveNode,
+		},
 	}
 
 	cfg, err := config.New(constants.ConfigFile)