#5211

Committed 08 Apr 2025 09:20AM UTC coverage: 59.343% (-0.001%) from 59.344%

Build # #5211

Build Type

Pull #3692

travis-ci

Committed by

Acedus

Commit Message

Fix cdi-deployment Deployment selector

The cdi-deployment selector has a matchLabels value of {"app":
"containerized-data-importer"} which is a common label that appears on
many CDI related Pods (e.g., cdi-apiserver, cdi-uploadproxy, importer,
etc.).

This causes certain functionalities such as the log subresource to
misbehave as it targets the first Pod of the Pod list according to the
kubectl ByLogging sorting conditions[1].

This commit changes the cdi-deployment selector to the common
"cdi.kubevirt.io" with the value "cdi-deployment".

Since deployment selectors are immutable[2], it also adds a reconcile
callback to delete the cdi-deployment Deployment in the event of
detecting a selector that differs from the operator's desired one.

[1] https://github.com/kubernetes/kubectl/blob/fec9d5b3d/pkg/util/podutils/podutils.go#L96
[2] https://kubernetes.io/docs/concepts/workloads/controllers/deployment/#selector

Signed-off-by: Adi Aloni <aaloni@redhat.com>

Pull Request Pull Request #3692: Fix cdi-deployment Deployment selector

Run Details

19 of 25 new or added lines in 2 files covered. (76.0%)

4 existing lines in 1 file now uncovered.

16822 of 28347 relevant lines covered (59.34%)

0.66 hits per line

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

52.36

/pkg/operator/controller/callbacks.go

