cozystack/internal/gpu/controller/binder.go

/*
Copyright 2026 The Cozystack Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

	http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package controller

import (
	"context"
	"errors"
	"fmt"
	"slices"
	"sort"

	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	gpuv1alpha1 "github.com/cozystack/cozystack/api/gpu/v1alpha1"
	"github.com/cozystack/cozystack/internal/gpu/model"
)

// ConditionWaiting is set on a device the binder cannot place yet; its
// reason names the shortage and it is removed once the device is placed.
const ConditionWaiting = "Waiting"

const (
	ReasonCapacity     = "Capacity"
	ReasonClassDeleted = "ClassDeleted"
	ReasonCordoned     = "Cordoned"
	ReasonBudgetsRead  = "BudgetsUnavailable"
)

// setWaiting records why a device waits and reports whether it changed.
func (w *world) setWaiting(d *gpuv1alpha1.GPUDevice, reason, message string) bool {
	return meta.SetStatusCondition(&d.Status.Conditions, metav1.Condition{
		Type: ConditionWaiting, Status: metav1.ConditionTrue, Reason: reason, Message: message,
		ObservedGeneration: d.Generation, LastTransitionTime: metav1.NewTime(w.now),
	})
}

func waitingCondition(d *gpuv1alpha1.GPUDevice) *metav1.Condition {
	return meta.FindStatusCondition(d.Status.Conditions, ConditionWaiting)
}

type bindRequest struct {
	namespace string
	req       *gpuv1alpha1.BindRequest
	devices   []*gpuv1alpha1.GPUDevice
}

// pendingRequests returns the complete, unanswered bind requests, oldest
// first. A request is complete when every device it names carries its ID.
func (w *world) pendingRequests() []bindRequest {
	byID := map[string]*bindRequest{}
	var order []*bindRequest
	for _, d := range w.devices {
		req := d.Status.BindRequest
		if req == nil || req.Result != "" {
			continue
		}
		k := d.Namespace + "/" + req.ID
		if byID[k] == nil {
			byID[k] = &bindRequest{namespace: d.Namespace, req: req}
			order = append(order, byID[k])
		}
	}
	var out []bindRequest
	for _, br := range order {
		complete := true
		for _, name := range br.req.Devices {
			d := w.device(br.namespace, name)
			if d == nil || d.Status.BindRequest == nil || d.Status.BindRequest.ID != br.req.ID {
				complete = false
				break
			}
			br.devices = append(br.devices, d)
		}
		if complete {
			out = append(out, *br)
		}
	}
	sort.SliceStable(out, func(i, j int) bool {
		return out[i].req.RequestedAt.Before(&out[j].req.RequestedAt)
	})
	return out
}

// bindRequests answers the webhook's requests: each set lands on one node
// with room for all of it, or nothing moves and the request is refused.
func (r *Reconciler) bindRequests(ctx context.Context, w *world) []error {
	var errs []error
	for _, br := range w.pendingRequests() {
		if w.aborted {
			break
		}
		result, message := r.decide(w, br)
		for _, d := range br.devices {
			d.Status.BindRequest.Result = result
			d.Status.BindRequest.Message = message
			if err := r.updateStatus(ctx, w, d); err != nil {
				errs = append(errs, err)
				break
			}
		}
	}
	return errs
}

// decide places one request's set and returns the answer. On refusal no
// placement is changed.
func (r *Reconciler) decide(w *world, br bindRequest) (gpuv1alpha1.BindResult, string) {
	refuse := func(format string, args ...any) (gpuv1alpha1.BindResult, string) {
		return gpuv1alpha1.BindRefused, fmt.Sprintf(format, args...)
	}
	var pending, held []*gpuv1alpha1.GPUDevice
	for _, d := range br.devices {
		class := w.classes[d.Spec.ClassName]
		switch {
		case d.Status.AttachedTo == nil || d.Status.AttachedTo.VMI != br.req.VMI:
			return refuse("GPUDevice %s is not attached to VMI %s", d.Name, br.req.VMI)
		case !d.DeletionTimestamp.IsZero():
			return refuse("GPUDevice %s is being deleted", d.Name)
		case holds(d):
			held = append(held, d)
		case d.Status.Phase == gpuv1alpha1.GPUDeviceLost || !isTemplate(d):
			return refuse("GPUDevice %s is %s", d.Name, phaseOrPending(d))
		case class == nil:
			return refuse("GPUClass %s of GPUDevice %s does not exist", d.Spec.ClassName, d.Name)
		case !class.DeletionTimestamp.IsZero():
			return refuse("GPUClass %s of GPUDevice %s is being deleted", d.Spec.ClassName, d.Name)
		default:
			pending = append(pending, d)
		}
	}

	if len(pending) > 0 {
		if w.quotaErr != nil {
			return refuse("tenant GPU budgets cannot be read: %v", w.quotaErr)
		}
		saved := slices.Clone(w.snap.Devices)
		for _, d := range pending {
			if err := w.snap.CheckOrder(d.Namespace, d.Spec.ClassName, int64(d.Spec.Count), w.entitled[d.Namespace]); err != nil {
				w.snap.Devices = saved
				return refuse("GPUDevice %s: %v", d.Name, err)
			}
			// Later template devices of the set count this one.
			probe := d.DeepCopy()
			probe.Status.Phase = gpuv1alpha1.GPUDeviceBound
			w.setDevice(probe)
		}
		w.snap.Devices = saved
	}

	if node, stay := w.stayPin(held); stay {
		if node == "" || len(pending) > 0 || !onOneNode(held, node) {
			return refuse("%s: a device of a class with onCordon Stay is on a cordoned node, so the set cannot be gathered", describe(br.devices))
		}
		return gpuv1alpha1.BindSucceeded, ""
	}

	cordonOnly := false
	if len(pending) == 0 && len(held) > 0 && onOneNode(held, held[0].Status.NodeName) {
		worst := problemNone
		for _, d := range held {
			worst = max(worst, w.problem(d))
		}
		if worst == problemNone {
			return gpuv1alpha1.BindSucceeded, ""
		}
		cordonOnly = worst == problemCordoned
	}

	target := w.pick(br.devices)
	switch {
	case target != "":
		w.placeOn(br.devices, target)
		return gpuv1alpha1.BindSucceeded, ""
	case cordonOnly:
		// With no room elsewhere a cordon alone does not move a set: the
		// VMI is pinned to the cordoned node and waits.
		return gpuv1alpha1.BindSucceeded, ""
	}
	return refuse("%s: no Ready node has room for all of them", describe(br.devices))
}

func phaseOrPending(d *gpuv1alpha1.GPUDevice) gpuv1alpha1.GPUDevicePhase {
	if d.Status.Phase == "" {
		return gpuv1alpha1.GPUDevicePending
	}
	return d.Status.Phase
}

func onOneNode(set []*gpuv1alpha1.GPUDevice, node string) bool {
	for _, d := range set {
		if d.Status.NodeName != node {
			return false
		}
	}
	return true
}

// stayPin returns the cordoned node a device of a Stay class pins the set
// to. The pin holds through a NotReady spell or a taint, and breaks when a
// member held on that node has lost capacity there, or the node is gone or
// no longer serves the member's class. It returns "" and true when two such
// devices sit on different nodes, and false when nothing pins the set.
// decide, relocateSet and relocateAttached all read the pin here, so the
// binder never admits a VMI that the full pass would then delete.
func (w *world) stayPin(held []*gpuv1alpha1.GPUDevice) (string, bool) {
	node, pinned := "", false
	for _, d := range held {
		n := w.nodes[d.Status.NodeName]
		if n == nil || !model.Cordoned(n) || w.onCordon(d.Spec.ClassName) != gpuv1alpha1.OnCordonStay {
			continue
		}
		if pinned && node != n.Name {
			return "", true
		}
		node, pinned = n.Name, true
	}
	if !pinned {
		return "", false
	}
	for _, d := range held {
		if d.Status.NodeName == node && w.problem(d) >= problemNotServed {
			return "", false
		}
	}
	return node, true
}

// relocate moves held devices off nodes that cannot keep them, as sets,
// and deletes unscheduled VMIs that would otherwise wait on such a node.
func (r *Reconciler) relocate(ctx context.Context, w *world) []error {
	var errs []error
	leaving := w.excessOnLostNodes()
	for _, set := range w.vmSets(func(d *gpuv1alpha1.GPUDevice) bool {
		return holds(d) && d.DeletionTimestamp.IsZero()
	}) {
		if w.aborted {
			break
		}
		var unattached []*gpuv1alpha1.GPUDevice
		for _, d := range set {
			if d.Status.AttachedTo == nil {
				unattached = append(unattached, d)
				continue
			}
			errs = append(errs, r.relocateAttached(ctx, w, d, set, leaving[d]))
		}
		if len(unattached) > 0 {
			errs = append(errs, r.relocateSet(ctx, w, unattached, leaving))
		}
	}
	return errs
}

// excessOnLostNodes picks, on each node that lost capacity for a resource
// but is still Ready, the devices that no longer fit there: unattached ones
// first, then those whose VMI has not been scheduled. A device whose VMI
// runs keeps it; it is refused at the VMI's next start.
func (w *world) excessOnLostNodes() map[*gpuv1alpha1.GPUDevice]bool {
	out := map[*gpuv1alpha1.GPUDevice]bool{}
	for node, loss := range w.lost {
		if loss.notReady {
			continue
		}
		for rn, healthy := range loss.healthy {
			var held int64
			var candidates []*gpuv1alpha1.GPUDevice
			for _, d := range w.devices {
				if d.Status.NodeName != node || !holds(d) || w.resourceName(d.Spec.ClassName) != rn {
					continue
				}
				held += int64(d.Spec.Count)
				candidates = append(candidates, d)
			}
			rank := func(d *gpuv1alpha1.GPUDevice) int {
				if d.Status.AttachedTo == nil {
					return 0
				}
				if v := w.attachedVMI(d); v != nil && v.unscheduled() {
					return 1
				}
				return 2
			}
			sort.SliceStable(candidates, func(i, j int) bool { return rank(candidates[i]) < rank(candidates[j]) })
			for _, d := range candidates {
				if held <= healthy || rank(d) == 2 {
					break
				}
				out[d] = true
				held -= int64(d.Spec.Count)
			}
		}
	}
	return out
}

// mustLeave reports whether an unattached device has to leave its node,
// and whether the reason is a cordon alone.
func (w *world) mustLeave(d *gpuv1alpha1.GPUDevice, excess bool) (leave, cordonOnly bool) {
	switch w.problem(d) {
	case problemLostCapacity:
		if w.lost[d.Status.NodeName].notReady || excess {
			return true, false
		}
		return false, false
	case problemGone, problemNotServed:
		return true, false
	case problemCordoned:
		return w.onCordon(d.Spec.ClassName) == gpuv1alpha1.OnCordonMove, true
	}
	return false, false
}

func (r *Reconciler) relocateSet(ctx context.Context, w *world, set []*gpuv1alpha1.GPUDevice, excess map[*gpuv1alpha1.GPUDevice]bool) error {
	hard, soft := false, false
	mustGo := map[*gpuv1alpha1.GPUDevice]bool{}
	for _, d := range set {
		leave, cordonOnly := w.mustLeave(d, excess[d])
		switch {
		case leave && cordonOnly:
			soft = true
		case leave:
			hard = true
			mustGo[d] = true
		}
	}
	_, stayPinned := w.stayPin(set)
	var errs []error
	if !hard && (!soft || stayPinned) {
		// Nothing to move, or a Stay device on a cordoned node keeps the set.
		for _, d := range set {
			if c := meta.FindStatusCondition(d.Status.Conditions, ConditionWaiting); c != nil && c.Reason == ReasonCordoned {
				meta.RemoveStatusCondition(&d.Status.Conditions, ConditionWaiting)
				errs = append(errs, r.updateStatus(ctx, w, d))
			}
		}
		return errors.Join(errs...)
	}
	if target := w.pick(set); target != "" {
		before := map[*gpuv1alpha1.GPUDevice]string{}
		for _, d := range set {
			before[d] = d.Status.NodeName
		}
		w.placeOn(set, target)
		return r.writeSet(ctx, w, set, func(d *gpuv1alpha1.GPUDevice) bool { return before[d] != target })
	}
	if hard {
		// The set goes to Lost together, so it is rebound together.
		for _, d := range set {
			reason := fmt.Sprintf("node %s %s and no node has room for the device's set", d.Status.NodeName, w.problem(d))
			if !mustGo[d] {
				reason = "another device of its set must leave its node and no node has room for the whole set"
			}
			d.Status.Phase = gpuv1alpha1.GPUDeviceLost
			w.setWaiting(d, ReasonCapacity, reason)
			w.setDevice(d)
		}
		return r.writeSet(ctx, w, set, func(*gpuv1alpha1.GPUDevice) bool { return true })
	}
	cordonedNode := ""
	for _, d := range set {
		if w.problem(d) == problemCordoned {
			cordonedNode = d.Status.NodeName
		}
	}
	for _, d := range set {
		msg := fmt.Sprintf("node %s is cordoned; waiting for room on another node", d.Status.NodeName)
		if w.problem(d) != problemCordoned {
			msg = fmt.Sprintf("node %s of another device of its set is cordoned; waiting for room for the whole set", cordonedNode)
		}
		if w.setWaiting(d, ReasonCordoned, msg) {
			errs = append(errs, r.updateStatus(ctx, w, d))
		}
	}
	return errors.Join(errs...)
}

// relocateAttached handles an attached device whose node cannot keep it.
// A VMI that has not been scheduled is deleted so the device can move; a
// running one keeps it, except on a node that is gone or NotReady past
// grace, where the device goes to Lost until the VMI ends.
func (r *Reconciler) relocateAttached(ctx context.Context, w *world, d *gpuv1alpha1.GPUDevice, set []*gpuv1alpha1.GPUDevice, excess bool) error {
	v := w.attachedVMI(d)
	if v == nil || v.final() {
		return nil
	}
	p := w.problem(d)
	loss := w.lost[d.Status.NodeName]
	hard := p == problemGone || p == problemNotServed || (p == problemLostCapacity && (loss.notReady || excess))
	switch {
	case hard && v.unscheduled():
		return r.deleteVMI(ctx, v, fmt.Sprintf("node %s of GPUDevice %s %s", d.Status.NodeName, d.Name, p))
	case p == problemGone || (p == problemLostCapacity && loss.notReady):
		if d.Status.Phase != gpuv1alpha1.GPUDeviceAttached {
			return nil
		}
		d.Status.Phase = gpuv1alpha1.GPUDeviceLost
		return r.updateStatus(ctx, w, d)
	case p == problemCordoned && v.unscheduled() && w.onCordon(d.Spec.ClassName) == gpuv1alpha1.OnCordonMove:
		owner := w.vms[key(v.Namespace, v.VM)]
		if owner == nil || !owner.restartsDeletedVMI() {
			// Deleting it would leave a Manual or RerunOnFailure VM stopped.
			return nil
		}
		var mine []*gpuv1alpha1.GPUDevice
		for _, m := range set {
			if m.Status.AttachedTo != nil && m.Status.AttachedTo.VMI == v.Name {
				mine = append(mine, m)
			}
		}
		// A Stay device among them pins the VMI here, as the binder answered.
		if _, pinned := w.stayPin(mine); pinned || w.pick(mine) == "" {
			return nil
		}
		return r.deleteVMI(ctx, v, fmt.Sprintf("node %s of GPUDevice %s is cordoned", d.Status.NodeName, d.Name))
	}
	return nil
}

// rebindLost places Lost devices that no VMI uses, before new orders, a
// VM's set onto one node. A set that does not fit does not hold up the
// sets after it. A rebind skips budgets: a Lost device already counts.
func (r *Reconciler) rebindLost(ctx context.Context, w *world) []error {
	var errs []error
	for _, set := range w.vmSets(unattachedSetMember) {
		if w.aborted {
			break
		}
		if !allLost(set) {
			continue
		}
		if target := w.pick(set); target != "" {
			w.placeOn(set, target)
			if err := r.writeSet(ctx, w, set, func(*gpuv1alpha1.GPUDevice) bool { return true }); err != nil {
				errs = append(errs, err)
			}
			continue
		}
		var changed []*gpuv1alpha1.GPUDevice
		for _, d := range set {
			// Keep the condition that says why the device left its node.
			if waitingCondition(d) == nil && w.setWaiting(d, ReasonCapacity, "no Ready node has room for the device's set") {
				changed = append(changed, d)
			}
		}
		if err := r.writeSet(ctx, w, changed, func(*gpuv1alpha1.GPUDevice) bool { return true }); err != nil {
			errs = append(errs, err)
		}
	}
	return errs
}

// writeSet writes the devices of a set that changed and stops at the first
// failure, so the rest of the pass places nothing on a view the cluster
// does not share.
func (r *Reconciler) writeSet(ctx context.Context, w *world, set []*gpuv1alpha1.GPUDevice, changed func(*gpuv1alpha1.GPUDevice) bool) error {
	for _, d := range set {
		if !changed(d) {
			continue
		}
		if err := r.updateStatus(ctx, w, d); err != nil {
			return err
		}
	}
	return nil
}

// bindOrders binds new orders, oldest first, each onto the fullest node
// with room. An order that does not fit waits with a condition and does
// not hold up the orders after it.
func (r *Reconciler) bindOrders(ctx context.Context, w *world) []error {
	var errs []error
	for _, d := range w.devices {
		if w.aborted {
			break
		}
		if holds(d) || d.Status.Phase == gpuv1alpha1.GPUDeviceLost || isTemplate(d) ||
			d.Status.AttachedTo != nil || !d.DeletionTimestamp.IsZero() {
			continue
		}
		changed := false
		if d.Status.Phase == "" {
			d.Status.Phase = gpuv1alpha1.GPUDevicePending
			changed = true
		}
		reason, message := r.order(w, d)
		if reason == "" {
			changed = true
		} else if w.setWaiting(d, reason, message) {
			changed = true
		}
		if changed {
			errs = append(errs, r.updateStatus(ctx, w, d))
		}
	}
	return errs
}

// order binds one pending device, or returns the reason it waits.
func (r *Reconciler) order(w *world, d *gpuv1alpha1.GPUDevice) (string, string) {
	class := w.classes[d.Spec.ClassName]
	switch {
	case class == nil:
		return string(model.ReasonClassNotFound), fmt.Sprintf("GPUClass %s does not exist", d.Spec.ClassName)
	case !class.DeletionTimestamp.IsZero():
		return ReasonClassDeleted, fmt.Sprintf("GPUClass %s is being deleted and takes no new orders", d.Spec.ClassName)
	case w.quotaErr != nil:
		return ReasonBudgetsRead, "tenant GPU budgets cannot be read"
	}
	if err := w.snap.CheckOrder(d.Namespace, d.Spec.ClassName, int64(d.Spec.Count), w.entitled[d.Namespace]); err != nil {
		var refusal *model.Refusal
		if errors.As(err, &refusal) {
			return string(refusal.Reason), refusal.Error()
		}
		return string(model.ReasonBudget), err.Error()
	}
	set := []*gpuv1alpha1.GPUDevice{d}
	if target := w.pick(set); target != "" {
		w.placeOn(set, target)
		return "", ""
	}
	return ReasonCapacity, fmt.Sprintf("no Ready node has %d free units of GPUClass %s", d.Spec.Count, d.Spec.ClassName)
}