21848362404

Committed 10 Feb 2026 01:46AM UTC coverage: 80.713% (-0.1%) from 80.83%

Build # 21848362404

Build Type

push

github

Committed by

lockwobr

Commit Message

fix: resolve webhook caBundle deadlock during helm upgrade

During helm upgrade, the webhook configurations' caBundle field was
reset to empty, causing new pods to fail readiness checks while the
old leader pod never detected the change (only watched the cert
Secret, with a 24h requeue). This created a deadlock where no pod
could fix the caBundle.

- Watch ValidatingWebhookConfiguration and MutatingWebhookConfiguration
  so the leader detects caBundle changes immediately
- Use bytes.Equal for caBundle comparison instead of len==0 so stale
  values are corrected, not just empty ones
- Remove caBundle from Helm webhook templates so upgrades stop
  resetting operator-managed values

Coverage Stats

4 of 15 new or added lines in 1 file covered. (26.67%)

4 existing lines in 1 file now uncovered.

6817 of 8446 relevant lines covered (80.71%)

4.27 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.83

/operator/internal/controller/skyhook_controller.go

/*
 * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 * SPDX-License-Identifier: Apache-2.0
 *
 *
 * 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 (
        "cmp"
        "context"
        "crypto/sha256"
        "encoding/hex"
        "encoding/json"
        "errors"
        "fmt"
        "reflect"
        "slices"
        "sort"
        "strings"
        "time"

        "github.com/NVIDIA/skyhook/operator/api/v1alpha1"
        "github.com/NVIDIA/skyhook/operator/internal/dal"
        "github.com/NVIDIA/skyhook/operator/internal/version"
        "github.com/NVIDIA/skyhook/operator/internal/wrapper"
        "github.com/go-logr/logr"

        corev1 "k8s.io/api/core/v1"
        policyv1 "k8s.io/api/policy/v1"
        apierrors "k8s.io/apimachinery/pkg/api/errors"
        "k8s.io/apimachinery/pkg/api/resource"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/runtime"
        "k8s.io/apimachinery/pkg/types"
        utilerrors "k8s.io/apimachinery/pkg/util/errors"
        "k8s.io/client-go/tools/record"
        "k8s.io/kubernetes/pkg/util/taints"
        ctrl "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"
        "sigs.k8s.io/controller-runtime/pkg/log"
        "sigs.k8s.io/controller-runtime/pkg/reconcile"
)

const (
        EventsReasonSkyhookApply       = "Apply"
        EventsReasonSkyhookInterrupt   = "Interrupt"
        EventsReasonSkyhookDrain       = "Drain"
        EventsReasonSkyhookStateChange = "State"
        EventsReasonNodeReboot         = "Reboot"
        EventTypeNormal                = "Normal"
        // EventTypeWarning = "Warning"
        TaintUnschedulable     = corev1.TaintNodeUnschedulable
        InterruptContainerName = "interrupt"

        SkyhookFinalizer = "skyhook.nvidia.com/skyhook"
)

type SkyhookOperatorOptions struct {
        Namespace            string        `env:"NAMESPACE, default=skyhook"`
        MaxInterval          time.Duration `env:"DEFAULT_INTERVAL, default=10m"`
        ImagePullSecret      string        `env:"IMAGE_PULL_SECRET"`
        CopyDirRoot          string        `env:"COPY_DIR_ROOT, default=/var/lib/skyhook"`
        ReapplyOnReboot      bool          `env:"REAPPLY_ON_REBOOT, default=false"`
        RuntimeRequiredTaint string        `env:"RUNTIME_REQUIRED_TAINT, default=skyhook.nvidia.com=runtime-required:NoSchedule"`
        PauseImage           string        `env:"PAUSE_IMAGE, default=registry.k8s.io/pause:3.10"`
        AgentImage           string        `env:"AGENT_IMAGE, default=ghcr.io/nvidia/skyhook/agent:latest"` // TODO: this needs to be updated with a working default
        AgentLogRoot         string        `env:"AGENT_LOG_ROOT, default=/var/log/skyhook"`
}

func (o *SkyhookOperatorOptions) Validate() error {

        messages := make([]string, 0)
        if o.Namespace == "" {
                messages = append(messages, "namespace must be set")
        }
        if o.CopyDirRoot == "" {
                messages = append(messages, "copy dir root must be set")
        }
        if o.RuntimeRequiredTaint == "" {
                messages = append(messages, "runtime required taint must be set")
        }
        if o.MaxInterval < time.Minute {
                messages = append(messages, "max interval must be at least 1 minute")
        }

        // CopyDirRoot must start with /
        if !strings.HasPrefix(o.CopyDirRoot, "/") {
                messages = append(messages, "copy dir root must start with /")
        }

        // RuntimeRequiredTaint must be parsable and must not be a deletion
        _, delete, err := taints.ParseTaints([]string{o.RuntimeRequiredTaint})
        if err != nil {
                messages = append(messages, fmt.Sprintf("runtime required taint is invalid: %s", err.Error()))
        }
        if len(delete) > 0 {
                messages = append(messages, "runtime required taint must not be a deletion")
        }

        if o.AgentImage == "" {
                messages = append(messages, "agent image must be set")
        }

        if !strings.Contains(o.AgentImage, ":") {
                messages = append(messages, "agent image must contain a tag")
        }

        if o.PauseImage == "" {
                messages = append(messages, "pause image must be set")
        }

        if !strings.Contains(o.PauseImage, ":") {
                messages = append(messages, "pause image must contain a tag")
        }

        if len(messages) > 0 {
                return errors.New(strings.Join(messages, ", "))
        }

        return nil
}

// AgentVersion returns the image tag portion of AgentImage
func (o *SkyhookOperatorOptions) AgentVersion() string {
        parts := strings.Split(o.AgentImage, ":")
        return parts[len(parts)-1]
}

func (o *SkyhookOperatorOptions) GetRuntimeRequiredTaint() corev1.Taint {
        to_add, _, _ := taints.ParseTaints([]string{o.RuntimeRequiredTaint})
        return to_add[0]
}

func (o *SkyhookOperatorOptions) GetRuntimeRequiredToleration() corev1.Toleration {
        taint := o.GetRuntimeRequiredTaint()
        return corev1.Toleration{
                Key:      taint.Key,
                Operator: corev1.TolerationOpEqual,
                Value:    taint.Value,
                Effect:   taint.Effect,
        }
}

// force type checking against this interface
var _ reconcile.Reconciler = &SkyhookReconciler{}

func NewSkyhookReconciler(schema *runtime.Scheme, c client.Client, recorder record.EventRecorder, opts SkyhookOperatorOptions) (*SkyhookReconciler, error) {

        err := opts.Validate()
        if err != nil {
                return nil, fmt.Errorf("invalid skyhook operator options: %w", err)
        }

        return &SkyhookReconciler{
                Client:   c,
                scheme:   schema,
                recorder: recorder,
                opts:     opts,
                dal:      dal.New(c),
        }, nil
}

// SkyhookReconciler reconciles a Skyhook object
type SkyhookReconciler struct {
        client.Client
        scheme   *runtime.Scheme
        recorder record.EventRecorder
        opts     SkyhookOperatorOptions
        dal      dal.DAL
}

// SetupWithManager sets up the controller with the Manager.
func (r *SkyhookReconciler) SetupWithManager(mgr ctrl.Manager) error {

        // indexes allow for query on fields to use the local cache
        indexer := mgr.GetFieldIndexer()
        err := indexer.
                IndexField(context.TODO(), &corev1.Pod{}, "spec.nodeName", func(o client.Object) []string {
                        pod, ok := o.(*corev1.Pod)
                        if !ok {
                                return nil
                        }
                        return []string{pod.Spec.NodeName}
                })

        if err != nil {
                return err
        }

        ehandler := &eventHandler{
                logger: mgr.GetLogger(),
                dal:    dal.New(r.Client),
        }

        return ctrl.NewControllerManagedBy(mgr).
                For(&v1alpha1.Skyhook{}).
                Watches(
                        &corev1.Pod{},
                        handler.EnqueueRequestsFromMapFunc(podHandlerFunc),
                ).
                Watches(
                        &corev1.Node{},
                        ehandler,
                ).
                Complete(r)
}

// CRD Permissions
//+kubebuilder:rbac:groups=skyhook.nvidia.com,resources=skyhooks,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=skyhook.nvidia.com,resources=skyhooks/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=skyhook.nvidia.com,resources=skyhooks/finalizers,verbs=update
//+kubebuilder:rbac:groups=skyhook.nvidia.com,resources=deploymentpolicies,verbs=get;list;watch

// core permissions
//+kubebuilder:rbac:groups=core,resources=nodes,verbs=get;list;update;patch;watch
//+kubebuilder:rbac:groups=core,resources=nodes/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=core,resources=pods/eviction,verbs=create
//+kubebuilder:rbac:groups=core,resources=events,verbs=create;patch
//+kubebuilder:rbac:groups=core,resources=configmaps,verbs=get;list;watch;create;update;patch;delete

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.16.3/pkg/reconcile
func (r *SkyhookReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
        logger := log.FromContext(ctx)
        // split off requests for pods
        if strings.HasPrefix(req.Name, "pod---") {
                name := strings.Split(req.Name, "pod---")[1]
                pod, err := r.dal.GetPod(ctx, req.Namespace, name)
                if err == nil && pod != nil { // if pod, then call other wise not a pod
                        return r.PodReconcile(ctx, pod)
                }
                return ctrl.Result{}, err
        }

        // get all skyhooks (SCR)
        skyhooks, err := r.dal.GetSkyhooks(ctx)
        if err != nil {
                // error, going to requeue and backoff
                logger.Error(err, "error getting skyhooks")
                return ctrl.Result{}, err
        }

        // if there are no skyhooks, so actually nothing to do, so don't requeue
        if skyhooks == nil || len(skyhooks.Items) == 0 {
                return ctrl.Result{}, nil
        }

        // get all nodes
        nodes, err := r.dal.GetNodes(ctx)
        if err != nil {
                // error, going to requeue and backoff
                logger.Error(err, "error getting nodes")
                return ctrl.Result{}, err
        }

        // if no nodes, well not work to do either
        if nodes == nil || len(nodes.Items) == 0 {
                // no nodes, so nothing to do
                return ctrl.Result{}, nil
        }

        // get all deployment policies
        deploymentPolicies, err := r.dal.GetDeploymentPolicies(ctx)
        if err != nil {
                logger.Error(err, "error getting deployment policies")
                return ctrl.Result{}, err
        }

        // TODO: this build state could error in a lot of ways, and I think we might want to move towards partial state
        // mean if we cant get on SCR state, great, process that one and error

        // BUILD cluster state from all skyhooks, and all nodes
        // this filters and pairs up nodes to skyhooks, also provides help methods for introspection and mutation
        clusterState, err := BuildState(skyhooks, nodes, deploymentPolicies)
        if err != nil {
                // error, going to requeue and backoff
                logger.Error(err, "error building cluster state")
                return ctrl.Result{}, err
        }

        if yes, result, err := shouldReturn(r.HandleMigrations(ctx, clusterState)); yes {
                return result, err
        }

        if yes, result, err := shouldReturn(r.TrackReboots(ctx, clusterState)); yes {
                return result, err
        }

        // node picker is for selecting nodes to do work, tries maintain a prior of nodes between SCRs
        nodePicker := NewNodePicker(logger, r.opts.GetRuntimeRequiredToleration())

        errs := make([]error, 0)
        var result *ctrl.Result

        for _, skyhook := range clusterState.skyhooks {
                if yes, result, err := shouldReturn(r.HandleFinalizer(ctx, skyhook)); yes {
                        return result, err
                }

                if yes, result, err := shouldReturn(r.ReportState(ctx, clusterState, skyhook)); yes {
                        return result, err
                }

                if skyhook.IsPaused() {
                        if yes, result, err := shouldReturn(r.UpdatePauseStatus(ctx, clusterState, skyhook)); yes {
                                return result, err
                        }
                        continue
                }

                if yes, result, err := r.validateAndUpsertSkyhookData(ctx, skyhook, clusterState); yes {
                        return result, err
                }

                changed := IntrospectSkyhook(skyhook, clusterState.skyhooks)
                if changed {
                        _, errs := r.SaveNodesAndSkyhook(ctx, clusterState, skyhook)
                        if len(errs) > 0 {
                                return ctrl.Result{RequeueAfter: time.Second * 2}, utilerrors.NewAggregate(errs)
                        }
                        return ctrl.Result{RequeueAfter: time.Second * 2}, nil
                }

                _, err := HandleVersionChange(skyhook)
                if err != nil {
                        return ctrl.Result{RequeueAfter: time.Second * 2}, fmt.Errorf("error getting packages to uninstall: %w", err)
                }
        }

        // Process all non-complete, non-disabled skyhooks (in priority order)
        // Each skyhook is processed only for nodes that are ready (all higher-priority skyhooks complete on that node)
        // This enables per-node priority ordering: nodes can progress independently
        result, err = r.processSkyhooksPerNode(ctx, clusterState, nodePicker, logger)
        if err != nil {
                errs = append(errs, err)
        }

        err = r.HandleRuntimeRequired(ctx, clusterState)
        if err != nil {
                errs = append(errs, err)
        }

        if len(errs) > 0 {
                err := utilerrors.NewAggregate(errs)
                return ctrl.Result{}, err
        }

        if result != nil {
                return *result, nil
        }

        // default happy retry after max
        return ctrl.Result{RequeueAfter: r.opts.MaxInterval}, nil
}

// processSkyhooksPerNode processes all skyhooks for nodes that are ready (per-node priority ordering).
// A node is ready for a skyhook if all higher-priority skyhooks are complete on that specific node.
func (r *SkyhookReconciler) processSkyhooksPerNode(ctx context.Context, clusterState *clusterState, nodePicker *NodePicker, logger logr.Logger) (*ctrl.Result, error) {
        var result *ctrl.Result
        var errs []error

        for _, skyhook := range clusterState.skyhooks {
                if skyhook.IsComplete() || skyhook.IsDisabled() || skyhook.IsPaused() {
                        continue
                }

                // Check if any nodes are ready for this skyhook
                if !hasReadyNodesForSkyhook(skyhook, clusterState.skyhooks) {
                        continue
                }

                res, err := r.RunSkyhookPackages(ctx, clusterState, nodePicker, skyhook)
                if err != nil {
                        logger.Error(err, "error processing skyhook", "skyhook", skyhook.GetSkyhook().Name)
                        errs = append(errs, err)
                }
                if res != nil {
                        result = res
                }
        }

        if len(errs) > 0 {
                return result, utilerrors.NewAggregate(errs)
        }
        return result, nil
}

// hasReadyNodesForSkyhook checks if any nodes are ready to process this skyhook.
// A node is ready if it's not complete and all higher-priority skyhooks are complete on that node.
func hasReadyNodesForSkyhook(skyhook SkyhookNodes, allSkyhooks []SkyhookNodes) bool {
        for _, node := range skyhook.GetNodes() {
                if !node.IsComplete() && IsNodeReadyForSkyhook(node.GetNode().Name, skyhook, allSkyhooks) {
                        return true
                }
        }
        return false
}

func shouldReturn(updates bool, err error) (bool, ctrl.Result, error) {
        if err != nil {
                return true, ctrl.Result{}, err
        }
        if updates {
                return true, ctrl.Result{RequeueAfter: time.Second * 2}, nil
        }
        return false, ctrl.Result{}, nil
}

func (r *SkyhookReconciler) HandleMigrations(ctx context.Context, clusterState *clusterState) (bool, error) {

        updates := false

        if version.VERSION == "" {
                // this means the binary was complied without version information
                return false, nil
        }

        logger := log.FromContext(ctx)
        errors := make([]error, 0)
        for _, skyhook := range clusterState.skyhooks {

                err := skyhook.Migrate(logger)
                if err != nil {
                        return false, fmt.Errorf("error migrating skyhook [%s]: %w", skyhook.GetSkyhook().Name, err)
                }

                if err := skyhook.GetSkyhook().Skyhook.Validate(); err != nil {
                        return false, fmt.Errorf("error validating skyhook [%s]: %w", skyhook.GetSkyhook().Name, err)
                }

                for _, node := range skyhook.GetNodes() {
                        if node.Changed() {
                                err := r.Status().Patch(ctx, node.GetNode(), client.MergeFrom(clusterState.tracker.GetOriginal(node.GetNode())))
                                if err != nil {
                                        errors = append(errors, fmt.Errorf("error patching node [%s]: %w", node.GetNode().Name, err))
                                }

                                err = r.Patch(ctx, node.GetNode(), client.MergeFrom(clusterState.tracker.GetOriginal(node.GetNode())))
                                if err != nil {
                                        errors = append(errors, fmt.Errorf("error patching node [%s]: %w", node.GetNode().Name, err))
                                }
                                updates = true
                        }
                }

                if skyhook.GetSkyhook().Updated {
                        // need to do this because SaveNodesAndSkyhook only saves skyhook status, not the main skyhook object where the annotations are
                        // additionally it needs to be an update, a patch nils out the annotations for some reason, which the save function does a patch

                        if err = r.Status().Update(ctx, skyhook.GetSkyhook().Skyhook); err != nil {
                                return false, fmt.Errorf("error updating during migration skyhook status [%s]: %w", skyhook.GetSkyhook().Name, err)
                        }

                        // because of conflict issues (409) we need to do things a bit differently here.
                        // We might be able to use server side apply in the future, but for now we need to do this
                        // https://kubernetes.io/docs/reference/using-api/server-side-apply/
                        // https://github.com/kubernetes-sigs/controller-runtime/issues/347

                        // work around for now is to grab a new copy of the object, and then patch it

                        newskyhook, err := r.dal.GetSkyhook(ctx, skyhook.GetSkyhook().Name)
                        if err != nil {
                                return false, fmt.Errorf("error getting skyhook to migrate [%s]: %w", skyhook.GetSkyhook().Name, err)
                        }
                        newPatch := client.MergeFrom(newskyhook.DeepCopy())

                        // set version
                        wrapper.NewSkyhookWrapper(newskyhook).SetVersion()

                        if err = r.Patch(ctx, newskyhook, newPatch); err != nil {
                                return false, fmt.Errorf("error updating during migration skyhook [%s]: %w", skyhook.GetSkyhook().Name, err)
                        }

                        updates = true
                }
        }

        if len(errors) > 0 {
                return false, utilerrors.NewAggregate(errors)
        }

        return updates, nil
}

// ReportState computes and puts important information into the skyhook status so that monitoring tools such as k9s
// can see the information at a glance. For example, the number of completed nodes and the list of packages in the skyhook.
func (r *SkyhookReconciler) ReportState(ctx context.Context, clusterState *clusterState, skyhook SkyhookNodes) (bool, error) {

        // save updated state to skyhook status
        skyhook.ReportState()

        if skyhook.GetSkyhook().Updated {
                _, errs := r.SaveNodesAndSkyhook(ctx, clusterState, skyhook)
                if len(errs) > 0 {
                        return false, utilerrors.NewAggregate(errs)
                }
                return true, nil
        }

        return false, nil
}

func (r *SkyhookReconciler) UpdatePauseStatus(ctx context.Context, clusterState *clusterState, skyhook SkyhookNodes) (bool, error) {
        changed := UpdateSkyhookPauseStatus(skyhook)

        if changed {
                _, errs := r.SaveNodesAndSkyhook(ctx, clusterState, skyhook)
                if len(errs) > 0 {
                        return false, utilerrors.NewAggregate(errs)
                }
                return true, nil
        }

        return false, nil
}

func (r *SkyhookReconciler) TrackReboots(ctx context.Context, clusterState *clusterState) (bool, error) {

        updates := false
        errs := make([]error, 0)

        for _, skyhook := range clusterState.skyhooks {
                if skyhook.GetSkyhook().Status.NodeBootIds == nil {
                        skyhook.GetSkyhook().Status.NodeBootIds = make(map[string]string)
                }

                for _, node := range skyhook.GetNodes() {
                        id, ok := skyhook.GetSkyhook().Status.NodeBootIds[node.GetNode().Name]

                        if !ok { // new node
                                skyhook.GetSkyhook().Status.NodeBootIds[node.GetNode().Name] = node.GetNode().Status.NodeInfo.BootID
                                skyhook.GetSkyhook().Updated = true
                        }

                        if id != "" && id != node.GetNode().Status.NodeInfo.BootID { // node rebooted
                                if r.opts.ReapplyOnReboot {
                                        r.recorder.Eventf(skyhook.GetSkyhook().Skyhook, EventTypeNormal, EventsReasonNodeReboot, "detected reboot, resetting node [%s] to be reapplied", node.GetNode().Name)
                                        r.recorder.Eventf(node.GetNode(), EventTypeNormal, EventsReasonNodeReboot, "detected reboot, resetting node for [%s] to be reapplied", node.GetSkyhook().Name)
                                        node.Reset()
                                }
                                skyhook.GetSkyhook().Status.NodeBootIds[node.GetNode().Name] = node.GetNode().Status.NodeInfo.BootID
                                skyhook.GetSkyhook().Updated = true
                        }

                        if node.Changed() { // update
                                updates = true
                                err := r.Update(ctx, node.GetNode())
                                if err != nil {
                                        errs = append(errs, fmt.Errorf("error updating node after reboot [%s]: %w", node.GetNode().Name, err))
                                }
                        }
                }
                if skyhook.GetSkyhook().Updated { // update
                        updates = true
                        err := r.Status().Update(ctx, skyhook.GetSkyhook().Skyhook)
                        if err != nil {
                                errs = append(errs, fmt.Errorf("error updating skyhook status after reboot [%s]: %w", skyhook.GetSkyhook().Name, err))
                        }
                }
        }

        return updates, utilerrors.NewAggregate(errs)
}

// RunSkyhookPackages runs all skyhook packages then saves and requeues if changes were made
func (r *SkyhookReconciler) RunSkyhookPackages(ctx context.Context, clusterState *clusterState, nodePicker *NodePicker, skyhook SkyhookNodes) (*ctrl.Result, error) {

        logger := log.FromContext(ctx)
        requeue := false

        toUninstall, err := HandleVersionChange(skyhook)
        if err != nil {
                return nil, fmt.Errorf("error getting packages to uninstall: %w", err)
        }

        changed := IntrospectSkyhook(skyhook, clusterState.skyhooks)
        if !changed && skyhook.IsComplete() {
                return nil, nil
        }

        selectedNode := nodePicker.SelectNodes(skyhook)

        for _, node := range selectedNode {
                // Skip nodes that are waiting on higher-priority skyhooks
                // This enables per-node priority ordering
                if !IsNodeReadyForSkyhook(node.GetNode().Name, skyhook, clusterState.skyhooks) {
                        continue
                }

                if node.IsComplete() && !node.Changed() {
                        continue
                }

                toRun, err := node.RunNext()
                if err != nil {
                        return nil, fmt.Errorf("error getting next packages to run: %w", err)
                }

                // prepend the uninstall packages so they are ran first
                toRun = append(toUninstall, toRun...)

                interrupt, pack := fudgeInterruptWithPriority(toRun, skyhook.GetSkyhook().GetConfigUpdates(), skyhook.GetSkyhook().GetConfigInterrupts())

                for _, f := range toRun {

                        ok, err := r.ProcessInterrupt(ctx, node, f, interrupt, interrupt != nil && f.Name == pack)
                        if err != nil {
                                // TODO: error handle
                                return nil, fmt.Errorf("error processing if we should interrupt [%s:%s]: %w", f.Name, f.Version, err)
                        }
                        if !ok {
                                requeue = true
                                continue
                        }

                        err = r.ApplyPackage(ctx, logger, clusterState, node, f, interrupt != nil && f.Name == pack)
                        if err != nil {
                                return nil, fmt.Errorf("error applying package [%s:%s]: %w", f.Name, f.Version, err)
                        }

                        // process one package at a time
                        if skyhook.GetSkyhook().Spec.Serial {
                                return &ctrl.Result{Requeue: true}, nil
                        }
                }
        }

        saved, errs := r.SaveNodesAndSkyhook(ctx, clusterState, skyhook)
        if len(errs) > 0 {
                return &ctrl.Result{}, utilerrors.NewAggregate(errs)
        }
        if saved {
                requeue = true
        }

        if !skyhook.IsComplete() || requeue {
                return &ctrl.Result{RequeueAfter: time.Second * 2}, nil // not sure this is better then just requeue bool
        }

        return nil, utilerrors.NewAggregate(errs)
}

// SaveNodesAndSkyhook saves nodes and skyhook and will update the events if the skyhook status changes
func (r *SkyhookReconciler) SaveNodesAndSkyhook(ctx context.Context, clusterState *clusterState, skyhook SkyhookNodes) (bool, []error) {
        saved := false
        errs := make([]error, 0)

        for _, node := range skyhook.GetNodes() {
                patch := client.StrategicMergeFrom(clusterState.tracker.GetOriginal(node.GetNode()))
                if node.Changed() {
                        err := r.Patch(ctx, node.GetNode(), patch)
                        if err != nil {
                                errs = append(errs, fmt.Errorf("error patching node [%s]: %w", node.GetNode().Name, err))
                        }
                        saved = true

                        err = r.UpsertNodeLabelsAnnotationsPackages(ctx, skyhook.GetSkyhook(), node.GetNode())
                        if err != nil {
                                errs = append(errs, fmt.Errorf("error upserting labels, annotations, and packages config map for node [%s]: %w", node.GetNode().Name, err))
                        }

                        if node.IsComplete() {
                                r.recorder.Eventf(node.GetNode(), EventTypeNormal, EventsReasonSkyhookStateChange, "Skyhook [%s] complete.", skyhook.GetSkyhook().Name)

                                // since node is complete remove from priority
                                if _, ok := skyhook.GetSkyhook().Status.NodePriority[node.GetNode().Name]; ok {
                                        delete(skyhook.GetSkyhook().Status.NodePriority, node.GetNode().Name)
                                        skyhook.GetSkyhook().Updated = true
                                }
                        }
                }

                // updates node's condition
                node.UpdateCondition()
                if node.Changed() {
                        // conditions are in status
                        err := r.Status().Patch(ctx, node.GetNode(), patch)
                        if err != nil {
                                errs = append(errs, fmt.Errorf("error patching node status [%s]: %w", node.GetNode().Name, err))
                        }
                        saved = true
                }

                if node.GetSkyhook() != nil && node.GetSkyhook().Updated {
                        skyhook.GetSkyhook().Updated = true
                }
        }

        if skyhook.GetSkyhook().Updated {
                patch := client.MergeFrom(clusterState.tracker.GetOriginal(skyhook.GetSkyhook().Skyhook))
                err := r.Status().Patch(ctx, skyhook.GetSkyhook().Skyhook, patch)
                if err != nil {
                        errs = append(errs, err)
                }
                saved = true

                if skyhook.GetPriorStatus() != "" && skyhook.GetPriorStatus() != skyhook.Status() {
                        // we transitioned, fire event
                        r.recorder.Eventf(skyhook.GetSkyhook(), EventTypeNormal, EventsReasonSkyhookStateChange, "Skyhook transitioned [%s] -> [%s]", skyhook.GetPriorStatus(), skyhook.Status())
                }
        }

        if len(errs) > 0 {
                saved = false
        }
        return saved, errs
}

// HandleVersionChange updates the state for the node or skyhook if a version is changed on a package
func HandleVersionChange(skyhook SkyhookNodes) ([]*v1alpha1.Package, error) {
        toUninstall := make([]*v1alpha1.Package, 0)
        versionChangeDetected := false

        for _, node := range skyhook.GetNodes() {
                nodeState, err := node.State()
                if err != nil {
                        return nil, err
                }

                for _, packageStatus := range nodeState {
                        upgrade := false

                        _package, exists := skyhook.GetSkyhook().Spec.Packages[packageStatus.Name]
                        if exists && _package.Version == packageStatus.Version {
                                continue // no uninstall needed for package
                        }

                        packageStatusRef := v1alpha1.PackageRef{
                                Name:    packageStatus.Name,
                                Version: packageStatus.Version,
                        }

                        if !exists && packageStatus.Stage != v1alpha1.StageUninstall {
                                // Start uninstall of old package
                                err := node.Upsert(packageStatusRef, packageStatus.Image, v1alpha1.StateInProgress, v1alpha1.StageUninstall, 0, "")
                                if err != nil {
                                        return nil, fmt.Errorf("error updating node status: %w", err)
                                }
                                versionChangeDetected = true
                        } else if exists && _package.Version != packageStatus.Version {
                                versionChangeDetected = true
                                comparison := version.Compare(_package.Version, packageStatus.Version)
                                if comparison == -2 {
                                        return nil, errors.New("error comparing package versions: invalid version string provided enabling webhooks validates versions before being applied")
                                }

                                if comparison == 1 {
                                        _packageStatus, found := node.PackageStatus(_package.GetUniqueName())
                                        if found && _packageStatus.Stage == v1alpha1.StageUpgrade {
                                                continue
                                        }

                                        // start upgrade of package
                                        err := node.Upsert(_package.PackageRef, _package.Image, v1alpha1.StateInProgress, v1alpha1.StageUpgrade, 0, _package.ContainerSHA)
                                        if err != nil {
                                                return nil, fmt.Errorf("error updating node status: %w", err)
                                        }

                                        upgrade = true
                                } else if comparison == -1 && packageStatus.Stage != v1alpha1.StageUninstall {
                                        // Start uninstall of old package
                                        err := node.Upsert(packageStatusRef, packageStatus.Image, v1alpha1.StateInProgress, v1alpha1.StageUninstall, 0, "")
                                        if err != nil {
                                                return nil, fmt.Errorf("error updating node status: %w", err)
                                        }

                                        // If version changed then update new version to wait
                                        err = node.Upsert(_package.PackageRef, _package.Image, v1alpha1.StateSkipped, v1alpha1.StageUninstall, 0, _package.ContainerSHA)
                                        if err != nil {
                                                return nil, fmt.Errorf("error updating node status: %w", err)
                                        }
                                }
                        }

                        // only need to create a feaux package for uninstall since it won't be in the DAG (Upgrade will)
                        newPackageStatus, found := node.PackageStatus(packageStatusRef.GetUniqueName())
                        if !upgrade && found && newPackageStatus.Stage == v1alpha1.StageUninstall && newPackageStatus.State == v1alpha1.StateInProgress {
                                // create fake package with the info we can salvage from the node state
                                newPackage := &v1alpha1.Package{
                                        PackageRef: packageStatusRef,
                                        Image:      packageStatus.Image,
                                }

                                // Add package to uninstall list if it's not already present
                                found := false
                                for _, uninstallPackage := range toUninstall {
                                        if reflect.DeepEqual(uninstallPackage, newPackage) {
                                                found = true
                                        }
                                }

                                if !found {
                                        toUninstall = append(toUninstall, newPackage)
                                }
                        }

                        // remove all config updates for the package since it's being uninstalled or
                        // upgraded. NOTE: The config updates must be removed whenever the version changes
                        // or else the package interrupt may be skipped if there is one
                        skyhook.GetSkyhook().RemoveConfigUpdates(_package.Name)

                        // set the node and skyhook status to in progress
                        node.SetStatus(v1alpha1.StatusInProgress)
                }
        }

        // Auto-reset batch state when version changes are detected (if configured)
        if versionChangeDetected {
                resetSkyhookBatchState(skyhook)
        }

        return toUninstall, nil
}

// helper for get a point to a ref
func ptr[E any](e E) *E {
        return &e
}

// generateSafeName generates a consistent name for Kubernetes resources that is unique
// while staying within the specified character limit
func generateSafeName(maxLen int, nameParts ...string) string {
        name := strings.Join(nameParts, "-")
        // Replace dots with dashes as they're not allowed in resource names
        name = strings.ReplaceAll(name, ".", "-")

        unique := sha256.Sum256([]byte(name))
        uniqueStr := hex.EncodeToString(unique[:])[:8]

        maxlen := maxLen - len(uniqueStr) - 1
        if len(name) > maxlen {
                name = name[:maxlen]
        }

        return strings.ToLower(fmt.Sprintf("%s-%s", name, uniqueStr))
}

func (r *SkyhookReconciler) UpsertNodeLabelsAnnotationsPackages(ctx context.Context, skyhook *wrapper.Skyhook, node *corev1.Node) error {
        // No work to do if there is no labels or annotations for node
        if len(node.Labels) == 0 && len(node.Annotations) == 0 {
                return nil
        }

        annotations, err := json.Marshal(node.Annotations)
        if err != nil {
                return fmt.Errorf("error converting annotations into byte array: %w", err)
        }

        labels, err := json.Marshal(node.Labels)
        if err != nil {
                return fmt.Errorf("error converting labels into byte array: %w", err)
        }

        // marshal intermediary package metadata for the agent
        metadata := NewSkyhookMetadata(r.opts, skyhook)
        packages, err := metadata.Marshal()
        if err != nil {
                return fmt.Errorf("error converting packages into byte array: %w", err)
        }

        configMapName := generateSafeName(253, skyhook.Name, node.Name, "metadata")
        newCM := &corev1.ConfigMap{
                ObjectMeta: metav1.ObjectMeta{
                        Name:      configMapName,
                        Namespace: r.opts.Namespace,
                        Labels: map[string]string{
                                fmt.Sprintf("%s/skyhook-node-meta", v1alpha1.METADATA_PREFIX): skyhook.Name,
                        },
                        Annotations: map[string]string{
                                fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX):      skyhook.Name,
                                fmt.Sprintf("%s/Node.name", v1alpha1.METADATA_PREFIX): node.Name,
                        },
                },
                Data: map[string]string{
                        "annotations.json": string(annotations),
                        "labels.json":      string(labels),
                        "packages.json":    string(packages),
                },
        }

        if err := ctrl.SetControllerReference(skyhook.Skyhook, newCM, r.scheme); err != nil {
                return fmt.Errorf("error setting ownership: %w", err)
        }

        existingConfigMap := &corev1.ConfigMap{}
        err = r.Get(ctx, client.ObjectKey{Namespace: r.opts.Namespace, Name: configMapName}, existingConfigMap)
        if err != nil {
                if apierrors.IsNotFound(err) {
                        // create
                        err := r.Create(ctx, newCM)
                        if err != nil {
                                return fmt.Errorf("error creating config map [%s]: %w", newCM.Name, err)
                        }
                } else {
                        return fmt.Errorf("error getting config map: %w", err)
                }
        } else {
                if !reflect.DeepEqual(existingConfigMap.Data, newCM.Data) {
                        // update
                        err := r.Update(ctx, newCM)
                        if err != nil {
                                return fmt.Errorf("error updating config map [%s]: %w", newCM.Name, err)
                        }
                }
        }

        return nil
}

// HandleConfigUpdates checks whether the configMap on a package was updated and if it was the configmap will
// be updated and the package will be put into config mode if the package is complete or erroring
func (r *SkyhookReconciler) HandleConfigUpdates(ctx context.Context, clusterState *clusterState, skyhook SkyhookNodes, _package v1alpha1.Package, oldConfigMap, newConfigMap *corev1.ConfigMap) (bool, error) {
        completedNodes, nodeCount := 0, len(skyhook.GetNodes())
        erroringNode := false

        // if configmap changed
        if !reflect.DeepEqual(oldConfigMap.Data, newConfigMap.Data) {
                for _, node := range skyhook.GetNodes() {
                        exists, err := r.PodExists(ctx, node.GetNode().Name, skyhook.GetSkyhook().Name, &_package)
                        if err != nil {
                                return false, err
                        }

                        if !exists && node.IsPackageComplete(_package) {
                                completedNodes++
                        }

                        // if we have an erroring node in the config, interrupt, or post-interrupt mode
                        // then we will restart the config changes
                        if packageStatus, found := node.PackageStatus(_package.GetUniqueName()); found {
                                switch packageStatus.Stage {
                                case v1alpha1.StageConfig, v1alpha1.StageInterrupt, v1alpha1.StagePostInterrupt:
                                        if packageStatus.State == v1alpha1.StateErroring {
                                                erroringNode = true

                                                // delete the erroring pod from the node so that it can be recreated
                                                // with the updated configmap
                                                pods, err := r.dal.GetPods(ctx,
                                                        client.MatchingFields{
                                                                "spec.nodeName": node.GetNode().Name,
                                                        },
                                                        client.MatchingLabels{
                                                                fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX):    skyhook.GetSkyhook().Name,
                                                                fmt.Sprintf("%s/package", v1alpha1.METADATA_PREFIX): fmt.Sprintf("%s-%s", _package.Name, _package.Version),
                                                        },
                                                )
                                                if err != nil {
                                                        return false, err
                                                }

                                                if pods != nil {
                                                        for _, pod := range pods.Items {
                                                                err := r.Delete(ctx, &pod)
                                                                if err != nil {
                                                                        return false, err
                                                                }
                                                        }
                                                }
                                        }
                                }
                        }
                }

                // if the update is complete or there is an erroring node put the package back into
                // the config mode and update the config map
                if completedNodes == nodeCount || erroringNode {
                        // get the keys in the configmap that changed
                        newConfigUpdates := make([]string, 0)
                        for key, new_val := range newConfigMap.Data {
                                if old_val, exists := oldConfigMap.Data[key]; !exists || old_val != new_val {
                                        newConfigUpdates = append(newConfigUpdates, key)
                                }
                        }

                        // if updates completed then clear out old config updates as they are finished
                        if completedNodes == nodeCount {
                                skyhook.GetSkyhook().RemoveConfigUpdates(_package.Name)
                        }

                        // Add the new changed keys to the config updates
                        skyhook.GetSkyhook().AddConfigUpdates(_package.Name, newConfigUpdates...)

                        for _, node := range skyhook.GetNodes() {
                                err := node.Upsert(_package.PackageRef, _package.Image, v1alpha1.StateInProgress, v1alpha1.StageConfig, 0, _package.ContainerSHA)
                                if err != nil {
                                        return false, fmt.Errorf("error upserting node status [%s]: %w", node.GetNode().Name, err)
                                }

                                node.SetStatus(v1alpha1.StatusInProgress)
                        }

                        _, errs := r.SaveNodesAndSkyhook(ctx, clusterState, skyhook)
                        if len(errs) > 0 {
                                return false, utilerrors.NewAggregate(errs)
                        }

                        // update config map
                        err := r.Update(ctx, newConfigMap)
                        if err != nil {
                                return false, fmt.Errorf("error updating config map [%s]: %w", newConfigMap.Name, err)
                        }

                        return true, nil
                }
        }

        return false, nil
}

func (r *SkyhookReconciler) UpsertConfigmaps(ctx context.Context, skyhook SkyhookNodes, clusterState *clusterState) (bool, error) {
        updated := false

        var list corev1.ConfigMapList
        err := r.List(ctx, &list, client.InNamespace(r.opts.Namespace), client.MatchingLabels{fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX): skyhook.GetSkyhook().Name})
        if err != nil {
                return false, fmt.Errorf("error listing config maps while upserting: %w", err)
        }

        existingCMs := make(map[string]corev1.ConfigMap)
        for _, cm := range list.Items {
                existingCMs[cm.Name] = cm
        }

        // clean up from an update
        shouldExist := make(map[string]struct{})
        for _, _package := range skyhook.GetSkyhook().Spec.Packages {
                shouldExist[strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhook.GetSkyhook().Name, _package.Name, _package.Version))] = struct{}{}
        }

        for k, v := range existingCMs {
                if _, ok := shouldExist[k]; !ok {
                        // delete
                        err := r.Delete(ctx, &v)
                        if err != nil {
                                return false, fmt.Errorf("error deleting existing config map [%s] while upserting: %w", v.Name, err)
                        }
                }
        }

        for _, _package := range skyhook.GetSkyhook().Spec.Packages {
                if len(_package.ConfigMap) > 0 {

                        newCM := &corev1.ConfigMap{
                                ObjectMeta: metav1.ObjectMeta{
                                        Name:      strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhook.GetSkyhook().Name, _package.Name, _package.Version)),
                                        Namespace: r.opts.Namespace,
                                        Labels: map[string]string{
                                                fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX): skyhook.GetSkyhook().Name,
                                        },
                                        Annotations: map[string]string{
                                                fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX):            skyhook.GetSkyhook().Name,
                                                fmt.Sprintf("%s/Package.Name", v1alpha1.METADATA_PREFIX):    _package.Name,
                                                fmt.Sprintf("%s/Package.Version", v1alpha1.METADATA_PREFIX): _package.Version,
                                        },
                                },
                                Data: _package.ConfigMap,
                        }
                        // set owner of CM to the SCR, which will clean up the CM in delete of the SCR
                        if err := ctrl.SetControllerReference(skyhook.GetSkyhook().Skyhook, newCM, r.scheme); err != nil {
                                return false, fmt.Errorf("error setting ownership of cm: %w", err)
                        }

                        if existingCM, ok := existingCMs[strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhook.GetSkyhook().Name, _package.Name, _package.Version))]; ok {
                                updatedConfigMap, err := r.HandleConfigUpdates(ctx, clusterState, skyhook, _package, &existingCM, newCM)
                                if err != nil {
                                        return false, fmt.Errorf("error updating config map [%s]: %s", newCM.Name, err)
                                }
                                if updatedConfigMap {
                                        updated = true
                                }
                        } else {
                                // create
                                err := r.Create(ctx, newCM)
                                if err != nil {
                                        return false, fmt.Errorf("error creating config map [%s]: %w", newCM.Name, err)
                                }
                        }
                }
        }

        return updated, nil
}

func (r *SkyhookReconciler) IsDrained(ctx context.Context, skyhookNode wrapper.SkyhookNode) (bool, error) {

        pods, err := r.dal.GetPods(ctx, client.MatchingFields{
                "spec.nodeName": skyhookNode.GetNode().Name,
        })
        if err != nil {
                return false, err
        }

        if pods == nil || len(pods.Items) == 0 {
                return true, nil
        }

        // checking for any running or pending pods with no toleration to unschedulable
        // if its has an unschedulable toleration we can ignore
        for _, pod := range pods.Items {

                if ShouldEvict(&pod) {
                        return false, nil
                }

        }

        return true, nil
}

func ShouldEvict(pod *corev1.Pod) bool {
        switch pod.Status.Phase {
        case corev1.PodRunning, corev1.PodPending:

                for _, taint := range pod.Spec.Tolerations {
                        switch taint.Key {
                        case "node.kubernetes.io/unschedulable": // ignoring
                                return false
                        }
                }

                if len(pod.ObjectMeta.OwnerReferences) > 1 {
                        for _, owner := range pod.ObjectMeta.OwnerReferences {
                                if owner.Kind == "DaemonSet" { // ignoring
                                        return false
                                }
                        }
                }

                if pod.GetNamespace() == "kube-system" {
                        return false
                }

                return true
        }
        return false
}

// HandleFinalizer returns true only if we container is deleted and we handled it completely, else false
func (r *SkyhookReconciler) HandleFinalizer(ctx context.Context, skyhook SkyhookNodes) (bool, error) {
        if skyhook.GetSkyhook().DeletionTimestamp.IsZero() { // if not deleted, and does not have our finalizer, add it
                if !controllerutil.ContainsFinalizer(skyhook.GetSkyhook().Skyhook, SkyhookFinalizer) {
                        controllerutil.AddFinalizer(skyhook.GetSkyhook().Skyhook, SkyhookFinalizer)

                        if err := r.Update(ctx, skyhook.GetSkyhook().Skyhook); err != nil {
                                return false, fmt.Errorf("error updating skyhook to add finalizer: %w", err)
                        }
                }
        } else { // being delete, time to handle our
                if controllerutil.ContainsFinalizer(skyhook.GetSkyhook().Skyhook, SkyhookFinalizer) {

                        errs := make([]error, 0)

                        // zero out all the metrics related to this skyhook both skyhook and packages
                        zeroOutSkyhookMetrics(skyhook)

                        for _, node := range skyhook.GetNodes() {
                                patch := client.StrategicMergeFrom(node.GetNode().DeepCopy())

                                node.Uncordon()

                                // if this doesn't change the node then don't patch
                                if !node.Changed() {
                                        continue
                                }

                                err := r.Patch(ctx, node.GetNode(), patch)
                                if err != nil {
                                        errs = append(errs, fmt.Errorf("error patching node [%s] in finalizer: %w", node.GetNode().Name, err))
                                }
                        }

                        if len(errs) > 0 { // we errored, so we need to return error, otherwise we would release the skyhook when we didnt finish
                                return false, utilerrors.NewAggregate(errs)
                        }

                        controllerutil.RemoveFinalizer(skyhook.GetSkyhook().Skyhook, SkyhookFinalizer)
                        if err := r.Update(ctx, skyhook.GetSkyhook().Skyhook); err != nil {
                                return false, fmt.Errorf("error updating skyhook removing finalizer: %w", err)
                        }
                        // should be 1, and now 2. we want to set ObservedGeneration up to not trigger an logic from this update adding the finalizer
                        skyhook.GetSkyhook().Status.ObservedGeneration = skyhook.GetSkyhook().Status.ObservedGeneration + 1

                        if err := r.Status().Update(ctx, skyhook.GetSkyhook().Skyhook); err != nil {
                                return false, fmt.Errorf("error updating skyhook status: %w", err)
                        }

                        return true, nil
                }
        }
        return false, nil
}

// HasNonInterruptWork returns true if pods are running on the node that are either packages, or matches the SCR selector
func (r *SkyhookReconciler) HasNonInterruptWork(ctx context.Context, skyhookNode wrapper.SkyhookNode) (bool, error) {

        selector, err := metav1.LabelSelectorAsSelector(&skyhookNode.GetSkyhook().Spec.PodNonInterruptLabels)
        if err != nil {
                return false, fmt.Errorf("error creating selector: %w", err)
        }

        if selector.Empty() { // when selector is empty it does not do any selecting, ie will return all pods on node.
                return false, nil
        }

        pods, err := r.dal.GetPods(ctx,
                client.MatchingLabelsSelector{Selector: selector},
                client.MatchingFields{
                        "spec.nodeName": skyhookNode.GetNode().Name,
                },
        )
        if err != nil {
                return false, fmt.Errorf("error getting pods: %w", err)
        }

        if pods == nil || len(pods.Items) == 0 {
                return false, nil
        }

        for _, pod := range pods.Items {
                switch pod.Status.Phase {
                case corev1.PodRunning, corev1.PodPending:
                        return true, nil
                }
        }

        return false, nil
}

func (r *SkyhookReconciler) HasRunningPackages(ctx context.Context, skyhookNode wrapper.SkyhookNode) (bool, error) {
        pods, err := r.dal.GetPods(ctx,
                client.HasLabels{fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX)},
                client.MatchingFields{
                        "spec.nodeName": skyhookNode.GetNode().Name,
                },
        )
        if err != nil {
                return false, fmt.Errorf("error getting pods: %w", err)
        }

        return pods != nil && len(pods.Items) > 0, nil
}

func (r *SkyhookReconciler) DrainNode(ctx context.Context, skyhookNode wrapper.SkyhookNode, _package *v1alpha1.Package) (bool, error) {
        drained, err := r.IsDrained(ctx, skyhookNode)
        if err != nil {
                return false, err
        }
        if drained {
                return true, nil
        }

        pods, err := r.dal.GetPods(ctx, client.MatchingFields{
                "spec.nodeName": skyhookNode.GetNode().Name,
        })
        if err != nil {
                return false, err
        }

        if pods == nil || len(pods.Items) == 0 {
                return true, nil
        }

        r.recorder.Eventf(skyhookNode.GetNode(), EventTypeNormal, EventsReasonSkyhookInterrupt,
                "draining node [%s] package [%s:%s] from [skyhook:%s]",
                skyhookNode.GetNode().Name,
                _package.Name,
                _package.Version,
                skyhookNode.GetSkyhook().Name,
        )

        errs := make([]error, 0)
        for _, pod := range pods.Items {

                if ShouldEvict(&pod) {
                        eviction := policyv1.Eviction{}
                        err := r.Client.SubResource("eviction").Create(ctx, &pod, &eviction)
                        if err != nil {
                                errs = append(errs, fmt.Errorf("error evicting pod [%s:%s]: %w", pod.Namespace, pod.Name, err))
                        }
                }
        }

        return len(errs) == 0, utilerrors.NewAggregate(errs)
}

// Interrupt should not be called unless safe to do so, IE already cordoned and drained
func (r *SkyhookReconciler) Interrupt(ctx context.Context, skyhookNode wrapper.SkyhookNode, _package *v1alpha1.Package, _interrupt *v1alpha1.Interrupt) error {

        hasPackagesRunning, err := r.HasRunningPackages(ctx, skyhookNode)
        if err != nil {
                return err
        }

        if hasPackagesRunning { // keep waiting...
                return nil
        }

        exists, err := r.PodExists(ctx, skyhookNode.GetNode().Name, skyhookNode.GetSkyhook().Name, _package)
        if err != nil {
                return err
        }
        if exists {
                // nothing to do here, already running
                return nil
        }

        // Ensure the node metadata configmap exists before creating the pod
        // This prevents a race where the pod starts before its required configmap is created
        if err := r.UpsertNodeLabelsAnnotationsPackages(ctx, skyhookNode.GetSkyhook(), skyhookNode.GetNode()); err != nil {
                return fmt.Errorf("error upserting node metadata configmap: %w", err)
        }

        argEncode, err := _interrupt.ToArgs()
        if err != nil {
                return fmt.Errorf("error creating interrupt args: %w", err)
        }

        pod := createInterruptPodForPackage(r.opts, _interrupt, argEncode, _package, skyhookNode.GetSkyhook(), skyhookNode.GetNode().Name)

        if err := SetPackages(pod, skyhookNode.GetSkyhook().Skyhook, _package.Image, v1alpha1.StageInterrupt, _package); err != nil {
                return fmt.Errorf("error setting package on interrupt: %w", err)
        }

        if err := ctrl.SetControllerReference(skyhookNode.GetSkyhook().Skyhook, pod, r.scheme); err != nil {
                return fmt.Errorf("error setting ownership: %w", err)
        }

        if err := r.Create(ctx, pod); err != nil {
                return fmt.Errorf("error creating interruption pod: %w", err)
        }

        _ = skyhookNode.Upsert(_package.PackageRef, _package.Image, v1alpha1.StateInProgress, v1alpha1.StageInterrupt, 0, _package.ContainerSHA)

        r.recorder.Eventf(skyhookNode.GetSkyhook().Skyhook, EventTypeNormal, EventsReasonSkyhookInterrupt,
                "Interrupting node [%s] package [%s:%s] from [skyhook:%s]",
                skyhookNode.GetNode().Name,
                _package.Name,
                _package.Version,
                skyhookNode.GetSkyhook().Name)

        return nil
}

// fudgeInterruptWithPriority takes a list of packages, interrupts, and configUpdates and returns the correct merged interrupt to run to handle all the packages
func fudgeInterruptWithPriority(next []*v1alpha1.Package, configUpdates map[string][]string, interrupts map[string][]*v1alpha1.Interrupt) (*v1alpha1.Interrupt, string) {
        var ret *v1alpha1.Interrupt
        var pack string

        // map interrupt to priority
        // A lower priority value means a higher priority and will be used in favor of anything with a higher value
        var priorities = map[v1alpha1.InterruptType]int{
                v1alpha1.REBOOT:               0,
                v1alpha1.RESTART_ALL_SERVICES: 1,
                v1alpha1.SERVICE:              2,
                v1alpha1.NOOP:                 3,
        }

        for _, _package := range next {

                if len(configUpdates[_package.Name]) == 0 {
                        interrupts[_package.Name] = []*v1alpha1.Interrupt{}
                        if _package.HasInterrupt() {
                                interrupts[_package.Name] = append(interrupts[_package.Name], _package.Interrupt)
                        }
                }
        }

        packageNames := make([]string, 0)
        for _, pkg := range next {
                packageNames = append(packageNames, pkg.Name)
        }
        sort.Strings(packageNames)

        for _, _package := range packageNames {
                _interrupts, ok := interrupts[_package]
                if !ok {
                        continue
                }

                for _, interrupt := range _interrupts {
                        if ret == nil { // prime ret, base case
                                ret = interrupt
                                pack = _package
                        }

                        // short circuit, reboot has highest priority
                        switch interrupt.Type {
                        case v1alpha1.REBOOT:
                                return interrupt, _package
                        }

                        // check if interrupt is higher priority using the priority_order
                        // A lower priority value means a higher priority
                        if priorities[interrupt.Type] < priorities[ret.Type] {
                                ret = interrupt
                                pack = _package
                        } else if priorities[interrupt.Type] == priorities[ret.Type] {
                                mergeInterrupt(ret, interrupt)
                        }
                }
        }

        return ret, pack // return merged interrupt and package
}

func mergeInterrupt(left, right *v1alpha1.Interrupt) {

        // make sure both are of type service
        if left.Type != v1alpha1.SERVICE || right.Type != v1alpha1.SERVICE {
                return
        }

        left.Services = merge(left.Services, right.Services)
}

func merge[T cmp.Ordered](left, right []T) []T {
        for _, r := range right {
                if !slices.Contains(left, r) {
                        left = append(left, r)
                }
        }
        slices.Sort(left)
        return left
}

// ValidateNodeConfigmaps validates that there are no orphaned or stale config maps for a node
func (r *SkyhookReconciler) ValidateNodeConfigmaps(ctx context.Context, skyhookName string, nodes []wrapper.SkyhookNode) (bool, error) {
        var list corev1.ConfigMapList
        err := r.List(ctx, &list, client.InNamespace(r.opts.Namespace), client.MatchingLabels{fmt.Sprintf("%s/skyhook-node-meta", v1alpha1.METADATA_PREFIX): skyhookName})
        if err != nil {
                return false, fmt.Errorf("error listing config maps: %w", err)
        }

        // No configmaps created by this skyhook, no work needs to be done
        if len(list.Items) == 0 {
                return false, nil
        }

        existingCMs := make(map[string]corev1.ConfigMap)
        for _, cm := range list.Items {
                existingCMs[cm.Name] = cm
        }

        shouldExist := make(map[string]struct{})
        for _, node := range nodes {
                shouldExist[generateSafeName(253, skyhookName, node.GetNode().Name, "metadata")] = struct{}{}
        }

        update := false
        errs := make([]error, 0)
        for k, v := range existingCMs {
                if _, ok := shouldExist[k]; !ok {
                        update = true
                        err := r.Delete(ctx, &v)
                        if err != nil {
                                errs = append(errs, fmt.Errorf("error deleting existing config map [%s]: %w", v.Name, err))
                        }
                }
        }

        // Ensure packages.json is present and up-to-date for expected configmaps
        skyhookCR, err := r.dal.GetSkyhook(ctx, skyhookName)
        if err != nil {
                return update, fmt.Errorf("error getting skyhook for metadata validation: %w", err)
        }
        skyhookWrapper := wrapper.NewSkyhookWrapper(skyhookCR)
        metadata := NewSkyhookMetadata(r.opts, skyhookWrapper)
        expectedBytes, err := metadata.Marshal()
        if err != nil {
                return update, fmt.Errorf("error marshalling metadata for validation: %w", err)
        }
        expected := string(expectedBytes)

        for i := range list.Items {
                cm := &list.Items[i]
                if _, ok := shouldExist[cm.Name]; !ok {
                        continue
                }
                if cm.Data == nil {
                        cm.Data = make(map[string]string)
                }
                if cm.Data["packages.json"] != expected {
                        cm.Data["packages.json"] = expected
                        if err := r.Update(ctx, cm); err != nil {
                                errs = append(errs, fmt.Errorf("error updating packages.json on config map [%s]: %w", cm.Name, err))
                        } else {
                                update = true
                        }
                }
        }

        return update, utilerrors.NewAggregate(errs)
}

// PodExists tests if this package is exists on a node.
func (r *SkyhookReconciler) PodExists(ctx context.Context, nodeName, skyhookName string, _package *v1alpha1.Package) (bool, error) {

        pods, err := r.dal.GetPods(ctx,
                client.MatchingFields{
                        "spec.nodeName": nodeName,
                },
                client.MatchingLabels{
                        fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX):    skyhookName,
                        fmt.Sprintf("%s/package", v1alpha1.METADATA_PREFIX): fmt.Sprintf("%s-%s", _package.Name, _package.Version),
                },
        )
        if err != nil {
                return false, fmt.Errorf("error check from existing pods: %w", err)
        }

        if pods == nil || len(pods.Items) == 0 {
                return false, nil
        }
        return true, nil
}

// createInterruptPodForPackage returns the pod spec for an interrupt pod given an package
func createInterruptPodForPackage(opts SkyhookOperatorOptions, _interrupt *v1alpha1.Interrupt, argEncode string, _package *v1alpha1.Package, skyhook *wrapper.Skyhook, nodeName string) *corev1.Pod {
        copyDir := fmt.Sprintf("%s/%s/%s-%s-%s-%d",
                opts.CopyDirRoot,
                skyhook.Name,
                _package.Name,
                _package.Version,
                skyhook.UID,
                skyhook.Generation,
        )

        volumes := []corev1.Volume{
                {
                        Name: "root-mount",
                        VolumeSource: corev1.VolumeSource{
                                HostPath: &corev1.HostPathVolumeSource{
                                        Path: "/",
                                },
                        },
                },
                {
                        // node names in different CSPs might include dots which isn't allowed in volume names
                        // so we have to replace all dots with dashes
                        Name: generateSafeName(63, skyhook.Name, nodeName, "metadata"),
                        VolumeSource: corev1.VolumeSource{
                                ConfigMap: &corev1.ConfigMapVolumeSource{
                                        LocalObjectReference: corev1.LocalObjectReference{
                                                Name: strings.ReplaceAll(fmt.Sprintf("%s-%s-metadata", skyhook.Name, nodeName), ".", "-"),
                                        },
                                },
                        },
                },
        }
        volumeMounts := []corev1.VolumeMount{
                {
                        Name:             "root-mount",
                        MountPath:        "/root",
                        MountPropagation: ptr(corev1.MountPropagationHostToContainer),
                },
        }

        pod := &corev1.Pod{
                ObjectMeta: metav1.ObjectMeta{
                        Name:      generateSafeName(63, skyhook.Name, "interrupt", string(_interrupt.Type), nodeName),
                        Namespace: opts.Namespace,
                        Labels: map[string]string{
                                fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX):      skyhook.Name,
                                fmt.Sprintf("%s/package", v1alpha1.METADATA_PREFIX):   fmt.Sprintf("%s-%s", _package.Name, _package.Version),
                                fmt.Sprintf("%s/interrupt", v1alpha1.METADATA_PREFIX): "True",
                        },
                },
                Spec: corev1.PodSpec{
                        NodeName:      nodeName,
                        RestartPolicy: corev1.RestartPolicyOnFailure,
                        InitContainers: []corev1.Container{
                                {
                                        Name:  InterruptContainerName,
                                        Image: getAgentImage(opts, _package),
                                        Args:  []string{"interrupt", "/root", copyDir, argEncode},
                                        Env:   getAgentConfigEnvVars(opts, _package.Name, _package.Version, skyhook.ResourceID(), skyhook.Name),
                                        SecurityContext: &corev1.SecurityContext{
                                                Privileged: ptr(true),
                                        },
                                        VolumeMounts: volumeMounts,
                                        Resources: corev1.ResourceRequirements{
                                                Limits: corev1.ResourceList{
                                                        corev1.ResourceCPU:    resource.MustParse("500m"),
                                                        corev1.ResourceMemory: resource.MustParse("64Mi"),
                                                },
                                                Requests: corev1.ResourceList{
                                                        corev1.ResourceCPU:    resource.MustParse("500m"),
                                                        corev1.ResourceMemory: resource.MustParse("64Mi"),
                                                },
                                        },
                                },
                        },
                        Containers: []corev1.Container{
                                {
                                        Name:  "pause",
                                        Image: opts.PauseImage,
                                        Resources: corev1.ResourceRequirements{
                                                Limits: corev1.ResourceList{
                                                        corev1.ResourceCPU:    resource.MustParse("100m"),
                                                        corev1.ResourceMemory: resource.MustParse("20Mi"),
                                                },
                                                Requests: corev1.ResourceList{
                                                        corev1.ResourceCPU:    resource.MustParse("100m"),
                                                        corev1.ResourceMemory: resource.MustParse("20Mi"),
                                                },
                                        },
                                },
                        },
                        HostPID:     true,
                        HostNetwork: true,
                        // If you change these go change the SelectNode toleration in cluster_state.go
                        Tolerations: append([]corev1.Toleration{ // tolerate all cordon
                                {
                                        Key:      TaintUnschedulable,
                                        Operator: corev1.TolerationOpExists,
                                },
                                opts.GetRuntimeRequiredToleration(),
                        }, skyhook.Spec.AdditionalTolerations...),
                        Volumes: volumes,
                },
        }
        if opts.ImagePullSecret != "" {
                pod.Spec.ImagePullSecrets = []corev1.LocalObjectReference{
                        {
                                Name: opts.ImagePullSecret,
                        },
                }
        }
        return pod
}

func trunstr(str string, length int) string {
        if len(str) > length {
                return str[:length]
        }
        return str
}

func getAgentImage(opts SkyhookOperatorOptions, _package *v1alpha1.Package) string {
        if _package.AgentImageOverride != "" {
                return _package.AgentImageOverride
        }
        return opts.AgentImage
}

// getPackageImage returns the full image reference for a package, using the digest if specified
func getPackageImage(_package *v1alpha1.Package) string {
        if _package.ContainerSHA != "" {
                // When containerSHA is specified, use it instead of the version tag for immutable image reference
                return fmt.Sprintf("%s@%s", _package.Image, _package.ContainerSHA)
        }
        // Fall back to version tag
        return fmt.Sprintf("%s:%s", _package.Image, _package.Version)
}

func getAgentConfigEnvVars(opts SkyhookOperatorOptions, packageName string, packageVersion string, resourceID string, skyhookName string) []corev1.EnvVar {
        return []corev1.EnvVar{
                {
                        Name:  "SKYHOOK_LOG_DIR",
                        Value: fmt.Sprintf("%s/%s", opts.AgentLogRoot, skyhookName),
                },
                {
                        Name:  "SKYHOOK_ROOT_DIR",
                        Value: fmt.Sprintf("%s/%s", opts.CopyDirRoot, skyhookName),
                },
                {
                        Name:  "COPY_RESOLV",
                        Value: "false",
                },
                {
                        Name:  "SKYHOOK_RESOURCE_ID",
                        Value: fmt.Sprintf("%s_%s_%s", resourceID, packageName, packageVersion),
                },
        }
}

// createPodFromPackage creates a pod spec for a skyhook pod for a given package
func createPodFromPackage(opts SkyhookOperatorOptions, _package *v1alpha1.Package, skyhook *wrapper.Skyhook, nodeName string, stage v1alpha1.Stage) *corev1.Pod {
        // Generate consistent names that won't exceed k8s limits
        volumeName := generateSafeName(63, "metadata", nodeName)
        configMapName := generateSafeName(253, skyhook.Name, nodeName, "metadata")

        volumes := []corev1.Volume{
                {
                        Name: "root-mount",
                        VolumeSource: corev1.VolumeSource{
                                HostPath: &corev1.HostPathVolumeSource{
                                        Path: "/",
                                },
                        },
                },
                {
                        Name: volumeName,
                        VolumeSource: corev1.VolumeSource{
                                ConfigMap: &corev1.ConfigMapVolumeSource{
                                        LocalObjectReference: corev1.LocalObjectReference{
                                                Name: configMapName,
                                        },
                                },
                        },
                },
        }

        volumeMounts := []corev1.VolumeMount{
                {
                        Name:             "root-mount",
                        MountPath:        "/root",
                        MountPropagation: ptr(corev1.MountPropagationHostToContainer),
                },
                {
                        Name:      volumeName,
                        MountPath: "/skyhook-package/node-metadata",
                },
        }

        if len(_package.ConfigMap) > 0 {
                volumeMounts = append(volumeMounts, corev1.VolumeMount{
                        Name:      _package.Name,
                        MountPath: "/skyhook-package/configmaps",
                })

                volumes = append(volumes, corev1.Volume{
                        Name: _package.Name,
                        VolumeSource: corev1.VolumeSource{
                                ConfigMap: &corev1.ConfigMapVolumeSource{
                                        LocalObjectReference: corev1.LocalObjectReference{
                                                Name: strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhook.Name, _package.Name, _package.Version)),
                                        },
                                },
                        },
                })
        }

        copyDir := fmt.Sprintf("%s/%s/%s-%s-%s-%d",
                opts.CopyDirRoot,
                skyhook.Name,
                _package.Name,
                _package.Version,
                skyhook.UID,
                skyhook.Generation,
        )
        applyargs := []string{strings.ToLower(string(stage)), "/root", copyDir}
        checkargs := []string{strings.ToLower(string(stage) + "-check"), "/root", copyDir}

        agentEnvs := append(
                _package.Env,
                getAgentConfigEnvVars(opts, _package.Name, _package.Version, skyhook.ResourceID(), skyhook.Name)...,
        )

        pod := &corev1.Pod{
                ObjectMeta: metav1.ObjectMeta{
                        Name:      generateSafeName(63, skyhook.Name, _package.Name, _package.Version, string(stage), nodeName),
                        Namespace: opts.Namespace,
                        Labels: map[string]string{
                                fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX):    skyhook.Name,
                                fmt.Sprintf("%s/package", v1alpha1.METADATA_PREFIX): fmt.Sprintf("%s-%s", _package.Name, _package.Version),
                        },
                },
                Spec: corev1.PodSpec{
                        NodeName:      nodeName,
                        RestartPolicy: corev1.RestartPolicyOnFailure,
                        InitContainers: []corev1.Container{
                                {
                                        Name:            fmt.Sprintf("%s-init", trunstr(_package.Name, 43)),
                                        Image:           getPackageImage(_package),
                                        ImagePullPolicy: "Always",
                                        Command:         []string{"/bin/sh"},
                                        Args: []string{
                                                "-c",
                                                "mkdir -p /root/${SKYHOOK_DIR} && cp -r /skyhook-package/* /root/${SKYHOOK_DIR}",
                                        },
                                        Env: []corev1.EnvVar{
                                                {
                                                        Name:  "SKYHOOK_DIR",
                                                        Value: copyDir,
                                                },
                                        },
                                        SecurityContext: &corev1.SecurityContext{
                                                Privileged: ptr(true),
                                        },
                                        VolumeMounts: volumeMounts,
                                },
                                {
                                        Name:            fmt.Sprintf("%s-%s", trunstr(_package.Name, 43), stage),
                                        Image:           getAgentImage(opts, _package),
                                        ImagePullPolicy: "Always",
                                        Args:            applyargs,
                                        Env:             agentEnvs,
                                        SecurityContext: &corev1.SecurityContext{
                                                Privileged: ptr(true),
                                        },
                                        VolumeMounts: volumeMounts,
                                },
                                {
                                        Name:            fmt.Sprintf("%s-%scheck", trunstr(_package.Name, 43), stage),
                                        Image:           getAgentImage(opts, _package),
                                        ImagePullPolicy: "Always",
                                        Args:            checkargs,
                                        Env:             agentEnvs,
                                        SecurityContext: &corev1.SecurityContext{
                                                Privileged: ptr(true),
                                        },
                                        VolumeMounts: volumeMounts,
                                },
                        },
                        Containers: []corev1.Container{
                                {
                                        Name:  "pause",
                                        Image: opts.PauseImage,
                                        Resources: corev1.ResourceRequirements{
                                                Limits: corev1.ResourceList{
                                                        corev1.ResourceCPU:    resource.MustParse("100m"),
                                                        corev1.ResourceMemory: resource.MustParse("20Mi"),
                                                },
                                                Requests: corev1.ResourceList{
                                                        corev1.ResourceCPU:    resource.MustParse("100m"),
                                                        corev1.ResourceMemory: resource.MustParse("20Mi"),
                                                },
                                        },
                                },
                        },
                        Volumes:     volumes,
                        HostPID:     true,
                        HostNetwork: true,
                        // If you change these go change the SelectNode toleration in cluster_state.go
                        Tolerations: append([]corev1.Toleration{ // tolerate all cordon
                                {
                                        Key:      TaintUnschedulable,
                                        Operator: corev1.TolerationOpExists,
                                },
                                opts.GetRuntimeRequiredToleration(),
                        }, skyhook.Spec.AdditionalTolerations...),
                },
        }
        if opts.ImagePullSecret != "" {
                pod.Spec.ImagePullSecrets = []corev1.LocalObjectReference{
                        {
                                Name: opts.ImagePullSecret,
                        },
                }
        }
        if _package.GracefulShutdown != nil {
                pod.Spec.TerminationGracePeriodSeconds = ptr(int64(_package.GracefulShutdown.Duration.Seconds()))
        }
        setPodResources(pod, _package.Resources)
        return pod
}

// FilterEnv removes the environment variables passed into exlude
func FilterEnv(envs []corev1.EnvVar, exclude ...string) []corev1.EnvVar {
        var filteredEnv []corev1.EnvVar

        // build map of exclude strings for faster lookup
        excludeMap := make(map[string]struct{})
        for _, name := range exclude {
                excludeMap[name] = struct{}{}
        }

        // If the environment variable name is in the exclude list, skip it
        // otherwise append it to the final list
        for _, env := range envs {
                if _, found := excludeMap[env.Name]; !found {
                        filteredEnv = append(filteredEnv, env)
                }
        }

        return filteredEnv
}

// PodMatchesPackage asserts that a given pod matches the given pod spec
func podMatchesPackage(opts SkyhookOperatorOptions, _package *v1alpha1.Package, pod corev1.Pod, skyhook *wrapper.Skyhook, stage v1alpha1.Stage) bool {
        var expectedPod *corev1.Pod

        // need to differentiate whether the pod is for an interrupt or not so we know
        // what to expect and how to compare them
        isInterrupt := false
        _, limitRange := pod.Annotations["kubernetes.io/limit-ranger"]

        if pod.Labels[fmt.Sprintf("%s/interrupt", v1alpha1.METADATA_PREFIX)] == "True" {
                expectedPod = createInterruptPodForPackage(opts, &v1alpha1.Interrupt{}, "", _package, skyhook, "")
                isInterrupt = true
        } else {
                expectedPod = createPodFromPackage(opts, _package, skyhook, "", stage)
        }

        actualPod := pod.DeepCopy()

        // check to see whether the name or the version of the package changed
        packageLabel := fmt.Sprintf("%s/package", v1alpha1.METADATA_PREFIX)
        if actualPod.Labels[packageLabel] != expectedPod.Labels[packageLabel] {
                return false
        }

        // compare initContainers since this is where a lot of the important info lives
        for i := range actualPod.Spec.InitContainers {
                expectedContainer := expectedPod.Spec.InitContainers[i]
                actualContainer := actualPod.Spec.InitContainers[i]

                if expectedContainer.Name != actualContainer.Name {
                        return false
                }

                if expectedContainer.Image != actualContainer.Image {
                        return false
                }

                // compare the containers env vars except for the ones that are inserted
                // by the operator by default as the SKYHOOK_RESOURCE_ID will change every
                // time the skyhook is updated and would cause every pod to be removed
                // TODO: This is ignoring all the static env vars that are set by operator config.
                // It probably should be just SKYHOOK_RESOURCE_ID that is ignored. Otherwise,
                // a user will have to manually delete the pod to update the package when operator is updated.
                dummyAgentEnv := getAgentConfigEnvVars(opts, "", "", "", "")
                excludedEnvs := make([]string, len(dummyAgentEnv))
                for i, env := range dummyAgentEnv {
                        excludedEnvs[i] = env.Name
                }
                expectedFilteredEnv := FilterEnv(expectedContainer.Env, excludedEnvs...)
                actualFilteredEnv := FilterEnv(actualContainer.Env, excludedEnvs...)
                if !reflect.DeepEqual(expectedFilteredEnv, actualFilteredEnv) {
                        return false
                }

                if !isInterrupt { // dont compare these since they are not configured on interrupt
                        // compare resource requests and limits (CPU, memory, etc.)
                        expectedResources := expectedContainer.Resources
                        actualResources := actualContainer.Resources
                        if skyhook.Spec.Packages[_package.Name].Resources != nil {
                                // If CR has resources specified, they should match exactly
                                if !reflect.DeepEqual(expectedResources, actualResources) {
                                        return false
                                }
                        } else {
                                // If CR has no resources specified, ensure pod has no resource overrides
                                if !limitRange {
                                        if actualResources.Requests != nil || actualResources.Limits != nil {
                                                return false
                                        }
                                }
                        }
                }
        }

        return true
}

// ValidateRunningPackages deletes pods that don't match the current spec and checks if there are pods running
// that don't match the node state and removes them if they exist
func (r *SkyhookReconciler) ValidateRunningPackages(ctx context.Context, skyhook SkyhookNodes) (bool, error) {

        update := false
        errs := make([]error, 0)
        // get all pods for this skyhook packages
        pods, err := r.dal.GetPods(ctx,
                client.MatchingLabels{
                        fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX): skyhook.GetSkyhook().Name,
                },
        )
        if err != nil {
                return false, fmt.Errorf("error getting pods while validating packages: %w", err)
        }
        if pods == nil || len(pods.Items) == 0 {
                return false, nil // nothing running for this skyhook on this node
        }

        // Initialize metrics for each stage
        stages := make(map[string]map[string]map[v1alpha1.Stage]int)

        // group pods by node
        podsbyNode := make(map[string][]corev1.Pod)
        for _, pod := range pods.Items {
                podsbyNode[pod.Spec.NodeName] = append(podsbyNode[pod.Spec.NodeName], pod)
        }

        for _, node := range skyhook.GetNodes() {
                nodeState, err := node.State()
                if err != nil {
                        return false, fmt.Errorf("error getting node state: %w", err)
                }

                for _, pod := range podsbyNode[node.GetNode().Name] {
                        found := false

                        runningPackage, err := GetPackage(&pod)
                        if err != nil {
                                errs = append(errs, fmt.Errorf("error getting package from pod [%s:%s] while validating packages: %w", pod.Namespace, pod.Name, err))
                        }

                        // check if the package is part of the skyhook spec, if not we need to delete it
                        for _, v := range skyhook.GetSkyhook().Spec.Packages {
                                if podMatchesPackage(r.opts, &v, pod, skyhook.GetSkyhook(), runningPackage.Stage) {
                                        found = true
                                }
                        }

                        // Increment the stage count for metrics
                        if _, ok := stages[runningPackage.Name]; !ok {
                                stages[runningPackage.Name] = make(map[string]map[v1alpha1.Stage]int)
                                if _, ok := stages[runningPackage.Name][runningPackage.Version]; !ok {
                                        stages[runningPackage.Name][runningPackage.Version] = make(map[v1alpha1.Stage]int)
                                        for _, stage := range v1alpha1.Stages {
                                                stages[runningPackage.Name][runningPackage.Version][stage] = 0
                                        }
                                }
                        }
                        stages[runningPackage.Name][runningPackage.Version][runningPackage.Stage]++

                        // uninstall is by definition not part of the skyhook spec, so we cant delete it (because it used to be but was removed, hence uninstalling it)
                        if runningPackage.Stage == v1alpha1.StageUninstall {
                                found = true
                        }

                        if !found {
                                update = true

                                err := r.InvalidPackage(ctx, &pod)
                                if err != nil {
                                        errs = append(errs, fmt.Errorf("error invalidating package: %w", err))
                                }
                                continue
                        }

                        // Check if package exists in node state, ie a package running that the node state doesn't know about
                        // something that is often done to try to fix bad node state is to clear the node state completely
                        // which if a package is running, we want to terminate it gracefully. Ofthen what leads to this is
                        // the package is in a crashloop and the operator want to restart it the whole package.
                        // when we apply a package it just check if there is a running package on the node for the state of the package
                        // this can cause to leave a pod running in say config mode, and it there is a depends on you might not correctly
                        // run thins in the correct order.
                        deleteMe := false
                        packageStatus, exists := nodeState[runningPackage.GetUniqueName()]
                        if !exists { // package not in node state, so we need to delete it
                                deleteMe = true
                        } else { // package in node state, so we need to check if it's running
                                // need check if the stats match, if not we need to delete it
                                if packageStatus.Stage != runningPackage.Stage {
                                        deleteMe = true
                                }
                        }

                        if deleteMe {
                                update = true
                                err := r.InvalidPackage(ctx, &pod)
                                if err != nil {
                                        errs = append(errs, fmt.Errorf("error invalidating package: %w", err))
                                }
                        }
                }
        }

        return update, utilerrors.NewAggregate(errs)
}

// InvalidPackage invalidates a package and updates the pod, which will trigger the pod to be deleted
func (r *SkyhookReconciler) InvalidPackage(ctx context.Context, pod *corev1.Pod) error {
        err := InvalidatePackage(pod)
        if err != nil {
                return fmt.Errorf("error invalidating package: %w", err)
        }

        err = r.Update(ctx, pod)
        if err != nil {
                return fmt.Errorf("error updating pod: %w", err)
        }

        return nil
}

// ProcessInterrupt will check and do the interrupt if need, and returns
// false means we are waiting
// true means we are good to proceed
func (r *SkyhookReconciler) ProcessInterrupt(ctx context.Context, skyhookNode wrapper.SkyhookNode, _package *v1alpha1.Package, interrupt *v1alpha1.Interrupt, runInterrupt bool) (bool, error) {

        if !skyhookNode.HasInterrupt(*_package) {
                return true, nil
        }

        // default starting stage
        stage := v1alpha1.StageApply
        nextStage := skyhookNode.NextStage(_package)
        if nextStage != nil {
                stage = *nextStage
        }

        // wait tell this is done if its happening
        status, found := skyhookNode.PackageStatus(_package.GetUniqueName())
        if found && status.State == v1alpha1.StateSkipped {
                return false, nil
        }

        // Theres is a race condition when a node reboots and api cleans up the interrupt pod
        // so we need to check if the pod exists and if it does, we need to recreate it
        if status != nil && (status.State == v1alpha1.StateInProgress || status.State == v1alpha1.StateErroring) && status.Stage == v1alpha1.StageInterrupt {
                // call interrupt to recreate the pod if missing
                err := r.Interrupt(ctx, skyhookNode, _package, interrupt)
                if err != nil {
                        return false, err
                }
        }

        // drain and cordon node before applying package that has an interrupt
        if stage == v1alpha1.StageApply {
                ready, err := r.EnsureNodeIsReadyForInterrupt(ctx, skyhookNode, _package)
                if err != nil {
                        return false, err
                }

                if !ready {
                        return false, nil
                }
        }

        // time to interrupt (once other packages have finished)
        if stage == v1alpha1.StageInterrupt && runInterrupt {
                err := r.Interrupt(ctx, skyhookNode, _package, interrupt)
                if err != nil {
                        return false, err
                }

                return false, nil
        }

        //skipping
        if stage == v1alpha1.StageInterrupt && !runInterrupt {
                err := skyhookNode.Upsert(_package.PackageRef, _package.Image, v1alpha1.StateSkipped, stage, 0, _package.ContainerSHA)
                if err != nil {
                        return false, fmt.Errorf("error upserting to skip interrupt: %w", err)
                }
                return false, nil
        }

        // wait tell this is done if its happening
        if status != nil && status.Stage == v1alpha1.StageInterrupt && status.State != v1alpha1.StateComplete {
                return false, nil
        }

        return true, nil
}

func (r *SkyhookReconciler) EnsureNodeIsReadyForInterrupt(ctx context.Context, skyhookNode wrapper.SkyhookNode, _package *v1alpha1.Package) (bool, error) {
        // cordon node
        skyhookNode.Cordon()

        hasWork, err := r.HasNonInterruptWork(ctx, skyhookNode)
        if err != nil {
                return false, err
        }
        if hasWork { // keep waiting...
                return false, nil
        }

        ready, err := r.DrainNode(ctx, skyhookNode, _package)
        if err != nil {
                return false, fmt.Errorf("error draining node [%s]: %w", skyhookNode.GetNode().Name, err)
        }

        return ready, nil
}

// ApplyPackage starts a pod on node for the package
func (r *SkyhookReconciler) ApplyPackage(ctx context.Context, logger logr.Logger, clusterState *clusterState, skyhookNode wrapper.SkyhookNode, _package *v1alpha1.Package, runInterrupt bool) error {

        if _package == nil {
                return errors.New("can not apply nil package")
        }

        // default starting stage
        stage := v1alpha1.StageApply

        // These modes don't have anything that comes before them so we must specify them as the
        // starting point. The next stage function will return nil until these modes complete.
        // Config is a special case as sometimes apply will come before it and other times it wont
        // which is why it needs to be here as well
        if packageStatus, found := skyhookNode.PackageStatus(_package.GetUniqueName()); found {
                switch packageStatus.Stage {
                case v1alpha1.StageConfig, v1alpha1.StageUpgrade, v1alpha1.StageUninstall:
                        stage = packageStatus.Stage
                }
        }

        // if stage != v1alpha1.StageApply {
        //         // If a node gets rest by a user, the about method will return the wrong node state. Above sources it from the skyhook status.
        //         // check if the node has nothing, reset it then apply the package.
        //         nodeState, err := skyhookNode.State()
        //         if err != nil {
        //                 return fmt.Errorf("error getting node state: %w", err)
        //         }

        //         _, found := nodeState[_package.GetUniqueName()]
        //         if !found {
        //                 stage = v1alpha1.StageApply
        //         }
        // }

        nextStage := skyhookNode.NextStage(_package)
        if nextStage != nil {
                stage = *nextStage
        }

        // test if pod exists, if so, bailout
        exists, err := r.PodExists(ctx, skyhookNode.GetNode().Name, skyhookNode.GetSkyhook().Name, _package)
        if err != nil {
                return err
        }

        // wait tell this is done if its happening
        status, found := skyhookNode.PackageStatus(_package.GetUniqueName())

        if found && status.State == v1alpha1.StateSkipped { // skipped, so nothing to do
                return nil
        }

        if found && status.State == v1alpha1.StateInProgress { // running, so do nothing atm
                if exists {
                        return nil
                }
        }

        if exists {
                // nothing to do here, already running
                return nil
        }

        // Ensure the node metadata configmap exists before creating the pod
        // This prevents a race where the pod starts before its required configmap is created
        if err := r.UpsertNodeLabelsAnnotationsPackages(ctx, skyhookNode.GetSkyhook(), skyhookNode.GetNode()); err != nil {
                return fmt.Errorf("error upserting node metadata configmap: %w", err)
        }

        pod := createPodFromPackage(r.opts, _package, skyhookNode.GetSkyhook(), skyhookNode.GetNode().Name, stage)

        if err := SetPackages(pod, skyhookNode.GetSkyhook().Skyhook, _package.Image, stage, _package); err != nil {
                return fmt.Errorf("error setting package on pod: %w", err)
        }

        // setup ownership of the pod we created
        // helps run time know what to do when something happens to this pod we are about to create
        if err := ctrl.SetControllerReference(skyhookNode.GetSkyhook().Skyhook, pod, r.scheme); err != nil {
                return fmt.Errorf("error setting ownership: %w", err)
        }

        if err := r.Create(ctx, pod); err != nil {
                return fmt.Errorf("error creating pod: %w", err)
        }

        if err = skyhookNode.Upsert(_package.PackageRef, _package.Image, v1alpha1.StateInProgress, stage, 0, _package.ContainerSHA); err != nil {
                err = fmt.Errorf("error upserting package: %w", err) // want to keep going in this case, but don't want to lose the err
        }

        skyhookNode.SetStatus(v1alpha1.StatusInProgress)

        skyhookNode.GetSkyhook().AddCondition(metav1.Condition{
                Type:               fmt.Sprintf("%s/ApplyPackage", v1alpha1.METADATA_PREFIX),
                Status:             metav1.ConditionTrue,
                ObservedGeneration: skyhookNode.GetSkyhook().Generation,
                LastTransitionTime: metav1.Now(),
                Reason:             "ApplyPackage",
                Message:            fmt.Sprintf("Applying package [%s:%s] to node [%s]", _package.Name, _package.Version, skyhookNode.GetNode().Name),
        })

        r.recorder.Eventf(skyhookNode.GetNode(), EventTypeNormal, EventsReasonSkyhookApply, "Applying package [%s:%s] from [skyhook:%s] stage [%s]", _package.Name, _package.Version, skyhookNode.GetSkyhook().Name, stage)
        r.recorder.Eventf(skyhookNode.GetSkyhook(), EventTypeNormal, EventsReasonSkyhookApply, "Applying package [%s:%s] to node [%s] stage [%s]", _package.Name, _package.Version, skyhookNode.GetNode().Name, stage)

        skyhookNode.GetSkyhook().Updated = true

        return err
}

// HandleRuntimeRequired finds any nodes for which all runtime required Skyhooks are complete and remove their runtime required taint
// Will return an error if the patching of the nodes is not possible
func (r *SkyhookReconciler) HandleRuntimeRequired(ctx context.Context, clusterState *clusterState) error {
        node_to_skyhooks, skyhook_node_map := groupSkyhooksByNode(clusterState)
        to_remove := getRuntimeRequiredTaintCompleteNodes(node_to_skyhooks, skyhook_node_map)
        // Remove the runtime required taint from nodes in to_remove
        taint_to_remove := r.opts.GetRuntimeRequiredTaint()
        errs := make([]error, 0)
        for _, node := range to_remove {
                // check before removing taint that it even exists to begin with
                if !taints.TaintExists(node.Spec.Taints, &taint_to_remove) {
                        continue
                }
                // RemoveTaint will ALWAYS return nil for its error so no need to check it
                new_node, updated, _ := taints.RemoveTaint(node, &taint_to_remove)
                if updated {
                        err := r.Patch(ctx, new_node, client.MergeFrom(node))
                        if err != nil {
                                errs = append(errs, err)
                        }
                }
        }
        if len(errs) > 0 {
                return utilerrors.NewAggregate(errs)
        }
        return nil
}

// Group Skyhooks by what node they target
func groupSkyhooksByNode(clusterState *clusterState) (map[types.UID][]SkyhookNodes, map[types.UID]*corev1.Node) {
        node_to_skyhooks := make(map[types.UID][]SkyhookNodes)
        nodes := make(map[types.UID]*corev1.Node)
        for _, skyhook := range clusterState.skyhooks {
                // Ignore skyhooks that don't have runtime required
                if !skyhook.GetSkyhook().Spec.RuntimeRequired {
                        continue
                }
                for _, node := range skyhook.GetNodes() {
                        if _, ok := node_to_skyhooks[node.GetNode().UID]; !ok {
                                node_to_skyhooks[node.GetNode().UID] = make([]SkyhookNodes, 0)
                                nodes[node.GetNode().UID] = node.GetNode()
                        }
                        node_to_skyhooks[node.GetNode().UID] = append(node_to_skyhooks[node.GetNode().UID], skyhook)
                }

        }
        return node_to_skyhooks, nodes
}

// Get the nodes to remove runtime required taint from node that all skyhooks targeting that node have completed
// Note: This checks per-node completion, not skyhook-level completion. A node's taint is removed when all
// runtime-required skyhooks are complete ON THAT SPECIFIC NODE, regardless of other nodes' completion status.
func getRuntimeRequiredTaintCompleteNodes(node_to_skyhooks map[types.UID][]SkyhookNodes, nodes map[types.UID]*corev1.Node) []*corev1.Node {
        to_remove := make([]*corev1.Node, 0)
        for node_uid, skyhooks := range node_to_skyhooks {
                node := nodes[node_uid]
                all_complete := true
                for _, skyhook := range skyhooks {
                        // Check if THIS specific node is complete for this skyhook (not all nodes)
                        _, nodeWrapper := skyhook.GetNode(node.Name)
                        if nodeWrapper == nil || !nodeWrapper.IsComplete() {
                                all_complete = false
                                break
                        }
                }
                if all_complete {
                        to_remove = append(to_remove, node)
                }
        }
        return to_remove
}

// setPodResources sets resources for all containers and init containers in the pod if override is set, else leaves empty for LimitRange
func setPodResources(pod *corev1.Pod, res *v1alpha1.ResourceRequirements) {
        if res == nil {
                return
        }
        if !res.CPURequest.IsZero() || !res.CPULimit.IsZero() || !res.MemoryRequest.IsZero() || !res.MemoryLimit.IsZero() {
                for i := range pod.Spec.InitContainers {
                        pod.Spec.InitContainers[i].Resources = corev1.ResourceRequirements{
                                Limits: corev1.ResourceList{
                                        corev1.ResourceCPU:    res.CPULimit,
                                        corev1.ResourceMemory: res.MemoryLimit,
                                },
                                Requests: corev1.ResourceList{
                                        corev1.ResourceCPU:    res.CPURequest,
                                        corev1.ResourceMemory: res.MemoryRequest,
                                },
                        }
                }
        }
}

// PartitionNodesIntoCompartments partitions nodes for each skyhook that uses deployment policies.
func partitionNodesIntoCompartments(clusterState *clusterState) error {
        for _, skyhook := range clusterState.skyhooks {
                // Skip skyhooks without a deployment policy (they use the default compartment created in BuildState)
                if skyhook.GetSkyhook().Spec.DeploymentPolicy == "" {
                        continue
                }

                // Skip if no compartments exist (e.g., deployment policy not found)
                // The webhook should prevent this at admission time, and the controller sets a condition at runtime,
                // but we guard here to prevent panics if the policy goes missing
                if len(skyhook.GetCompartments()) == 0 {
                        continue
                }

                // Clear all compartments before reassigning nodes to prevent stale nodes
                // This ensures nodes are only in their current compartment based on current labels
                for _, compartment := range skyhook.GetCompartments() {
                        compartment.ClearNodes()
                }

                for _, node := range skyhook.GetNodes() {
                        compartmentName, err := skyhook.AssignNodeToCompartment(node)
                        if err != nil {
                                return fmt.Errorf("error assigning node %s: %w", node.GetNode().Name, err)
                        }
                        if err := skyhook.AddCompartmentNode(compartmentName, node); err != nil {
                                return fmt.Errorf("error adding node %s to compartment %s: %w", node.GetNode().Name, compartmentName, err)
                        }
                }
        }

        return nil
}

// validateAndUpsertSkyhookData performs validation and configmap operations for a skyhook
func (r *SkyhookReconciler) validateAndUpsertSkyhookData(ctx context.Context, skyhook SkyhookNodes, clusterState *clusterState) (bool, ctrl.Result, error) {
        if yes, result, err := shouldReturn(r.ValidateRunningPackages(ctx, skyhook)); yes {
                return yes, result, err
        }

        if yes, result, err := shouldReturn(r.ValidateNodeConfigmaps(ctx, skyhook.GetSkyhook().Name, skyhook.GetNodes())); yes {
                return yes, result, err
        }

        if yes, result, err := shouldReturn(r.UpsertConfigmaps(ctx, skyhook, clusterState)); yes {
                return yes, result, err
        }

        return false, ctrl.Result{}, nil
}

NVIDIA / skyhook / 21848362404

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous