3598967339

Committed 02 Dec 2022 04:55AM UTC coverage: 70.077% (+24.9%) from 45.203%

Build Type

push

github

Committed by GitHub

Commit Message

Merge pull request #1615 from hccheng72/fix-orphaned-azdrivernode

Run Details

49 of 49 new or added lines in 2 files covered. (100.0%)

7021 of 10019 relevant lines covered (70.08%)

6.84 hits per line

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

43.08

/pkg/controller/node.go

/*
Copyright 2021 The Kubernetes Authors.

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

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

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

package controller

import (
        "context"
        "fmt"

        "github.com/go-logr/logr"
        corev1 "k8s.io/api/core/v1"
        "k8s.io/apimachinery/pkg/api/errors"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

        azdiskv1beta2 "sigs.k8s.io/azuredisk-csi-driver/pkg/apis/azuredisk/v1beta2"
        consts "sigs.k8s.io/azuredisk-csi-driver/pkg/azureconstants"
        "sigs.k8s.io/azuredisk-csi-driver/pkg/workflow"

        "sigs.k8s.io/azuredisk-csi-driver/pkg/azureutils"
        "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/manager"
        "sigs.k8s.io/controller-runtime/pkg/predicate"
        "sigs.k8s.io/controller-runtime/pkg/reconcile"
        "sigs.k8s.io/controller-runtime/pkg/source"
)

// ReconcileNode reconciles AzDriverNode
type ReconcileNode struct {
        *SharedState
        logger logr.Logger
}

// Implement reconcile.Reconciler so the controller can reconcile objects
var _ reconcile.Reconciler = &ReconcileNode{}

func (r *ReconcileNode) Reconcile(ctx context.Context, request reconcile.Request) (reconcile.Result, error) {
        if !r.isRecoveryComplete() {
                return reconcile.Result{Requeue: true}, nil
        }
        logger := r.logger.WithValues(consts.NodeNameLabel, request.Name)

        n := &corev1.Node{}
        err := r.cachedClient.Get(ctx, request.NamespacedName, n)

        // If the node is not found, delete the corresponding AzDriverNode
        if errors.IsNotFound(err) {
                // Delete the azDriverNode, since corresponding node is deleted
                azN := r.azClient.DiskV1beta2().AzDriverNodes(r.config.ObjectNamespace)
                err = azN.Delete(ctx, request.Name, metav1.DeleteOptions{})

                // If there is an issue in deleting the AzDriverNode, requeue
                if err != nil && !errors.IsNotFound(err) {
                        return reconcile.Result{Requeue: true}, err
                }
                r.deleteNodeFromAvailableAttachmentsMap(ctx, request.Name)

                // Delete all volumeAttachments attached to this node, if failed, requeue
                if _, err = r.cleanUpAzVolumeAttachmentByNode(ctx, request.Name, azdrivernode, azureutils.AllRoles, cleanUpAttachment); err != nil {
                        return reconcile.Result{Requeue: true}, err
                }
                return reconcile.Result{}, nil
        }

        if err != nil {
                return reconcile.Result{Requeue: true}, err
        }

        // If the node has no DeletionTimestamp, it means it's either create event or update event
        if n.ObjectMeta.DeletionTimestamp.IsZero() && !n.Spec.Unschedulable {
                // Add the new schedulable node in availableAttachmentsMap
                r.addNodeToAvailableAttachmentsMap(ctx, n.Name, n.GetLabels())

                // Node is schedulable, proceed to attempt creation of replica attachment
                logger.Info("Node is now available. Will requeue failed replica creation requests.")
                r.tryCreateFailedReplicas(ctx, nodeavailability)
        }

        return reconcile.Result{}, nil
}

// run an update on existing azdrivernode objects to store them under new version if necessary
func (r *ReconcileNode) Recover(ctx context.Context) error {
        var err error
        ctx, w := workflow.New(ctx)
        defer func() { w.Finish(err) }()

        var azNodes *azdiskv1beta2.AzDriverNodeList
        if azNodes, err = r.azClient.DiskV1beta2().AzDriverNodes(r.config.ObjectNamespace).List(ctx, metav1.ListOptions{}); err != nil {
                return err
        }

        errCount := 0
        for _, azNode := range azNodes.Items {
                // if the corresponding node has been deleted, delete the azdrivernode object
                _, err = r.kubeClient.CoreV1().Nodes().Get(ctx, azNode.Spec.NodeName, metav1.GetOptions{})
                if err != nil {
                        if errors.IsNotFound(err) {
                                n := r.azClient.DiskV1beta2().AzDriverNodes(r.config.ObjectNamespace)
                                if err = n.Delete(ctx, azNode.Name, metav1.DeleteOptions{}); err != nil {
                                        w.Logger().Errorf(err, "failed to delete azDriverNode (%s)", azNode.Name)
                                        errCount++
                                }
                        } else {
                                w.Logger().Errorf(err, "failed to find node (%s)", azNode.Spec.NodeName)
                                errCount++
                        }
                } else {
                        updateFunc := func(obj client.Object) error {
                                azNode := obj.(*azdiskv1beta2.AzDriverNode)
                                azNode.Annotations = azureutils.AddToMap(azNode.Annotations, consts.RecoverAnnotation, "azDriverNode")
                                return nil
                        }
                        if _, err = azureutils.UpdateCRIWithRetry(ctx, nil, r.cachedClient, r.azClient, &azNode, updateFunc, consts.NormalUpdateMaxNetRetry, azureutils.UpdateCRI); err != nil {
                                w.Logger().Errorf(err, "failed to recover azDriverNode (%s) with annotation", azNode.Name)
                                errCount++
                                continue
                        }
                        r.addNodeToAvailableAttachmentsMap(ctx, azNode.Name, azNode.GetLabels())
                }
        }

        if errCount > 0 {
                err = fmt.Errorf("failed to recover all azDriverNodes")
        }
        return err
}

// NewNodeController initializes node-controller
func NewNodeController(mgr manager.Manager, controllerSharedState *SharedState) (*ReconcileNode, error) {
        logger := mgr.GetLogger().WithValues("controller", "node")
        reconciler := ReconcileNode{
                SharedState: controllerSharedState,
                logger:      logger,
        }

        c, err := controller.New("node-controller", mgr, controller.Options{
                MaxConcurrentReconciles: consts.DefaultWorkerThreads,
                Reconciler:              &reconciler,
                LogConstructor:          func(req *reconcile.Request) logr.Logger { return logger },
        })

        if err != nil {
                logger.Error(err, "failed to create controller")
                return nil, err
        }

        // Predicate to reconcile created, new schedulable and deleted nodes
        p := predicate.Funcs{
                CreateFunc: func(e event.CreateEvent) bool {
                        return true
                },
                UpdateFunc: func(e event.UpdateEvent) bool {
                        // make sure only update event from node taint changed from "unschedulable" gets enqueued to reconciler queue
                        old, oldOk := e.ObjectOld.(*corev1.Node)
                        new, newOk := e.ObjectNew.(*corev1.Node)

                        wasUnschedulable := false
                        nowSchedulable := true
                        for _, taint := range old.Spec.Taints {
                                if taint.Key == "node.kubernetes.io/unschedulable" {
                                        wasUnschedulable = true
                                }
                        }
                        for _, taint := range new.Spec.Taints {
                                if taint.Key == "node.kubernetes.io/unschedulable" {
                                        nowSchedulable = false
                                }
                        }

                        if oldOk && newOk && wasUnschedulable && nowSchedulable {
                                return true
                        }
                        return false
                },
                GenericFunc: func(e event.GenericEvent) bool {
                        return false
                },
                DeleteFunc: func(e event.DeleteEvent) bool {
                        return true
                },
        }

        logger.V(2).Info("Starting to watch cluster nodes.")
        // Watch the nodes
        err = c.Watch(&source.Kind{Type: &corev1.Node{}}, &handler.EnqueueRequestForObject{}, p)
        if err != nil {
                logger.Error(err, "failed to initialize watch for Node")
                return nil, err
        }
        logger.V(2).Info("Controller set-up successful.")

        return &reconciler, err
}

1	/*
2	Copyright 2021 The Kubernetes Authors.
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 controller
18
19	import (
20	"context"
21	"fmt"
22
23	"github.com/go-logr/logr"
24	corev1 "k8s.io/api/core/v1"
25	"k8s.io/apimachinery/pkg/api/errors"
26	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
27
28	azdiskv1beta2 "sigs.k8s.io/azuredisk-csi-driver/pkg/apis/azuredisk/v1beta2"
29	consts "sigs.k8s.io/azuredisk-csi-driver/pkg/azureconstants"
30	"sigs.k8s.io/azuredisk-csi-driver/pkg/workflow"
31
32	"sigs.k8s.io/azuredisk-csi-driver/pkg/azureutils"
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/manager"
38	"sigs.k8s.io/controller-runtime/pkg/predicate"
39	"sigs.k8s.io/controller-runtime/pkg/reconcile"
40	"sigs.k8s.io/controller-runtime/pkg/source"
41	)
42
43	// ReconcileNode reconciles AzDriverNode
44	type ReconcileNode struct {
45	*SharedState
46	logger logr.Logger
47	}
48
49	// Implement reconcile.Reconciler so the controller can reconcile objects
50	var _ reconcile.Reconciler = &ReconcileNode{}
51
52	func (r *ReconcileNode) Reconcile(ctx context.Context, request reconcile.Request) (reconcile.Result, error) {	6✔
53	if !r.isRecoveryComplete() {	6✔
54	return reconcile.Result{Requeue: true}, nil	×
55	}	×
56	logger := r.logger.WithValues(consts.NodeNameLabel, request.Name)	6✔
57		6✔
58	n := &corev1.Node{}	6✔
59	err := r.cachedClient.Get(ctx, request.NamespacedName, n)	6✔
60		6✔
61	// If the node is not found, delete the corresponding AzDriverNode	6✔
62	if errors.IsNotFound(err) {	8✔
63	// Delete the azDriverNode, since corresponding node is deleted	2✔
64	azN := r.azClient.DiskV1beta2().AzDriverNodes(r.config.ObjectNamespace)	2✔
65	err = azN.Delete(ctx, request.Name, metav1.DeleteOptions{})	2✔
66		2✔
67	// If there is an issue in deleting the AzDriverNode, requeue	2✔
68	if err != nil && !errors.IsNotFound(err) {	2✔
69	return reconcile.Result{Requeue: true}, err	×
70	}	×
71	r.deleteNodeFromAvailableAttachmentsMap(ctx, request.Name)	2✔
72		2✔
73	// Delete all volumeAttachments attached to this node, if failed, requeue	2✔
74	if _, err = r.cleanUpAzVolumeAttachmentByNode(ctx, request.Name, azdrivernode, azureutils.AllRoles, cleanUpAttachment); err != nil {	2✔
75	return reconcile.Result{Requeue: true}, err	×
76	}	×
77	return reconcile.Result{}, nil	2✔
78	}
79
80	if err != nil {	5✔
81	return reconcile.Result{Requeue: true}, err	1✔
82	}	1✔
83
84	// If the node has no DeletionTimestamp, it means it's either create event or update event
85	if n.ObjectMeta.DeletionTimestamp.IsZero() && !n.Spec.Unschedulable {	6✔
86	// Add the new schedulable node in availableAttachmentsMap	3✔
87	r.addNodeToAvailableAttachmentsMap(ctx, n.Name, n.GetLabels())	3✔
88		3✔
89	// Node is schedulable, proceed to attempt creation of replica attachment	3✔
90	logger.Info("Node is now available. Will requeue failed replica creation requests.")	3✔
91	r.tryCreateFailedReplicas(ctx, nodeavailability)	3✔
92	}	3✔
93
94	return reconcile.Result{}, nil	3✔
95	}
96
97	// run an update on existing azdrivernode objects to store them under new version if necessary
98	func (r *ReconcileNode) Recover(ctx context.Context) error {	2✔
99	var err error	2✔
100	ctx, w := workflow.New(ctx)	2✔
101	defer func() { w.Finish(err) }()	4✔
102
103	var azNodes *azdiskv1beta2.AzDriverNodeList	2✔
104	if azNodes, err = r.azClient.DiskV1beta2().AzDriverNodes(r.config.ObjectNamespace).List(ctx, metav1.ListOptions{}); err != nil {	2✔
105	return err	×
106	}	×
107
108	errCount := 0	2✔
109	for _, azNode := range azNodes.Items {	5✔
110	// if the corresponding node has been deleted, delete the azdrivernode object	3✔
111	_, err = r.kubeClient.CoreV1().Nodes().Get(ctx, azNode.Spec.NodeName, metav1.GetOptions{})	3✔
112	if err != nil {	4✔
113	if errors.IsNotFound(err) {	2✔
114	n := r.azClient.DiskV1beta2().AzDriverNodes(r.config.ObjectNamespace)	1✔
115	if err = n.Delete(ctx, azNode.Name, metav1.DeleteOptions{}); err != nil {	1✔
116	w.Logger().Errorf(err, "failed to delete azDriverNode (%s)", azNode.Name)	×
117	errCount++	×
118	}	×
119	} else {	×
120	w.Logger().Errorf(err, "failed to find node (%s)", azNode.Spec.NodeName)	×
121	errCount++	×
122	}	×
123	} else {	2✔
124	updateFunc := func(obj client.Object) error {	4✔
125	azNode := obj.(*azdiskv1beta2.AzDriverNode)	2✔
126	azNode.Annotations = azureutils.AddToMap(azNode.Annotations, consts.RecoverAnnotation, "azDriverNode")	2✔
127	return nil	2✔
128	}	2✔
129	if _, err = azureutils.UpdateCRIWithRetry(ctx, nil, r.cachedClient, r.azClient, &azNode, updateFunc, consts.NormalUpdateMaxNetRetry, azureutils.UpdateCRI); err != nil {	2✔
130	w.Logger().Errorf(err, "failed to recover azDriverNode (%s) with annotation", azNode.Name)	×
131	errCount++	×
132	continue	×
133	}
134	r.addNodeToAvailableAttachmentsMap(ctx, azNode.Name, azNode.GetLabels())	2✔
135	}
136	}
137
138	if errCount > 0 {	2✔
139	err = fmt.Errorf("failed to recover all azDriverNodes")	×
140	}	×
141	return err	2✔
142	}
143
144	// NewNodeController initializes node-controller
145	func NewNodeController(mgr manager.Manager, controllerSharedState SharedState) (ReconcileNode, error) {	×
146	logger := mgr.GetLogger().WithValues("controller", "node")	×
147	reconciler := ReconcileNode{	×
148	SharedState: controllerSharedState,	×
149	logger: logger,	×
150	}	×
151		×
152	c, err := controller.New("node-controller", mgr, controller.Options{	×
153	MaxConcurrentReconciles: consts.DefaultWorkerThreads,	×
154	Reconciler: &reconciler,	×
155	LogConstructor: func(req *reconcile.Request) logr.Logger { return logger },	×
156	})
157
158	if err != nil {	×
159	logger.Error(err, "failed to create controller")	×
160	return nil, err	×
161	}	×
162
163	// Predicate to reconcile created, new schedulable and deleted nodes
164	p := predicate.Funcs{	×
165	CreateFunc: func(e event.CreateEvent) bool {	×
166	return true	×
167	},	×
168	UpdateFunc: func(e event.UpdateEvent) bool {	×
169	// make sure only update event from node taint changed from "unschedulable" gets enqueued to reconciler queue	×
170	old, oldOk := e.ObjectOld.(*corev1.Node)	×
171	new, newOk := e.ObjectNew.(*corev1.Node)	×
172		×
173	wasUnschedulable := false	×
174	nowSchedulable := true	×
175	for _, taint := range old.Spec.Taints {	×
176	if taint.Key == "node.kubernetes.io/unschedulable" {	×
177	wasUnschedulable = true	×
178	}	×
179	}
180	for _, taint := range new.Spec.Taints {	×
181	if taint.Key == "node.kubernetes.io/unschedulable" {	×
182	nowSchedulable = false	×
183	}	×
184	}
185
186	if oldOk && newOk && wasUnschedulable && nowSchedulable {	×
187	return true	×
188	}	×
189	return false	×
190	},
191	GenericFunc: func(e event.GenericEvent) bool {	×
192	return false	×
193	},	×
194	DeleteFunc: func(e event.DeleteEvent) bool {	×
195	return true	×
196	},	×
197	}
198
199	logger.V(2).Info("Starting to watch cluster nodes.")	×
200	// Watch the nodes	×
201	err = c.Watch(&source.Kind{Type: &corev1.Node{}}, &handler.EnqueueRequestForObject{}, p)	×
202	if err != nil {	×
203	logger.Error(err, "failed to initialize watch for Node")	×
204	return nil, err	×
205	}	×
206	logger.V(2).Info("Controller set-up successful.")	×
207		×
208	return &reconciler, err	×
209	}

kubernetes-sigs / azuredisk-csi-driver / 3598967339

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