Merge pull request kubernetes#70344 from andrewsykim/consolidate-node-delete

consolidate node deletion logic between kube-controller-manager and cloud-controller-manager

Author: k8s-ci-robot

cmd/cloud-controller-manager/app/controllermanager.go

@@ -226,12 +226,23 @@ func startControllers(c *cloudcontrollerconfig.CompletedConfig, stop <-chan stru
 	nodeController := cloudcontrollers.NewCloudNodeController(
 		c.SharedInformers.Core().V1().Nodes(),
 		client("cloud-node-controller"), cloud,
-		c.ComponentConfig.KubeCloudShared.NodeMonitorPeriod.Duration,
 		c.ComponentConfig.NodeStatusUpdateFrequency.Duration)
 
-	nodeController.Run(stop)
+	go nodeController.Run(stop)
 	time.Sleep(wait.Jitter(c.ComponentConfig.Generic.ControllerStartInterval.Duration, ControllerStartJitter))
 
+	cloudNodeLifecycleController, err := cloudcontrollers.NewCloudNodeLifecycleController(
+		c.SharedInformers.Core().V1().Nodes(),
+		client("cloud-node-lifecycle-controller"), cloud,
+		c.ComponentConfig.KubeCloudShared.NodeMonitorPeriod.Duration,
+	)
+	if err != nil {
+		klog.Errorf("failed to start cloud node lifecycle controller: %s", err)
+	} else {
+		go cloudNodeLifecycleController.Run(stop)
+		time.Sleep(wait.Jitter(c.ComponentConfig.Generic.ControllerStartInterval.Duration, ControllerStartJitter))
+	}
+
 	// Start the PersistentVolumeLabelController
 	pvlController := cloudcontrollers.NewPersistentVolumeLabelController(client("pvl-controller"), cloud)
 	go pvlController.Run(5, stop)

cmd/kube-controller-manager/app/BUILD

@@ -46,6 +46,7 @@ go_library(
         "//pkg/controller/certificates/cleaner:go_default_library",
         "//pkg/controller/certificates/rootcacertpublisher:go_default_library",
         "//pkg/controller/certificates/signer:go_default_library",
+        "//pkg/controller/cloud:go_default_library",
         "//pkg/controller/clusterroleaggregation:go_default_library",
         "//pkg/controller/cronjob:go_default_library",
         "//pkg/controller/daemon:go_default_library",

cmd/kube-controller-manager/app/controllermanager.go

@@ -383,13 +383,13 @@ func NewControllerInitializers(loopMode ControllerLoopMode) map[string]InitFunc
 	controllers["bootstrapsigner"] = startBootstrapSignerController
 	controllers["tokencleaner"] = startTokenCleanerController
 	controllers["nodeipam"] = startNodeIpamController
+	controllers["nodelifecycle"] = startNodeLifecycleController
 	if loopMode == IncludeCloudLoops {
 		controllers["service"] = startServiceController
 		controllers["route"] = startRouteController
+		controllers["cloudnodelifecycle"] = startCloudNodeLifecycleController
 		// TODO: volume controller into the IncludeCloudLoops only set.
-		// TODO: Separate cluster in cloud check from node lifecycle controller.
 	}
-	controllers["nodelifecycle"] = startNodeLifecycleController
 	controllers["persistentvolume-binder"] = startPersistentVolumeBinderController
 	controllers["attachdetach"] = startAttachDetachController
 	controllers["persistentvolume-expander"] = startVolumeExpandController

cmd/kube-controller-manager/app/core.go

@@ -38,6 +38,7 @@ import (
 	clientset "k8s.io/client-go/kubernetes"
 	csiclientset "k8s.io/csi-api/pkg/client/clientset/versioned"
 	"k8s.io/kubernetes/pkg/controller"
+	cloudcontroller "k8s.io/kubernetes/pkg/controller/cloud"
 	endpointcontroller "k8s.io/kubernetes/pkg/controller/endpoint"
 	"k8s.io/kubernetes/pkg/controller/garbagecollector"
 	namespacecontroller "k8s.io/kubernetes/pkg/controller/namespace"
@@ -125,7 +126,6 @@ func startNodeLifecycleController(ctx ControllerContext) (http.Handler, bool, er
 		ctx.InformerFactory.Core().V1().Pods(),
 		ctx.InformerFactory.Core().V1().Nodes(),
 		ctx.InformerFactory.Extensions().V1beta1().DaemonSets(),
-		ctx.Cloud,
 		ctx.ClientBuilder.ClientOrDie("node-controller"),
 		ctx.ComponentConfig.KubeCloudShared.NodeMonitorPeriod.Duration,
 		ctx.ComponentConfig.NodeLifecycleController.NodeStartupGracePeriod.Duration,
@@ -146,6 +146,24 @@ func startNodeLifecycleController(ctx ControllerContext) (http.Handler, bool, er
 	return nil, true, nil
 }
 
+func startCloudNodeLifecycleController(ctx ControllerContext) (http.Handler, bool, error) {
+	cloudNodeLifecycleController, err := cloudcontroller.NewCloudNodeLifecycleController(
+		ctx.InformerFactory.Core().V1().Nodes(),
+		ctx.ClientBuilder.ClientOrDie("cloud-node-lifecycle-controller"),
+		ctx.Cloud,
+		ctx.ComponentConfig.KubeCloudShared.NodeMonitorPeriod.Duration,
+	)
+	if err != nil {
+		// the controller manager should continue to run if the "Instances" interface is not
+		// supported, though it's unlikely for a cloud provider to not support it
+		klog.Errorf("failed to start cloud node lifecycle controller: %v", err)
+		return nil, false, nil
+	}
+
+	go cloudNodeLifecycleController.Run(ctx.Stop)
+	return nil, true, nil
+}
+
 func startRouteController(ctx ControllerContext) (http.Handler, bool, error) {
 	if !ctx.ComponentConfig.KubeCloudShared.AllocateNodeCIDRs || !ctx.ComponentConfig.KubeCloudShared.ConfigureCloudRoutes {
 		klog.Infof("Will not configure cloud provider routes for allocate-node-cidrs: %v, configure-cloud-routes: %v.", ctx.ComponentConfig.KubeCloudShared.AllocateNodeCIDRs, ctx.ComponentConfig.KubeCloudShared.ConfigureCloudRoutes)

pkg/controller/BUILD

@@ -55,7 +55,6 @@ go_library(
         "//pkg/apis/core:go_default_library",
         "//pkg/apis/core/install:go_default_library",
         "//pkg/apis/core/validation:go_default_library",
-        "//pkg/scheduler/api:go_default_library",
         "//pkg/serviceaccount:go_default_library",
         "//pkg/util/hash:go_default_library",
         "//pkg/util/taints:go_default_library",

pkg/controller/cloud/BUILD

@@ -10,14 +10,14 @@ go_library(
     name = "go_default_library",
     srcs = [
         "node_controller.go",
+        "node_lifecycle_controller.go",
         "pvlcontroller.go",
     ],
     importpath = "k8s.io/kubernetes/pkg/controller/cloud",
     deps = [
         "//pkg/api/v1/node:go_default_library",
         "//pkg/apis/core/v1/helper:go_default_library",
         "//pkg/controller:go_default_library",
-        "//pkg/controller/util/node:go_default_library",
         "//pkg/features:go_default_library",
         "//pkg/kubelet/apis:go_default_library",
         "//pkg/scheduler/api:go_default_library",
@@ -26,6 +26,7 @@ go_library(
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/labels:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
@@ -51,6 +52,7 @@ go_test(
     name = "go_default_test",
     srcs = [
         "node_controller_test.go",
+        "node_lifecycle_controller_test.go",
         "pvlcontroller_test.go",
     ],
     embed = [":go_default_library"],
@@ -67,10 +69,10 @@ go_test(
         "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
-        "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/util/feature/testing:go_default_library",
         "//staging/src/k8s.io/client-go/informers:go_default_library",
+        "//staging/src/k8s.io/client-go/informers/core/v1:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes/fake:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes/scheme:go_default_library",
         "//staging/src/k8s.io/client-go/testing:go_default_library",

pkg/controller/cloud/node_controller.go

@@ -37,9 +37,6 @@ import (
 	"k8s.io/client-go/tools/record"
 	clientretry "k8s.io/client-go/util/retry"
 	cloudprovider "k8s.io/cloud-provider"
-	nodeutilv1 "k8s.io/kubernetes/pkg/api/v1/node"
-	"k8s.io/kubernetes/pkg/controller"
-	nodectrlutil "k8s.io/kubernetes/pkg/controller/util/node"
 	kubeletapis "k8s.io/kubernetes/pkg/kubelet/apis"
 	schedulerapi "k8s.io/kubernetes/pkg/scheduler/api"
 	nodeutil "k8s.io/kubernetes/pkg/util/node"
@@ -58,11 +55,6 @@ type CloudNodeController struct {
 
 	cloud cloudprovider.Interface
 
-	// Value controlling NodeController monitoring period, i.e. how often does NodeController
-	// check node status posted from kubelet. This value should be lower than nodeMonitorGracePeriod
-	// set in controller-manager
-	nodeMonitorPeriod time.Duration
-
 	nodeStatusUpdateFrequency time.Duration
 }
 
@@ -79,7 +71,6 @@ func NewCloudNodeController(
 	nodeInformer coreinformers.NodeInformer,
 	kubeClient clientset.Interface,
 	cloud cloudprovider.Interface,
-	nodeMonitorPeriod time.Duration,
 	nodeStatusUpdateFrequency time.Duration) *CloudNodeController {
 
 	eventBroadcaster := record.NewBroadcaster()
@@ -97,7 +88,6 @@ func NewCloudNodeController(
 		kubeClient:                kubeClient,
 		recorder:                  recorder,
 		cloud:                     cloud,
-		nodeMonitorPeriod:         nodeMonitorPeriod,
 		nodeStatusUpdateFrequency: nodeStatusUpdateFrequency,
 	}
 
@@ -111,8 +101,9 @@ func NewCloudNodeController(
 	return cnc
 }
 
-// This controller deletes a node if kubelet is not reporting
-// and the node is gone from the cloud provider.
+// This controller updates newly registered nodes with information
+// from the cloud provider. This call is blocking so should be called
+// via a goroutine
 func (cnc *CloudNodeController) Run(stopCh <-chan struct{}) {
 	defer utilruntime.HandleCrash()
 
@@ -121,10 +112,7 @@ func (cnc *CloudNodeController) Run(stopCh <-chan struct{}) {
 	// very infrequently. DO NOT MODIFY this to perform frequent operations.
 
 	// Start a loop to periodically update the node addresses obtained from the cloud
-	go wait.Until(cnc.UpdateNodeStatus, cnc.nodeStatusUpdateFrequency, stopCh)
-
-	// Start a loop to periodically check if any nodes have been deleted from cloudprovider
-	go wait.Until(cnc.MonitorNode, cnc.nodeMonitorPeriod, stopCh)
+	wait.Until(cnc.UpdateNodeStatus, cnc.nodeStatusUpdateFrequency, stopCh)
 }
 
 // UpdateNodeStatus updates the node status, such as node addresses
@@ -210,108 +198,6 @@ func (cnc *CloudNodeController) updateNodeAddress(node *v1.Node, instances cloud
 	}
 }
 
-// Monitor node queries the cloudprovider for non-ready nodes and deletes them
-// if they cannot be found in the cloud provider
-func (cnc *CloudNodeController) MonitorNode() {
-	instances, ok := cnc.cloud.Instances()
-	if !ok {
-		utilruntime.HandleError(fmt.Errorf("failed to get instances from cloud provider"))
-		return
-	}
-
-	nodes, err := cnc.kubeClient.CoreV1().Nodes().List(metav1.ListOptions{ResourceVersion: "0"})
-	if err != nil {
-		klog.Errorf("Error monitoring node status: %v", err)
-		return
-	}
-
-	for i := range nodes.Items {
-		var currentReadyCondition *v1.NodeCondition
-		node := &nodes.Items[i]
-		// Try to get the current node status
-		// If node status is empty, then kubelet has not posted ready status yet. In this case, process next node
-		for rep := 0; rep < nodeStatusUpdateRetry; rep++ {
-			_, currentReadyCondition = nodeutilv1.GetNodeCondition(&node.Status, v1.NodeReady)
-			if currentReadyCondition != nil {
-				break
-			}
-			name := node.Name
-			node, err = cnc.kubeClient.CoreV1().Nodes().Get(name, metav1.GetOptions{})
-			if err != nil {
-				klog.Errorf("Failed while getting a Node to retry updating NodeStatus. Probably Node %s was deleted.", name)
-				break
-			}
-			time.Sleep(retrySleepTime)
-		}
-		if currentReadyCondition == nil {
-			klog.Errorf("Update status of Node %v from CloudNodeController exceeds retry count or the Node was deleted.", node.Name)
-			continue
-		}
-		// If the known node status says that Node is NotReady, then check if the node has been removed
-		// from the cloud provider. If node cannot be found in cloudprovider, then delete the node immediately
-		if currentReadyCondition != nil {
-			if currentReadyCondition.Status != v1.ConditionTrue {
-				// we need to check this first to get taint working in similar in all cloudproviders
-				// current problem is that shutdown nodes are not working in similar way ie. all cloudproviders
-				// does not delete node from kubernetes cluster when instance it is shutdown see issue #46442
-				shutdown, err := nodectrlutil.ShutdownInCloudProvider(context.TODO(), cnc.cloud, node)
-				if err != nil {
-					klog.Errorf("Error checking if node %s is shutdown: %v", node.Name, err)
-				}
-
-				if shutdown && err == nil {
-					// if node is shutdown add shutdown taint
-					err = controller.AddOrUpdateTaintOnNode(cnc.kubeClient, node.Name, controller.ShutdownTaint)
-					if err != nil {
-						klog.Errorf("Error patching node taints: %v", err)
-					}
-					// Continue checking the remaining nodes since the current one is shutdown.
-					continue
-				}
-
-				// Check with the cloud provider to see if the node still exists. If it
-				// doesn't, delete the node immediately.
-				exists, err := ensureNodeExistsByProviderID(instances, node)
-				if err != nil {
-					klog.Errorf("Error checking if node %s exists: %v", node.Name, err)
-					continue
-				}
-
-				if exists {
-					// Continue checking the remaining nodes since the current one is fine.
-					continue
-				}
-
-				klog.V(2).Infof("Deleting node since it is no longer present in cloud provider: %s", node.Name)
-
-				ref := &v1.ObjectReference{
-					Kind:      "Node",
-					Name:      node.Name,
-					UID:       types.UID(node.UID),
-					Namespace: "",
-				}
-				klog.V(2).Infof("Recording %s event message for node %s", "DeletingNode", node.Name)
-
-				cnc.recorder.Eventf(ref, v1.EventTypeNormal, fmt.Sprintf("Deleting Node %v because it's not present according to cloud provider", node.Name), "Node %s event: %s", node.Name, "DeletingNode")
-
-				go func(nodeName string) {
-					defer utilruntime.HandleCrash()
-					if err := cnc.kubeClient.CoreV1().Nodes().Delete(nodeName, nil); err != nil {
-						klog.Errorf("unable to delete node %q: %v", nodeName, err)
-					}
-				}(node.Name)
-
-			} else {
-				// if taint exist remove taint
-				err = controller.RemoveTaintOffNode(cnc.kubeClient, node.Name, node, controller.ShutdownTaint)
-				if err != nil {
-					klog.Errorf("Error patching node taints: %v", err)
-				}
-			}
-		}
-	}
-}
-
 func (cnc *CloudNodeController) UpdateCloudNode(_, newObj interface{}) {
 	if _, ok := newObj.(*v1.Node); !ok {
 		utilruntime.HandleError(fmt.Errorf("unexpected object type: %v", newObj))