/*
Copyright 2018 The CDI 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"
        "reflect"

        "github.com/go-logr/logr"

        admissionregistrationv1 "k8s.io/api/admissionregistration/v1"
        appsv1 "k8s.io/api/apps/v1"
        corev1 "k8s.io/api/core/v1"
        extv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
        "k8s.io/apimachinery/pkg/api/errors"
        "k8s.io/apimachinery/pkg/api/meta"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
        "k8s.io/apimachinery/pkg/labels"
        "k8s.io/apimachinery/pkg/runtime"
        "k8s.io/apimachinery/pkg/runtime/schema"

        "sigs.k8s.io/controller-runtime/pkg/client"
        "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

        cdiv1 "kubevirt.io/containerized-data-importer-api/pkg/apis/core/v1beta1"
        "kubevirt.io/containerized-data-importer/pkg/common"
        cdicontroller "kubevirt.io/containerized-data-importer/pkg/controller"
        cc "kubevirt.io/containerized-data-importer/pkg/controller/common"
        featuregates "kubevirt.io/containerized-data-importer/pkg/feature-gates"
        "kubevirt.io/containerized-data-importer/pkg/operator/resources/cluster"
        "kubevirt.io/containerized-data-importer/pkg/operator/resources/utils"
        sdk "kubevirt.io/controller-lifecycle-operator-sdk/pkg/sdk"
        "kubevirt.io/controller-lifecycle-operator-sdk/pkg/sdk/callbacks"
)

func addReconcileCallbacks(r *ReconcileCDI) {
        r.reconciler.AddCallback(&appsv1.Deployment{}, reconcileDeleteControllerDeployment)
        r.reconciler.AddCallback(&appsv1.Deployment{}, reconcileUpdateControllerDeploymentSelector)
        r.reconciler.AddCallback(&corev1.ServiceAccount{}, reconcileSCC)
        r.reconciler.AddCallback(&appsv1.Deployment{}, reconcileCreatePrometheusInfra)
        r.reconciler.AddCallback(&appsv1.Deployment{}, reconcileRemainingRelationshipLabels)
        r.reconciler.AddCallback(&appsv1.Deployment{}, reconcileDeleteDeprecatedResources)
        r.reconciler.AddCallback(&appsv1.Deployment{}, reconcileCDICRD)
        r.reconciler.AddCallback(&appsv1.Deployment{}, reconcilePvcMutatingWebhook)
        r.reconciler.AddCallback(&extv1.CustomResourceDefinition{}, reconcileSetConfigAuthority)
        r.reconciler.AddCallback(&extv1.CustomResourceDefinition{}, reconcileHandleOldVersion)
        if r.haveRoutes {
                r.reconciler.AddCallback(&appsv1.Deployment{}, reconcileRoute)
        }
}

func isControllerDeployment(d *appsv1.Deployment) bool {
        return d.Name == "cdi-deployment"
}

func reconcileUpdateControllerDeploymentSelector(args *callbacks.ReconcileCallbackArgs) error {
        if args.State != callbacks.ReconcileStatePostRead {
                return nil
        }

        if args.CurrentObject == nil || args.DesiredObject == nil {
                return nil
        }

        deployment := args.DesiredObject.(*appsv1.Deployment)

        if !isControllerDeployment(deployment) {
                return nil
        }

        if !reflect.DeepEqual(args.CurrentObject.(*appsv1.Deployment).Spec.Selector.MatchLabels, deployment.Spec.Selector.MatchLabels) {
                args.Logger.Info("Mismatching selector detected for CDI deployment, cleaning up and requeuing")
                if err := deleteControllerDeployment(deployment, args); err != nil {
                        return err
                }
        }

        return nil
}

func reconcileDeleteControllerDeployment(args *callbacks.ReconcileCallbackArgs) error {
        switch args.State {
        case callbacks.ReconcileStatePostDelete, callbacks.ReconcileStateOperatorDelete:
        default:
                return nil
        }

        var deployment *appsv1.Deployment
        if args.DesiredObject != nil {
                deployment = args.DesiredObject.(*appsv1.Deployment)
        } else if args.CurrentObject != nil {
                deployment = args.CurrentObject.(*appsv1.Deployment)
        } else {
                args.Logger.Info("Received callback with no desired/current object")
                return nil
        }

        if !isControllerDeployment(deployment) {
                return nil
        }

        if err := deleteControllerDeployment(deployment, args); err != nil {
                return err
        }

        return nil
}

func deleteControllerDeployment(deployment *appsv1.Deployment, args *callbacks.ReconcileCallbackArgs) error {
        args.Logger.Info("Deleting CDI deployment and all import/upload/clone pods/services")
        err := args.Client.Delete(context.TODO(), deployment, &client.DeleteOptions{
                PropagationPolicy: &[]metav1.DeletionPropagation{metav1.DeletePropagationForeground}[0],
        })
        cr := args.Resource.(runtime.Object)
        if err != nil && !errors.IsNotFound(err) {
                args.Logger.Error(err, "Error deleting cdi controller deployment")
                args.Recorder.Event(cr, corev1.EventTypeWarning, deleteResourceFailed, fmt.Sprintf("Failed to delete deployment %s, %v", deployment.Name, err))
                return err
        }
        args.Recorder.Event(cr, corev1.EventTypeNormal, deleteResourceSuccess, fmt.Sprintf("Deleted deployment %s successfully", deployment.Name))

        if err = deleteWorkerResources(args.Logger, args.Client); err != nil {
                args.Logger.Error(err, "Error deleting worker resources")
                args.Recorder.Event(cr, corev1.EventTypeWarning, deleteResourceFailed, fmt.Sprintf("Failed to deleted worker resources %v", err))
                return err
        }
        args.Recorder.Event(cr, corev1.EventTypeNormal, deleteResourceSuccess, "Deleted worker resources successfully")

        return nil
}

func reconcileRoute(args *callbacks.ReconcileCallbackArgs) error {
        if args.State != callbacks.ReconcileStatePostRead {
                return nil
        }

        deployment := args.CurrentObject.(*appsv1.Deployment)
        if !isControllerDeployment(deployment) || !sdk.CheckDeploymentReady(deployment) {
                return nil
        }

        cr := args.Resource.(runtime.Object)
        updated, err := ensureUploadProxyRouteExists(context.TODO(), args.Logger, args.Client, args.Scheme, deployment)
        if err != nil {
                args.Recorder.Event(cr, corev1.EventTypeWarning, createResourceFailed, fmt.Sprintf("Failed to ensure upload proxy route exists, %v", err))
                return err
        }

        if updated {
                args.Recorder.Event(cr, corev1.EventTypeNormal, createResourceSuccess, "Successfully ensured upload proxy route exists")
        }

        if err := updateUserRoutes(context.TODO(), args.Logger, args.Client, args.Recorder); err != nil {
                return err
        }

        return nil
}

func reconcileSCC(args *callbacks.ReconcileCallbackArgs) error {
        switch args.State {
        case callbacks.ReconcileStatePreCreate, callbacks.ReconcileStatePostRead:
        default:
                return nil
        }

        sa := args.DesiredObject.(*corev1.ServiceAccount)
        if sa.Name != common.ControllerServiceAccountName {
                return nil
        }

        cr := args.Resource.(runtime.Object)
        updated, err := ensureSCCExists(context.TODO(), args.Logger, args.Client, args.Namespace, common.ControllerServiceAccountName, common.CronJobServiceAccountName)
        if err != nil {
                args.Recorder.Event(cr, corev1.EventTypeWarning, createResourceFailed, fmt.Sprintf("Failed to ensure SecurityContextConstraint exists, %v", err))
                return err
        }

        if updated {
                args.Recorder.Event(cr, corev1.EventTypeNormal, createResourceSuccess, "Successfully ensured SecurityContextConstraint exists")
        }

        return nil
}

func reconcileCreatePrometheusInfra(args *callbacks.ReconcileCallbackArgs) error {
        if args.State != callbacks.ReconcileStatePostRead {
                return nil
        }

        deployment := args.CurrentObject.(*appsv1.Deployment)
        // we don't check sdk.CheckDeploymentReady(deployment) since we want Prometheus to cover NotReady state as well
        if !isControllerDeployment(deployment) {
                return nil
        }

        cr := args.Resource.(runtime.Object)
        namespace := deployment.GetNamespace()
        if namespace == "" {
                return fmt.Errorf("cluster scoped owner not supported")
        }

        if deployed, err := isPrometheusDeployed(args.Logger, args.Client, namespace); err != nil {
                return err
        } else if !deployed {
                return nil
        }
        if err := ensurePrometheusResourcesExist(context.TODO(), args.Client, args.Scheme, deployment); err != nil {
                args.Recorder.Event(cr, corev1.EventTypeWarning, createResourceFailed, fmt.Sprintf("Failed to ensure prometheus resources exists, %v", err))
                return err
        }

        return nil
}

func deleteWorkerResources(l logr.Logger, c client.Client) error {
        listTypes := []client.ObjectList{&corev1.PodList{}, &corev1.ServiceList{}}

        ls, err := labels.Parse(fmt.Sprintf("cdi.kubevirt.io in (%s, %s, %s)",
                common.ImporterPodName, common.UploadServerCDILabel, common.ClonerSourcePodName))
        if err != nil {
                return err
        }

        for _, lt := range listTypes {
                lo := &client.ListOptions{
                        LabelSelector: ls,
                }

                l.V(1).Info("Deleting worker resources", "type", reflect.TypeOf(lt).Elem().Name())

                if err := cc.BulkDeleteResources(context.TODO(), c, lt, lo); err != nil {
                        return err
                }
        }

        return nil
}

func reconcileSetConfigAuthority(args *callbacks.ReconcileCallbackArgs) error {
        if args.State != callbacks.ReconcileStatePostRead {
                return nil
        }

        crd := args.CurrentObject.(*extv1.CustomResourceDefinition)
        if crd.Name != "cdiconfigs.cdi.kubevirt.io" {
                return nil
        }

        cdi, ok := args.Resource.(*cdiv1.CDI)
        if !ok {
                return nil
        }

        if _, ok = cdi.Annotations[cdicontroller.AnnConfigAuthority]; ok {
                return nil
        }

        if cdi.Spec.Config == nil {
                cl := &cdiv1.CDIConfigList{}
                err := args.Client.List(context.TODO(), cl)
                if err != nil {
                        if meta.IsNoMatchError(err) {
                                return nil
                        }

                        return err
                }

                if len(cl.Items) != 1 {
                        return nil
                }

                cs := cl.Items[0].Spec.DeepCopy()
                if !reflect.DeepEqual(cs, &cdiv1.CDIConfigSpec{}) {
                        cdi.Spec.Config = cs
                }
        }

        if cdi.Annotations == nil {
                cdi.Annotations = map[string]string{}
        }
        cdi.Annotations[cdicontroller.AnnConfigAuthority] = ""

        return args.Client.Update(context.TODO(), cdi)
}

func getSpecVersion(version string, crd *extv1.CustomResourceDefinition) *extv1.CustomResourceDefinitionVersion {
        for _, v := range crd.Spec.Versions {
                if v.Name == version {
                        return &v
                }
        }
        return nil
}

func rewriteOldObjects(args *callbacks.ReconcileCallbackArgs, version string, crd *extv1.CustomResourceDefinition) error {
        args.Logger.Info("Rewriting old objects")
        kind := crd.Spec.Names.Kind
        gvk := schema.GroupVersionKind{
                Group:   crd.Spec.Group,
                Version: version,
                Kind:    kind,
        }
        ul := &unstructured.UnstructuredList{}
        ul.SetGroupVersionKind(gvk)
        err := args.Client.List(context.TODO(), ul, &client.ListOptions{})
        if err != nil {
                return err
        }
        for _, item := range ul.Items {
                nn := client.ObjectKey{Namespace: item.GetNamespace(), Name: item.GetName()}
                u := &unstructured.Unstructured{}
                u.SetGroupVersionKind(item.GetObjectKind().GroupVersionKind())
                err = args.Client.Get(context.TODO(), nn, u)
                if err != nil {
                        return err
                }
                err = args.Client.Update(context.TODO(), u)
                if err != nil {
                        return err
                }
        }
        return nil
}

func removeStoredVersion(args *callbacks.ReconcileCallbackArgs, desiredVersion string, crd *extv1.CustomResourceDefinition) error {
        args.Logger.Info("Removing stored version")
        crd.Status.StoredVersions = []string{desiredVersion}
        return args.Client.Status().Update(context.TODO(), crd)
}

// Handle upgrade from clusters that had v1alpha1 as a storage version
// and remove it from all CRDs managed by us
func reconcileHandleOldVersion(args *callbacks.ReconcileCallbackArgs) error {
        if args.State != callbacks.ReconcileStatePostRead {
                return nil
        }
        currentCrd := args.CurrentObject.(*extv1.CustomResourceDefinition)
        desiredCrd := args.DesiredObject.(*extv1.CustomResourceDefinition)
        desiredVersion := newestVersion(desiredCrd)

        if olderVersionsExist(desiredVersion, currentCrd) {
                restoreOlderVersions(currentCrd, desiredCrd)
                if !desiredIsStorage(desiredVersion, currentCrd) {
                        // Let kubernetes add it
                        return nil
                }
                if err := rewriteOldObjects(args, desiredVersion, currentCrd); err != nil {
                        return err
                }
                if err := removeStoredVersion(args, desiredVersion, currentCrd); err != nil {
                        return err
                }
        }
        return nil
}

func olderVersionsExist(desiredVersion string, crd *extv1.CustomResourceDefinition) bool {
        for _, version := range crd.Status.StoredVersions {
                if version != desiredVersion {
                        return true
                }
        }
        return false
}

func desiredIsStorage(desiredVersion string, crd *extv1.CustomResourceDefinition) bool {
        specVersion := getSpecVersion(desiredVersion, crd)
        return specVersion != nil && specVersion.Storage
}

func newestVersion(crd *extv1.CustomResourceDefinition) string {
        orderedVersions := []string{"v1", "v1beta1", "v1alpha1"}
        for _, version := range orderedVersions {
                specVersion := getSpecVersion(version, crd)
                if specVersion != nil {
                        return version
                }
        }
        return ""
}

// Merge both old and new versions into new CRD, so we have both in the desiredCrd object
func restoreOlderVersions(currentCrd, desiredCrd *extv1.CustomResourceDefinition) *extv1.CustomResourceDefinition {
        for _, version := range currentCrd.Status.StoredVersions {
                specVersion := getSpecVersion(version, desiredCrd)
                if specVersion == nil {
                        // Not available in desired CRD, restore from current CRD
                        specVersion := getSpecVersion(version, currentCrd)
                        // We are only allowed one storage version
                        // The desired CRD already has one
                        specVersion.Storage = false

                        desiredCrd.Spec.Versions = append(desiredCrd.Spec.Versions, *specVersion)
                }
        }
        return desiredCrd
}

func reconcilePvcMutatingWebhook(args *callbacks.ReconcileCallbackArgs) error {
        if args.State != callbacks.ReconcileStatePostRead {
                return nil
        }

        deployment, ok := args.DesiredObject.(*appsv1.Deployment)
        if !ok || deployment.Name != common.CDIApiServerResourceName {
                return nil
        }

        enabled, err := featuregates.IsWebhookPvcRenderingEnabled(args.Client)
        if err != nil {
                return cc.IgnoreNotFound(err)
        }

        whc := &admissionregistrationv1.MutatingWebhookConfiguration{}
        key := client.ObjectKey{Name: "cdi-api-pvc-mutate"}
        err = args.Client.Get(context.TODO(), key, whc)
        if err != nil && !errors.IsNotFound(err) {
                return err
        }

        exists := err == nil
        if !enabled {
                if !exists {
                        return nil
                }
                err = args.Client.Delete(context.TODO(), whc)
                return client.IgnoreNotFound(err)
        }

        if !exists {
                if err := initPvcMutatingWebhook(whc, args); err != nil {
                        return err
                }
                return args.Client.Create(context.TODO(), whc)
        }

        whcCopy := whc.DeepCopy()
        if err := initPvcMutatingWebhook(whc, args); err != nil {
                return err
        }
        if !reflect.DeepEqual(whc, whcCopy) {
                return args.Client.Update(context.TODO(), whc)
        }

        return nil
}

func initPvcMutatingWebhook(whc *admissionregistrationv1.MutatingWebhookConfiguration, args *callbacks.ReconcileCallbackArgs) error {
        path := "/pvc-mutate"
        defaultServicePort := int32(443)
        allScopes := admissionregistrationv1.AllScopes
        exactPolicy := admissionregistrationv1.Exact
        failurePolicy := admissionregistrationv1.Fail
        defaultTimeoutSeconds := int32(10)
        reinvocationNever := admissionregistrationv1.NeverReinvocationPolicy
        sideEffect := admissionregistrationv1.SideEffectClassNone
        bundle := cluster.GetAPIServerCABundle(args.Namespace, args.Client, args.Logger)

        whc.Name = "cdi-api-pvc-mutate"
        whc.Labels = map[string]string{utils.CDILabel: cluster.APIServerServiceName}
        whc.Webhooks = []admissionregistrationv1.MutatingWebhook{
                {
                        Name: "pvc-mutate.cdi.kubevirt.io",
                        Rules: []admissionregistrationv1.RuleWithOperations{{
                                Operations: []admissionregistrationv1.OperationType{
                                        admissionregistrationv1.Create,
                                },
                                Rule: admissionregistrationv1.Rule{
                                        APIGroups:   []string{corev1.SchemeGroupVersion.Group},
                                        APIVersions: []string{corev1.SchemeGroupVersion.Version},
                                        Resources:   []string{"persistentvolumeclaims"},
                                        Scope:       &allScopes,
                                },
                        }},
                        ClientConfig: admissionregistrationv1.WebhookClientConfig{
                                Service: &admissionregistrationv1.ServiceReference{
                                        Namespace: args.Namespace,
                                        Name:      cluster.APIServerServiceName,
                                        Path:      &path,
                                        Port:      &defaultServicePort,
                                },
                                CABundle: bundle,
                        },
                        FailurePolicy:     &failurePolicy,
                        SideEffects:       &sideEffect,
                        MatchPolicy:       &exactPolicy,
                        NamespaceSelector: &metav1.LabelSelector{},
                        TimeoutSeconds:    &defaultTimeoutSeconds,
                        AdmissionReviewVersions: []string{
                                "v1",
                        },
                        ObjectSelector: &metav1.LabelSelector{
                                MatchLabels: map[string]string{
                                        common.PvcApplyStorageProfileLabel: "true",
                                },
                        },
                        ReinvocationPolicy: &reinvocationNever,
                },
        }

        cdi, err := cc.GetActiveCDI(context.TODO(), args.Client)
        if err != nil {
                return err
        }

        return controllerutil.SetControllerReference(cdi, whc, args.Scheme)
}

1	/*
2	Copyright 2018 The CDI 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	"reflect"
23
24	"github.com/go-logr/logr"
25
26	admissionregistrationv1 "k8s.io/api/admissionregistration/v1"
27	appsv1 "k8s.io/api/apps/v1"
28	corev1 "k8s.io/api/core/v1"
29	extv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
30	"k8s.io/apimachinery/pkg/api/errors"
31	"k8s.io/apimachinery/pkg/api/meta"
32	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
33	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
34	"k8s.io/apimachinery/pkg/labels"
35	"k8s.io/apimachinery/pkg/runtime"
36	"k8s.io/apimachinery/pkg/runtime/schema"
37
38	"sigs.k8s.io/controller-runtime/pkg/client"
39	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
40
41	cdiv1 "kubevirt.io/containerized-data-importer-api/pkg/apis/core/v1beta1"
42	"kubevirt.io/containerized-data-importer/pkg/common"
43	cdicontroller "kubevirt.io/containerized-data-importer/pkg/controller"
44	cc "kubevirt.io/containerized-data-importer/pkg/controller/common"
45	featuregates "kubevirt.io/containerized-data-importer/pkg/feature-gates"
46	"kubevirt.io/containerized-data-importer/pkg/operator/resources/cluster"
47	"kubevirt.io/containerized-data-importer/pkg/operator/resources/utils"
48	sdk "kubevirt.io/controller-lifecycle-operator-sdk/pkg/sdk"
49	"kubevirt.io/controller-lifecycle-operator-sdk/pkg/sdk/callbacks"
50	)
51
52	func addReconcileCallbacks(r *ReconcileCDI) {	1✔
53	r.reconciler.AddCallback(&appsv1.Deployment{}, reconcileDeleteControllerDeployment)	1✔
54	r.reconciler.AddCallback(&appsv1.Deployment{}, reconcileUpdateControllerDeploymentSelector)	1✔
55	r.reconciler.AddCallback(&corev1.ServiceAccount{}, reconcileSCC)	1✔
56	r.reconciler.AddCallback(&appsv1.Deployment{}, reconcileCreatePrometheusInfra)	1✔
57	r.reconciler.AddCallback(&appsv1.Deployment{}, reconcileRemainingRelationshipLabels)	1✔
58	r.reconciler.AddCallback(&appsv1.Deployment{}, reconcileDeleteDeprecatedResources)	1✔
59	r.reconciler.AddCallback(&appsv1.Deployment{}, reconcileCDICRD)	1✔
60	r.reconciler.AddCallback(&appsv1.Deployment{}, reconcilePvcMutatingWebhook)	1✔
61	r.reconciler.AddCallback(&extv1.CustomResourceDefinition{}, reconcileSetConfigAuthority)	1✔
62	r.reconciler.AddCallback(&extv1.CustomResourceDefinition{}, reconcileHandleOldVersion)	1✔
63	if r.haveRoutes {	2✔
64	r.reconciler.AddCallback(&appsv1.Deployment{}, reconcileRoute)	1✔
65	}	1✔
66	}
67
68	func isControllerDeployment(d *appsv1.Deployment) bool {	1✔
69	return d.Name == "cdi-deployment"	1✔
70	}	1✔
71
72	func reconcileUpdateControllerDeploymentSelector(args *callbacks.ReconcileCallbackArgs) error {	1✔
73	if args.State != callbacks.ReconcileStatePostRead {	2✔
74	return nil	1✔
75	}	1✔
76
77	if args.CurrentObject == nil \|\| args.DesiredObject == nil {	1✔
NEW 78	return nil	×
NEW 79	}	×
80
81	deployment := args.DesiredObject.(*appsv1.Deployment)	1✔
82		1✔
83	if !isControllerDeployment(deployment) {	2✔
84	return nil	1✔
85	}	1✔
86
87	if !reflect.DeepEqual(args.CurrentObject.(*appsv1.Deployment).Spec.Selector.MatchLabels, deployment.Spec.Selector.MatchLabels) {	2✔
88	args.Logger.Info("Mismatching selector detected for CDI deployment, cleaning up and requeuing")	1✔
89	if err := deleteControllerDeployment(deployment, args); err != nil {	1✔
NEW 90	return err	×
NEW 91	}	×
92	}
93
94	return nil	1✔
95	}
96
97	func reconcileDeleteControllerDeployment(args *callbacks.ReconcileCallbackArgs) error {	1✔
98	switch args.State {	1✔
99	case callbacks.ReconcileStatePostDelete, callbacks.ReconcileStateOperatorDelete:	1✔
100	default:	1✔
101	return nil	1✔
102	}
103
104	var deployment *appsv1.Deployment	1✔
105	if args.DesiredObject != nil {	2✔
106	deployment = args.DesiredObject.(*appsv1.Deployment)	1✔
107	} else if args.CurrentObject != nil {	3✔
108	deployment = args.CurrentObject.(*appsv1.Deployment)	1✔
109	} else {	1✔
110	args.Logger.Info("Received callback with no desired/current object")	×
111	return nil	×
112	}	×
113
114	if !isControllerDeployment(deployment) {	2✔
115	return nil	1✔
116	}	1✔
117
118	if err := deleteControllerDeployment(deployment, args); err != nil {	1✔
NEW 119	return err	×
NEW 120	}	×
121
122	return nil	1✔
123	}
124
125	func deleteControllerDeployment(deployment appsv1.Deployment, args callbacks.ReconcileCallbackArgs) error {	1✔
126	args.Logger.Info("Deleting CDI deployment and all import/upload/clone pods/services")	1✔
127	err := args.Client.Delete(context.TODO(), deployment, &client.DeleteOptions{	1✔
128	PropagationPolicy: &[]metav1.DeletionPropagation{metav1.DeletePropagationForeground}[0],	1✔
129	})	1✔
130	cr := args.Resource.(runtime.Object)	1✔
131	if err != nil && !errors.IsNotFound(err) {	1✔
132	args.Logger.Error(err, "Error deleting cdi controller deployment")	×
133	args.Recorder.Event(cr, corev1.EventTypeWarning, deleteResourceFailed, fmt.Sprintf("Failed to delete deployment %s, %v", deployment.Name, err))	×
134	return err	×
135	}	×
136	args.Recorder.Event(cr, corev1.EventTypeNormal, deleteResourceSuccess, fmt.Sprintf("Deleted deployment %s successfully", deployment.Name))	1✔
137		1✔
138	if err = deleteWorkerResources(args.Logger, args.Client); err != nil {	1✔
139	args.Logger.Error(err, "Error deleting worker resources")	×
140	args.Recorder.Event(cr, corev1.EventTypeWarning, deleteResourceFailed, fmt.Sprintf("Failed to deleted worker resources %v", err))	×
141	return err	×
142	}	×
143	args.Recorder.Event(cr, corev1.EventTypeNormal, deleteResourceSuccess, "Deleted worker resources successfully")	1✔
144		1✔
145	return nil	1✔
146	}
147
148	func reconcileRoute(args *callbacks.ReconcileCallbackArgs) error {	1✔
149	if args.State != callbacks.ReconcileStatePostRead {	2✔
150	return nil	1✔
151	}	1✔
152
153	deployment := args.CurrentObject.(*appsv1.Deployment)	1✔
154	if !isControllerDeployment(deployment) \|\| !sdk.CheckDeploymentReady(deployment) {	2✔
155	return nil	1✔
156	}	1✔
157
158	cr := args.Resource.(runtime.Object)	1✔
159	updated, err := ensureUploadProxyRouteExists(context.TODO(), args.Logger, args.Client, args.Scheme, deployment)	1✔
160	if err != nil {	1✔
161	args.Recorder.Event(cr, corev1.EventTypeWarning, createResourceFailed, fmt.Sprintf("Failed to ensure upload proxy route exists, %v", err))	×
162	return err	×
163	}	×
164
165	if updated {	2✔
166	args.Recorder.Event(cr, corev1.EventTypeNormal, createResourceSuccess, "Successfully ensured upload proxy route exists")	1✔
167	}	1✔
168
169	if err := updateUserRoutes(context.TODO(), args.Logger, args.Client, args.Recorder); err != nil {	1✔
170	return err	×
171	}	×
172
173	return nil	1✔
174	}
175
176	func reconcileSCC(args *callbacks.ReconcileCallbackArgs) error {	1✔
177	switch args.State {	1✔
178	case callbacks.ReconcileStatePreCreate, callbacks.ReconcileStatePostRead:	1✔
179	default:	1✔
180	return nil	1✔
181	}
182
183	sa := args.DesiredObject.(*corev1.ServiceAccount)	1✔
184	if sa.Name != common.ControllerServiceAccountName {	2✔
185	return nil	1✔
186	}	1✔
187
188	cr := args.Resource.(runtime.Object)	1✔
189	updated, err := ensureSCCExists(context.TODO(), args.Logger, args.Client, args.Namespace, common.ControllerServiceAccountName, common.CronJobServiceAccountName)	1✔
190	if err != nil {	1✔
191	args.Recorder.Event(cr, corev1.EventTypeWarning, createResourceFailed, fmt.Sprintf("Failed to ensure SecurityContextConstraint exists, %v", err))	×
192	return err	×
193	}	×
194
195	if updated {	2✔
196	args.Recorder.Event(cr, corev1.EventTypeNormal, createResourceSuccess, "Successfully ensured SecurityContextConstraint exists")	1✔
197	}	1✔
198
199	return nil	1✔
200	}
201
202	func reconcileCreatePrometheusInfra(args *callbacks.ReconcileCallbackArgs) error {	1✔
203	if args.State != callbacks.ReconcileStatePostRead {	2✔
204	return nil	1✔
205	}	1✔
206
207	deployment := args.CurrentObject.(*appsv1.Deployment)	1✔
208	// we don't check sdk.CheckDeploymentReady(deployment) since we want Prometheus to cover NotReady state as well	1✔
209	if !isControllerDeployment(deployment) {	2✔
210	return nil	1✔
211	}	1✔
212
213	cr := args.Resource.(runtime.Object)	1✔
214	namespace := deployment.GetNamespace()	1✔
215	if namespace == "" {	1✔
216	return fmt.Errorf("cluster scoped owner not supported")	×
217	}	×
218
219	if deployed, err := isPrometheusDeployed(args.Logger, args.Client, namespace); err != nil {	1✔
220	return err	×
221	} else if !deployed {	1✔
222	return nil	×
223	}	×
224	if err := ensurePrometheusResourcesExist(context.TODO(), args.Client, args.Scheme, deployment); err != nil {	1✔
225	args.Recorder.Event(cr, corev1.EventTypeWarning, createResourceFailed, fmt.Sprintf("Failed to ensure prometheus resources exists, %v", err))	×
226	return err	×
227	}	×
228
229	return nil	1✔
230	}
231
232	func deleteWorkerResources(l logr.Logger, c client.Client) error {	1✔
233	listTypes := []client.ObjectList{&corev1.PodList{}, &corev1.ServiceList{}}	1✔
234		1✔
235	ls, err := labels.Parse(fmt.Sprintf("cdi.kubevirt.io in (%s, %s, %s)",	1✔
236	common.ImporterPodName, common.UploadServerCDILabel, common.ClonerSourcePodName))	1✔
237	if err != nil {	1✔
238	return err	×
239	}	×
240
241	for _, lt := range listTypes {	2✔
242	lo := &client.ListOptions{	1✔
243	LabelSelector: ls,	1✔
244	}	1✔
245		1✔
246	l.V(1).Info("Deleting worker resources", "type", reflect.TypeOf(lt).Elem().Name())	1✔
247		1✔
248	if err := cc.BulkDeleteResources(context.TODO(), c, lt, lo); err != nil {	1✔
249	return err	×
250	}	×
251	}
252
253	return nil	1✔
254	}
255
256	func reconcileSetConfigAuthority(args *callbacks.ReconcileCallbackArgs) error {	1✔
257	if args.State != callbacks.ReconcileStatePostRead {	2✔
258	return nil	1✔
259	}	1✔
260
261	crd := args.CurrentObject.(*extv1.CustomResourceDefinition)	1✔
262	if crd.Name != "cdiconfigs.cdi.kubevirt.io" {	2✔
263	return nil	1✔
264	}	1✔
265
266	cdi, ok := args.Resource.(*cdiv1.CDI)	1✔
267	if !ok {	1✔
268	return nil	×
269	}	×
270
271	if _, ok = cdi.Annotations[cdicontroller.AnnConfigAuthority]; ok {	2✔
272	return nil	1✔
273	}	1✔
274
275	if cdi.Spec.Config == nil {	2✔
276	cl := &cdiv1.CDIConfigList{}	1✔
277	err := args.Client.List(context.TODO(), cl)	1✔
278	if err != nil {	1✔
279	if meta.IsNoMatchError(err) {	×
280	return nil	×
281	}	×
282
283	return err	×
284	}
285
286	if len(cl.Items) != 1 {	2✔
287	return nil	1✔
288	}	1✔
289
290	cs := cl.Items[0].Spec.DeepCopy()	1✔
291	if !reflect.DeepEqual(cs, &cdiv1.CDIConfigSpec{}) {	2✔
292	cdi.Spec.Config = cs	1✔
293	}	1✔
294	}
295
296	if cdi.Annotations == nil {	2✔
297	cdi.Annotations = map[string]string{}	1✔
298	}	1✔
299	cdi.Annotations[cdicontroller.AnnConfigAuthority] = ""	1✔
300		1✔
301	return args.Client.Update(context.TODO(), cdi)	1✔
302	}
303
304	func getSpecVersion(version string, crd extv1.CustomResourceDefinition) extv1.CustomResourceDefinitionVersion {	1✔
305	for _, v := range crd.Spec.Versions {	2✔
306	if v.Name == version {	2✔
307	return &v	1✔
308	}	1✔
309	}
310	return nil	1✔
311	}
312
313	func rewriteOldObjects(args callbacks.ReconcileCallbackArgs, version string, crd extv1.CustomResourceDefinition) error {	×
314	args.Logger.Info("Rewriting old objects")	×
315	kind := crd.Spec.Names.Kind	×
316	gvk := schema.GroupVersionKind{	×
317	Group: crd.Spec.Group,	×
318	Version: version,	×
319	Kind: kind,	×
320	}	×
321	ul := &unstructured.UnstructuredList{}	×
322	ul.SetGroupVersionKind(gvk)	×
323	err := args.Client.List(context.TODO(), ul, &client.ListOptions{})	×
324	if err != nil {	×
325	return err	×
326	}	×
327	for _, item := range ul.Items {	×
328	nn := client.ObjectKey{Namespace: item.GetNamespace(), Name: item.GetName()}	×
329	u := &unstructured.Unstructured{}	×
330	u.SetGroupVersionKind(item.GetObjectKind().GroupVersionKind())	×
331	err = args.Client.Get(context.TODO(), nn, u)	×
332	if err != nil {	×
333	return err	×
334	}	×
335	err = args.Client.Update(context.TODO(), u)	×
336	if err != nil {	×
337	return err	×
338	}	×
339	}
340	return nil	×
341	}
342
343	func removeStoredVersion(args callbacks.ReconcileCallbackArgs, desiredVersion string, crd extv1.CustomResourceDefinition) error {	×
344	args.Logger.Info("Removing stored version")	×
345	crd.Status.StoredVersions = []string{desiredVersion}	×
346	return args.Client.Status().Update(context.TODO(), crd)	×
347	}	×
348
349	// Handle upgrade from clusters that had v1alpha1 as a storage version
350	// and remove it from all CRDs managed by us
351	func reconcileHandleOldVersion(args *callbacks.ReconcileCallbackArgs) error {	1✔
352	if args.State != callbacks.ReconcileStatePostRead {	2✔
353	return nil	1✔
354	}	1✔
355	currentCrd := args.CurrentObject.(*extv1.CustomResourceDefinition)	1✔
356	desiredCrd := args.DesiredObject.(*extv1.CustomResourceDefinition)	1✔
357	desiredVersion := newestVersion(desiredCrd)	1✔
358		1✔
359	if olderVersionsExist(desiredVersion, currentCrd) {	1✔
360	restoreOlderVersions(currentCrd, desiredCrd)	×
361	if !desiredIsStorage(desiredVersion, currentCrd) {	×
362	// Let kubernetes add it	×
363	return nil	×
364	}	×
365	if err := rewriteOldObjects(args, desiredVersion, currentCrd); err != nil {	×
366	return err	×
367	}	×
368	if err := removeStoredVersion(args, desiredVersion, currentCrd); err != nil {	×
369	return err	×
370	}	×
371	}
372	return nil	1✔
373	}
374
375	func olderVersionsExist(desiredVersion string, crd *extv1.CustomResourceDefinition) bool {	1✔
376	for _, version := range crd.Status.StoredVersions {	1✔
377	if version != desiredVersion {	×
378	return true	×
379	}	×
380	}
381	return false	1✔
382	}
383
384	func desiredIsStorage(desiredVersion string, crd *extv1.CustomResourceDefinition) bool {	×
385	specVersion := getSpecVersion(desiredVersion, crd)	×
386	return specVersion != nil && specVersion.Storage	×
387	}	×
388
389	func newestVersion(crd *extv1.CustomResourceDefinition) string {	1✔
390	orderedVersions := []string{"v1", "v1beta1", "v1alpha1"}	1✔
391	for _, version := range orderedVersions {	2✔
392	specVersion := getSpecVersion(version, crd)	1✔
393	if specVersion != nil {	2✔
394	return version	1✔
395	}	1✔
396	}
397	return ""	×
398	}
399
400	// Merge both old and new versions into new CRD, so we have both in the desiredCrd object
401	func restoreOlderVersions(currentCrd, desiredCrd extv1.CustomResourceDefinition) extv1.CustomResourceDefinition {	×
402	for _, version := range currentCrd.Status.StoredVersions {	×
403	specVersion := getSpecVersion(version, desiredCrd)	×
404	if specVersion == nil {	×
405	// Not available in desired CRD, restore from current CRD	×
406	specVersion := getSpecVersion(version, currentCrd)	×
407	// We are only allowed one storage version	×
408	// The desired CRD already has one	×
409	specVersion.Storage = false	×
410		×
411	desiredCrd.Spec.Versions = append(desiredCrd.Spec.Versions, *specVersion)	×
412	}	×
413	}
414	return desiredCrd	×
415	}
416
417	func reconcilePvcMutatingWebhook(args *callbacks.ReconcileCallbackArgs) error {	1✔
418	if args.State != callbacks.ReconcileStatePostRead {	2✔
419	return nil	1✔
420	}	1✔
421
422	deployment, ok := args.DesiredObject.(*appsv1.Deployment)	1✔
423	if !ok \|\| deployment.Name != common.CDIApiServerResourceName {	2✔
424	return nil	1✔
425	}	1✔
426
427	enabled, err := featuregates.IsWebhookPvcRenderingEnabled(args.Client)	1✔
428	if err != nil {	2✔
429	return cc.IgnoreNotFound(err)	1✔
430	}	1✔
431
432	whc := &admissionregistrationv1.MutatingWebhookConfiguration{}	1✔
433	key := client.ObjectKey{Name: "cdi-api-pvc-mutate"}	1✔
434	err = args.Client.Get(context.TODO(), key, whc)	1✔
435	if err != nil && !errors.IsNotFound(err) {	1✔
436	return err	×
437	}	×
438
439	exists := err == nil	1✔
440	if !enabled {	2✔
441	if !exists {	2✔
442	return nil	1✔
443	}	1✔
444	err = args.Client.Delete(context.TODO(), whc)	×
445	return client.IgnoreNotFound(err)	×
446	}
447
448	if !exists {	×
449	if err := initPvcMutatingWebhook(whc, args); err != nil {	×
450	return err	×
451	}	×
452	return args.Client.Create(context.TODO(), whc)	×
453	}
454
455	whcCopy := whc.DeepCopy()	×
456	if err := initPvcMutatingWebhook(whc, args); err != nil {	×
457	return err	×
458	}	×
459	if !reflect.DeepEqual(whc, whcCopy) {	×
460	return args.Client.Update(context.TODO(), whc)	×
461	}	×
462
463	return nil	×
464	}
465
466	func initPvcMutatingWebhook(whc admissionregistrationv1.MutatingWebhookConfiguration, args callbacks.ReconcileCallbackArgs) error {	×
467	path := "/pvc-mutate"	×
468	defaultServicePort := int32(443)	×
469	allScopes := admissionregistrationv1.AllScopes	×
470	exactPolicy := admissionregistrationv1.Exact	×
471	failurePolicy := admissionregistrationv1.Fail	×
472	defaultTimeoutSeconds := int32(10)	×
473	reinvocationNever := admissionregistrationv1.NeverReinvocationPolicy	×
474	sideEffect := admissionregistrationv1.SideEffectClassNone	×
475	bundle := cluster.GetAPIServerCABundle(args.Namespace, args.Client, args.Logger)	×
476		×
477	whc.Name = "cdi-api-pvc-mutate"	×
478	whc.Labels = map[string]string{utils.CDILabel: cluster.APIServerServiceName}	×
479	whc.Webhooks = []admissionregistrationv1.MutatingWebhook{	×
480	{	×
481	Name: "pvc-mutate.cdi.kubevirt.io",	×
482	Rules: []admissionregistrationv1.RuleWithOperations{{	×
483	Operations: []admissionregistrationv1.OperationType{	×
484	admissionregistrationv1.Create,	×
485	},	×
486	Rule: admissionregistrationv1.Rule{	×
487	APIGroups: []string{corev1.SchemeGroupVersion.Group},	×
488	APIVersions: []string{corev1.SchemeGroupVersion.Version},	×
489	Resources: []string{"persistentvolumeclaims"},	×
490	Scope: &allScopes,	×
491	},	×
492	}},	×
493	ClientConfig: admissionregistrationv1.WebhookClientConfig{	×
494	Service: &admissionregistrationv1.ServiceReference{	×
495	Namespace: args.Namespace,	×
496	Name: cluster.APIServerServiceName,	×
497	Path: &path,	×
498	Port: &defaultServicePort,	×
499	},	×
500	CABundle: bundle,	×
501	},	×
502	FailurePolicy: &failurePolicy,	×
503	SideEffects: &sideEffect,	×
504	MatchPolicy: &exactPolicy,	×
505	NamespaceSelector: &metav1.LabelSelector{},	×
506	TimeoutSeconds: &defaultTimeoutSeconds,	×
507	AdmissionReviewVersions: []string{	×
508	"v1",	×
509	},	×
510	ObjectSelector: &metav1.LabelSelector{	×
511	MatchLabels: map[string]string{	×
512	common.PvcApplyStorageProfileLabel: "true",	×
513	},	×
514	},	×
515	ReinvocationPolicy: &reinvocationNever,	×
516	},	×
517	}	×
518		×
519	cdi, err := cc.GetActiveCDI(context.TODO(), args.Client)	×
520	if err != nil {	×
521	return err	×
522	}	×
523
524	return controllerutil.SetControllerReference(cdi, whc, args.Scheme)	×
525	}

kubevirt / containerized-data-importer / #5211

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