#5221

Committed 09 Apr 2025 11:53AM UTC coverage: 59.346% (+0.002%) from 59.344%

Build # #5221

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

24 of 34 new or added lines in 2 files covered. (70.59%)

4 existing lines in 1 file now uncovered.

16824 of 28349 relevant lines covered (59.35%)

0.66 hits per line

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

52.6

/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 fmt.Errorf("requeuing CDI deployment selector update")
        }

        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
        }

        cr := args.Resource.(runtime.Object)
        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 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))

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

kubevirt / containerized-data-importer / #5221

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