johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package scale

import (
	"fmt"
	"slices"
	"time"

	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

const HostnameLabel = "kubernetes.io/hostname"

type NodeLoad struct {
	Name string

	Created time.Time

	Held bool

	Busy int

	Evictable int

	Incoming bool

	Ready bool

	ReadySince time.Time
}

type Placement struct {
	Nodes []NodeLoad

	Unplaced []time.Time

	Unschedulable []Unschedulable

	Spread Spread
}

func (p Placement) Load(name string) (NodeLoad, bool) {
	for _, n := range p.Nodes {
		if n.Name == name {
			return n, true
		}
	}
	return NodeLoad{}, false
}

func (p Placement) placing(now time.Time, hold time.Duration) (time.Duration, bool) {
	for _, created := range p.Unplaced {
		if created.IsZero() {
			return 0, true
		}
		if age := now.Sub(created); age < hold {
			return age, true
		}
	}
	return 0, false
}

func ControlledByADaemonSet(p *corev1.Pod) bool {
	c := metav1.GetControllerOf(p)
	return c != nil && c.Kind == KindDaemonSet
}

func MirrorPod(p *corev1.Pod) bool {
	_, ok := p.Annotations[corev1.MirrorPodAnnotationKey]
	return ok
}

func FollowsTheNodes(p *corev1.Pod) bool {
	return ControlledByADaemonSet(p) || MirrorPod(p)
}

func live(p *corev1.Pod) bool {
	switch p.Status.Phase {
	case corev1.PodPending, corev1.PodRunning, corev1.PodUnknown:
		return true
	}
	return false
}

func Loads(nodes []corev1.Node, o Observation, replicaSets []appsv1.ReplicaSet, b Baseline) Placement {
	ours := baselineWorkloads(o, b)
	deploymentOf := deploymentsOfReplicaSets(replicaSets)
	byName := map[string]*NodeLoad{}
	out := Placement{Nodes: make([]NodeLoad, 0, len(nodes))}
	for i := range nodes {
		ready, since := NodeReady(&nodes[i])
		out.Nodes = append(out.Nodes, NodeLoad{Name: nodes[i].Name, Created: nodes[i].CreationTimestamp.Time,
			Ready: ready, ReadySince: since})
	}
	var workload []*corev1.Pod
	for i := range out.Nodes {
		byName[out.Nodes[i].Name] = &out.Nodes[i]
	}
	for i := range o.Pods {
		p := &o.Pods[i]
		if !live(p) || FollowsTheNodes(p) {
			continue
		}
		baseline := AppliedByTheControlPlane(p.ObjectMeta)
		if k, ok := workloadOf(p, deploymentOf); ok && ours[k] {
			baseline = true
		}
		if !baseline && p.DeletionTimestamp == nil {
			workload = append(workload, p)
		}
		if p.Spec.NodeName != "" {
			n, ok := byName[p.Spec.NodeName]
			if !ok {
				continue
			}
			if baseline {
				n.Evictable++
			} else {
				n.Busy++
			}
			continue
		}
		if baseline || p.Status.Phase != corev1.PodPending {
			continue
		}
		if target := destinedFor(p); target != "" {
			if n, ok := byName[target]; ok {
				n.Incoming = true
				continue
			}
		}
		out.Unplaced = append(out.Unplaced, p.CreationTimestamp.Time)
		if u, ok := unschedulable(p); ok {
			out.Unschedulable = append(out.Unschedulable, u)
		}
	}
	out.Spread = spreadOf(workload)
	return out
}

func destinedFor(p *corev1.Pod) string {
	if p.Status.NominatedNodeName != "" {
		return p.Status.NominatedNodeName
	}
	if h := p.Spec.NodeSelector[HostnameLabel]; h != "" {
		return h
	}
	a := p.Spec.Affinity
	if a == nil || a.NodeAffinity == nil || a.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution == nil {
		return ""
	}
	for _, term := range a.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms {
		for _, r := range term.MatchExpressions {
			if r.Key == HostnameLabel && r.Operator == corev1.NodeSelectorOpIn && len(r.Values) == 1 {
				return r.Values[0]
			}
		}
		for _, r := range term.MatchFields {
			if r.Key == "metadata.name" && r.Operator == corev1.NodeSelectorOpIn && len(r.Values) == 1 {
				return r.Values[0]
			}
		}
	}
	return ""
}

func baselineWorkloads(o Observation, b Baseline) map[WorkloadKey]bool {
	out := map[WorkloadKey]bool{}
	for _, d := range o.Deployments {
		k := WorkloadKey{Kind: KindDeployment, Namespace: d.Namespace, Name: d.Name}
		if AppliedByTheControlPlane(d.ObjectMeta) || b.Holds(k, d.Spec) {
			out[k] = true
		}
	}
	for _, s := range o.StatefulSets {
		k := WorkloadKey{Kind: KindStatefulSet, Namespace: s.Namespace, Name: s.Name}
		if AppliedByTheControlPlane(s.ObjectMeta) || b.Holds(k, s.Spec) {
			out[k] = true
		}
	}
	for _, j := range o.Jobs {
		k := WorkloadKey{Kind: KindJob, Namespace: j.Namespace, Name: j.Name}
		if AppliedByTheControlPlane(j.ObjectMeta) || b.Holds(k, j.Spec) {
			out[k] = true
		}
	}
	return out
}

func deploymentsOfReplicaSets(replicaSets []appsv1.ReplicaSet) map[WorkloadKey]WorkloadKey {
	out := map[WorkloadKey]WorkloadKey{}
	for i := range replicaSets {
		rs := &replicaSets[i]
		if c := metav1.GetControllerOf(rs); c != nil && c.Kind == KindDeployment {
			out[WorkloadKey{Kind: KindReplicaSet, Namespace: rs.Namespace, Name: rs.Name}] =
				WorkloadKey{Kind: KindDeployment, Namespace: rs.Namespace, Name: c.Name}
		}
	}
	return out
}

func (n NodeLoad) idle() bool {
	return n.Busy == 0 && !n.Incoming
}

func nodeClocks(p Placement, before map[string]time.Time, now time.Time) map[string]time.Time {
	out := make(map[string]time.Time, len(p.Nodes))
	for _, n := range p.Nodes {
		if !n.idle() {
			continue
		}
		if since, ok := before[n.Name]; ok {
			out[n.Name] = since
			continue
		}
		out[n.Name] = now
	}
	return out
}

func restartedClocks(clocks map[string]time.Time, released string, now time.Time) map[string]time.Time {
	out := make(map[string]time.Time, len(clocks))
	for name := range clocks {
		if name == released {
			continue
		}
		out[name] = now
	}
	return out
}

func releasable(p Placement, clocks map[string]time.Time, now time.Time, hold time.Duration) []NodeLoad {
	var out []NodeLoad
	for _, n := range p.Nodes {
		if !n.Held || !n.idle() {
			continue
		}
		since, ok := clocks[n.Name]
		if !ok || now.Sub(since) < hold {
			continue
		}
		out = append(out, n)
	}
	slices.SortFunc(out, func(a, b NodeLoad) int {
		switch {
		case a.Evictable != b.Evictable:
			return a.Evictable - b.Evictable
		case !a.Created.Equal(b.Created):
			return b.Created.Compare(a.Created)
		}
		if a.Name < b.Name {
			return -1
		}
		if a.Name > b.Name {
			return 1
		}
		return 0
	})
	return out
}

func scaleIn(o Observation, st State, now time.Time, opts Options, target int, clocks map[string]time.Time) Decision {
	hold := opts.hold()
	keep := Decision{Action: None, Nodes: st.Nodes, Intent: true, NodeIdleSince: clocks}
	if age, placing := o.Placement.placing(now, hold); placing {
		keep.Reason = fmt.Sprintf("intent, and %d nodes are counted for a target of %d, and a pod "+
			"created %s ago is Pending with no node yet: the scheduler may place it on any idle "+
			"node, so no node is released until it is placed or has waited out the %s hold",
			st.Nodes, target, age.Round(time.Second), hold)
		return keep
	}
	candidates := releasable(o.Placement, clocks, now, hold)
	if len(candidates) == 0 {
		keep.Reason = fmt.Sprintf("intent, and %d nodes are counted for a target of %d, and no "+
			"node this provider holds has held nothing beyond the baseline for the %s hold",
			st.Nodes, target, hold)
		return keep
	}
	pick := candidates[0]
	return Decision{
		Action: Release, Nodes: st.Nodes - 1, Intent: true, Release: pick.Name,
		Idle: now.Sub(clocks[pick.Name]), NodeIdleSince: clocks,
		Reason: fmt.Sprintf("intent, and %d nodes are counted for a target of %d; %s has held "+
			"nothing beyond the baseline for %s, past the %s hold, and is the emptiest and newest "+
			"of %d such nodes, so it alone is released this hold",
			st.Nodes, target, pick.Name, now.Sub(clocks[pick.Name]).Round(time.Second), hold,
			len(candidates)),
	}
}