Kubernetes 节点生命周期控制器

$ k get node vtester0 -o json | jq '.status.conditions[] | select(.type == "Ready")'
{
  "lastHeartbeatTime": "2025-02-02T15:32:41Z",
  "lastTransitionTime": "2024-12-31T08:54:57Z",
  "message": "kubelet is posting ready status",
  "reason": "KubeletReady",
  "status": "True",
  "type": "Ready"
}

func SetDefaults_KubeletConfiguration(obj *kubeletconfigv1beta1.KubeletConfiguration) {
    // a lot of code here
    if obj.NodeStatusReportFrequency == zeroDuration {
        // For backward compatibility, NodeStatusReportFrequency's default value is
        // set to NodeStatusUpdateFrequency if NodeStatusUpdateFrequency is set
        // explicitly.
        if obj.NodeStatusUpdateFrequency == zeroDuration {
            obj.NodeStatusReportFrequency = metav1.Duration{Duration: 5 * time.Minute}
        } else {
            obj.NodeStatusReportFrequency = obj.NodeStatusUpdateFrequency
        }
    }
}

$ kubectl get node vtester1 -w
NAME       STATUS   ROLES    AGE   VERSION
vtester1   Ready    <none>   38d   v1.26.10
vtester1   NotReady   <none>   38d   v1.26.10
vtester1   NotReady   <none>   38d   v1.26.10
vtester1   NotReady   <none>   38d   v1.26.10


$ kubectl get node vtester1 -o yaml | yq .spec.taints
- effect: NoSchedule
key: node.kubernetes.io/unreachable
timeAdded: "2025-02-07T15:44:35Z"
- effect: NoExecute
key: node.kubernetes.io/unreachable
timeAdded: "2025-02-07T15:44:40Z"

$ kubectl get node vtester1 -o yaml | yq .status.conditions
- lastHeartbeatTime: "2025-02-05T03:55:06Z"
  lastTransitionTime: "2025-02-05T03:55:06Z"
  message: Flannel is running on this node
  reason: FlannelIsUp
  status: "False"
  type: NetworkUnavailable
- lastHeartbeatTime: "2025-02-07T15:46:10Z"
  lastTransitionTime: "2025-02-07T15:51:10Z"
  message: Kubelet stopped posting node status.
  reason: NodeStatusUnknown
  status: Unknown
  type: MemoryPressure
- lastHeartbeatTime: "2025-02-07T15:46:10Z"
  lastTransitionTime: "2025-02-07T15:51:10Z"
  message: Kubelet stopped posting node status.
  reason: NodeStatusUnknown
  status: Unknown
  type: DiskPressure
- lastHeartbeatTime: "2025-02-07T15:46:10Z"
  lastTransitionTime: "2025-02-07T15:51:10Z"
  message: Kubelet stopped posting node status.
  reason: NodeStatusUnknown
  status: Unknown
  type: PIDPressure
- lastHeartbeatTime: "2025-02-07T15:46:10Z"
  lastTransitionTime: "2025-02-07T15:51:10Z"
  message: Kubelet stopped posting node status.
  reason: NodeStatusUnknown
  status: Unknown
  type: Ready

var (
    // UnreachableTaintTemplate is the taint for when a node becomes unreachable.
    UnreachableTaintTemplate = &v1.Taint{
        Key:    v1.TaintNodeUnreachable,
        Effect: v1.TaintEffectNoExecute,
    }
)

func (nc *Controller) doNoExecuteTaintingPass(ctx context.Context) {
        // a lot of code here
        zoneNoExecuteTainterWorker.Try(func(value scheduler.TimedValue) (bool, time.Duration) {
            node, err := nc.nodeLister.Get(value.Value)
            if apierrors.IsNotFound(err) {
                klog.Warningf("Node %v no longer present in nodeLister!", value.Value)
                return true, 0
            } else if err != nil {
                klog.Warningf("Failed to get Node %v from the nodeLister: %v", value.Value, err)
                // retry in 50 millisecond
                return false, 50 * time.Millisecond
            }
            _, condition := controllerutil.GetNodeCondition(&node.Status, v1.NodeReady)
            // Because we want to mimic NodeStatus.Condition["Ready"] we make "unreachable" and "not ready" taints mutually exclusive.
            taintToAdd := v1.Taint{}
            oppositeTaint := v1.Taint{}
            switch condition.Status {
            case v1.ConditionFalse:
                taintToAdd = *NotReadyTaintTemplate
                oppositeTaint = *UnreachableTaintTemplate
            case v1.ConditionUnknown:
                taintToAdd = *UnreachableTaintTemplate
                oppositeTaint = *NotReadyTaintTemplate
            default:
                // It seems that the Node is ready again, so there's no need to taint it.
                klog.V(4).Infof("Node %v was in a taint queue, but it's ready now. Ignoring taint request.", value.Value)
                return true, 0
            }
            result := controllerutil.SwapNodeControllerTaint(ctx, nc.kubeClient, []*v1.Taint{&taintToAdd}, []*v1.Taint{&oppositeTaint}, node)
            if result {
                //count the evictionsNumber
                zone := nodetopology.GetZoneKey(node)
                evictionsNumber.WithLabelValues(zone).Inc()
                evictionsTotal.WithLabelValues(zone).Inc()
            }

            return result, 0
        })
}

func (nc *Controller) Run(ctx context.Context) {
    if nc.runTaintManager {
        // Handling taint based evictions. Because we don't want a dedicated logic in TaintManager for NC-originated
        // taints and we normally don't rate limit evictions caused by taints, we need to rate limit adding taints.
        go wait.UntilWithContext(ctx, nc.doNoExecuteTaintingPass, scheduler.NodeEvictionPeriod)
    }
}

var (
    nodeConditionToTaintKeyStatusMap = map[v1.NodeConditionType]map[v1.ConditionStatus]string{
        v1.NodeReady: {
            v1.ConditionFalse:   v1.TaintNodeNotReady,
            v1.ConditionUnknown: v1.TaintNodeUnreachable,
        },
        v1.NodeMemoryPressure: {
            v1.ConditionTrue: v1.TaintNodeMemoryPressure,
        },
        v1.NodeDiskPressure: {
            v1.ConditionTrue: v1.TaintNodeDiskPressure,
        },
        v1.NodeNetworkUnavailable: {
            v1.ConditionTrue: v1.TaintNodeNetworkUnavailable,
        },
        v1.NodePIDPressure: {
            v1.ConditionTrue: v1.TaintNodePIDPressure,
        },
    }
)

func (nc *Controller) doNoScheduleTaintingPass(ctx context.Context, nodeName string) error {
    // a lot of code here
    var taints []v1.Taint
    for _, condition := range node.Status.Conditions {
        if taintMap, found := nodeConditionToTaintKeyStatusMap[condition.Type]; found {
            if taintKey, found := taintMap[condition.Status]; found {
                taints = append(taints, v1.Taint{
                    Key:    taintKey,
                    Effect: v1.TaintEffectNoSchedule,
                })
            }
        }
    }
}

