kubeedge/cloud/pkg/controllermanager/nodegroup/nodegroupcontroller.go

package nodegroup

import (
	"context"
	"fmt"
	"sort"

	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/equality"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/fields"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/types"
	utilerrors "k8s.io/apimachinery/pkg/util/errors"
	"k8s.io/apimachinery/pkg/util/sets"
	"k8s.io/klog/v2"
	controllerruntime "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/handler"

	appsv1alpha1 "github.com/kubeedge/api/apis/apps/v1alpha1"
)

const (
	// ControllerName is the controller name that will be used when reporting events.
	ControllerName = "nodegroup-controller"

	LabelBelongingTo              = "apps.kubeedge.io/belonging-to"
	NodeGroupControllerFinalizer  = "apps.kubeedge.io/nodegroup-controller"
	ServiceTopologyAnnotation     = "apps.kubeedge.io/service-topology"
	ServiceTopologyRangeNodegroup = "range-nodegroup"
	LabelTopologyZone             = "topology.kubernetes.io/zone"
)

var (
	conditionStatusReadyStatusMap = map[corev1.ConditionStatus]appsv1alpha1.ReadyStatus{
		corev1.ConditionTrue:    appsv1alpha1.NodeReady,
		corev1.ConditionFalse:   appsv1alpha1.NodeNotReady,
		corev1.ConditionUnknown: appsv1alpha1.Unknown,
		// for the convenience of processing the situation that node has no ready condition
		"": appsv1alpha1.Unknown,
	}
)

// Controller is to sync NodeGroup.
type Controller struct {
	client.Client
}

func NewController(cli client.Client) *Controller {
	return &Controller{
		cli,
	}
}

// Reconcile performs a full reconciliation for the object referred to by the Request.
// The Controller will requeue the Request to be processed again if an error is non-nil or
// Result.Requeue is true, otherwise upon completion it will remove the work from the queue.
func (c *Controller) Reconcile(ctx context.Context, req controllerruntime.Request) (controllerruntime.Result, error) {
	klog.Infof("Reconciling nodeGroup %s", req.NamespacedName.Name)

	nodeGroup := &appsv1alpha1.NodeGroup{}
	if err := c.Client.Get(ctx, req.NamespacedName, nodeGroup); err != nil {
		// The resource may no longer exist, in which case we stop processing.
		if apierrors.IsNotFound(err) {
			return controllerruntime.Result{}, nil
		}

		return controllerruntime.Result{Requeue: true}, err
	}

	if !nodeGroup.DeletionTimestamp.IsZero() {
		// remove labels it added to nodes before deleting this NodeGroup
		klog.Infof("begin to remove node group label on nodes selected by nodegroup %s", nodeGroup.Name)
		if err := c.evictNodesInNodegroup(ctx, nodeGroup.Name); err != nil {
			return controllerruntime.Result{Requeue: true}, err
		}
		// this NodeGroup can be deleted now
		if err := c.removeFinalizer(ctx, nodeGroup); err != nil {
			return controllerruntime.Result{Requeue: true}, err
		}
		return controllerruntime.Result{}, nil
	}

	if !controllerutil.ContainsFinalizer(nodeGroup, NodeGroupControllerFinalizer) {
		controllerutil.AddFinalizer(nodeGroup, NodeGroupControllerFinalizer)
		if err := c.Client.Update(ctx, nodeGroup); err != nil {
			klog.Errorf("failed to add finalizer for nodegroup %s, %s", nodeGroup.Name, err)
			return controllerruntime.Result{Requeue: true}, err
		}
	}

	return c.syncNodeGroup(ctx, nodeGroup)
}

