28108216358

Committed 24 Jun 2026 02:33PM UTC coverage: 29.476%. First build

Build # 28108216358

Build Type

Pull #2572

github

Committed by

karthikvetrivel

Commit Message

Requeue disabled GPUClusterConfig to detect ClusterPolicy removal

Signed-off-by: Karthik Vetrivel <kvetrivel@nvidia.com>

Pull Request Pull Request #2572: Support NVIDIADriver reconciliation without a ClusterPolicy

Coverage Stats

70 of 128 new or added lines in 4 files covered. (54.69%)

4040 of 13706 relevant lines covered (29.48%)

0.33 hits per line

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

49.53

/controllers/gpuclusterconfig_controller.go

/*
Copyright 2025.

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 controllers

import (
        "context"
        "fmt"
        "time"

        corev1 "k8s.io/api/core/v1"
        apierrors "k8s.io/apimachinery/pkg/api/errors"
        "k8s.io/apimachinery/pkg/runtime"
        "k8s.io/apimachinery/pkg/types"
        "k8s.io/client-go/util/workqueue"
        ctrl "sigs.k8s.io/controller-runtime"
        "sigs.k8s.io/controller-runtime/pkg/client"
        "sigs.k8s.io/controller-runtime/pkg/controller"
        "sigs.k8s.io/controller-runtime/pkg/event"
        "sigs.k8s.io/controller-runtime/pkg/handler"
        "sigs.k8s.io/controller-runtime/pkg/log"
        "sigs.k8s.io/controller-runtime/pkg/predicate"
        "sigs.k8s.io/controller-runtime/pkg/reconcile"
        "sigs.k8s.io/controller-runtime/pkg/source"

        gpuv1 "github.com/NVIDIA/gpu-operator/api/nvidia/v1"
        nvidiav1alpha1 "github.com/NVIDIA/gpu-operator/api/nvidia/v1alpha1"
        "github.com/NVIDIA/gpu-operator/controllers/clusterinfo"
        "github.com/NVIDIA/gpu-operator/internal/conditions"
        "github.com/NVIDIA/gpu-operator/internal/consts"
        "github.com/NVIDIA/gpu-operator/internal/state"
)

// GPUClusterConfigReconciler reconciles a GPUClusterConfig object
type GPUClusterConfigReconciler struct {
        client.Client
        Scheme      *runtime.Scheme
        ClusterInfo clusterinfo.Interface
        Namespace   string

        stateManager     state.Manager
        conditionUpdater conditions.Updater

        // singleton is the GPUClusterConfig that owns reconciliation; the first instance to
        // reconcile claims it (first-wins), mirroring ClusterPolicy.
        singleton *nvidiav1alpha1.GPUClusterConfig
}

//+kubebuilder:rbac:groups=nvidia.com,resources=gpuclusterconfigs,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=nvidia.com,resources=gpuclusterconfigs/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=nvidia.com,resources=gpuclusterconfigs/finalizers,verbs=update
//+kubebuilder:rbac:groups=nvidia.com,resources=clusterpolicies,verbs=get;list;watch
//+kubebuilder:rbac:groups="",resources=namespaces,verbs=get;update;patch
//+kubebuilder:rbac:groups="",resources=nodes,verbs=get;list;watch;update;patch
//+kubebuilder:rbac:groups=resource.k8s.io,resources=resourceclaimtemplates,verbs=get;list;watch;create;update;delete

func (r *GPUClusterConfigReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
        logger := log.FromContext(ctx)
        logger.V(consts.LogLevelInfo).Info("Reconciling GPUClusterConfig")

        instance := &nvidiav1alpha1.GPUClusterConfig{}
        if err := r.Get(ctx, req.NamespacedName, instance); err != nil {
                if apierrors.IsNotFound(err) {
                        // Deleted; owned objects are garbage-collected, so there is nothing to clean up.
                        return ctrl.Result{}, nil
                }
                // instance was not populated by the failed Get, so there is no object to
                // update status on; just surface the error for requeue.
                logger.Error(err, "error getting GPUClusterConfig object")
                return ctrl.Result{}, fmt.Errorf("error getting GPUClusterConfig object: %w", err)
        }

        // GPUClusterConfig (DRA path) is mutually exclusive with ClusterPolicy: if one
        // exists, yield to it rather than deploying the DRA stack alongside it.
        clusterPolicies := &gpuv1.ClusterPolicyList{}
        if err := r.List(ctx, clusterPolicies); err != nil {
                return ctrl.Result{}, fmt.Errorf("error listing ClusterPolicies: %w", err)
        }
        if len(clusterPolicies.Items) > 0 {
                logger.V(consts.LogLevelWarning).Info("ClusterPolicy present, skipping mutually exclusive GPUClusterConfig")
                if err := r.updateCrStatus(ctx, instance, nvidiav1alpha1.Disabled); err != nil {
                        return ctrl.Result{}, err
                }
                msg := "GPUClusterConfig is mutually exclusive with ClusterPolicy; remove the ClusterPolicy or disable GPUClusterConfig"
                if condErr := r.conditionUpdater.SetConditionsError(ctx, instance, conditions.ReconcileFailed, msg); condErr != nil {
                        logger.Error(condErr, "failed to set condition")
                }
                // Requeue so the ClusterPolicy's deletion is noticed and the instance
                // recovers; nothing watches ClusterPolicy here, mirroring the ready-path
                // resync below.
                return ctrl.Result{RequeueAfter: time.Minute}, nil
        }

        // Singleton, first-wins (mirroring ClusterPolicy): the first instance to reconcile
        // claims ownership; any other instance is marked Ignored and skipped. The owner is
        // held in memory, so the choice resets on operator restart.
        if r.singleton != nil && r.singleton.Name != instance.Name {
                logger.V(consts.LogLevelWarning).Info("Multiple GPUClusterConfig instances found, ignoring this one",
                        "name", instance.Name, "owner", r.singleton.Name)
                if err := r.updateCrStatus(ctx, instance, nvidiav1alpha1.Ignored); err != nil {
                        return ctrl.Result{}, err
                }
                return ctrl.Result{}, nil
        }
        r.singleton = instance

        if err := r.labelGPUNodes(ctx); err != nil {
                return ctrl.Result{}, fmt.Errorf("error labeling GPU nodes: %w", err)
        }

        infoCatalog := state.NewInfoCatalog()
        infoCatalog.Add(state.InfoTypeClusterInfo, r.ClusterInfo)

        managerStatus := r.stateManager.SyncState(ctx, instance, infoCatalog)

        if err := r.updateCrStatus(ctx, instance, nvidiav1alpha1.State(managerStatus.Status)); err != nil {
                return ctrl.Result{}, err
        }

        if managerStatus.Status != state.SyncStateReady {
                logger.Info("GPUClusterConfig instance is not ready")
                for _, result := range managerStatus.StatesStatus {
                        if result.Status != state.SyncStateReady && result.ErrInfo != nil {
                                if condErr := r.conditionUpdater.SetConditionsError(ctx, instance, conditions.ReconcileFailed, fmt.Sprintf("Error syncing state %s: %v", result.StateName, result.ErrInfo)); condErr != nil {
                                        logger.Error(condErr, "failed to set condition")
                                }
                                return ctrl.Result{RequeueAfter: time.Second * 5}, nil
                        }
                }
                // no state reported an error, so we are waiting on operand pods
                if condErr := r.conditionUpdater.SetConditionsError(ctx, instance, conditions.OperandNotReady, "Waiting for operand pods to be ready"); condErr != nil {
                        logger.Error(condErr, "failed to set condition")
                }
                return ctrl.Result{RequeueAfter: time.Second * 5}, nil
        }

        if condErr := r.conditionUpdater.SetConditionsReady(ctx, instance, conditions.Reconciled, "All resources have been successfully reconciled"); condErr != nil {
                logger.Error(condErr, "failed to set condition")
                return ctrl.Result{}, condErr
        }
        // Resync periodically so out-of-band changes (a deleted DeviceClass/VAP, or a
        // newly-created ClusterPolicy) are detected and reconciled even while ready;
        // only DaemonSets are watched, and the ready path is otherwise event-driven.
        return ctrl.Result{RequeueAfter: time.Minute}, nil
}

// labelGPUNodes labels nodes NFD discovered NVIDIA GPUs on with the common
// nvidia.com/gpu.present label, and resets it to "false" when a node no longer has
// GPUs. This is a slimmed version of the ClusterPolicy controller's labeler: the
// operands match on the common label directly, so per-component state labels are
// not managed, except the driver deploy label, which the driver DaemonSet
// rendered for NVIDIADriver hardcodes in its nodeSelector and which only the
// ClusterPolicy controller would otherwise apply.
func (r *GPUClusterConfigReconciler) labelGPUNodes(ctx context.Context) error {
        logger := log.FromContext(ctx)

        list := &corev1.NodeList{}
        if err := r.List(ctx, list); err != nil {
                return fmt.Errorf("unable to list nodes to check labels: %w", err)
        }
        for i := range list.Items {
                node := &list.Items[i]
                labels := node.GetLabels()
                patch := client.MergeFrom(node.DeepCopy())
                modified := false
                if hasGPULabels(labels) {
                        if !hasCommonGPULabel(labels) {
                                labels[commonGPULabelKey] = commonGPULabelValue
                                modified = true
                        }
                        if labels[driverDeployLabelKey] != "true" {
                                labels[driverDeployLabelKey] = "true"
                                modified = true
                        }
                        // Set the DRA kubelet-plugin deploy label (which the plugin selects on) only
                        // when absent, so the k8s-driver-manager can pause it during a driver reload
                        // to drain the plugin without this controller fighting the pause.
                        if _, ok := labels[state.DraDriverDeployLabelKey]; !ok {
                                labels[state.DraDriverDeployLabelKey] = "true"
                                modified = true
                        }
                        // Set the GFD deploy label (which the GFD DaemonSet selects on) only when
                        // absent, so the k8s-driver-manager can pause it during a driver reload to
                        // drain GFD off the node without this controller fighting the pause.
                        if _, ok := labels[gfdDeployLabelKey]; !ok {
                                labels[gfdDeployLabelKey] = "true"
                                modified = true
                        }
                } else if hasCommonGPULabel(labels) {
                        labels[commonGPULabelKey] = "false"
                        delete(labels, driverDeployLabelKey)
                        delete(labels, state.DraDriverDeployLabelKey)
                        delete(labels, gfdDeployLabelKey)
                        modified = true
                }
                if !modified {
                        continue
                }
                node.SetLabels(labels)
                logger.Info("Updating GPU node labels", "node", node.Name)
                if err := r.Patch(ctx, node, patch); err != nil {
                        return fmt.Errorf("unable to label node %s: %w", node.Name, err)
                }
        }
        return nil
}

// updateCrStatus writes desired to the CR's status, skipping the write when it is already current.
func (r *GPUClusterConfigReconciler) updateCrStatus(ctx context.Context, cr *nvidiav1alpha1.GPUClusterConfig, desired nvidiav1alpha1.State) error {
        reqLogger := log.FromContext(ctx)

        // Refetch to avoid a resourceVersion conflict.
        instance := &nvidiav1alpha1.GPUClusterConfig{}
        if err := r.Get(ctx, types.NamespacedName{Name: cr.Name}, instance); err != nil {
                reqLogger.Error(err, "Failed to get GPUClusterConfig instance for status update")
                return err
        }

        if instance.Status.State == desired && instance.Status.Namespace == r.Namespace {
                return nil
        }
        instance.Status.State = desired
        instance.Status.Namespace = r.Namespace

        reqLogger.V(consts.LogLevelInfo).Info("Updating CR Status", "Status", instance.Status)
        if err := r.Status().Update(ctx, instance); err != nil {
                reqLogger.Error(err, "Failed to update CR status")
                return err
        }
        cr.Status.State = instance.Status.State
        cr.Status.Namespace = instance.Status.Namespace
        return nil
}

// enqueueAllGPUClusterConfigs enqueues every instance so each is reconciled when any
// instance or owned resource changes.
func (r *GPUClusterConfigReconciler) enqueueAllGPUClusterConfigs(ctx context.Context, _ *nvidiav1alpha1.GPUClusterConfig) []reconcile.Request {
        logger := log.FromContext(ctx)
        list := &nvidiav1alpha1.GPUClusterConfigList{}

        if err := r.List(ctx, list); err != nil {
                logger.Error(err, "Unable to list GPUClusterConfig resources")
                return []reconcile.Request{}
        }

        reconcileRequests := make([]reconcile.Request, 0, len(list.Items))
        for _, config := range list.Items {
                reconcileRequests = append(reconcileRequests,
                        reconcile.Request{
                                NamespacedName: types.NamespacedName{
                                        Name: config.GetName(),
                                },
                        })
        }

        return reconcileRequests
}

func (r *GPUClusterConfigReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager) error {
        // The state manager renders the DRA driver operand for the GPUClusterConfig.
        stateManager, err := state.NewManager(
                nvidiav1alpha1.GPUClusterConfigCRDName,
                r.Namespace,
                mgr.GetClient(),
                mgr.GetScheme())
        if err != nil {
                return fmt.Errorf("error creating state manager: %v", err)
        }
        r.stateManager = stateManager

        r.conditionUpdater = conditions.NewGPUClusterConfigUpdater(mgr.GetClient())

        c, err := controller.New("gpu-cluster-config-controller", mgr, controller.Options{
                Reconciler:              r,
                MaxConcurrentReconciles: 1,
                RateLimiter:             workqueue.NewTypedItemExponentialFailureRateLimiter[reconcile.Request](minDelayCR, maxDelayCR),
        })
        if err != nil {
                return err
        }

        err = c.Watch(source.Kind(
                mgr.GetCache(),
                &nvidiav1alpha1.GPUClusterConfig{},
                handler.TypedEnqueueRequestsFromMapFunc(r.enqueueAllGPUClusterConfigs),
                predicate.TypedGenerationChangedPredicate[*nvidiav1alpha1.GPUClusterConfig]{},
        ),
        )
        if err != nil {
                return err
        }

        // Watch Nodes so GPU nodes are labeled as NFD discovers GPUs (and the label is
        // reset when they disappear). Only label transitions the labeler acts on trigger
        // a reconcile; the 1-minute Ready resync covers any remaining drift.
        nodeMapFn := func(ctx context.Context, _ *corev1.Node) []reconcile.Request {
                return r.enqueueAllGPUClusterConfigs(ctx, nil)
        }
        nodePredicate := predicate.TypedFuncs[*corev1.Node]{
                CreateFunc: func(e event.TypedCreateEvent[*corev1.Node]) bool {
                        return hasGPULabels(e.Object.GetLabels())
                },
                UpdateFunc: func(e event.TypedUpdateEvent[*corev1.Node]) bool {
                        labels := e.ObjectNew.GetLabels()
                        return hasCommonGPULabel(labels) != hasGPULabels(labels)
                },
                DeleteFunc: func(_ event.TypedDeleteEvent[*corev1.Node]) bool {
                        return false
                },
        }
        err = c.Watch(source.Kind(
                mgr.GetCache(),
                &corev1.Node{},
                handler.TypedEnqueueRequestsFromMapFunc(nodeMapFn),
                nodePredicate,
        ))
        if err != nil {
                return err
        }

        // Watch the secondary resources each state manager owns.
        watchSources := stateManager.GetWatchSources(mgr)
        for _, watchSource := range watchSources {
                err = c.Watch(
                        watchSource,
                )
                if err != nil {
                        return fmt.Errorf("error setting up Watch for source type %v: %w", watchSource, err)
                }
        }

        return nil
}

1	/*
2	Copyright 2025.
3
4	Licensed under the Apache License, Version 2.0 (the "License");
5	you may not use this file except in compliance with the License.
6	You may obtain a copy of the License at
7
8	http://www.apache.org/licenses/LICENSE-2.0
9
10	Unless required by applicable law or agreed to in writing, software
11	distributed under the License is distributed on an "AS IS" BASIS,
12	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	See the License for the specific language governing permissions and
14	limitations under the License.
15	*/
16
17	package controllers
18
19	import (
20	"context"
21	"fmt"
22	"time"
23
24	corev1 "k8s.io/api/core/v1"
25	apierrors "k8s.io/apimachinery/pkg/api/errors"
26	"k8s.io/apimachinery/pkg/runtime"
27	"k8s.io/apimachinery/pkg/types"
28	"k8s.io/client-go/util/workqueue"
29	ctrl "sigs.k8s.io/controller-runtime"
30	"sigs.k8s.io/controller-runtime/pkg/client"
31	"sigs.k8s.io/controller-runtime/pkg/controller"
32	"sigs.k8s.io/controller-runtime/pkg/event"
33	"sigs.k8s.io/controller-runtime/pkg/handler"
34	"sigs.k8s.io/controller-runtime/pkg/log"
35	"sigs.k8s.io/controller-runtime/pkg/predicate"
36	"sigs.k8s.io/controller-runtime/pkg/reconcile"
37	"sigs.k8s.io/controller-runtime/pkg/source"
38
39	gpuv1 "github.com/NVIDIA/gpu-operator/api/nvidia/v1"
40	nvidiav1alpha1 "github.com/NVIDIA/gpu-operator/api/nvidia/v1alpha1"
41	"github.com/NVIDIA/gpu-operator/controllers/clusterinfo"
42	"github.com/NVIDIA/gpu-operator/internal/conditions"
43	"github.com/NVIDIA/gpu-operator/internal/consts"
44	"github.com/NVIDIA/gpu-operator/internal/state"
45	)
46
47	// GPUClusterConfigReconciler reconciles a GPUClusterConfig object
48	type GPUClusterConfigReconciler struct {
49	client.Client
50	Scheme *runtime.Scheme
51	ClusterInfo clusterinfo.Interface
52	Namespace string
53
54	stateManager state.Manager
55	conditionUpdater conditions.Updater
56
57	// singleton is the GPUClusterConfig that owns reconciliation; the first instance to
58	// reconcile claims it (first-wins), mirroring ClusterPolicy.
59	singleton *nvidiav1alpha1.GPUClusterConfig
60	}
61
62	//+kubebuilder:rbac:groups=nvidia.com,resources=gpuclusterconfigs,verbs=get;list;watch;create;update;patch;delete
63	//+kubebuilder:rbac:groups=nvidia.com,resources=gpuclusterconfigs/status,verbs=get;update;patch
64	//+kubebuilder:rbac:groups=nvidia.com,resources=gpuclusterconfigs/finalizers,verbs=update
65	//+kubebuilder:rbac:groups=nvidia.com,resources=clusterpolicies,verbs=get;list;watch
66	//+kubebuilder:rbac:groups="",resources=namespaces,verbs=get;update;patch
67	//+kubebuilder:rbac:groups="",resources=nodes,verbs=get;list;watch;update;patch
68	//+kubebuilder:rbac:groups=resource.k8s.io,resources=resourceclaimtemplates,verbs=get;list;watch;create;update;delete
69
70	func (r *GPUClusterConfigReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {	1✔
71	logger := log.FromContext(ctx)	1✔
72	logger.V(consts.LogLevelInfo).Info("Reconciling GPUClusterConfig")	1✔
73		1✔
74	instance := &nvidiav1alpha1.GPUClusterConfig{}	1✔
75	if err := r.Get(ctx, req.NamespacedName, instance); err != nil {	2✔
76	if apierrors.IsNotFound(err) {	2✔
77	// Deleted; owned objects are garbage-collected, so there is nothing to clean up.	1✔
78	return ctrl.Result{}, nil	1✔
79	}	1✔
80	// instance was not populated by the failed Get, so there is no object to
81	// update status on; just surface the error for requeue.
82	logger.Error(err, "error getting GPUClusterConfig object")	×
83	return ctrl.Result{}, fmt.Errorf("error getting GPUClusterConfig object: %w", err)	×
84	}
85
86	// GPUClusterConfig (DRA path) is mutually exclusive with ClusterPolicy: if one
87	// exists, yield to it rather than deploying the DRA stack alongside it.
88	clusterPolicies := &gpuv1.ClusterPolicyList{}	1✔
89	if err := r.List(ctx, clusterPolicies); err != nil {	1✔
90	return ctrl.Result{}, fmt.Errorf("error listing ClusterPolicies: %w", err)	×
91	}	×
92	if len(clusterPolicies.Items) > 0 {	2✔
93	logger.V(consts.LogLevelWarning).Info("ClusterPolicy present, skipping mutually exclusive GPUClusterConfig")	1✔
94	if err := r.updateCrStatus(ctx, instance, nvidiav1alpha1.Disabled); err != nil {	1✔
95	return ctrl.Result{}, err	×
96	}	×
97	msg := "GPUClusterConfig is mutually exclusive with ClusterPolicy; remove the ClusterPolicy or disable GPUClusterConfig"	1✔
98	if condErr := r.conditionUpdater.SetConditionsError(ctx, instance, conditions.ReconcileFailed, msg); condErr != nil {	1✔
99	logger.Error(condErr, "failed to set condition")	×
100	}	×
101	// Requeue so the ClusterPolicy's deletion is noticed and the instance
102	// recovers; nothing watches ClusterPolicy here, mirroring the ready-path
103	// resync below.
104	return ctrl.Result{RequeueAfter: time.Minute}, nil	1✔
105	}
106
107	// Singleton, first-wins (mirroring ClusterPolicy): the first instance to reconcile
108	// claims ownership; any other instance is marked Ignored and skipped. The owner is
109	// held in memory, so the choice resets on operator restart.
110	if r.singleton != nil && r.singleton.Name != instance.Name {	2✔
111	logger.V(consts.LogLevelWarning).Info("Multiple GPUClusterConfig instances found, ignoring this one",	1✔
112	"name", instance.Name, "owner", r.singleton.Name)	1✔
113	if err := r.updateCrStatus(ctx, instance, nvidiav1alpha1.Ignored); err != nil {	1✔
114	return ctrl.Result{}, err	×
115	}	×
116	return ctrl.Result{}, nil	1✔
117	}
118	r.singleton = instance	1✔
119		1✔
120	if err := r.labelGPUNodes(ctx); err != nil {	1✔
NEW 121	return ctrl.Result{}, fmt.Errorf("error labeling GPU nodes: %w", err)	×
NEW 122	}	×
123
124	infoCatalog := state.NewInfoCatalog()	1✔
125	infoCatalog.Add(state.InfoTypeClusterInfo, r.ClusterInfo)	1✔
126		1✔
127	managerStatus := r.stateManager.SyncState(ctx, instance, infoCatalog)	1✔
128		1✔
129	if err := r.updateCrStatus(ctx, instance, nvidiav1alpha1.State(managerStatus.Status)); err != nil {	1✔
130	return ctrl.Result{}, err	×
131	}	×
132
133	if managerStatus.Status != state.SyncStateReady {	1✔
134	logger.Info("GPUClusterConfig instance is not ready")	×
135	for _, result := range managerStatus.StatesStatus {	×
136	if result.Status != state.SyncStateReady && result.ErrInfo != nil {	×
137	if condErr := r.conditionUpdater.SetConditionsError(ctx, instance, conditions.ReconcileFailed, fmt.Sprintf("Error syncing state %s: %v", result.StateName, result.ErrInfo)); condErr != nil {	×
138	logger.Error(condErr, "failed to set condition")	×
139	}	×
140	return ctrl.Result{RequeueAfter: time.Second * 5}, nil	×
141	}
142	}
143	// no state reported an error, so we are waiting on operand pods
144	if condErr := r.conditionUpdater.SetConditionsError(ctx, instance, conditions.OperandNotReady, "Waiting for operand pods to be ready"); condErr != nil {	×
145	logger.Error(condErr, "failed to set condition")	×
146	}	×
147	return ctrl.Result{RequeueAfter: time.Second * 5}, nil	×
148	}
149
150	if condErr := r.conditionUpdater.SetConditionsReady(ctx, instance, conditions.Reconciled, "All resources have been successfully reconciled"); condErr != nil {	1✔
151	logger.Error(condErr, "failed to set condition")	×
152	return ctrl.Result{}, condErr	×
153	}	×
154	// Resync periodically so out-of-band changes (a deleted DeviceClass/VAP, or a
155	// newly-created ClusterPolicy) are detected and reconciled even while ready;
156	// only DaemonSets are watched, and the ready path is otherwise event-driven.
157	return ctrl.Result{RequeueAfter: time.Minute}, nil	1✔
158	}
159
160	// labelGPUNodes labels nodes NFD discovered NVIDIA GPUs on with the common
161	// nvidia.com/gpu.present label, and resets it to "false" when a node no longer has
162	// GPUs. This is a slimmed version of the ClusterPolicy controller's labeler: the
163	// operands match on the common label directly, so per-component state labels are
164	// not managed, except the driver deploy label, which the driver DaemonSet
165	// rendered for NVIDIADriver hardcodes in its nodeSelector and which only the
166	// ClusterPolicy controller would otherwise apply.
167	func (r *GPUClusterConfigReconciler) labelGPUNodes(ctx context.Context) error {	1✔
168	logger := log.FromContext(ctx)	1✔
169		1✔
170	list := &corev1.NodeList{}	1✔
171	if err := r.List(ctx, list); err != nil {	1✔
NEW 172	return fmt.Errorf("unable to list nodes to check labels: %w", err)	×
NEW 173	}	×
174	for i := range list.Items {	2✔
175	node := &list.Items[i]	1✔
176	labels := node.GetLabels()	1✔
177	patch := client.MergeFrom(node.DeepCopy())	1✔
178	modified := false	1✔
179	if hasGPULabels(labels) {	2✔
180	if !hasCommonGPULabel(labels) {	2✔
181	labels[commonGPULabelKey] = commonGPULabelValue	1✔
182	modified = true	1✔
183	}	1✔
184	if labels[driverDeployLabelKey] != "true" {	2✔
185	labels[driverDeployLabelKey] = "true"	1✔
186	modified = true	1✔
187	}	1✔
188	// Set the DRA kubelet-plugin deploy label (which the plugin selects on) only
189	// when absent, so the k8s-driver-manager can pause it during a driver reload
190	// to drain the plugin without this controller fighting the pause.
191	if _, ok := labels[state.DraDriverDeployLabelKey]; !ok {	2✔
192	labels[state.DraDriverDeployLabelKey] = "true"	1✔
193	modified = true	1✔
194	}	1✔
195	// Set the GFD deploy label (which the GFD DaemonSet selects on) only when
196	// absent, so the k8s-driver-manager can pause it during a driver reload to
197	// drain GFD off the node without this controller fighting the pause.
198	if _, ok := labels[gfdDeployLabelKey]; !ok {	2✔
199	labels[gfdDeployLabelKey] = "true"	1✔
200	modified = true	1✔
201	}	1✔
202	} else if hasCommonGPULabel(labels) {	2✔
203	labels[commonGPULabelKey] = "false"	1✔
204	delete(labels, driverDeployLabelKey)	1✔
205	delete(labels, state.DraDriverDeployLabelKey)	1✔
206	delete(labels, gfdDeployLabelKey)	1✔
207	modified = true	1✔
208	}	1✔
209	if !modified {	2✔
210	continue	1✔
211	}
212	node.SetLabels(labels)	1✔
213	logger.Info("Updating GPU node labels", "node", node.Name)	1✔
214	if err := r.Patch(ctx, node, patch); err != nil {	1✔
NEW 215	return fmt.Errorf("unable to label node %s: %w", node.Name, err)	×
NEW 216	}	×
217	}
218	return nil	1✔
219	}
220
221	// updateCrStatus writes desired to the CR's status, skipping the write when it is already current.
222	func (r GPUClusterConfigReconciler) updateCrStatus(ctx context.Context, cr nvidiav1alpha1.GPUClusterConfig, desired nvidiav1alpha1.State) error {	1✔
223	reqLogger := log.FromContext(ctx)	1✔
224		1✔
225	// Refetch to avoid a resourceVersion conflict.	1✔
226	instance := &nvidiav1alpha1.GPUClusterConfig{}	1✔
227	if err := r.Get(ctx, types.NamespacedName{Name: cr.Name}, instance); err != nil {	1✔
228	reqLogger.Error(err, "Failed to get GPUClusterConfig instance for status update")	×
229	return err	×
230	}	×
231
232	if instance.Status.State == desired && instance.Status.Namespace == r.Namespace {	1✔
233	return nil	×
234	}	×
235	instance.Status.State = desired	1✔
236	instance.Status.Namespace = r.Namespace	1✔
237		1✔
238	reqLogger.V(consts.LogLevelInfo).Info("Updating CR Status", "Status", instance.Status)	1✔
239	if err := r.Status().Update(ctx, instance); err != nil {	1✔
240	reqLogger.Error(err, "Failed to update CR status")	×
241	return err	×
242	}	×
243	cr.Status.State = instance.Status.State	1✔
244	cr.Status.Namespace = instance.Status.Namespace	1✔
245	return nil	1✔
246	}
247
248	// enqueueAllGPUClusterConfigs enqueues every instance so each is reconciled when any
249	// instance or owned resource changes.
250	func (r GPUClusterConfigReconciler) enqueueAllGPUClusterConfigs(ctx context.Context, _ nvidiav1alpha1.GPUClusterConfig) []reconcile.Request {	1✔
251	logger := log.FromContext(ctx)	1✔
252	list := &nvidiav1alpha1.GPUClusterConfigList{}	1✔
253		1✔
254	if err := r.List(ctx, list); err != nil {	1✔
255	logger.Error(err, "Unable to list GPUClusterConfig resources")	×
256	return []reconcile.Request{}	×
257	}	×
258
259	reconcileRequests := make([]reconcile.Request, 0, len(list.Items))	1✔
260	for _, config := range list.Items {	2✔
261	reconcileRequests = append(reconcileRequests,	1✔
262	reconcile.Request{	1✔
263	NamespacedName: types.NamespacedName{	1✔
264	Name: config.GetName(),	1✔
265	},	1✔
266	})	1✔
267	}	1✔
268
269	return reconcileRequests	1✔
270	}
271
272	func (r *GPUClusterConfigReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager) error {	×
273	// The state manager renders the DRA driver operand for the GPUClusterConfig.	×
274	stateManager, err := state.NewManager(	×
275	nvidiav1alpha1.GPUClusterConfigCRDName,	×
276	r.Namespace,	×
277	mgr.GetClient(),	×
278	mgr.GetScheme())	×
279	if err != nil {	×
280	return fmt.Errorf("error creating state manager: %v", err)	×
281	}	×
282	r.stateManager = stateManager	×
283		×
284	r.conditionUpdater = conditions.NewGPUClusterConfigUpdater(mgr.GetClient())	×
285		×
286	c, err := controller.New("gpu-cluster-config-controller", mgr, controller.Options{	×
287	Reconciler: r,	×
288	MaxConcurrentReconciles: 1,	×
289	RateLimiter: workqueue.NewTypedItemExponentialFailureRateLimiter[reconcile.Request](minDelayCR, maxDelayCR),	×
290	})	×
291	if err != nil {	×
292	return err	×
293	}	×
294
295	err = c.Watch(source.Kind(	×
296	mgr.GetCache(),	×
297	&nvidiav1alpha1.GPUClusterConfig{},	×
298	handler.TypedEnqueueRequestsFromMapFunc(r.enqueueAllGPUClusterConfigs),	×
299	predicate.TypedGenerationChangedPredicate[*nvidiav1alpha1.GPUClusterConfig]{},	×
300	),	×
301	)	×
302	if err != nil {	×
303	return err	×
304	}	×
305
306	// Watch Nodes so GPU nodes are labeled as NFD discovers GPUs (and the label is
307	// reset when they disappear). Only label transitions the labeler acts on trigger
308	// a reconcile; the 1-minute Ready resync covers any remaining drift.
NEW 309	nodeMapFn := func(ctx context.Context, _ *corev1.Node) []reconcile.Request {	×
NEW 310	return r.enqueueAllGPUClusterConfigs(ctx, nil)	×
NEW 311	}	×
NEW 312	nodePredicate := predicate.TypedFuncs[*corev1.Node]{	×
NEW 313	CreateFunc: func(e event.TypedCreateEvent[*corev1.Node]) bool {	×
NEW 314	return hasGPULabels(e.Object.GetLabels())	×
NEW 315	},	×
NEW 316	UpdateFunc: func(e event.TypedUpdateEvent[*corev1.Node]) bool {	×
NEW 317	labels := e.ObjectNew.GetLabels()	×
NEW 318	return hasCommonGPULabel(labels) != hasGPULabels(labels)	×
NEW 319	},	×
NEW 320	DeleteFunc: func(_ event.TypedDeleteEvent[*corev1.Node]) bool {	×
NEW 321	return false	×
NEW 322	},	×
323	}
NEW 324	err = c.Watch(source.Kind(	×
NEW 325	mgr.GetCache(),	×
NEW 326	&corev1.Node{},	×
NEW 327	handler.TypedEnqueueRequestsFromMapFunc(nodeMapFn),	×
NEW 328	nodePredicate,	×
NEW 329	))	×
NEW 330	if err != nil {	×
NEW 331	return err	×
NEW 332	}	×
333
334	// Watch the secondary resources each state manager owns.
335	watchSources := stateManager.GetWatchSources(mgr)	×
336	for _, watchSource := range watchSources {	×
337	err = c.Watch(	×
338	watchSource,	×
339	)	×
340	if err != nil {	×
341	return fmt.Errorf("error setting up Watch for source type %v: %w", watchSource, err)	×
342	}	×
343	}
344
345	return nil	×
346	}

NVIDIA / gpu-operator / 28108216358

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