13526205039

Committed 25 Feb 2025 04:34PM UTC coverage: 48.882% (+0.9%) from 48.008%

Build # 13526205039

Build Type

Pull #788

github

Committed by

web-flow

Commit Message

Merge 3e6b91cf3 into d7c9458e0

Pull Request Pull Request #788: Daemon redesign - using controller-runtime

Run Details

273 of 615 new or added lines in 18 files covered. (44.39%)

199 existing lines in 12 files now uncovered.

7324 of 14983 relevant lines covered (48.88%)

0.54 hits per line

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

84.62

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

        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/platforms"
        "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, platformHelper platforms.Interface) (*DrainReconcile, error) {
        drainer, err := drain.NewDrainer(platformHelper)
        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) {
        reqLogger := log.FromContext(ctx)
        reqLogger.Info("Reconciling Drain")

        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, &reqLogger, node, nodeNetworkState)
                }
        }

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

k8snetworkplumbingwg / sriov-network-operator / 13526205039

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