func (c *Controller) syncNodeGroup(ctx context.Context, nodeGroup *appsv1alpha1.NodeGroup) (controllerruntime.Result, error) {
	debugLogNodes := func(msg string, nodes []corev1.Node) {
		if klog.V(4).Enabled() {
			if len(nodes) == 0 {
				klog.Infof("%s: get no nodes when syncing nodegroup %s", msg, nodeGroup.Name)
				return
			}
			nodeNames := []string{}
			for i := range nodes {
				nodeNames = append(nodeNames, nodes[i].Name)
			}
			klog.Infof("%s: get %d nodes %v when syncing nodegroup %s", msg, len(nodes), nodeNames, nodeGroup.Name)
		}
	}

	newNodes, err := c.getNodesSelectedBy(ctx, nodeGroup)
	if err != nil {
		klog.Errorf("failed to get all new nodes, %s, continue with what have found.", err)
	}
	debugLogNodes("get new nodes", newNodes)

	oldNodes, err := c.getNodesByLabels(ctx, map[string]string{LabelBelongingTo: nodeGroup.Name})
	if err != nil {
		klog.Errorf("failed to get old nodes for nodegroup %s, %s.", nodeGroup.Name, err)
		return controllerruntime.Result{Requeue: true}, err
	}
	debugLogNodes("get current nodes", oldNodes)

	// delete belonging label on nodes that do not belong to this node group
	nodesDeleted, _ := nodesDiff(oldNodes, newNodes)
	debugLogNodes("get nodes to delete label", nodesDeleted)

	if err := c.evictNodes(ctx, nodesDeleted); err != nil {
		klog.Errorf("failed to evict nodes that do not belong to this nodegroup anymore, %s", err)
		return controllerruntime.Result{Requeue: true}, err
	}

	// This loop will
	// 1. add or update belonging label for nodes
	// 2. prepare NodeStatus for NodeGroup
	nodeStatusList := []appsv1alpha1.NodeStatus{}
	existingNodes := sets.NewString()
	for _, node := range newNodes {
		existingNodes = existingNodes.Insert(node.Name)
		nodeStatus := appsv1alpha1.NodeStatus{
			NodeName: node.Name,
		}
		// update ReadyStatus
		nodeReadyConditionStatus, _ := getNodeReadyConditionFromNode(&node)
		nodeStatus.ReadyStatus = conditionStatusReadyStatusMap[nodeReadyConditionStatus]
		klog.V(4).Infof("get status %s for node %s, when reconciling nodegroup %s", nodeStatus.ReadyStatus, node.Name, nodeGroup.Name)

		// try to add node group label to this node
		if err := c.addOrUpdateNodeLabel(ctx, &node, nodeGroup.Name, nodeGroup.Spec.TopologyEnabled); err != nil {
			klog.Errorf("failed to update belonging label for node %s in nodegroup %s, %s, continue to reconcile other nodes", node.Name, nodeGroup.Name, err)
			nodeStatus.SelectionStatus = appsv1alpha1.FailedSelection
			nodeStatus.SelectionStatusReason = err.Error()
		} else {
			nodeStatus.SelectionStatus = appsv1alpha1.SucceededSelection
		}
		nodeStatusList = append(nodeStatusList, nodeStatus)
	}
	// update status for nodes that do not exist but specified by node name.
	nonExistingNodes := sets.NewString(nodeGroup.Spec.Nodes...).Difference(existingNodes)
	for node := range nonExistingNodes {
		nodeStatusList = append(nodeStatusList, appsv1alpha1.NodeStatus{
			NodeName:              node,
			SelectionStatus:       appsv1alpha1.FailedSelection,
			SelectionStatusReason: "node does not exist",
			ReadyStatus:           appsv1alpha1.Unknown,
		})
	}
	sort.Slice(nodeStatusList, func(i, j int) bool {
		return nodeStatusList[i].NodeName < nodeStatusList[j].NodeName
	})
	if equality.Semantic.DeepEqual(nodeGroup.Status.NodeStatuses, nodeStatusList) {
		klog.V(4).Infof("status of nodegroup is unchanged, skip update")
		return controllerruntime.Result{}, nil
	}
	klog.V(4).Infof("status of nodegroup has changed, old: %v, new: %v", nodeGroup.Status.NodeStatuses, nodeStatusList)
	nodeGroup.Status.NodeStatuses = nodeStatusList
	if err := c.Status().Update(ctx, nodeGroup); err != nil {
		klog.Errorf("failed to update status for nodegroup %s, %s", nodeGroup.Name, err)
		return controllerruntime.Result{Requeue: true}, nil
	}
	return controllerruntime.Result{}, nil
}

