11229466077

Committed 08 Oct 2024 05:59AM UTC coverage: 45.063% (-0.1%) from 45.177%

Build # 11229466077

Build Type

Pull #666

github

Committed by

web-flow

Commit Message

Merge 60432e00c into aecf4730f

Pull Request Pull Request #666: Implement RDMA subsystem mode change

Run Details

84 of 189 new or added lines in 11 files covered. (44.44%)

2 existing lines in 1 file now uncovered.

6700 of 14868 relevant lines covered (45.06%)

0.5 hits per line

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

69.05

/controllers/helper.go

/*


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 (
        "bytes"
        "context"
        "encoding/json"
        "fmt"
        "os"
        "sort"
        "strings"

        errs "github.com/pkg/errors"
        appsv1 "k8s.io/api/apps/v1"
        corev1 "k8s.io/api/core/v1"
        "k8s.io/apimachinery/pkg/api/equality"
        "k8s.io/apimachinery/pkg/api/errors"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        uns "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
        "k8s.io/apimachinery/pkg/labels"
        "k8s.io/apimachinery/pkg/runtime"
        "k8s.io/apimachinery/pkg/types"
        "k8s.io/apimachinery/pkg/util/intstr"
        kscheme "k8s.io/client-go/kubernetes/scheme"
        k8sclient "sigs.k8s.io/controller-runtime/pkg/client"
        "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
        "sigs.k8s.io/controller-runtime/pkg/event"
        "sigs.k8s.io/controller-runtime/pkg/log"
        "sigs.k8s.io/controller-runtime/pkg/predicate"

        sriovnetworkv1 "github.com/k8snetworkplumbingwg/sriov-network-operator/api/v1"
        "github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/apply"
        constants "github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/consts"
        "github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/render"
        "github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/vars"
)

var (
        webhooks = map[string](string){
                constants.InjectorWebHookName: constants.InjectorWebHookPath,
                constants.OperatorWebHookName: constants.OperatorWebHookPath,
        }
        oneNode           = intstr.FromInt32(1)
        defaultPoolConfig = &sriovnetworkv1.SriovNetworkPoolConfig{Spec: sriovnetworkv1.SriovNetworkPoolConfigSpec{
                MaxUnavailable: &oneNode,
                NodeSelector:   &metav1.LabelSelector{},
                RdmaMode:       ""}}
)

const (
        clusterRoleResourceName               = "ClusterRole"
        clusterRoleBindingResourceName        = "ClusterRoleBinding"
        mutatingWebhookConfigurationCRDName   = "MutatingWebhookConfiguration"
        validatingWebhookConfigurationCRDName = "ValidatingWebhookConfiguration"
        machineConfigCRDName                  = "MachineConfig"
        trueString                            = "true"
)

type DrainAnnotationPredicate struct {
        predicate.Funcs
}

func (DrainAnnotationPredicate) Create(e event.CreateEvent) bool {
        if e.Object == nil {
                return false
        }

        if _, hasAnno := e.Object.GetAnnotations()[constants.NodeDrainAnnotation]; hasAnno {
                return true
        }
        return false
}

func (DrainAnnotationPredicate) Update(e event.UpdateEvent) bool {
        if e.ObjectOld == nil {
                return false
        }
        if e.ObjectNew == nil {
                return false
        }

        oldAnno, hasOldAnno := e.ObjectOld.GetAnnotations()[constants.NodeDrainAnnotation]
        newAnno, hasNewAnno := e.ObjectNew.GetAnnotations()[constants.NodeDrainAnnotation]

        if !hasOldAnno && hasNewAnno {
                return true
        }

        if oldAnno != newAnno {
                return true
        }

        return false
}

type DrainStateAnnotationPredicate struct {
        predicate.Funcs
}

func (DrainStateAnnotationPredicate) Create(e event.CreateEvent) bool {
        if e.Object == nil {
                return false
        }

        if _, hasAnno := e.Object.GetLabels()[constants.NodeStateDrainAnnotationCurrent]; hasAnno {
                return true
        }
        return false
}

func (DrainStateAnnotationPredicate) Update(e event.UpdateEvent) bool {
        if e.ObjectOld == nil {
                return false
        }
        if e.ObjectNew == nil {
                return false
        }

        oldAnno, hasOldAnno := e.ObjectOld.GetLabels()[constants.NodeStateDrainAnnotationCurrent]
        newAnno, hasNewAnno := e.ObjectNew.GetLabels()[constants.NodeStateDrainAnnotationCurrent]

        if !hasOldAnno || !hasNewAnno {
                return true
        }

        if oldAnno != newAnno {
                return true
        }

        return oldAnno != newAnno
}

func GetImagePullSecrets() []string {
        imagePullSecrets := os.Getenv("IMAGE_PULL_SECRETS")
        if imagePullSecrets != "" {
                return strings.Split(imagePullSecrets, ",")
        } else {
                return []string{}
        }
}

func formatJSON(str string) (string, error) {
        var prettyJSON bytes.Buffer
        if err := json.Indent(&prettyJSON, []byte(str), "", "    "); err != nil {
                return "", err
        }
        return prettyJSON.String(), nil
}

func GetDefaultNodeSelector() map[string]string {
        return map[string]string{"node-role.kubernetes.io/worker": "",
                "kubernetes.io/os": "linux"}
}

// hasNoValidPolicy returns true if no SriovNetworkNodePolicy
// or only the (deprecated) "default" policy is present
func hasNoValidPolicy(pl []sriovnetworkv1.SriovNetworkNodePolicy) bool {
        switch len(pl) {
        case 0:
                return true
        case 1:
                return pl[0].Name == constants.DefaultPolicyName
        default:
                return false
        }
}

func syncPluginDaemonObjs(ctx context.Context,
        client k8sclient.Client,
        scheme *runtime.Scheme,
        dc *sriovnetworkv1.SriovOperatorConfig,
        pl *sriovnetworkv1.SriovNetworkNodePolicyList) error {
        logger := log.Log.WithName("syncPluginDaemonObjs")
        logger.V(1).Info("Start to sync sriov daemons objects")

        // render plugin manifests
        data := render.MakeRenderData()
        data.Data["Namespace"] = vars.Namespace
        data.Data["SRIOVDevicePluginImage"] = os.Getenv("SRIOV_DEVICE_PLUGIN_IMAGE")
        data.Data["ReleaseVersion"] = os.Getenv("RELEASEVERSION")
        data.Data["ResourcePrefix"] = vars.ResourcePrefix
        data.Data["ImagePullSecrets"] = GetImagePullSecrets()
        data.Data["NodeSelectorField"] = GetDefaultNodeSelector()
        data.Data["UseCDI"] = dc.Spec.UseCDI
        objs, err := renderDsForCR(constants.PluginPath, &data)
        if err != nil {
                logger.Error(err, "Fail to render SR-IoV manifests")
                return err
        }

        if hasNoValidPolicy(pl.Items) {
                for _, obj := range objs {
                        err := deleteK8sResource(ctx, client, obj)
                        if err != nil {
                                return err
                        }
                }
                return nil
        }

        // Sync DaemonSets
        for _, obj := range objs {
                if obj.GetKind() == constants.DaemonSet && len(dc.Spec.ConfigDaemonNodeSelector) > 0 {
                        scheme := kscheme.Scheme
                        ds := &appsv1.DaemonSet{}
                        err = scheme.Convert(obj, ds, nil)
                        if err != nil {
                                logger.Error(err, "Fail to convert to DaemonSet")
                                return err
                        }
                        ds.Spec.Template.Spec.NodeSelector = dc.Spec.ConfigDaemonNodeSelector
                        err = scheme.Convert(ds, obj, nil)
                        if err != nil {
                                logger.Error(err, "Fail to convert to Unstructured")
                                return err
                        }
                }
                err = syncDsObject(ctx, client, scheme, dc, pl, obj)
                if err != nil {
                        logger.Error(err, "Couldn't sync SR-IoV daemons objects")
                        return err
                }
        }

        return nil
}

func deleteK8sResource(ctx context.Context, client k8sclient.Client, in *uns.Unstructured) error {
        if err := apply.DeleteObject(ctx, client, in); err != nil {
                return fmt.Errorf("failed to delete object %v with err: %v", in, err)
        }
        return nil
}

func syncDsObject(ctx context.Context, client k8sclient.Client, scheme *runtime.Scheme, dc *sriovnetworkv1.SriovOperatorConfig, pl *sriovnetworkv1.SriovNetworkNodePolicyList, obj *uns.Unstructured) error {
        logger := log.Log.WithName("syncDsObject")
        kind := obj.GetKind()
        logger.V(1).Info("Start to sync Objects", "Kind", kind)
        switch kind {
        case constants.ServiceAccount, constants.Role, constants.RoleBinding:
                if err := controllerutil.SetControllerReference(dc, obj, scheme); err != nil {
                        return err
                }
                if err := apply.ApplyObject(ctx, client, obj); err != nil {
                        logger.Error(err, "Fail to sync", "Kind", kind)
                        return err
                }
        case constants.DaemonSet:
                ds := &appsv1.DaemonSet{}
                err := scheme.Convert(obj, ds, nil)
                if err != nil {
                        logger.Error(err, "Fail to convert to DaemonSet")
                        return err
                }
                err = syncDaemonSet(ctx, client, scheme, dc, pl, ds)
                if err != nil {
                        logger.Error(err, "Fail to sync DaemonSet", "Namespace", ds.Namespace, "Name", ds.Name)
                        return err
                }
        }
        return nil
}

func setDsNodeAffinity(pl *sriovnetworkv1.SriovNetworkNodePolicyList, ds *appsv1.DaemonSet) error {
        terms := nodeSelectorTermsForPolicyList(pl.Items)
        if len(terms) > 0 {
                ds.Spec.Template.Spec.Affinity = &corev1.Affinity{
                        NodeAffinity: &corev1.NodeAffinity{
                                RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{
                                        NodeSelectorTerms: terms,
                                },
                        },
                }
        }
        return nil
}

func nodeSelectorTermsForPolicyList(policies []sriovnetworkv1.SriovNetworkNodePolicy) []corev1.NodeSelectorTerm {
        terms := []corev1.NodeSelectorTerm{}
        for _, p := range policies {
                // Note(adrianc): default policy is deprecated and ignored.
                if p.Name == constants.DefaultPolicyName {
                        continue
                }

                if len(p.Spec.NodeSelector) == 0 {
                        continue
                }
                expressions := []corev1.NodeSelectorRequirement{}
                for k, v := range p.Spec.NodeSelector {
                        exp := corev1.NodeSelectorRequirement{
                                Operator: corev1.NodeSelectorOpIn,
                                Key:      k,
                                Values:   []string{v},
                        }
                        expressions = append(expressions, exp)
                }
                // sorting is needed to keep the daemon spec stable.
                // the items are popped in a random order from the map
                sort.Slice(expressions, func(i, j int) bool {
                        return expressions[i].Key < expressions[j].Key
                })
                nodeSelector := corev1.NodeSelectorTerm{
                        MatchExpressions: expressions,
                }
                terms = append(terms, nodeSelector)
        }

        return terms
}

// renderDsForCR returns a busybox pod with the same name/namespace as the cr
func renderDsForCR(path string, data *render.RenderData) ([]*uns.Unstructured, error) {
        logger := log.Log.WithName("renderDsForCR")
        logger.V(1).Info("Start to render objects")

        objs, err := render.RenderDir(path, data)
        if err != nil {
                return nil, errs.Wrap(err, "failed to render SR-IOV Network Operator manifests")
        }
        return objs, nil
}

func syncDaemonSet(ctx context.Context, client k8sclient.Client, scheme *runtime.Scheme, dc *sriovnetworkv1.SriovOperatorConfig, pl *sriovnetworkv1.SriovNetworkNodePolicyList, in *appsv1.DaemonSet) error {
        logger := log.Log.WithName("syncDaemonSet")
        logger.V(1).Info("Start to sync DaemonSet", "Namespace", in.Namespace, "Name", in.Name)
        var err error

        if pl != nil {
                if err = setDsNodeAffinity(pl, in); err != nil {
                        return err
                }
        }
        if err = controllerutil.SetControllerReference(dc, in, scheme); err != nil {
                return err
        }
        ds := &appsv1.DaemonSet{}
        err = client.Get(ctx, types.NamespacedName{Namespace: in.Namespace, Name: in.Name}, ds)
        if err != nil {
                if errors.IsNotFound(err) {
                        logger.V(1).Info("Created DaemonSet", in.Namespace, in.Name)
                        err = client.Create(ctx, in)
                        if err != nil {
                                logger.Error(err, "Fail to create Daemonset", "Namespace", in.Namespace, "Name", in.Name)
                                return err
                        }
                } else {
                        logger.Error(err, "Fail to get Daemonset", "Namespace", in.Namespace, "Name", in.Name)
                        return err
                }
        } else {
                logger.V(1).Info("DaemonSet already exists, updating")
                // DeepDerivative checks for changes only comparing non-zero fields in the source struct.
                // This skips default values added by the api server.
                // References in https://github.com/kubernetes-sigs/kubebuilder/issues/592#issuecomment-625738183

                // Note(Adrianc): we check Equality of OwnerReference as we changed sriov-device-plugin owner ref
                // from SriovNetworkNodePolicy to SriovOperatorConfig, hence even if there is no change in spec,
                // we need to update the obj's owner reference.

                if equality.Semantic.DeepEqual(in.OwnerReferences, ds.OwnerReferences) &&
                        equality.Semantic.DeepDerivative(in.Spec, ds.Spec) {
                        logger.V(1).Info("Daemonset spec did not change, not updating")
                        return nil
                }
                err = client.Update(ctx, in)
                if err != nil {
                        logger.Error(err, "Fail to update DaemonSet", "Namespace", in.Namespace, "Name", in.Name)
                        return err
                }
        }
        return nil
}

func updateDaemonsetNodeSelector(obj *uns.Unstructured, nodeSelector map[string]string) error {
        if len(nodeSelector) == 0 {
                return nil
        }

        ds := &appsv1.DaemonSet{}
        scheme := kscheme.Scheme
        err := scheme.Convert(obj, ds, nil)
        if err != nil {
                return fmt.Errorf("failed to convert Unstructured [%s] to DaemonSet: %v", obj.GetName(), err)
        }

        ds.Spec.Template.Spec.NodeSelector = nodeSelector

        err = scheme.Convert(ds, obj, nil)
        if err != nil {
                return fmt.Errorf("failed to convert DaemonSet [%s] to Unstructured: %v", obj.GetName(), err)
        }
        return nil
}

func findNodePoolConfig(ctx context.Context, node *corev1.Node, c k8sclient.Client) (*sriovnetworkv1.SriovNetworkPoolConfig, []corev1.Node, error) {
        logger := log.FromContext(ctx)
        logger.Info("FindNodePoolConfig():")
        // get all the sriov network pool configs
        npcl := &sriovnetworkv1.SriovNetworkPoolConfigList{}
        err := c.List(ctx, npcl)
        if err != nil {
                logger.Error(err, "failed to list sriovNetworkPoolConfig")
                return nil, nil, err
        }

        selectedNpcl := []*sriovnetworkv1.SriovNetworkPoolConfig{}
        nodesInPools := map[string]interface{}{}

        for _, npc := range npcl.Items {
                // we skip hw offload objects
                if npc.Spec.OvsHardwareOffloadConfig.Name != "" {
                        continue
                }

                if npc.Spec.NodeSelector == nil {
                        npc.Spec.NodeSelector = &metav1.LabelSelector{}
                }

                selector, err := metav1.LabelSelectorAsSelector(npc.Spec.NodeSelector)
                if err != nil {
                        logger.Error(err, "failed to create label selector from nodeSelector", "nodeSelector", npc.Spec.NodeSelector)
                        return nil, nil, err
                }

                if selector.Matches(labels.Set(node.Labels)) {
                        selectedNpcl = append(selectedNpcl, npc.DeepCopy())
                }

                nodeList := &corev1.NodeList{}
                err = c.List(ctx, nodeList, &k8sclient.ListOptions{LabelSelector: selector})
                if err != nil {
                        logger.Error(err, "failed to list all the nodes matching the pool with label selector from nodeSelector",
                                "machineConfigPoolName", npc,
                                "nodeSelector", npc.Spec.NodeSelector)
                        return nil, nil, err
                }

                for _, nodeName := range nodeList.Items {
                        nodesInPools[nodeName.Name] = nil
                }
        }

        if len(selectedNpcl) > 1 {
                // don't allow the node to be part of multiple pools
                err = fmt.Errorf("node is part of more then one pool")
                logger.Error(err, "multiple pools founded for a specific node", "numberOfPools", len(selectedNpcl), "pools", selectedNpcl)
                return nil, nil, err
        } else if len(selectedNpcl) == 1 {
                // found one pool for our node
                logger.V(2).Info("found sriovNetworkPool", "pool", *selectedNpcl[0])
                selector, err := metav1.LabelSelectorAsSelector(selectedNpcl[0].Spec.NodeSelector)
                if err != nil {
                        logger.Error(err, "failed to create label selector from nodeSelector", "nodeSelector", selectedNpcl[0].Spec.NodeSelector)
                        return nil, nil, err
                }

                // list all the nodes that are also part of this pool and return them
                nodeList := &corev1.NodeList{}
                err = c.List(ctx, nodeList, &k8sclient.ListOptions{LabelSelector: selector})
                if err != nil {
                        logger.Error(err, "failed to list nodes using with label selector", "labelSelector", selector)
                        return nil, nil, err
                }

                return selectedNpcl[0], nodeList.Items, nil
        } else {
                // in this case we get all the nodes and remove the ones that already part of any pool
                logger.V(1).Info("node doesn't belong to any pool, using default drain configuration with MaxUnavailable of one", "pool", *defaultPoolConfig)
                nodeList := &corev1.NodeList{}
                err = c.List(ctx, nodeList)
                if err != nil {
                        logger.Error(err, "failed to list all the nodes")
                        return nil, nil, err
                }

                defaultNodeLists := []corev1.Node{}
                for _, nodeObj := range nodeList.Items {
                        if _, exist := nodesInPools[nodeObj.Name]; !exist {
                                defaultNodeLists = append(defaultNodeLists, nodeObj)
                        }
                }
                return defaultPoolConfig, defaultNodeLists, nil
        }
}

1	/*
2
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	"bytes"
21	"context"
22	"encoding/json"
23	"fmt"
24	"os"
25	"sort"
26	"strings"
27
28	errs "github.com/pkg/errors"
29	appsv1 "k8s.io/api/apps/v1"
30	corev1 "k8s.io/api/core/v1"
31	"k8s.io/apimachinery/pkg/api/equality"
32	"k8s.io/apimachinery/pkg/api/errors"
33	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
34	uns "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
35	"k8s.io/apimachinery/pkg/labels"
36	"k8s.io/apimachinery/pkg/runtime"
37	"k8s.io/apimachinery/pkg/types"
38	"k8s.io/apimachinery/pkg/util/intstr"
39	kscheme "k8s.io/client-go/kubernetes/scheme"
40	k8sclient "sigs.k8s.io/controller-runtime/pkg/client"
41	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
42	"sigs.k8s.io/controller-runtime/pkg/event"
43	"sigs.k8s.io/controller-runtime/pkg/log"
44	"sigs.k8s.io/controller-runtime/pkg/predicate"
45
46	sriovnetworkv1 "github.com/k8snetworkplumbingwg/sriov-network-operator/api/v1"
47	"github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/apply"
48	constants "github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/consts"
49	"github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/render"
50	"github.com/k8snetworkplumbingwg/sriov-network-operator/pkg/vars"
51	)
52
53	var (
54	webhooks = map[string](string){
55	constants.InjectorWebHookName: constants.InjectorWebHookPath,
56	constants.OperatorWebHookName: constants.OperatorWebHookPath,
57	}
58	oneNode = intstr.FromInt32(1)
59	defaultPoolConfig = &sriovnetworkv1.SriovNetworkPoolConfig{Spec: sriovnetworkv1.SriovNetworkPoolConfigSpec{
60	MaxUnavailable: &oneNode,
61	NodeSelector: &metav1.LabelSelector{},
62	RdmaMode: ""}}
63	)
64
65	const (
66	clusterRoleResourceName = "ClusterRole"
67	clusterRoleBindingResourceName = "ClusterRoleBinding"
68	mutatingWebhookConfigurationCRDName = "MutatingWebhookConfiguration"
69	validatingWebhookConfigurationCRDName = "ValidatingWebhookConfiguration"
70	machineConfigCRDName = "MachineConfig"
71	trueString = "true"
72	)
73
74	type DrainAnnotationPredicate struct {
75	predicate.Funcs
76	}
77
78	func (DrainAnnotationPredicate) Create(e event.CreateEvent) bool {	1✔
79	if e.Object == nil {	1✔
80	return false	×
81	}	×
82
83	if _, hasAnno := e.Object.GetAnnotations()[constants.NodeDrainAnnotation]; hasAnno {	2✔
84	return true	1✔
85	}	1✔
86	return false	×
87	}
88
89	func (DrainAnnotationPredicate) Update(e event.UpdateEvent) bool {	1✔
90	if e.ObjectOld == nil {	1✔
91	return false	×
92	}	×
93	if e.ObjectNew == nil {	1✔
94	return false	×
95	}	×
96
97	oldAnno, hasOldAnno := e.ObjectOld.GetAnnotations()[constants.NodeDrainAnnotation]	1✔
98	newAnno, hasNewAnno := e.ObjectNew.GetAnnotations()[constants.NodeDrainAnnotation]	1✔
99		1✔
100	if !hasOldAnno && hasNewAnno {	1✔
101	return true	×
102	}	×
103
104	if oldAnno != newAnno {	2✔
105	return true	1✔
106	}	1✔
107
108	return false	1✔
109	}
110
111	type DrainStateAnnotationPredicate struct {
112	predicate.Funcs
113	}
114
115	func (DrainStateAnnotationPredicate) Create(e event.CreateEvent) bool {	1✔
116	if e.Object == nil {	1✔
117	return false	×
118	}	×
119
120	if _, hasAnno := e.Object.GetLabels()[constants.NodeStateDrainAnnotationCurrent]; hasAnno {	2✔
121	return true	1✔
122	}	1✔
123	return false	×
124	}
125
126	func (DrainStateAnnotationPredicate) Update(e event.UpdateEvent) bool {	1✔
127	if e.ObjectOld == nil {	1✔
128	return false	×
129	}	×
130	if e.ObjectNew == nil {	1✔
131	return false	×
132	}	×
133
134	oldAnno, hasOldAnno := e.ObjectOld.GetLabels()[constants.NodeStateDrainAnnotationCurrent]	1✔
135	newAnno, hasNewAnno := e.ObjectNew.GetLabels()[constants.NodeStateDrainAnnotationCurrent]	1✔
136		1✔
137	if !hasOldAnno \|\| !hasNewAnno {	1✔
138	return true	×
139	}	×
140
141	if oldAnno != newAnno {	1✔
142	return true	×
143	}	×
144
145	return oldAnno != newAnno	1✔
146	}
147
148	func GetImagePullSecrets() []string {	1✔
149	imagePullSecrets := os.Getenv("IMAGE_PULL_SECRETS")	1✔
150	if imagePullSecrets != "" {	1✔
151	return strings.Split(imagePullSecrets, ",")	×
152	} else {	1✔
153	return []string{}	1✔
154	}	1✔
155	}
156
157	func formatJSON(str string) (string, error) {	1✔
158	var prettyJSON bytes.Buffer	1✔
159	if err := json.Indent(&prettyJSON, []byte(str), "", " "); err != nil {	1✔
160	return "", err	×
161	}	×
162	return prettyJSON.String(), nil	1✔
163	}
164
165	func GetDefaultNodeSelector() map[string]string {	1✔
166	return map[string]string{"node-role.kubernetes.io/worker": "",	1✔
167	"kubernetes.io/os": "linux"}	1✔
168	}	1✔
169
170	// hasNoValidPolicy returns true if no SriovNetworkNodePolicy
171	// or only the (deprecated) "default" policy is present
172	func hasNoValidPolicy(pl []sriovnetworkv1.SriovNetworkNodePolicy) bool {	1✔
173	switch len(pl) {	1✔
174	case 0:	×
175	return true	×
176	case 1:	1✔
177	return pl[0].Name == constants.DefaultPolicyName	1✔
178	default:	1✔
179	return false	1✔
180	}
181	}
182
183	func syncPluginDaemonObjs(ctx context.Context,
184	client k8sclient.Client,
185	scheme *runtime.Scheme,
186	dc *sriovnetworkv1.SriovOperatorConfig,
187	pl *sriovnetworkv1.SriovNetworkNodePolicyList) error {	1✔
188	logger := log.Log.WithName("syncPluginDaemonObjs")	1✔
189	logger.V(1).Info("Start to sync sriov daemons objects")	1✔
190		1✔
191	// render plugin manifests	1✔
192	data := render.MakeRenderData()	1✔
193	data.Data["Namespace"] = vars.Namespace	1✔
194	data.Data["SRIOVDevicePluginImage"] = os.Getenv("SRIOV_DEVICE_PLUGIN_IMAGE")	1✔
195	data.Data["ReleaseVersion"] = os.Getenv("RELEASEVERSION")	1✔
196	data.Data["ResourcePrefix"] = vars.ResourcePrefix	1✔
197	data.Data["ImagePullSecrets"] = GetImagePullSecrets()	1✔
198	data.Data["NodeSelectorField"] = GetDefaultNodeSelector()	1✔
199	data.Data["UseCDI"] = dc.Spec.UseCDI	1✔
200	objs, err := renderDsForCR(constants.PluginPath, &data)	1✔
201	if err != nil {	1✔
202	logger.Error(err, "Fail to render SR-IoV manifests")	×
203	return err	×
204	}	×
205
206	if hasNoValidPolicy(pl.Items) {	1✔
207	for _, obj := range objs {	×
208	err := deleteK8sResource(ctx, client, obj)	×
209	if err != nil {	×
210	return err	×
211	}	×
212	}
213	return nil	×
214	}
215
216	// Sync DaemonSets
217	for _, obj := range objs {	2✔
218	if obj.GetKind() == constants.DaemonSet && len(dc.Spec.ConfigDaemonNodeSelector) > 0 {	2✔
219	scheme := kscheme.Scheme	1✔
220	ds := &appsv1.DaemonSet{}	1✔
221	err = scheme.Convert(obj, ds, nil)	1✔
222	if err != nil {	1✔
223	logger.Error(err, "Fail to convert to DaemonSet")	×
224	return err	×
225	}	×
226	ds.Spec.Template.Spec.NodeSelector = dc.Spec.ConfigDaemonNodeSelector	1✔
227	err = scheme.Convert(ds, obj, nil)	1✔
228	if err != nil {	1✔
229	logger.Error(err, "Fail to convert to Unstructured")	×
230	return err	×
231	}	×
232	}
233	err = syncDsObject(ctx, client, scheme, dc, pl, obj)	1✔
234	if err != nil {	1✔
235	logger.Error(err, "Couldn't sync SR-IoV daemons objects")	×
236	return err	×
237	}	×
238	}
239
240	return nil	1✔
241	}
242
243	func deleteK8sResource(ctx context.Context, client k8sclient.Client, in *uns.Unstructured) error {	×
244	if err := apply.DeleteObject(ctx, client, in); err != nil {	×
245	return fmt.Errorf("failed to delete object %v with err: %v", in, err)	×
246	}	×
247	return nil	×
248	}
249
250	func syncDsObject(ctx context.Context, client k8sclient.Client, scheme runtime.Scheme, dc sriovnetworkv1.SriovOperatorConfig, pl sriovnetworkv1.SriovNetworkNodePolicyList, obj uns.Unstructured) error {	1✔
251	logger := log.Log.WithName("syncDsObject")	1✔
252	kind := obj.GetKind()	1✔
253	logger.V(1).Info("Start to sync Objects", "Kind", kind)	1✔
254	switch kind {	1✔
255	case constants.ServiceAccount, constants.Role, constants.RoleBinding:	1✔
256	if err := controllerutil.SetControllerReference(dc, obj, scheme); err != nil {	1✔
257	return err	×
258	}	×
259	if err := apply.ApplyObject(ctx, client, obj); err != nil {	1✔
260	logger.Error(err, "Fail to sync", "Kind", kind)	×
261	return err	×
262	}	×
263	case constants.DaemonSet:	1✔
264	ds := &appsv1.DaemonSet{}	1✔
265	err := scheme.Convert(obj, ds, nil)	1✔
266	if err != nil {	1✔
267	logger.Error(err, "Fail to convert to DaemonSet")	×
268	return err	×
269	}	×
270	err = syncDaemonSet(ctx, client, scheme, dc, pl, ds)	1✔
271	if err != nil {	1✔
272	logger.Error(err, "Fail to sync DaemonSet", "Namespace", ds.Namespace, "Name", ds.Name)	×
273	return err	×
274	}	×
275	}
276	return nil	1✔
277	}
278
279	func setDsNodeAffinity(pl sriovnetworkv1.SriovNetworkNodePolicyList, ds appsv1.DaemonSet) error {	1✔
280	terms := nodeSelectorTermsForPolicyList(pl.Items)	1✔
281	if len(terms) > 0 {	2✔
282	ds.Spec.Template.Spec.Affinity = &corev1.Affinity{	1✔
283	NodeAffinity: &corev1.NodeAffinity{	1✔
284	RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{	1✔
285	NodeSelectorTerms: terms,	1✔
286	},	1✔
287	},	1✔
288	}	1✔
289	}	1✔
290	return nil	1✔
291	}
292
293	func nodeSelectorTermsForPolicyList(policies []sriovnetworkv1.SriovNetworkNodePolicy) []corev1.NodeSelectorTerm {	1✔
294	terms := []corev1.NodeSelectorTerm{}	1✔
295	for _, p := range policies {	2✔
296	// Note(adrianc): default policy is deprecated and ignored.	1✔
297	if p.Name == constants.DefaultPolicyName {	1✔
298	continue	×
299	}
300
301	if len(p.Spec.NodeSelector) == 0 {	2✔
302	continue	1✔
303	}
304	expressions := []corev1.NodeSelectorRequirement{}	1✔
305	for k, v := range p.Spec.NodeSelector {	2✔
306	exp := corev1.NodeSelectorRequirement{	1✔
307	Operator: corev1.NodeSelectorOpIn,	1✔
308	Key: k,	1✔
309	Values: []string{v},	1✔
310	}	1✔
311	expressions = append(expressions, exp)	1✔
312	}	1✔
313	// sorting is needed to keep the daemon spec stable.
314	// the items are popped in a random order from the map
315	sort.Slice(expressions, func(i, j int) bool {	2✔
316	return expressions[i].Key < expressions[j].Key	1✔
317	})	1✔
318	nodeSelector := corev1.NodeSelectorTerm{	1✔
319	MatchExpressions: expressions,	1✔
320	}	1✔
321	terms = append(terms, nodeSelector)	1✔
322	}
323
324	return terms	1✔
325	}
326
327	// renderDsForCR returns a busybox pod with the same name/namespace as the cr
328	func renderDsForCR(path string, data render.RenderData) ([]uns.Unstructured, error) {	1✔
329	logger := log.Log.WithName("renderDsForCR")	1✔
330	logger.V(1).Info("Start to render objects")	1✔
331		1✔
332	objs, err := render.RenderDir(path, data)	1✔
333	if err != nil {	1✔
334	return nil, errs.Wrap(err, "failed to render SR-IOV Network Operator manifests")	×
335	}	×
336	return objs, nil	1✔
337	}
338
339	func syncDaemonSet(ctx context.Context, client k8sclient.Client, scheme runtime.Scheme, dc sriovnetworkv1.SriovOperatorConfig, pl sriovnetworkv1.SriovNetworkNodePolicyList, in appsv1.DaemonSet) error {	1✔
340	logger := log.Log.WithName("syncDaemonSet")	1✔
341	logger.V(1).Info("Start to sync DaemonSet", "Namespace", in.Namespace, "Name", in.Name)	1✔
342	var err error	1✔
343		1✔
344	if pl != nil {	2✔
345	if err = setDsNodeAffinity(pl, in); err != nil {	1✔
346	return err	×
347	}	×
348	}
349	if err = controllerutil.SetControllerReference(dc, in, scheme); err != nil {	1✔
350	return err	×
351	}	×
352	ds := &appsv1.DaemonSet{}	1✔
353	err = client.Get(ctx, types.NamespacedName{Namespace: in.Namespace, Name: in.Name}, ds)	1✔
354	if err != nil {	2✔
355	if errors.IsNotFound(err) {	2✔
356	logger.V(1).Info("Created DaemonSet", in.Namespace, in.Name)	1✔
357	err = client.Create(ctx, in)	1✔
358	if err != nil {	1✔
359	logger.Error(err, "Fail to create Daemonset", "Namespace", in.Namespace, "Name", in.Name)	×
360	return err	×
361	}	×
362	} else {	×
363	logger.Error(err, "Fail to get Daemonset", "Namespace", in.Namespace, "Name", in.Name)	×
364	return err	×
365	}	×
366	} else {	1✔
367	logger.V(1).Info("DaemonSet already exists, updating")	1✔
368	// DeepDerivative checks for changes only comparing non-zero fields in the source struct.	1✔
369	// This skips default values added by the api server.	1✔
370	// References in https://github.com/kubernetes-sigs/kubebuilder/issues/592#issuecomment-625738183	1✔
371		1✔
372	// Note(Adrianc): we check Equality of OwnerReference as we changed sriov-device-plugin owner ref	1✔
373	// from SriovNetworkNodePolicy to SriovOperatorConfig, hence even if there is no change in spec,	1✔
374	// we need to update the obj's owner reference.	1✔
375		1✔
376	if equality.Semantic.DeepEqual(in.OwnerReferences, ds.OwnerReferences) &&	1✔
377	equality.Semantic.DeepDerivative(in.Spec, ds.Spec) {	2✔
378	logger.V(1).Info("Daemonset spec did not change, not updating")	1✔
379	return nil	1✔
380	}	1✔
381	err = client.Update(ctx, in)	1✔
382	if err != nil {	1✔
383	logger.Error(err, "Fail to update DaemonSet", "Namespace", in.Namespace, "Name", in.Name)	×
384	return err	×
385	}	×
386	}
387	return nil	1✔
388	}
389
390	func updateDaemonsetNodeSelector(obj *uns.Unstructured, nodeSelector map[string]string) error {	1✔
391	if len(nodeSelector) == 0 {	2✔
392	return nil	1✔
393	}	1✔
394
395	ds := &appsv1.DaemonSet{}	1✔
396	scheme := kscheme.Scheme	1✔
397	err := scheme.Convert(obj, ds, nil)	1✔
398	if err != nil {	1✔
399	return fmt.Errorf("failed to convert Unstructured [%s] to DaemonSet: %v", obj.GetName(), err)	×
400	}	×
401
402	ds.Spec.Template.Spec.NodeSelector = nodeSelector	1✔
403		1✔
404	err = scheme.Convert(ds, obj, nil)	1✔
405	if err != nil {	1✔
406	return fmt.Errorf("failed to convert DaemonSet [%s] to Unstructured: %v", obj.GetName(), err)	×
407	}	×
408	return nil	1✔
409	}
410
411	func findNodePoolConfig(ctx context.Context, node corev1.Node, c k8sclient.Client) (sriovnetworkv1.SriovNetworkPoolConfig, []corev1.Node, error) {	1✔
412	logger := log.FromContext(ctx)	1✔
413	logger.Info("FindNodePoolConfig():")	1✔
414	// get all the sriov network pool configs	1✔
415	npcl := &sriovnetworkv1.SriovNetworkPoolConfigList{}	1✔
416	err := c.List(ctx, npcl)	1✔
417	if err != nil {	1✔
NEW 418	logger.Error(err, "failed to list sriovNetworkPoolConfig")	×
NEW 419	return nil, nil, err	×
NEW 420	}	×
421
422	selectedNpcl := []*sriovnetworkv1.SriovNetworkPoolConfig{}	1✔
423	nodesInPools := map[string]interface{}{}	1✔
424		1✔
425	for _, npc := range npcl.Items {	2✔
426	// we skip hw offload objects	1✔
427	if npc.Spec.OvsHardwareOffloadConfig.Name != "" {	2✔
428	continue	1✔
429	}
430
431	if npc.Spec.NodeSelector == nil {	2✔
432	npc.Spec.NodeSelector = &metav1.LabelSelector{}	1✔
433	}	1✔
434
435	selector, err := metav1.LabelSelectorAsSelector(npc.Spec.NodeSelector)	1✔
436	if err != nil {	1✔
NEW 437	logger.Error(err, "failed to create label selector from nodeSelector", "nodeSelector", npc.Spec.NodeSelector)	×
NEW 438	return nil, nil, err	×
NEW 439	}	×
440
441	if selector.Matches(labels.Set(node.Labels)) {	2✔
442	selectedNpcl = append(selectedNpcl, npc.DeepCopy())	1✔
443	}	1✔
444
445	nodeList := &corev1.NodeList{}	1✔
446	err = c.List(ctx, nodeList, &k8sclient.ListOptions{LabelSelector: selector})	1✔
447	if err != nil {	1✔
NEW 448	logger.Error(err, "failed to list all the nodes matching the pool with label selector from nodeSelector",	×
NEW 449	"machineConfigPoolName", npc,	×
NEW 450	"nodeSelector", npc.Spec.NodeSelector)	×
NEW 451	return nil, nil, err	×
NEW 452	}	×
453
454	for _, nodeName := range nodeList.Items {	2✔
455	nodesInPools[nodeName.Name] = nil	1✔
456	}	1✔
457	}
458
459	if len(selectedNpcl) > 1 {	1✔
NEW 460	// don't allow the node to be part of multiple pools	×
NEW 461	err = fmt.Errorf("node is part of more then one pool")	×
NEW 462	logger.Error(err, "multiple pools founded for a specific node", "numberOfPools", len(selectedNpcl), "pools", selectedNpcl)	×
NEW 463	return nil, nil, err	×
464	} else if len(selectedNpcl) == 1 {	2✔
465	// found one pool for our node	1✔
466	logger.V(2).Info("found sriovNetworkPool", "pool", *selectedNpcl[0])	1✔
467	selector, err := metav1.LabelSelectorAsSelector(selectedNpcl[0].Spec.NodeSelector)	1✔
468	if err != nil {	1✔
NEW 469	logger.Error(err, "failed to create label selector from nodeSelector", "nodeSelector", selectedNpcl[0].Spec.NodeSelector)	×
NEW 470	return nil, nil, err	×
NEW 471	}	×
472
473	// list all the nodes that are also part of this pool and return them
474	nodeList := &corev1.NodeList{}	1✔
475	err = c.List(ctx, nodeList, &k8sclient.ListOptions{LabelSelector: selector})	1✔
476	if err != nil {	1✔
NEW 477	logger.Error(err, "failed to list nodes using with label selector", "labelSelector", selector)	×
NEW 478	return nil, nil, err	×
NEW 479	}	×
480
481	return selectedNpcl[0], nodeList.Items, nil	1✔
482	} else {	1✔
483	// in this case we get all the nodes and remove the ones that already part of any pool	1✔
484	logger.V(1).Info("node doesn't belong to any pool, using default drain configuration with MaxUnavailable of one", "pool", *defaultPoolConfig)	1✔
485	nodeList := &corev1.NodeList{}	1✔
486	err = c.List(ctx, nodeList)	1✔
487	if err != nil {	1✔
NEW 488	logger.Error(err, "failed to list all the nodes")	×
NEW 489	return nil, nil, err	×
NEW 490	}	×
491
492	defaultNodeLists := []corev1.Node{}	1✔
493	for _, nodeObj := range nodeList.Items {	2✔
494	if _, exist := nodesInPools[nodeObj.Name]; !exist {	2✔
495	defaultNodeLists = append(defaultNodeLists, nodeObj)	1✔
496	}	1✔
497	}
498	return defaultPoolConfig, defaultNodeLists, nil	1✔
499	}
500	}

k8snetworkplumbingwg / sriov-network-operator / 11229466077

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