func (nc *Controller) tryUpdateNodeHealth(ctx context.Context, node *v1.Node) (time.Duration, v1.NodeCondition, *v1.NodeCondition, error) {
    nodeHealth := nc.nodeHealthMap.getDeepCopy(node.Name)
    defer func() {
        nc.nodeHealthMap.set(node.Name, nodeHealth)
    }()

    _, currentReadyCondition := controllerutil.GetNodeCondition(&node.Status, v1.NodeReady)
    // a lot of code here
    if nc.now().After(nodeHealth.probeTimestamp.Add(gracePeriod)) {
        // NodeReady condition or lease was last set longer ago than gracePeriod, so
        // update it to Unknown (regardless of its current value) in the master.

        nodeConditionTypes := []v1.NodeConditionType{
            v1.NodeReady,
            v1.NodeMemoryPressure,
            v1.NodeDiskPressure,
            v1.NodePIDPressure,
            // We don't change 'NodeNetworkUnavailable' condition, as it's managed on a control plane level.
            // v1.NodeNetworkUnavailable,
        }

        nowTimestamp := nc.now()
        for _, nodeConditionType := range nodeConditionTypes {
            _, currentCondition := controllerutil.GetNodeCondition(&node.Status, nodeConditionType)
            if currentCondition == nil {
                // a lot of code here
            } else {
                klog.V(2).Infof("node %v hasn't been updated for %+v. Last %v is: %+v",
                    node.Name, nc.now().Time.Sub(nodeHealth.probeTimestamp.Time), nodeConditionType, currentCondition)
                if currentCondition.Status != v1.ConditionUnknown {
                    currentCondition.Status = v1.ConditionUnknown
                    currentCondition.Reason = "NodeStatusUnknown"
                    currentCondition.Message = "Kubelet stopped posting node status."
                    currentCondition.LastTransitionTime = nowTimestamp
                }
            }
        }
        // We need to update currentReadyCondition due to its value potentially changed.
        _, currentReadyCondition = controllerutil.GetNodeCondition(&node.Status, v1.NodeReady)

        if !apiequality.Semantic.DeepEqual(currentReadyCondition, &observedReadyCondition) {
            if _, err := nc.kubeClient.CoreV1().Nodes().UpdateStatus(ctx, node, metav1.UpdateOptions{}); err != nil {
                klog.Errorf("Error updating node %s: %v", node.Name, err)
                return gracePeriod, observedReadyCondition, currentReadyCondition, err
            }
            nodeHealth = &nodeHealthData{
                status:                   &node.Status,
                probeTimestamp:           nodeHealth.probeTimestamp,
                readyTransitionTimestamp: nc.now(),
                lease:                    observedLease,
            }
            return gracePeriod, observedReadyCondition, currentReadyCondition, nil
        }
    }
}

func RecommendedDefaultNodeLifecycleControllerConfiguration(obj *kubectrlmgrconfigv1alpha1.NodeLifecycleControllerConfiguration) {
    // a lot of code here
    if obj.NodeMonitorGracePeriod == zero {
        obj.NodeMonitorGracePeriod = metav1.Duration{Duration: 40 * time.Second}
    }
}

func (nc *Controller) Run(ctx context.Context) {
    // a lot of code here
    if nc.runTaintManager {
        // Handling taint based evictions. Because we don't want a dedicated logic in TaintManager for NC-originated
        // taints and we normally don't rate limit evictions caused by taints, we need to rate limit adding taints.
        go wait.UntilWithContext(ctx, nc.doNoExecuteTaintingPass, scheduler.NodeEvictionPeriod)
    } else {
        // Managing eviction of nodes:
        // When we delete pods off a node, if the node was not empty at the time we then
        // queue an eviction watcher. If we hit an error, retry deletion.
        go wait.UntilWithContext(ctx, nc.doEvictionPass, scheduler.NodeEvictionPeriod)
    }
}

func NewNoExecuteTaintManager(ctx context.Context, c clientset.Interface, podLister corelisters.PodLister, nodeLister corelisters.NodeLister, getPodsAssignedToNode GetPodsByNodeNameFunc) *NoExecuteTaintManager {
    eventBroadcaster := record.NewBroadcaster()
    recorder := eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "taint-controller"})

    tm := &NoExecuteTaintManager{
        client:                c,
        broadcaster:           eventBroadcaster,
        recorder:              recorder,
        podLister:             podLister,
        nodeLister:            nodeLister,
        getPodsAssignedToNode: getPodsAssignedToNode,
        taintedNodes:          make(map[string][]v1.Taint),

        nodeUpdateQueue: workqueue.NewNamed("noexec_taint_node"),
        podUpdateQueue:  workqueue.NewNamed("noexec_taint_pod"),
    }
    tm.taintEvictionQueue = CreateWorkerQueue(deletePodHandler(c, tm.emitPodDeletionEvent))

    return tm
}

func addConditionAndDeletePod(ctx context.Context, c clientset.Interface, name, ns string) (err error) {
    // a lot of code here
    return c.CoreV1().Pods(ns).Delete(ctx, name, metav1.DeleteOptions{})
}

import (
    policyv1 "k8s.io/api/policy/v1"
    "sigs.k8s.io/controller-runtime/pkg/client"
)

// new a client instance firstly
// client :=
// a lot of code here
client.SubResource("eviction").Create(ctx, &pod, &policyv1.Eviction{
    // custom options
})