#5694

Committed 02 Dec 2025 10:12AM UTC coverage: 58.669% (+0.02%) from 58.652%

Build # #5694

Build Type

Pull #3962

travis-ci

Committed by

noamasu

Commit Message

Move PvcApplyStorageProfileLabel constant to API package

- Add LabelApplyStorageProfile to containerized-data-importer-api
  package for external consumer access
- Update CDI code to use cdiv1.LabelApplyStorageProfile

This enables external projects (e.g., KubeVirt) to import the constant
from the API package instead of duplicating it.

Signed-off-by: Noam Assouline <nassouli@redhat.com>

Pull Request Pull Request #3962: Move applyStorageProfile constant to API package

Run Details

3 of 5 new or added lines in 4 files covered. (60.0%)

4 existing lines in 1 file now uncovered.

17396 of 29651 relevant lines covered (58.67%)

0.65 hits per line

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

54.21

/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)
        }
        if r.haveVolumeDataSourceValidator {
                r.reconciler.AddCallback(&appsv1.Deployment{}, reconcileVolumePopulators)
        }
}

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")
                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
        }

        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 reconcileVolumePopulators(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 := ensureVolumePopulatorsExist(context.TODO(), args.Client, args.Scheme)
        if err != nil {
                args.Recorder.Event(cr, corev1.EventTypeWarning, createResourceFailed, fmt.Sprintf("Failed to ensure volume populators exist, %v", err))
                return err
        }

        if updated {
                args.Recorder.Event(cr, corev1.EventTypeNormal, createResourceSuccess, "Successfully ensured volume populators exist")
        }

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

kubevirt / containerized-data-importer / #5694

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