4369939980

Committed 09 Mar 2023 01:00AM UTC coverage: 69.522% (+25.0%) from 44.538%

Build Type

push

github

Committed by GitHub

Commit Message

Merge pull request #1753 from alice-zheyan-yu/delete_duplicated_import_package_api_v1

Run Details

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

7142 of 10273 relevant lines covered (69.52%)

6.72 hits per line

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

42.11

/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, deleteOnly); 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() {
                // Add the new schedulable node in availableAttachmentsMap
                if ok := r.addNodeToAvailableAttachmentsMap(ctx, n.Name, n.GetLabels()); ok && !n.Spec.Unschedulable {
                        // 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, recoveryID string) 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
                var node *corev1.Node
                node, 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, recoveryID)
                                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, node.Name, node.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 {

                        old, oldOk := e.ObjectOld.(*corev1.Node)
                        new, newOk := e.ObjectNew.(*corev1.Node)

                        if oldOk && newOk {
                                // update event from node taint changed from "unschedulable" gets enqueued to reconciler queue
                                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
                                        }
                                }

                                // update event when LabelInstanceTypeStable is added to node
                                _, oldLableOk := old.GetLabels()[corev1.LabelInstanceTypeStable]
                                _, newLabelOk := new.GetLabels()[corev1.LabelInstanceTypeStable]

                                return (wasUnschedulable && nowSchedulable) || (!oldLableOk && newLabelOk)
                        }
                        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, deleteOnly); 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() {	6✔
86	// Add the new schedulable node in availableAttachmentsMap	3✔
87	if ok := r.addNodeToAvailableAttachmentsMap(ctx, n.Name, n.GetLabels()); ok && !n.Spec.Unschedulable {	5✔
88	// Node is schedulable, proceed to attempt creation of replica attachment	2✔
89	logger.Info("Node is now available. Will requeue failed replica creation requests.")	2✔
90	r.tryCreateFailedReplicas(ctx, nodeavailability)	2✔
91	}	2✔
92	}
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, recoveryID string) 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	var node *corev1.Node	3✔
112	node, err = r.kubeClient.CoreV1().Nodes().Get(ctx, azNode.Spec.NodeName, metav1.GetOptions{})	3✔
113	if err != nil {	4✔
114	if errors.IsNotFound(err) {	2✔
115	n := r.azClient.DiskV1beta2().AzDriverNodes(r.config.ObjectNamespace)	1✔
116	if err = n.Delete(ctx, azNode.Name, metav1.DeleteOptions{}); err != nil {	1✔
117	w.Logger().Errorf(err, "failed to delete azDriverNode (%s)", azNode.Name)	×
118	errCount++	×
119	}	×
120	} else {	×
121	w.Logger().Errorf(err, "failed to find node (%s)", azNode.Spec.NodeName)	×
122	errCount++	×
123	}	×
124	} else {	2✔
125	updateFunc := func(obj client.Object) error {	4✔
126	azNode := obj.(*azdiskv1beta2.AzDriverNode)	2✔
127	azNode.Annotations = azureutils.AddToMap(azNode.Annotations, consts.RecoverAnnotation, recoveryID)	2✔
128	return nil	2✔
129	}	2✔
130	if _, err = azureutils.UpdateCRIWithRetry(ctx, nil, r.cachedClient, r.azClient, &azNode, updateFunc, consts.NormalUpdateMaxNetRetry, azureutils.UpdateCRI); err != nil {	2✔
131	w.Logger().Errorf(err, "failed to recover azDriverNode (%s) with annotation", azNode.Name)	×
132	errCount++	×
133	continue	×
134	}
135	r.addNodeToAvailableAttachmentsMap(ctx, node.Name, node.GetLabels())	2✔
136	}
137	}
138
139	if errCount > 0 {	2✔
140	err = fmt.Errorf("failed to recover all azDriverNodes")	×
141	}	×
142	return err	2✔
143	}
144
145	// NewNodeController initializes node-controller
146	func NewNodeController(mgr manager.Manager, controllerSharedState SharedState) (ReconcileNode, error) {	×
147	logger := mgr.GetLogger().WithValues("controller", "node")	×
148	reconciler := ReconcileNode{	×
149	SharedState: controllerSharedState,	×
150	logger: logger,	×
151	}	×
152		×
153	c, err := controller.New("node-controller", mgr, controller.Options{	×
154	MaxConcurrentReconciles: consts.DefaultWorkerThreads,	×
155	Reconciler: &reconciler,	×
156	LogConstructor: func(req *reconcile.Request) logr.Logger { return logger },	×
157	})
158
159	if err != nil {	×
160	logger.Error(err, "failed to create controller")	×
161	return nil, err	×
162	}	×
163
164	// Predicate to reconcile created, new schedulable and deleted nodes
165	p := predicate.Funcs{	×
166	CreateFunc: func(e event.CreateEvent) bool {	×
167	return true	×
168	},	×
169	UpdateFunc: func(e event.UpdateEvent) bool {	×
170		×
171	old, oldOk := e.ObjectOld.(*corev1.Node)	×
172	new, newOk := e.ObjectNew.(*corev1.Node)	×
173		×
174	if oldOk && newOk {	×
175	// update event from node taint changed from "unschedulable" gets enqueued to reconciler queue	×
176	wasUnschedulable := false	×
177	nowSchedulable := true	×
178	for _, taint := range old.Spec.Taints {	×
179	if taint.Key == "node.kubernetes.io/unschedulable" {	×
180	wasUnschedulable = true	×
181	}	×
182	}
183	for _, taint := range new.Spec.Taints {	×
184	if taint.Key == "node.kubernetes.io/unschedulable" {	×
185	nowSchedulable = false	×
186	}	×
187	}
188
189	// update event when LabelInstanceTypeStable is added to node
190	_, oldLableOk := old.GetLabels()[corev1.LabelInstanceTypeStable]	×
191	_, newLabelOk := new.GetLabels()[corev1.LabelInstanceTypeStable]	×
192		×
193	return (wasUnschedulable && nowSchedulable) \|\| (!oldLableOk && newLabelOk)	×
194	}
195	return false	×
196	},
197	GenericFunc: func(e event.GenericEvent) bool {	×
198	return false	×
199	},	×
200	DeleteFunc: func(e event.DeleteEvent) bool {	×
201	return true	×
202	},	×
203	}
204
205	logger.V(2).Info("Starting to watch cluster nodes.")	×
206	// Watch the nodes	×
207	err = c.Watch(&source.Kind{Type: &corev1.Node{}}, &handler.EnqueueRequestForObject{}, p)	×
208	if err != nil {	×
209	logger.Error(err, "failed to initialize watch for Node")	×
210	return nil, err	×
211	}	×
212	logger.V(2).Info("Controller set-up successful.")	×
213		×
214	return &reconciler, err	×
215	}

kubernetes-sigs / azuredisk-csi-driver / 4369939980

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