// SetupWithManager creates a controller and register to controller manager.
func (c *Controller) SetupWithManager(ctx context.Context, mgr controllerruntime.Manager) error {
	if err := mgr.GetFieldIndexer().IndexField(ctx, &corev1.Pod{}, "spec.nodeName", func(o client.Object) []string {
		pod := o.(*corev1.Pod)
		return []string{pod.Spec.NodeName}
	}); err != nil {
		return fmt.Errorf("failed to set nodeName field selector for manager, %v", err)
	}
	return controllerruntime.NewControllerManagedBy(mgr).
		For(&appsv1alpha1.NodeGroup{}).
		Watches(&corev1.Node{}, handler.EnqueueRequestsFromMapFunc(c.nodeMapFunc)).
		Complete(c)
}

// evictNodes will remove the belonging-to label from nodes and evict pods
// that should run in the nodegroup which the node was used to belong to.
func (c *Controller) evictNodes(ctx context.Context, nodes []corev1.Node) error {
	errs := []error{}
	for _, node := range nodes {
		n := node.DeepCopy()
		ng := n.Labels[LabelBelongingTo]
		delete(n.Labels, LabelBelongingTo)
		delete(n.Labels, LabelTopologyZone)
		if err := c.Client.Patch(ctx, n, client.MergeFrom(&node)); err != nil {
			klog.Errorf("failed to remove belonging label of nodegroup %s on node %s, %v", ng, node.Name, err)
			errs = append(errs, err)
		}

		if err := c.evictPodsShouldNotRunOnNode(ctx, n, ng); err != nil {
			klog.Errorf("failed to evict pods running on node %s in nodegroup %s, %v", node.Name, ng, err)
			errs = append(errs, err)
		}
	}
	return utilerrors.NewAggregate(errs)
}

func (c *Controller) evictPodsShouldNotRunOnNode(ctx context.Context, node *corev1.Node, nodegroup string) error {
	// find all pods running on this node
	runningPods := &corev1.PodList{}
	nodeNameSelector := fields.OneTermEqualSelector("spec.nodeName", node.Name)
	if err := c.Client.List(ctx, runningPods, client.MatchingFieldsSelector{Selector: nodeNameSelector}); err != nil {
		return fmt.Errorf("failed to get pods running on node %s, %v", node.Name, err)
	}

	// evict pods
	errs := []error{}
	for i := range runningPods.Items {
		pod := &runningPods.Items[i]
		nodeSelector := pod.Spec.NodeSelector
		if v, ok := nodeSelector[LabelBelongingTo]; ok && v == nodegroup {
			// TODO: in an async way?
			// Delete pod seems to block until the pod has actually stopped
			klog.V(4).Infof("try to evict pod %s/%s running on node %s", pod.Namespace, pod.Name, node.Name)
			if err := c.Client.Delete(ctx, pod); err != nil {
				errs = append(errs, fmt.Errorf("failed to delete pod %s/%s, %v", pod.Namespace, pod.Name, err))
			}
		}
	}
	return utilerrors.NewAggregate(errs)
}

func (c *Controller) removeFinalizer(ctx context.Context, nodeGroup *appsv1alpha1.NodeGroup) error {
	if !controllerutil.ContainsFinalizer(nodeGroup, NodeGroupControllerFinalizer) {
		return nil
	}
	controllerutil.RemoveFinalizer(nodeGroup, NodeGroupControllerFinalizer)
	if err := c.Client.Update(ctx, nodeGroup); err != nil {
		klog.Errorf("failed to remove finalizer on nodegroup %s, %s", nodeGroup.Name, err)
		return err
	}
	return nil
}

