20490873192

Committed 24 Dec 2025 05:19PM UTC coverage: 63.224% (-0.007%) from 63.231%

Build # 20490873192

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #989 from k8snetworkplumbingwg/dependabot/go_modules/general-dependencies-d732c02c82

build(deps): bump the general-dependencies group with 4 updates

Run Details

9120 of 14425 relevant lines covered (63.22%)

0.7 hits per line

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

82.79

/controllers/drain_controller.go

/*
Copyright 2021.

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"
        "sync"

        "github.com/go-logr/logr"
        corev1 "k8s.io/api/core/v1"
        "k8s.io/apimachinery/pkg/api/errors"
        "k8s.io/apimachinery/pkg/runtime"
        "k8s.io/apimachinery/pkg/types"
        "k8s.io/client-go/tools/record"
        "k8s.io/client-go/util/workqueue"
        ctrl "sigs.k8s.io/controller-runtime"
        "sigs.k8s.io/controller-runtime/pkg/builder"
        "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/reconcile"

        sriovnetworkv1 "github.com/k8snetworkplumbingwg/sriov-network-operator/api/v1"
        constants "github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/consts"
        "github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/drain"
        "github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/orchestrator"
        "github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/utils"
        "github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/vars"
)

type DrainReconcile struct {
        client.Client
        Scheme   *runtime.Scheme
        recorder record.EventRecorder
        drainer  drain.DrainInterface

        drainCheckMutex sync.Mutex
}

func NewDrainReconcileController(client client.Client, Scheme *runtime.Scheme, recorder record.EventRecorder, orchestrator orchestrator.Interface) (*DrainReconcile, error) {
        drainer, err := drain.NewDrainer(orchestrator)
        if err != nil {
                return nil, err
        }

        return &DrainReconcile{
                client,
                Scheme,
                recorder,
                drainer,
                sync.Mutex{}}, nil
}

//+kubebuilder:rbac:groups="",resources=nodes,verbs=get;list;watch;update;patch
//+kubebuilder:rbac:groups=sriovnetwork.openshift.io,resources=sriovnodestates,verbs=get;list;watch
//+kubebuilder:rbac:groups=core,resources=events,verbs=create;patch

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
// For more details, check Reconcile and its Result here:
// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.8.3/pkg/reconcile
func (dr *DrainReconcile) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
        ctx = context.WithValue(ctx, constants.LoggerContextKey, log.FromContext(ctx))
        reqLogger := log.FromContext(ctx).WithName("Drain Reconcile")

        req.Namespace = vars.Namespace

        // get node object
        node := &corev1.Node{}
        found, err := dr.getObject(ctx, req, node)
        if err != nil {
                reqLogger.Error(err, "failed to get node object")
                return ctrl.Result{}, err
        }
        if !found {
                reqLogger.Info("node not found don't, requeue the request")
                return ctrl.Result{}, nil
        }

        // get sriovNodeNodeState object
        nodeNetworkState := &sriovnetworkv1.SriovNetworkNodeState{}
        found, err = dr.getObject(ctx, req, nodeNetworkState)
        if err != nil {
                reqLogger.Error(err, "failed to get sriovNetworkNodeState object")
                return ctrl.Result{}, err
        }
        if !found {
                reqLogger.Info("sriovNetworkNodeState not found, don't requeue the request")
                return ctrl.Result{}, nil
        }

        // create the drain state annotation if it doesn't exist in the sriovNetworkNodeState object
        nodeStateDrainAnnotationCurrent, currentNodeStateExist, err := dr.ensureAnnotationExists(ctx, nodeNetworkState, constants.NodeStateDrainAnnotationCurrent)
        if err != nil {
                reqLogger.Error(err, "failed to ensure nodeStateDrainAnnotationCurrent")
                return ctrl.Result{}, err
        }
        _, desireNodeStateExist, err := dr.ensureAnnotationExists(ctx, nodeNetworkState, constants.NodeStateDrainAnnotation)
        if err != nil {
                reqLogger.Error(err, "failed to ensure nodeStateDrainAnnotation")
                return ctrl.Result{}, err
        }

        // create the drain state annotation if it doesn't exist in the node object
        nodeDrainAnnotation, nodeExist, err := dr.ensureAnnotationExists(ctx, node, constants.NodeDrainAnnotation)
        if err != nil {
                reqLogger.Error(err, "failed to ensure nodeStateDrainAnnotation")
                return ctrl.Result{}, err
        }

        // requeue the request if we needed to add any of the annotations
        if !nodeExist || !currentNodeStateExist || !desireNodeStateExist {
                return ctrl.Result{Requeue: true}, nil
        }
        reqLogger.V(2).Info("Drain annotations", "nodeAnnotation", nodeDrainAnnotation, "nodeStateAnnotation", nodeStateDrainAnnotationCurrent)

        // Check the node request
        if nodeDrainAnnotation == constants.DrainIdle {
                // this cover the case the node is on idle

                // node request to be on idle and the currect state is idle
                // we don't do anything
                if nodeStateDrainAnnotationCurrent == constants.DrainIdle {
                        reqLogger.Info("node and nodeState are on idle nothing todo")
                        return reconcile.Result{}, nil
                }

                // we have two options here:
                // 1. node request idle and the current status is drain complete
                // this means the daemon finish is work, so we need to clean the drain
                //
                // 2. the operator is still draining the node but maybe the sriov policy changed and the daemon
                //  doesn't need to drain anymore, so we can stop the drain
                if nodeStateDrainAnnotationCurrent == constants.DrainComplete ||
                        nodeStateDrainAnnotationCurrent == constants.Draining {
                        return dr.handleNodeIdleNodeStateDrainingOrCompleted(ctx, node, nodeNetworkState)
                }
        }

        // this cover the case a node request to drain or reboot
        if nodeDrainAnnotation == constants.DrainRequired ||
                nodeDrainAnnotation == constants.RebootRequired {
                return dr.handleNodeDrainOrReboot(ctx, node, nodeNetworkState, nodeDrainAnnotation, nodeStateDrainAnnotationCurrent)
        }

        reqLogger.Error(nil, "unexpected node drain annotation")
        return reconcile.Result{}, fmt.Errorf("unexpected node drain annotation")
}

func (dr *DrainReconcile) getObject(ctx context.Context, req ctrl.Request, object client.Object) (bool, error) {
        err := dr.Get(ctx, req.NamespacedName, object)
        if err != nil {
                if errors.IsNotFound(err) {
                        return false, nil
                }
                return false, err
        }
        return true, nil
}

func (dr *DrainReconcile) ensureAnnotationExists(ctx context.Context, object client.Object, key string) (string, bool, error) {
        value, exist := object.GetAnnotations()[key]
        if !exist {
                err := utils.AnnotateObject(ctx, object, key, constants.DrainIdle, dr.Client)
                if err != nil {
                        return "", false, err
                }
                return constants.DrainIdle, false, nil
        }

        return value, true, nil
}

// SetupWithManager sets up the controller with the Manager.
func (dr *DrainReconcile) SetupWithManager(mgr ctrl.Manager) error {
        createUpdateEnqueue := handler.Funcs{
                CreateFunc: func(c context.Context, e event.TypedCreateEvent[client.Object], w workqueue.TypedRateLimitingInterface[reconcile.Request]) {
                        w.Add(reconcile.Request{NamespacedName: types.NamespacedName{
                                Namespace: vars.Namespace,
                                Name:      e.Object.GetName(),
                        }})
                },
                UpdateFunc: func(ctx context.Context, e event.TypedUpdateEvent[client.Object], w workqueue.TypedRateLimitingInterface[reconcile.Request]) {
                        w.Add(reconcile.Request{NamespacedName: types.NamespacedName{
                                Namespace: vars.Namespace,
                                Name:      e.ObjectNew.GetName(),
                        }})
                },
        }

        // Watch for spec and annotation changes
        nodePredicates := builder.WithPredicates(DrainAnnotationPredicate{})
        nodeStatePredicates := builder.WithPredicates(DrainStateAnnotationPredicate{})

        return ctrl.NewControllerManagedBy(mgr).
                WithOptions(controller.Options{
                        MaxConcurrentReconciles: 50,
                        LogConstructor: func(request *reconcile.Request) logr.Logger {
                                logger := mgr.GetLogger().WithValues("Function", "Drain")
                                // Inspired by https://github.com/kubernetes-sigs/controller-runtime/blob/52b17917caa97ec546423867d9637f1787830f3e/pkg/builder/controller.go#L447
                                if req, ok := any(request).(*reconcile.Request); ok && req != nil {
                                        logger = logger.WithValues("node", request.Name)
                                }
                                return logger
                        },
                }).
                For(&corev1.Node{}, nodePredicates).
                Watches(&sriovnetworkv1.SriovNetworkNodeState{}, createUpdateEnqueue, nodeStatePredicates).
                Complete(dr)
}

1	/*
2	Copyright 2021.
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	"sync"
23
24	"github.com/go-logr/logr"
25	corev1 "k8s.io/api/core/v1"
26	"k8s.io/apimachinery/pkg/api/errors"
27	"k8s.io/apimachinery/pkg/runtime"
28	"k8s.io/apimachinery/pkg/types"
29	"k8s.io/client-go/tools/record"
30	"k8s.io/client-go/util/workqueue"
31	ctrl "sigs.k8s.io/controller-runtime"
32	"sigs.k8s.io/controller-runtime/pkg/builder"
33	"sigs.k8s.io/controller-runtime/pkg/client"
34	"sigs.k8s.io/controller-runtime/pkg/controller"
35	"sigs.k8s.io/controller-runtime/pkg/event"
36	"sigs.k8s.io/controller-runtime/pkg/handler"
37	"sigs.k8s.io/controller-runtime/pkg/log"
38	"sigs.k8s.io/controller-runtime/pkg/reconcile"
39
40	sriovnetworkv1 "github.com/k8snetworkplumbingwg/sriov-network-operator/api/v1"
41	constants "github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/consts"
42	"github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/drain"
43	"github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/orchestrator"
44	"github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/utils"
45	"github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/vars"
46	)
47
48	type DrainReconcile struct {
49	client.Client
50	Scheme *runtime.Scheme
51	recorder record.EventRecorder
52	drainer drain.DrainInterface
53
54	drainCheckMutex sync.Mutex
55	}
56
57	func NewDrainReconcileController(client client.Client, Scheme runtime.Scheme, recorder record.EventRecorder, orchestrator orchestrator.Interface) (DrainReconcile, error) {	1✔
58	drainer, err := drain.NewDrainer(orchestrator)	1✔
59	if err != nil {	1✔
60	return nil, err	×
61	}	×
62
63	return &DrainReconcile{	1✔
64	client,	1✔
65	Scheme,	1✔
66	recorder,	1✔
67	drainer,	1✔
68	sync.Mutex{}}, nil	1✔
69	}
70
71	//+kubebuilder:rbac:groups="",resources=nodes,verbs=get;list;watch;update;patch
72	//+kubebuilder:rbac:groups=sriovnetwork.openshift.io,resources=sriovnodestates,verbs=get;list;watch
73	//+kubebuilder:rbac:groups=core,resources=events,verbs=create;patch
74
75	// Reconcile is part of the main kubernetes reconciliation loop which aims to
76	// move the current state of the cluster closer to the desired state.
77	// For more details, check Reconcile and its Result here:
78	// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.8.3/pkg/reconcile
79	func (dr *DrainReconcile) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {	1✔
80	ctx = context.WithValue(ctx, constants.LoggerContextKey, log.FromContext(ctx))	1✔
81	reqLogger := log.FromContext(ctx).WithName("Drain Reconcile")	1✔
82		1✔
83	req.Namespace = vars.Namespace	1✔
84		1✔
85	// get node object	1✔
86	node := &corev1.Node{}	1✔
87	found, err := dr.getObject(ctx, req, node)	1✔
88	if err != nil {	2✔
89	reqLogger.Error(err, "failed to get node object")	1✔
90	return ctrl.Result{}, err	1✔
91	}	1✔
92	if !found {	2✔
93	reqLogger.Info("node not found don't, requeue the request")	1✔
94	return ctrl.Result{}, nil	1✔
95	}	1✔
96
97	// get sriovNodeNodeState object
98	nodeNetworkState := &sriovnetworkv1.SriovNetworkNodeState{}	1✔
99	found, err = dr.getObject(ctx, req, nodeNetworkState)	1✔
100	if err != nil {	1✔
101	reqLogger.Error(err, "failed to get sriovNetworkNodeState object")	×
102	return ctrl.Result{}, err	×
103	}	×
104	if !found {	1✔
105	reqLogger.Info("sriovNetworkNodeState not found, don't requeue the request")	×
106	return ctrl.Result{}, nil	×
107	}	×
108
109	// create the drain state annotation if it doesn't exist in the sriovNetworkNodeState object
110	nodeStateDrainAnnotationCurrent, currentNodeStateExist, err := dr.ensureAnnotationExists(ctx, nodeNetworkState, constants.NodeStateDrainAnnotationCurrent)	1✔
111	if err != nil {	1✔
112	reqLogger.Error(err, "failed to ensure nodeStateDrainAnnotationCurrent")	×
113	return ctrl.Result{}, err	×
114	}	×
115	_, desireNodeStateExist, err := dr.ensureAnnotationExists(ctx, nodeNetworkState, constants.NodeStateDrainAnnotation)	1✔
116	if err != nil {	1✔
117	reqLogger.Error(err, "failed to ensure nodeStateDrainAnnotation")	×
118	return ctrl.Result{}, err	×
119	}	×
120
121	// create the drain state annotation if it doesn't exist in the node object
122	nodeDrainAnnotation, nodeExist, err := dr.ensureAnnotationExists(ctx, node, constants.NodeDrainAnnotation)	1✔
123	if err != nil {	1✔
124	reqLogger.Error(err, "failed to ensure nodeStateDrainAnnotation")	×
125	return ctrl.Result{}, err	×
126	}	×
127
128	// requeue the request if we needed to add any of the annotations
129	if !nodeExist \|\| !currentNodeStateExist \|\| !desireNodeStateExist {	2✔
130	return ctrl.Result{Requeue: true}, nil	1✔
131	}	1✔
132	reqLogger.V(2).Info("Drain annotations", "nodeAnnotation", nodeDrainAnnotation, "nodeStateAnnotation", nodeStateDrainAnnotationCurrent)	1✔
133		1✔
134	// Check the node request	1✔
135	if nodeDrainAnnotation == constants.DrainIdle {	2✔
136	// this cover the case the node is on idle	1✔
137		1✔
138	// node request to be on idle and the currect state is idle	1✔
139	// we don't do anything	1✔
140	if nodeStateDrainAnnotationCurrent == constants.DrainIdle {	2✔
141	reqLogger.Info("node and nodeState are on idle nothing todo")	1✔
142	return reconcile.Result{}, nil	1✔
143	}	1✔
144
145	// we have two options here:
146	// 1. node request idle and the current status is drain complete
147	// this means the daemon finish is work, so we need to clean the drain
148	//
149	// 2. the operator is still draining the node but maybe the sriov policy changed and the daemon
150	// doesn't need to drain anymore, so we can stop the drain
151	if nodeStateDrainAnnotationCurrent == constants.DrainComplete \|\|	1✔
152	nodeStateDrainAnnotationCurrent == constants.Draining {	2✔
153	return dr.handleNodeIdleNodeStateDrainingOrCompleted(ctx, node, nodeNetworkState)	1✔
154	}	1✔
155	}
156
157	// this cover the case a node request to drain or reboot
158	if nodeDrainAnnotation == constants.DrainRequired \|\|	1✔
159	nodeDrainAnnotation == constants.RebootRequired {	2✔
160	return dr.handleNodeDrainOrReboot(ctx, node, nodeNetworkState, nodeDrainAnnotation, nodeStateDrainAnnotationCurrent)	1✔
161	}	1✔
162
163	reqLogger.Error(nil, "unexpected node drain annotation")	×
164	return reconcile.Result{}, fmt.Errorf("unexpected node drain annotation")	×
165	}
166
167	func (dr *DrainReconcile) getObject(ctx context.Context, req ctrl.Request, object client.Object) (bool, error) {	1✔
168	err := dr.Get(ctx, req.NamespacedName, object)	1✔
169	if err != nil {	2✔
170	if errors.IsNotFound(err) {	2✔
171	return false, nil	1✔
172	}	1✔
173	return false, err	1✔
174	}
175	return true, nil	1✔
176	}
177
178	func (dr *DrainReconcile) ensureAnnotationExists(ctx context.Context, object client.Object, key string) (string, bool, error) {	1✔
179	value, exist := object.GetAnnotations()[key]	1✔
180	if !exist {	2✔
181	err := utils.AnnotateObject(ctx, object, key, constants.DrainIdle, dr.Client)	1✔
182	if err != nil {	1✔
183	return "", false, err	×
184	}	×
185	return constants.DrainIdle, false, nil	1✔
186	}
187
188	return value, true, nil	1✔
189	}
190
191	// SetupWithManager sets up the controller with the Manager.
192	func (dr *DrainReconcile) SetupWithManager(mgr ctrl.Manager) error {	1✔
193	createUpdateEnqueue := handler.Funcs{	1✔
194	CreateFunc: func(c context.Context, e event.TypedCreateEvent[client.Object], w workqueue.TypedRateLimitingInterface[reconcile.Request]) {	2✔
195	w.Add(reconcile.Request{NamespacedName: types.NamespacedName{	1✔
196	Namespace: vars.Namespace,	1✔
197	Name: e.Object.GetName(),	1✔
198	}})	1✔
199	},	1✔
200	UpdateFunc: func(ctx context.Context, e event.TypedUpdateEvent[client.Object], w workqueue.TypedRateLimitingInterface[reconcile.Request]) {	1✔
201	w.Add(reconcile.Request{NamespacedName: types.NamespacedName{	1✔
202	Namespace: vars.Namespace,	1✔
203	Name: e.ObjectNew.GetName(),	1✔
204	}})	1✔
205	},	1✔
206	}
207
208	// Watch for spec and annotation changes
209	nodePredicates := builder.WithPredicates(DrainAnnotationPredicate{})	1✔
210	nodeStatePredicates := builder.WithPredicates(DrainStateAnnotationPredicate{})	1✔
211		1✔
212	return ctrl.NewControllerManagedBy(mgr).	1✔
213	WithOptions(controller.Options{	1✔
214	MaxConcurrentReconciles: 50,	1✔
215	LogConstructor: func(request *reconcile.Request) logr.Logger {	2✔
216	logger := mgr.GetLogger().WithValues("Function", "Drain")	1✔
217	// Inspired by https://github.com/kubernetes-sigs/controller-runtime/blob/52b17917caa97ec546423867d9637f1787830f3e/pkg/builder/controller.go#L447	1✔
218	if req, ok := any(request).(*reconcile.Request); ok && req != nil {	2✔
219	logger = logger.WithValues("node", request.Name)	1✔
220	}	1✔
221	return logger	1✔
222	},
223	}).
224	For(&corev1.Node{}, nodePredicates).
225	Watches(&sriovnetworkv1.SriovNetworkNodeState{}, createUpdateEnqueue, nodeStatePredicates).
226	Complete(dr)
227	}

k8snetworkplumbingwg / sriov-network-operator / 20490873192

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