#4852

Committed 13 Aug 2024 12:06AM UTC coverage: 59.141% (-0.03%) from 59.171%

Build # #4852

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

nbdkit datapath debug (#3361)

* pkg/image/nbdkit: Enable controlpath debugging

Disabling datapath debugging reduces noise, but controlpath debugging
is actually useful, not very noisy, and usually you would want it to
be enabled.

The difference is explained here:

https://libguestfs.org/nbdkit.1.html#SERVER-DEBUG-FLAGS

Virt-v2v disables datapath debugging but keeps controlpath debugging
enabled.

Signed-off-by: Richard W.M. Jones <rjones@redhat.com>

* pkg/image/nbdkit: Enable recommended nbdkit VDDK debugging options

VDDK datapath debugging logs every VDDK read and write which is noisy
and unnecessary.

Enabling VDDK stats gives useful information about how long each VDDK
call took, for virtually no overhead.  This information is printed to
stderr when nbdkit closes the connection.

This matches the upstream recommendations here, and also the flags
used by virt-v2v:

https://libguestfs.org/nbdkit-vddk-plugin.1.html#Troubleshooting-performance-problems

Signed-off-by: Richard W.M. Jones <rjones@redhat.com>

---------

Signed-off-by: Richard W.M. Jones <rjones@redhat.com>

Run Details

0 of 2 new or added lines in 1 file covered. (0.0%)

152 existing lines in 5 files now uncovered.

16488 of 27879 relevant lines covered (59.14%)

0.65 hits per line

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

50.28

/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(&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 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
        }

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

kubevirt / containerized-data-importer / #4852

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