johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package providermain

import (
	"context"
	"errors"
	"fmt"
	"time"

	corev1 "k8s.io/api/core/v1"
	policyv1 "k8s.io/api/policy/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"

	"github.com/johnandersen777/socialweb-computer-kcp/internal/scale"
)

type observeCluster func(ctx context.Context, cluster string) (scale.Observation, int, error)

var _ scale.Releaser = (*tenantNodes)(nil)

var errNoReleaseObserver = errors.New(
	"clusterprovider: releasing one node re-reads the cluster after the cordon, and no observer was " +
		"supplied -- a pod the scheduler bound between the decision's read and the cordon would be " +
		"evicted and its node released under it")

var errReleaseNodeBusy = errors.New(
	"clusterprovider: the node took work between the decision's read and the cordon, so it is " +
		"uncordoned and kept; the next pass decides again")

func (h *nodeHosts) observeHeld(observe observeCluster) observeCluster {
	return func(ctx context.Context, cluster string) (scale.Observation, int, error) {
		o, nodes, err := observe(ctx, cluster)
		if err != nil {
			return o, nodes, err
		}
		held := h.heldNames(cluster)
		now := time.Now()
		reported := make(map[string]bool, len(o.Placement.Nodes))
		for i := range o.Placement.Nodes {
			nd := &o.Placement.Nodes[i]
			reported[nd.Name] = true
			nd.Held = held[nd.Name]
			if seen := h.observedReadySince(cluster, nd.Name, nd.Ready, now); nd.Ready && seen.After(nd.ReadySince) {
				nd.ReadySince = seen
			}
		}
		o.InFlight = h.inFlight(cluster, reported)
		return o, nodes + o.InFlight, nil
	}
}

func (h *nodeHosts) heldNames(cluster string) map[string]bool {
	h.mu.Lock()
	n, ok := h.hosts[cluster]
	h.mu.Unlock()
	if !ok {
		return nil
	}
	n.mu.Lock()
	defer n.mu.Unlock()
	out := map[string]bool{}
	for _, node := range n.heldNodesLocked() {
		out[node.name] = true
	}
	return out
}

func (n *tenantNodes) Release(ctx context.Context, name string) error {
	n.mu.Lock()
	var node heldNode
	found := false
	for _, h := range n.heldNodesLocked() {
		if h.name == name {
			node, found = h, true
			break
		}
	}
	n.mu.Unlock()
	if !found {
		return fmt.Errorf("clusterprovider: node %s of %s is not held by this provider process, so "+
			"there is no claim here to delete and no guest to release", name, n.cluster)
	}
	if n.opts.Observe == nil {
		return errNoReleaseObserver
	}
	client, err := n.client(ctx)
	if err != nil {
		return err
	}
	cordoned, err := setUnschedulable(ctx, client, name, true)
	if err != nil {
		return fmt.Errorf("clusterprovider: cordoning node %s of %s before releasing it: %w", name, n.cluster, err)
	}
	uncordon := func(cause error) error {
		if !cordoned {
			return cause
		}
		if _, err := setUnschedulable(ctx, client, name, false); err != nil {
			return errors.Join(cause, fmt.Errorf("clusterprovider: uncordoning node %s of %s after "+
				"its release stopped: %w", name, n.cluster, err))
		}
		return cause
	}
	o, _, err := n.opts.Observe(ctx, n.cluster)
	if err != nil {
		return uncordon(fmt.Errorf("clusterprovider: re-reading %s after cordoning %s: %w", n.cluster, name, err))
	}
	load, seen := o.Placement.Load(name)
	if !seen || load.Busy > 0 || load.Incoming {
		n.log.Info("cluster provider: a node chosen for release took work before the cordon, so it is kept",
			"cluster", n.cluster, "node", name, "busy", load.Busy, "incoming", load.Incoming, "seen", seen)
		return uncordon(fmt.Errorf("%w: node %s of %s", errReleaseNodeBusy, name, n.cluster))
	}
	evicted, err := drainNode(ctx, client, name)
	if err != nil {
		return uncordon(fmt.Errorf("clusterprovider: draining node %s of %s: %w -- the node is "+
			"uncordoned and kept, and the next pass decides again", name, n.cluster, err))
	}
	n.log.Info("cluster provider: a node held nothing beyond the baseline for the hold, so it is "+
		"cordoned, drained and released",
		"cluster", n.cluster, "node", name, "evicted", evicted)
	if err := n.teardownHeld(ctx, node); err != nil {
		return err
	}
	return n.writeSnapshot()
}

func setUnschedulable(ctx context.Context, client kubernetes.Interface, name string, unschedulable bool) (bool, error) {
	node, err := client.CoreV1().Nodes().Get(ctx, name, metav1.GetOptions{})
	if err != nil {
		return false, err
	}
	if node.Spec.Unschedulable == unschedulable {
		return false, nil
	}
	node.Spec.Unschedulable = unschedulable
	if _, err := client.CoreV1().Nodes().Update(ctx, node, metav1.UpdateOptions{}); err != nil {
		return false, err
	}
	return true, nil
}

func drainNode(ctx context.Context, client kubernetes.Interface, name string) ([]string, error) {
	pods, err := client.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{
		FieldSelector: "spec.nodeName=" + name,
	})
	if err != nil {
		return nil, fmt.Errorf("listing the pods on %s: %w", name, err)
	}
	var evicted []string
	for i := range pods.Items {
		p := &pods.Items[i]
		if p.Spec.NodeName != name || p.DeletionTimestamp != nil || scale.FollowsTheNodes(p) {
			continue
		}
		if p.Status.Phase == corev1.PodSucceeded || p.Status.Phase == corev1.PodFailed {
			continue
		}
		err := client.CoreV1().Pods(p.Namespace).EvictV1(ctx, &policyv1.Eviction{
			ObjectMeta: metav1.ObjectMeta{Namespace: p.Namespace, Name: p.Name},
		})
		if err != nil && !apierrors.IsNotFound(err) {
			return evicted, fmt.Errorf("evicting %s/%s: %w", p.Namespace, p.Name, err)
		}
		evicted = append(evicted, p.Namespace+"/"+p.Name)
	}
	return evicted, nil
}