func (c *Controller) getNodesSelectedBy(ctx context.Context, nodeGroup *appsv1alpha1.NodeGroup) ([]corev1.Node, error) {
	errs := []error{}
	nodesByLabel, err := c.getNodesByLabels(ctx, nodeGroup.Spec.MatchLabels)
	if err != nil {
		klog.Errorf("failed to get nodes by MatchLabels %v, %s", nodeGroup.Spec.MatchLabels, err)
		errs = append(errs, err)
	}
	klog.V(4).Infof("get %d nodes that match labels in nodegroup %s", len(nodesByLabel), nodeGroup.Name)

	nodesByName, err := c.getNodesByNodeName(ctx, nodeGroup.Spec.Nodes)
	if err != nil {
		klog.Errorf("failed to get all nodes specified in the NodeGroup.Spec.Nodes, %s.", err)
		errs = append(errs, err)
	}
	klog.V(4).Infof("get %d nodes that specified by name in nodegroup %s", len(nodesByName), nodeGroup.Name)
	// remove duplicate nodes
	return nodesUnion(nodesByLabel, nodesByName), utilerrors.NewAggregate(errs)
}

// We can assume that one node can only be in one of following conditions:
// 1. This node is an orphan, do not and will not belong to any NodeGroup.
// 2. This node is or will be a member of one NodeGroup.
func (c *Controller) nodeMapFunc(_ context.Context, obj client.Object) []controllerruntime.Request {
	node := obj.(*corev1.Node)
	if nodeGroupName, ok := node.Labels[LabelBelongingTo]; ok {
		return []controllerruntime.Request{
			{
				NamespacedName: types.NamespacedName{
					Name: nodeGroupName,
				},
			},
		}
	}
	// node do not have belonging label, either a new node will be added to a node group or an orphan node
	nodegroupList := &appsv1alpha1.NodeGroupList{}
	if err := c.Client.List(context.TODO(), nodegroupList); err != nil {
		klog.Errorf("failed to list all nodegroups, %s", err)
		return nil
	}

	for _, nodegroup := range nodegroupList.Items {
		if IfMatchNodeGroup(node, &nodegroup) {
			// this node will be added into a node group
			return []controllerruntime.Request{
				{
					NamespacedName: types.NamespacedName{
						Name: nodegroup.Name,
					},
				},
			}
		}
	}

	// an orphan node, do not reconcile
	return []controllerruntime.Request{}
}

func (c *Controller) evictNodesInNodegroup(ctx context.Context, nodeGroupName string) error {
	selector := labels.SelectorFromSet(map[string]string{LabelBelongingTo: nodeGroupName})
	nodeList := &corev1.NodeList{}
	err := c.Client.List(ctx, nodeList, &client.ListOptions{LabelSelector: selector})
	if err != nil {
		return err
	}
	return c.evictNodes(ctx, nodeList.Items)
}

// getNodesByLabels can get all nodes matching these labels.
func (c *Controller) getNodesByLabels(ctx context.Context, matchLabels map[string]string) ([]corev1.Node, error) {
	if matchLabels == nil {
		// Return empty when matchLabels is nil
		// Otherwise, it will select all nodes, it's not what we want
		return []corev1.Node{}, nil
	}
	selector := labels.SelectorFromSet(matchLabels)
	nodeList := &corev1.NodeList{}
	err := c.Client.List(ctx, nodeList, &client.ListOptions{LabelSelector: selector})
	if err != nil {
		return nil, err
	}
	return nodeList.Items, nil
}

