3591601521

Committed 01 Dec 2022 10:24AM UTC coverage: 70.418% (+0.06%) from 70.358%

Build Type

push

github

Committed by GitHub

Commit Message

Merge pull request #1612 from alice-zheyan-yu/combine_two_controllers

Run Details

36 of 36 new or added lines in 1 file covered. (100.0%)

7008 of 9952 relevant lines covered (70.42%)

6.87 hits per line

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

40.0

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

        "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/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 nodes *azdiskv1beta2.AzDriverNodeList
        if nodes, err = r.azClient.DiskV1beta2().AzDriverNodes(r.config.ObjectNamespace).List(ctx, metav1.ListOptions{}); err != nil {
                if errors.IsNotFound(err) {
                        return nil
                }
                return err
        }

        for _, node := range nodes.Items {
                updated := node.DeepCopy()
                updated.Annotations = azureutils.AddToMap(updated.Annotations, consts.RecoverAnnotation, "azDriverNode")
                if _, err = r.azClient.DiskV1beta2().AzDriverNodes(r.config.ObjectNamespace).Update(ctx, updated, metav1.UpdateOptions{}); err != nil {
                        return err
                }
                r.addNodeToAvailableAttachmentsMap(ctx, node.Name, node.GetLabels())
        }
        return nil
}

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

kubernetes-sigs / azuredisk-csi-driver / 3591601521

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