// getNodesByNodeName can get all nodes specified by node names.
func (c *Controller) getNodesByNodeName(ctx context.Context, nodeNames []string) ([]corev1.Node, error) {
	errs := []error{}
	nodes := []corev1.Node{}
	for _, name := range nodeNames {
		node := &corev1.Node{}
		if err := c.Client.Get(ctx, types.NamespacedName{Name: name}, node); err != nil {
			klog.Errorf("failed to get node %s, %s", name, err)
			errs = append(errs, err)
			continue
		}
		nodes = append(nodes, *node)
	}

	return nodes, utilerrors.NewAggregate(errs)
}

func (c *Controller) addOrUpdateNodeLabel(ctx context.Context,
	node *corev1.Node,
	nodeGroupName string,
	topologyEnabled bool,
) error {
	if v, ok := node.Labels[LabelBelongingTo]; ok {
		if v == nodeGroupName {
			// Check if the topology label needs to be updated.
			newNode := node.DeepCopy()
			if topologyEnabled {
				newNode.Labels[LabelTopologyZone] = nodeGroupName
			} else {
				delete(newNode.Labels, LabelTopologyZone)
			}
			if equality.Semantic.DeepEqual(node.Labels, newNode.Labels) {
				// The label has not changed and is returned directly.
				return nil
			}
			if err := c.Client.Patch(ctx, newNode, client.MergeFrom(node)); err != nil {
				klog.Errorf("failed to update labels on node %s for nodeGroup %s: %v", node.Name, nodeGroupName, err)
				return err
			}
			return nil
		}
		return fmt.Errorf("node %s has already belonged to NodeGroup %s", node.Name, v)
	}

	// !ok
	// add new label to this node
	newnode := node.DeepCopy()
	if newnode.Labels == nil {
		newnode.Labels = map[string]string{}
	}

	newnode.Labels[LabelBelongingTo] = nodeGroupName
	if topologyEnabled {
		newnode.Labels[LabelTopologyZone] = nodeGroupName
	}
	if err := c.Client.Patch(ctx, newnode, client.MergeFrom(node)); err != nil {
		klog.Errorf("failed to add label %s=%s to node %s, %s", LabelBelongingTo, nodeGroupName, node.Name, err)
		return err
	}
	return nil
}

// IfMatchNodeGroup will check if the node is selected by the nodegroup.
func IfMatchNodeGroup(node *corev1.Node, nodegroup *appsv1alpha1.NodeGroup) bool {
	// check if nodename is in the nodegroup.Spec.Nodes
	for _, nodeName := range nodegroup.Spec.Nodes {
		if nodeName == node.Name {
			return true
		}
	}
	// check if labels of this node selected by nodegroup.Spec.MatchLabels
	selector := labels.SelectorFromSet(nodegroup.Spec.MatchLabels)
	return selector.Matches(labels.Set(node.Labels))
}

func getNodeReadyConditionFromNode(node *corev1.Node) (corev1.ConditionStatus, bool) {
	for _, condition := range node.Status.Conditions {
		if condition.Type == corev1.NodeReady {
			return condition.Status, true
		}
	}
	return "", false
}

func nodesDiff(oldNodes []corev1.Node, newNodes []corev1.Node) ([]corev1.Node, []corev1.Node) {
	nodesDeleted, nodesAdded := []corev1.Node{}, []corev1.Node{}
	m := map[string]corev1.Node{}
	for _, n := range oldNodes {
		m[n.Name] = n
	}
	for _, n := range newNodes {
		_, exist := m[n.Name]
		if exist {
			delete(m, n.Name)
		} else {
			nodesAdded = append(nodesAdded, n)
		}
	}
	for _, n := range m {
		nodesDeleted = append(nodesDeleted, n)
	}
	return nodesDeleted, nodesAdded
}

func nodesUnion(a []corev1.Node, b []corev1.Node) []corev1.Node {
	nodesMap := map[string]*corev1.Node{}
	for i := range a {
		nodesMap[a[i].Name] = &a[i]
	}
	for i := range b {
		nodesMap[b[i].Name] = &b[i]
	}

	nodes := []corev1.Node{}
	for _, node := range nodesMap {
		nodes = append(nodes, *node)
	}
	return nodes
}