#4747

Committed 20 Jun 2024 06:20PM UTC coverage: 59.036% (-0.02%) from 59.052%

Build # #4747

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Update csi-cloning prerequisites (#3325)

The CSI clone supports cross storage classes now. See https://github.com/kubernetes-csi/external-provisioner/pull/699

The limitation in CDI was removed in https://github.com/kubevirt/containerized-data-importer/pull/2750/files#diff-6a637cfe840e080e03603830a5504a0c9L754

Signed-off-by: halfcrazy <hackzhuyan@gmail.com>

Run Details

16294 of 27600 relevant lines covered (59.04%)

0.65 hits per line

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

67.41

/pkg/controller/storageprofile-controller.go

package controller

import (
        "context"
        "errors"
        "fmt"
        "reflect"
        "strconv"

        "github.com/go-logr/logr"
        snapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"
        ocpconfigv1 "github.com/openshift/api/config/v1"
        "github.com/prometheus/client_golang/prometheus"

        v1 "k8s.io/api/core/v1"
        storagev1 "k8s.io/api/storage/v1"
        k8serrors "k8s.io/apimachinery/pkg/api/errors"
        "k8s.io/apimachinery/pkg/api/meta"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/runtime"
        "k8s.io/apimachinery/pkg/types"
        "k8s.io/client-go/tools/record"
        storagehelpers "k8s.io/component-helpers/storage/volume"

        "sigs.k8s.io/controller-runtime/pkg/client"
        "sigs.k8s.io/controller-runtime/pkg/controller"
        "sigs.k8s.io/controller-runtime/pkg/event"
        "sigs.k8s.io/controller-runtime/pkg/handler"
        "sigs.k8s.io/controller-runtime/pkg/manager"
        "sigs.k8s.io/controller-runtime/pkg/predicate"
        "sigs.k8s.io/controller-runtime/pkg/reconcile"
        "sigs.k8s.io/controller-runtime/pkg/source"

        cdiv1 "kubevirt.io/containerized-data-importer-api/pkg/apis/core/v1beta1"
        "kubevirt.io/containerized-data-importer/pkg/common"
        cc "kubevirt.io/containerized-data-importer/pkg/controller/common"
        metrics "kubevirt.io/containerized-data-importer/pkg/monitoring/metrics/cdi-controller"
        "kubevirt.io/containerized-data-importer/pkg/operator"
        "kubevirt.io/containerized-data-importer/pkg/storagecapabilities"
        "kubevirt.io/containerized-data-importer/pkg/util"
)

const (
        storageProfileControllerName = "storageprofile-controller"
        counterLabelStorageClass     = "storageclass"
        counterLabelProvisioner      = "provisioner"
        counterLabelComplete         = "complete"
        counterLabelDefault          = "default"
        counterLabelVirtDefault      = "virtdefault"
        counterLabelRWX              = "rwx"
        counterLabelSmartClone       = "smartclone"
        counterLabelDegraded         = "degraded"
)

// StorageProfileReconciler members
type StorageProfileReconciler struct {
        client client.Client
        // use this for getting any resources not in the install namespace or cluster scope
        uncachedClient  client.Client
        recorder        record.EventRecorder
        scheme          *runtime.Scheme
        log             logr.Logger
        installerLabels map[string]string
}

// Reconcile the reconcile.Reconciler implementation for the StorageProfileReconciler object.
func (r *StorageProfileReconciler) Reconcile(_ context.Context, req reconcile.Request) (reconcile.Result, error) {
        log := r.log.WithValues("StorageProfile", req.NamespacedName)
        log.Info("reconciling StorageProfile")

        storageClass := &storagev1.StorageClass{}
        if err := r.client.Get(context.TODO(), req.NamespacedName, storageClass); err != nil {
                if k8serrors.IsNotFound(err) {
                        return reconcile.Result{}, r.deleteStorageProfile(req.NamespacedName.Name, log)
                }
                return reconcile.Result{}, err
        } else if storageClass.GetDeletionTimestamp() != nil {
                return reconcile.Result{}, r.deleteStorageProfile(req.NamespacedName.Name, log)
        }

        return r.reconcileStorageProfile(storageClass)
}

func (r *StorageProfileReconciler) reconcileStorageProfile(sc *storagev1.StorageClass) (reconcile.Result, error) {
        log := r.log.WithValues("StorageProfile", sc.Name)

        storageProfile, prevStorageProfile, err := r.getStorageProfile(sc)
        if err != nil {
                log.Error(err, "Unable to create StorageProfile")
                return reconcile.Result{}, err
        }

        storageProfile.Status.StorageClass = &sc.Name
        storageProfile.Status.Provisioner = &sc.Provisioner
        snapClass, err := cc.GetSnapshotClassForSmartClone(nil, &sc.Name, storageProfile.Spec.SnapshotClass, r.log, r.client, r.recorder)
        if err != nil {
                return reconcile.Result{}, err
        }
        if snapClass != "" {
                storageProfile.Status.SnapshotClass = &snapClass
        }
        storageProfile.Status.CloneStrategy = r.reconcileCloneStrategy(sc, storageProfile.Spec.CloneStrategy, snapClass)
        storageProfile.Status.DataImportCronSourceFormat = r.reconcileDataImportCronSourceFormat(sc, storageProfile.Spec.DataImportCronSourceFormat, snapClass)

        var claimPropertySets []cdiv1.ClaimPropertySet

        if len(storageProfile.Spec.ClaimPropertySets) > 0 {
                for _, cps := range storageProfile.Spec.ClaimPropertySets {
                        if cps.VolumeMode == nil || len(cps.AccessModes) == 0 {
                                err = errors.New("each ClaimPropertySet must provide both volume mode and access modes")
                                log.Error(err, "Unable to update StorageProfile")
                                return reconcile.Result{}, err
                        }
                }
                claimPropertySets = storageProfile.Spec.ClaimPropertySets
        } else {
                claimPropertySets = r.reconcilePropertySets(sc)
        }

        storageProfile.Status.ClaimPropertySets = claimPropertySets

        util.SetRecommendedLabels(storageProfile, r.installerLabels, "cdi-controller")
        if err := r.updateStorageProfile(prevStorageProfile, storageProfile, log); err != nil {
                return reconcile.Result{}, err
        }

        return reconcile.Result{}, r.computeMetrics(storageProfile, sc)
}

func (r *StorageProfileReconciler) updateStorageProfile(prevStorageProfile runtime.Object, storageProfile *cdiv1.StorageProfile, log logr.Logger) error {
        if prevStorageProfile == nil {
                return r.client.Create(context.TODO(), storageProfile)
        } else if !reflect.DeepEqual(prevStorageProfile, storageProfile) {
                // Updates have happened, update StorageProfile.
                log.Info("Updating StorageProfile", "StorageProfile.Name", storageProfile.Name, "storageProfile", storageProfile)
                return r.client.Update(context.TODO(), storageProfile)
        }

        return nil
}

func (r *StorageProfileReconciler) getStorageProfile(sc *storagev1.StorageClass) (*cdiv1.StorageProfile, runtime.Object, error) {
        var prevStorageProfile runtime.Object
        storageProfile := &cdiv1.StorageProfile{}

        if err := r.client.Get(context.TODO(), types.NamespacedName{Name: sc.Name}, storageProfile); err != nil {
                if k8serrors.IsNotFound(err) {
                        storageProfile, err = r.createEmptyStorageProfile(sc)
                        if err != nil {
                                return nil, nil, err
                        }
                } else {
                        return nil, nil, err
                }
        } else {
                prevStorageProfile = storageProfile.DeepCopyObject()
        }

        return storageProfile, prevStorageProfile, nil
}

func (r *StorageProfileReconciler) reconcilePropertySets(sc *storagev1.StorageClass) []cdiv1.ClaimPropertySet {
        claimPropertySets := []cdiv1.ClaimPropertySet{}
        capabilities, found := storagecapabilities.GetCapabilities(r.client, sc)
        if found {
                for i := range capabilities {
                        claimPropertySet := cdiv1.ClaimPropertySet{
                                AccessModes: []v1.PersistentVolumeAccessMode{capabilities[i].AccessMode},
                                VolumeMode:  &capabilities[i].VolumeMode,
                        }
                        claimPropertySets = append(claimPropertySets, claimPropertySet)
                }
        }
        return claimPropertySets
}

func (r *StorageProfileReconciler) reconcileCloneStrategy(sc *storagev1.StorageClass, desiredCloneStrategy *cdiv1.CDICloneStrategy, snapClass string) *cdiv1.CDICloneStrategy {
        if desiredCloneStrategy != nil {
                return desiredCloneStrategy
        }

        if annStrategyVal, ok := sc.Annotations["cdi.kubevirt.io/clone-strategy"]; ok {
                return r.getCloneStrategyFromStorageClass(annStrategyVal)
        }

        // Default to trying snapshot clone unless volume snapshot class missing
        hostAssistedStrategy := cdiv1.CloneStrategyHostAssisted
        strategy := hostAssistedStrategy
        if snapClass != "" {
                strategy = cdiv1.CloneStrategySnapshot
        }

        if knownStrategy, ok := storagecapabilities.GetAdvisedCloneStrategy(sc); ok {
                strategy = knownStrategy
        }

        if strategy == cdiv1.CloneStrategySnapshot && snapClass == "" {
                r.log.Info("No VolumeSnapshotClass found for storage class, falling back to host assisted cloning", "StorageClass.Name", sc.Name)
                return &hostAssistedStrategy
        }

        return &strategy
}

func (r *StorageProfileReconciler) getCloneStrategyFromStorageClass(annStrategyVal string) *cdiv1.CDICloneStrategy {
        var strategy cdiv1.CDICloneStrategy

        switch annStrategyVal {
        case "copy":
                strategy = cdiv1.CloneStrategyHostAssisted
        case "snapshot":
                strategy = cdiv1.CloneStrategySnapshot
        case "csi-clone":
                strategy = cdiv1.CloneStrategyCsiClone
        }

        return &strategy
}

func (r *StorageProfileReconciler) reconcileDataImportCronSourceFormat(sc *storagev1.StorageClass, desiredFormat *cdiv1.DataImportCronSourceFormat, snapClass string) *cdiv1.DataImportCronSourceFormat {
        if desiredFormat != nil {
                return desiredFormat
        }

        // This can be changed later on
        // for example, if at some point we're confident snapshot sources should be the default
        pvcFormat := cdiv1.DataImportCronSourceFormatPvc
        format := pvcFormat

        if knownFormat, ok := storagecapabilities.GetAdvisedSourceFormat(sc); ok {
                format = knownFormat
        }

        if format == cdiv1.DataImportCronSourceFormatSnapshot && snapClass == "" {
                // No point using snapshots without a corresponding snapshot class
                r.log.Info("No VolumeSnapshotClass found for storage class, falling back to pvc sources for DataImportCrons", "StorageClass.Name", sc.Name)
                return &pvcFormat
        }

        return &format
}

func (r *StorageProfileReconciler) createEmptyStorageProfile(sc *storagev1.StorageClass) (*cdiv1.StorageProfile, error) {
        storageProfile := MakeEmptyStorageProfileSpec(sc.Name)
        util.SetRecommendedLabels(storageProfile, r.installerLabels, "cdi-controller")
        // uncachedClient is used to directly get the config map
        // the controller runtime client caches objects that are read once, and thus requires a list/watch
        // should be cheaper than watching
        if err := operator.SetOwnerRuntime(r.uncachedClient, storageProfile); err != nil {
                return nil, err
        }
        return storageProfile, nil
}

func (r *StorageProfileReconciler) deleteStorageProfile(name string, log logr.Logger) error {
        log.Info("Cleaning up StorageProfile that corresponds to deleted StorageClass", "StorageClass.Name", name)
        profile := &cdiv1.StorageProfile{
                ObjectMeta: metav1.ObjectMeta{
                        Name: name,
                },
        }

        if err := r.client.Delete(context.TODO(), profile); cc.IgnoreNotFound(err) != nil {
                return err
        }

        labels := prometheus.Labels{
                counterLabelStorageClass: name,
        }
        metrics.DeleteStorageProfileStatus(labels)
        return nil
}

func isNoProvisioner(name string, cl client.Client) bool {
        storageClass := &storagev1.StorageClass{}
        if err := cl.Get(context.TODO(), types.NamespacedName{Name: name}, storageClass); err != nil {
                return false
        }
        return storageClass.Provisioner == storagehelpers.NotSupportedProvisioner
}

func (r *StorageProfileReconciler) computeMetrics(profile *cdiv1.StorageProfile, sc *storagev1.StorageClass) error {
        if profile.Status.StorageClass == nil || profile.Status.Provisioner == nil {
                return nil
        }

        storageClass := *profile.Status.StorageClass
        provisioner := *profile.Status.Provisioner

        // We don't count explicitly unsupported provisioners as incomplete
        _, found := storagecapabilities.UnsupportedProvisioners[*profile.Status.Provisioner]
        isComplete := found || !isIncomplete(profile.Status.ClaimPropertySets)
        isDefault := sc.Annotations[cc.AnnDefaultStorageClass] == "true"
        isVirtDefault := sc.Annotations[cc.AnnDefaultVirtStorageClass] == "true"
        isRWX := hasRWX(profile.Status.ClaimPropertySets)
        isSmartClone, err := r.hasSmartClone(profile)
        if err != nil {
                return err
        }

        isSNO := false
        clusterInfra := &ocpconfigv1.Infrastructure{}
        if err := r.client.Get(context.TODO(), types.NamespacedName{Name: "cluster"}, clusterInfra); err != nil {
                if !meta.IsNoMatchError(err) && !k8serrors.IsNotFound(err) {
                        return err
                }
        } else {
                isSNO = clusterInfra.Status.ControlPlaneTopology == ocpconfigv1.SingleReplicaTopologyMode &&
                        clusterInfra.Status.InfrastructureTopology == ocpconfigv1.SingleReplicaTopologyMode
        }

        isDegraded := (!isSNO && !isRWX) || !isSmartClone

        // Setting the labeled Gauge to 1 will not delete older metric, so we need to explicitly delete them
        scLabels := prometheus.Labels{counterLabelStorageClass: storageClass, counterLabelProvisioner: provisioner}
        metricsDeleted := metrics.DeleteStorageProfileStatus(scLabels)
        scLabels = createLabels(storageClass, provisioner, isComplete, isDefault, isVirtDefault, isRWX, isSmartClone, isDegraded)
        metrics.SetStorageProfileStatus(scLabels, 1)
        r.log.Info(fmt.Sprintf("Set metric:%s complete:%t default:%t vdefault:%t rwx:%t smartclone:%t degraded:%t (deleted %d)",
                storageClass, isComplete, isDefault, isVirtDefault, isRWX, isSmartClone, isDegraded, metricsDeleted))

        return nil
}

func (r *StorageProfileReconciler) hasSmartClone(sp *cdiv1.StorageProfile) (bool, error) {
        strategy := sp.Status.CloneStrategy
        provisioner := sp.Status.Provisioner

        if strategy != nil {
                if *strategy == cdiv1.CloneStrategyHostAssisted {
                        return false, nil
                }
                if *strategy == cdiv1.CloneStrategyCsiClone && provisioner != nil {
                        driver := &storagev1.CSIDriver{}
                        if err := r.client.Get(context.TODO(), types.NamespacedName{Name: *provisioner}, driver); err != nil {
                                return false, cc.IgnoreNotFound(err)
                        }
                        return true, nil
                }
        }

        if (strategy == nil || *strategy == cdiv1.CloneStrategySnapshot) && provisioner != nil {
                vscs := &snapshotv1.VolumeSnapshotClassList{}
                if err := r.client.List(context.TODO(), vscs); err != nil {
                        return false, err
                }
                return hasDriver(vscs, *provisioner), nil
        }

        return false, nil
}

func createLabels(storageClass, provisioner string, isComplete, isDefault, isVirtDefault, isRWX, isSmartClone, isDegraded bool) prometheus.Labels {
        return prometheus.Labels{
                counterLabelStorageClass: storageClass,
                counterLabelProvisioner:  provisioner,
                counterLabelComplete:     strconv.FormatBool(isComplete),
                counterLabelDefault:      strconv.FormatBool(isDefault),
                counterLabelVirtDefault:  strconv.FormatBool(isVirtDefault),
                counterLabelRWX:          strconv.FormatBool(isRWX),
                counterLabelSmartClone:   strconv.FormatBool(isSmartClone),
                counterLabelDegraded:     strconv.FormatBool(isDegraded),
        }
}

// MakeEmptyStorageProfileSpec creates StorageProfile manifest
func MakeEmptyStorageProfileSpec(name string) *cdiv1.StorageProfile {
        return &cdiv1.StorageProfile{
                TypeMeta: metav1.TypeMeta{
                        Kind:       "StorageProfile",
                        APIVersion: "cdi.kubevirt.io/v1beta1",
                },
                ObjectMeta: metav1.ObjectMeta{
                        Name: name,
                        Labels: map[string]string{
                                common.CDILabelKey:       common.CDILabelValue,
                                common.CDIComponentLabel: "",
                        },
                },
        }
}

// NewStorageProfileController creates a new instance of the StorageProfile controller.
func NewStorageProfileController(mgr manager.Manager, log logr.Logger, installerLabels map[string]string) (controller.Controller, error) {
        uncachedClient, err := client.New(mgr.GetConfig(), client.Options{
                Scheme: mgr.GetScheme(),
                Mapper: mgr.GetRESTMapper(),
        })
        if err != nil {
                return nil, err
        }

        reconciler := &StorageProfileReconciler{
                client:          mgr.GetClient(),
                uncachedClient:  uncachedClient,
                recorder:        mgr.GetEventRecorderFor(storageProfileControllerName),
                scheme:          mgr.GetScheme(),
                log:             log.WithName(storageProfileControllerName),
                installerLabels: installerLabels,
        }

        storageProfileController, err := controller.New(
                storageProfileControllerName,
                mgr,
                controller.Options{Reconciler: reconciler, MaxConcurrentReconciles: 3})
        if err != nil {
                return nil, err
        }
        if err := addStorageProfileControllerWatches(mgr, storageProfileController, log); err != nil {
                return nil, err
        }

        log.Info("Initialized StorageProfile controller")
        return storageProfileController, nil
}

func addStorageProfileControllerWatches(mgr manager.Manager, c controller.Controller, log logr.Logger) error {
        if err := c.Watch(source.Kind(mgr.GetCache(), &storagev1.StorageClass{}), &handler.EnqueueRequestForObject{}); err != nil {
                return err
        }

        if err := c.Watch(source.Kind(mgr.GetCache(), &cdiv1.StorageProfile{}), &handler.EnqueueRequestForObject{}); err != nil {
                return err
        }

        if err := c.Watch(source.Kind(mgr.GetCache(), &v1.PersistentVolume{}), handler.EnqueueRequestsFromMapFunc(
                func(_ context.Context, obj client.Object) []reconcile.Request {
                        return []reconcile.Request{{
                                NamespacedName: types.NamespacedName{Name: scName(obj)},
                        }}
                },
        ),
                predicate.Funcs{
                        CreateFunc: func(e event.CreateEvent) bool { return isNoProvisioner(scName(e.Object), mgr.GetClient()) },
                        UpdateFunc: func(e event.UpdateEvent) bool { return isNoProvisioner(scName(e.ObjectNew), mgr.GetClient()) },
                        DeleteFunc: func(e event.DeleteEvent) bool { return isNoProvisioner(scName(e.Object), mgr.GetClient()) },
                }); err != nil {
                return err
        }

        mapSnapshotClassToProfile := func(ctx context.Context, obj client.Object) []reconcile.Request {
                var scList storagev1.StorageClassList
                if err := mgr.GetClient().List(ctx, &scList); err != nil {
                        c.GetLogger().Error(err, "Unable to list StorageClasses")
                        return nil
                }
                vsc := obj.(*snapshotv1.VolumeSnapshotClass)
                var reqs []reconcile.Request
                for _, sc := range scList.Items {
                        if sc.Provisioner == vsc.Driver {
                                reqs = append(reqs, reconcile.Request{NamespacedName: types.NamespacedName{Name: sc.Name}})
                        }
                }
                return reqs
        }
        if err := mgr.GetClient().List(context.TODO(), &snapshotv1.VolumeSnapshotClassList{}, &client.ListOptions{Limit: 1}); err != nil {
                if meta.IsNoMatchError(err) {
                        // Back out if there's no point to attempt watch
                        return nil
                }
                if !cc.IsErrCacheNotStarted(err) {
                        return err
                }
        }
        if err := c.Watch(source.Kind(mgr.GetCache(), &snapshotv1.VolumeSnapshotClass{}),
                handler.EnqueueRequestsFromMapFunc(mapSnapshotClassToProfile),
        ); err != nil {
                return err
        }

        return nil
}

func scName(obj client.Object) string {
        return obj.(*v1.PersistentVolume).Spec.StorageClassName
}

func isIncomplete(sets []cdiv1.ClaimPropertySet) bool {
        if len(sets) > 0 {
                for _, cps := range sets {
                        if len(cps.AccessModes) == 0 || cps.VolumeMode == nil {
                                return true
                        }
                }
        } else {
                return true
        }

        return false
}

func hasRWX(cpSets []cdiv1.ClaimPropertySet) bool {
        for _, cpSet := range cpSets {
                for _, am := range cpSet.AccessModes {
                        if am == v1.ReadWriteMany {
                                return true
                        }
                }
        }
        return false
}

func hasDriver(vscs *snapshotv1.VolumeSnapshotClassList, driver string) bool {
        for i := range vscs.Items {
                vsc := vscs.Items[i]
                if vsc.Driver == driver {
                        return true
                }
        }
        return false
}

1	package controller
2
3	import (
4	"context"
5	"errors"
6	"fmt"
7	"reflect"
8	"strconv"
9
10	"github.com/go-logr/logr"
11	snapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"
12	ocpconfigv1 "github.com/openshift/api/config/v1"
13	"github.com/prometheus/client_golang/prometheus"
14
15	v1 "k8s.io/api/core/v1"
16	storagev1 "k8s.io/api/storage/v1"
17	k8serrors "k8s.io/apimachinery/pkg/api/errors"
18	"k8s.io/apimachinery/pkg/api/meta"
19	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
20	"k8s.io/apimachinery/pkg/runtime"
21	"k8s.io/apimachinery/pkg/types"
22	"k8s.io/client-go/tools/record"
23	storagehelpers "k8s.io/component-helpers/storage/volume"
24
25	"sigs.k8s.io/controller-runtime/pkg/client"
26	"sigs.k8s.io/controller-runtime/pkg/controller"
27	"sigs.k8s.io/controller-runtime/pkg/event"
28	"sigs.k8s.io/controller-runtime/pkg/handler"
29	"sigs.k8s.io/controller-runtime/pkg/manager"
30	"sigs.k8s.io/controller-runtime/pkg/predicate"
31	"sigs.k8s.io/controller-runtime/pkg/reconcile"
32	"sigs.k8s.io/controller-runtime/pkg/source"
33
34	cdiv1 "kubevirt.io/containerized-data-importer-api/pkg/apis/core/v1beta1"
35	"kubevirt.io/containerized-data-importer/pkg/common"
36	cc "kubevirt.io/containerized-data-importer/pkg/controller/common"
37	metrics "kubevirt.io/containerized-data-importer/pkg/monitoring/metrics/cdi-controller"
38	"kubevirt.io/containerized-data-importer/pkg/operator"
39	"kubevirt.io/containerized-data-importer/pkg/storagecapabilities"
40	"kubevirt.io/containerized-data-importer/pkg/util"
41	)
42
43	const (
44	storageProfileControllerName = "storageprofile-controller"
45	counterLabelStorageClass = "storageclass"
46	counterLabelProvisioner = "provisioner"
47	counterLabelComplete = "complete"
48	counterLabelDefault = "default"
49	counterLabelVirtDefault = "virtdefault"
50	counterLabelRWX = "rwx"
51	counterLabelSmartClone = "smartclone"
52	counterLabelDegraded = "degraded"
53	)
54
55	// StorageProfileReconciler members
56	type StorageProfileReconciler struct {
57	client client.Client
58	// use this for getting any resources not in the install namespace or cluster scope
59	uncachedClient client.Client
60	recorder record.EventRecorder
61	scheme *runtime.Scheme
62	log logr.Logger
63	installerLabels map[string]string
64	}
65
66	// Reconcile the reconcile.Reconciler implementation for the StorageProfileReconciler object.
67	func (r *StorageProfileReconciler) Reconcile(_ context.Context, req reconcile.Request) (reconcile.Result, error) {	1✔
68	log := r.log.WithValues("StorageProfile", req.NamespacedName)	1✔
69	log.Info("reconciling StorageProfile")	1✔
70		1✔
71	storageClass := &storagev1.StorageClass{}	1✔
72	if err := r.client.Get(context.TODO(), req.NamespacedName, storageClass); err != nil {	2✔
73	if k8serrors.IsNotFound(err) {	2✔
74	return reconcile.Result{}, r.deleteStorageProfile(req.NamespacedName.Name, log)	1✔
75	}	1✔
76	return reconcile.Result{}, err	×
77	} else if storageClass.GetDeletionTimestamp() != nil {	1✔
78	return reconcile.Result{}, r.deleteStorageProfile(req.NamespacedName.Name, log)	×
79	}	×
80
81	return r.reconcileStorageProfile(storageClass)	1✔
82	}
83
84	func (r StorageProfileReconciler) reconcileStorageProfile(sc storagev1.StorageClass) (reconcile.Result, error) {	1✔
85	log := r.log.WithValues("StorageProfile", sc.Name)	1✔
86		1✔
87	storageProfile, prevStorageProfile, err := r.getStorageProfile(sc)	1✔
88	if err != nil {	1✔
89	log.Error(err, "Unable to create StorageProfile")	×
90	return reconcile.Result{}, err	×
91	}	×
92
93	storageProfile.Status.StorageClass = &sc.Name	1✔
94	storageProfile.Status.Provisioner = &sc.Provisioner	1✔
95	snapClass, err := cc.GetSnapshotClassForSmartClone(nil, &sc.Name, storageProfile.Spec.SnapshotClass, r.log, r.client, r.recorder)	1✔
96	if err != nil {	2✔
97	return reconcile.Result{}, err	1✔
98	}	1✔
99	if snapClass != "" {	2✔
100	storageProfile.Status.SnapshotClass = &snapClass	1✔
101	}	1✔
102	storageProfile.Status.CloneStrategy = r.reconcileCloneStrategy(sc, storageProfile.Spec.CloneStrategy, snapClass)	1✔
103	storageProfile.Status.DataImportCronSourceFormat = r.reconcileDataImportCronSourceFormat(sc, storageProfile.Spec.DataImportCronSourceFormat, snapClass)	1✔
104		1✔
105	var claimPropertySets []cdiv1.ClaimPropertySet	1✔
106		1✔
107	if len(storageProfile.Spec.ClaimPropertySets) > 0 {	2✔
108	for _, cps := range storageProfile.Spec.ClaimPropertySets {	2✔
109	if cps.VolumeMode == nil \|\| len(cps.AccessModes) == 0 {	2✔
110	err = errors.New("each ClaimPropertySet must provide both volume mode and access modes")	1✔
111	log.Error(err, "Unable to update StorageProfile")	1✔
112	return reconcile.Result{}, err	1✔
113	}	1✔
114	}
115	claimPropertySets = storageProfile.Spec.ClaimPropertySets	1✔
116	} else {	1✔
117	claimPropertySets = r.reconcilePropertySets(sc)	1✔
118	}	1✔
119
120	storageProfile.Status.ClaimPropertySets = claimPropertySets	1✔
121		1✔
122	util.SetRecommendedLabels(storageProfile, r.installerLabels, "cdi-controller")	1✔
123	if err := r.updateStorageProfile(prevStorageProfile, storageProfile, log); err != nil {	1✔
124	return reconcile.Result{}, err	×
125	}	×
126
127	return reconcile.Result{}, r.computeMetrics(storageProfile, sc)	1✔
128	}
129
130	func (r StorageProfileReconciler) updateStorageProfile(prevStorageProfile runtime.Object, storageProfile cdiv1.StorageProfile, log logr.Logger) error {	1✔
131	if prevStorageProfile == nil {	2✔
132	return r.client.Create(context.TODO(), storageProfile)	1✔
133	} else if !reflect.DeepEqual(prevStorageProfile, storageProfile) {	3✔
134	// Updates have happened, update StorageProfile.	1✔
135	log.Info("Updating StorageProfile", "StorageProfile.Name", storageProfile.Name, "storageProfile", storageProfile)	1✔
136	return r.client.Update(context.TODO(), storageProfile)	1✔
137	}	1✔
138
139	return nil	1✔
140	}
141
142	func (r StorageProfileReconciler) getStorageProfile(sc storagev1.StorageClass) (*cdiv1.StorageProfile, runtime.Object, error) {	1✔
143	var prevStorageProfile runtime.Object	1✔
144	storageProfile := &cdiv1.StorageProfile{}	1✔
145		1✔
146	if err := r.client.Get(context.TODO(), types.NamespacedName{Name: sc.Name}, storageProfile); err != nil {	2✔
147	if k8serrors.IsNotFound(err) {	2✔
148	storageProfile, err = r.createEmptyStorageProfile(sc)	1✔
149	if err != nil {	1✔
150	return nil, nil, err	×
151	}	×
152	} else {	×
153	return nil, nil, err	×
154	}	×
155	} else {	1✔
156	prevStorageProfile = storageProfile.DeepCopyObject()	1✔
157	}	1✔
158
159	return storageProfile, prevStorageProfile, nil	1✔
160	}
161
162	func (r StorageProfileReconciler) reconcilePropertySets(sc storagev1.StorageClass) []cdiv1.ClaimPropertySet {	1✔
163	claimPropertySets := []cdiv1.ClaimPropertySet{}	1✔
164	capabilities, found := storagecapabilities.GetCapabilities(r.client, sc)	1✔
165	if found {	2✔
166	for i := range capabilities {	2✔
167	claimPropertySet := cdiv1.ClaimPropertySet{	1✔
168	AccessModes: []v1.PersistentVolumeAccessMode{capabilities[i].AccessMode},	1✔
169	VolumeMode: &capabilities[i].VolumeMode,	1✔
170	}	1✔
171	claimPropertySets = append(claimPropertySets, claimPropertySet)	1✔
172	}	1✔
173	}
174	return claimPropertySets	1✔
175	}
176
177	func (r StorageProfileReconciler) reconcileCloneStrategy(sc storagev1.StorageClass, desiredCloneStrategy cdiv1.CDICloneStrategy, snapClass string) cdiv1.CDICloneStrategy {	1✔
178	if desiredCloneStrategy != nil {	2✔
179	return desiredCloneStrategy	1✔
180	}	1✔
181
182	if annStrategyVal, ok := sc.Annotations["cdi.kubevirt.io/clone-strategy"]; ok {	2✔
183	return r.getCloneStrategyFromStorageClass(annStrategyVal)	1✔
184	}	1✔
185
186	// Default to trying snapshot clone unless volume snapshot class missing
187	hostAssistedStrategy := cdiv1.CloneStrategyHostAssisted	1✔
188	strategy := hostAssistedStrategy	1✔
189	if snapClass != "" {	2✔
190	strategy = cdiv1.CloneStrategySnapshot	1✔
191	}	1✔
192
193	if knownStrategy, ok := storagecapabilities.GetAdvisedCloneStrategy(sc); ok {	2✔
194	strategy = knownStrategy	1✔
195	}	1✔
196
197	if strategy == cdiv1.CloneStrategySnapshot && snapClass == "" {	1✔
198	r.log.Info("No VolumeSnapshotClass found for storage class, falling back to host assisted cloning", "StorageClass.Name", sc.Name)	×
199	return &hostAssistedStrategy	×
200	}	×
201
202	return &strategy	1✔
203	}
204
205	func (r StorageProfileReconciler) getCloneStrategyFromStorageClass(annStrategyVal string) cdiv1.CDICloneStrategy {	1✔
206	var strategy cdiv1.CDICloneStrategy	1✔
207		1✔
208	switch annStrategyVal {	1✔
209	case "copy":	1✔
210	strategy = cdiv1.CloneStrategyHostAssisted	1✔
211	case "snapshot":	1✔
212	strategy = cdiv1.CloneStrategySnapshot	1✔
213	case "csi-clone":	1✔
214	strategy = cdiv1.CloneStrategyCsiClone	1✔
215	}
216
217	return &strategy	1✔
218	}
219
220	func (r StorageProfileReconciler) reconcileDataImportCronSourceFormat(sc storagev1.StorageClass, desiredFormat cdiv1.DataImportCronSourceFormat, snapClass string) cdiv1.DataImportCronSourceFormat {	1✔
221	if desiredFormat != nil {	1✔
222	return desiredFormat	×
223	}	×
224
225	// This can be changed later on
226	// for example, if at some point we're confident snapshot sources should be the default
227	pvcFormat := cdiv1.DataImportCronSourceFormatPvc	1✔
228	format := pvcFormat	1✔
229		1✔
230	if knownFormat, ok := storagecapabilities.GetAdvisedSourceFormat(sc); ok {	2✔
231	format = knownFormat	1✔
232	}	1✔
233
234	if format == cdiv1.DataImportCronSourceFormatSnapshot && snapClass == "" {	2✔
235	// No point using snapshots without a corresponding snapshot class	1✔
236	r.log.Info("No VolumeSnapshotClass found for storage class, falling back to pvc sources for DataImportCrons", "StorageClass.Name", sc.Name)	1✔
237	return &pvcFormat	1✔
238	}	1✔
239
240	return &format	1✔
241	}
242
243	func (r StorageProfileReconciler) createEmptyStorageProfile(sc storagev1.StorageClass) (*cdiv1.StorageProfile, error) {	1✔
244	storageProfile := MakeEmptyStorageProfileSpec(sc.Name)	1✔
245	util.SetRecommendedLabels(storageProfile, r.installerLabels, "cdi-controller")	1✔
246	// uncachedClient is used to directly get the config map	1✔
247	// the controller runtime client caches objects that are read once, and thus requires a list/watch	1✔
248	// should be cheaper than watching	1✔
249	if err := operator.SetOwnerRuntime(r.uncachedClient, storageProfile); err != nil {	1✔
250	return nil, err	×
251	}	×
252	return storageProfile, nil	1✔
253	}
254
255	func (r *StorageProfileReconciler) deleteStorageProfile(name string, log logr.Logger) error {	1✔
256	log.Info("Cleaning up StorageProfile that corresponds to deleted StorageClass", "StorageClass.Name", name)	1✔
257	profile := &cdiv1.StorageProfile{	1✔
258	ObjectMeta: metav1.ObjectMeta{	1✔
259	Name: name,	1✔
260	},	1✔
261	}	1✔
262		1✔
263	if err := r.client.Delete(context.TODO(), profile); cc.IgnoreNotFound(err) != nil {	1✔
264	return err	×
265	}	×
266
267	labels := prometheus.Labels{	1✔
268	counterLabelStorageClass: name,	1✔
269	}	1✔
270	metrics.DeleteStorageProfileStatus(labels)	1✔
271	return nil	1✔
272	}
273
274	func isNoProvisioner(name string, cl client.Client) bool {	×
275	storageClass := &storagev1.StorageClass{}	×
276	if err := cl.Get(context.TODO(), types.NamespacedName{Name: name}, storageClass); err != nil {	×
277	return false	×
278	}	×
279	return storageClass.Provisioner == storagehelpers.NotSupportedProvisioner	×
280	}
281
282	func (r StorageProfileReconciler) computeMetrics(profile cdiv1.StorageProfile, sc *storagev1.StorageClass) error {	1✔
283	if profile.Status.StorageClass == nil \|\| profile.Status.Provisioner == nil {	1✔
284	return nil	×
285	}	×
286
287	storageClass := *profile.Status.StorageClass	1✔
288	provisioner := *profile.Status.Provisioner	1✔
289		1✔
290	// We don't count explicitly unsupported provisioners as incomplete	1✔
291	_, found := storagecapabilities.UnsupportedProvisioners[*profile.Status.Provisioner]	1✔
292	isComplete := found \|\| !isIncomplete(profile.Status.ClaimPropertySets)	1✔
293	isDefault := sc.Annotations[cc.AnnDefaultStorageClass] == "true"	1✔
294	isVirtDefault := sc.Annotations[cc.AnnDefaultVirtStorageClass] == "true"	1✔
295	isRWX := hasRWX(profile.Status.ClaimPropertySets)	1✔
296	isSmartClone, err := r.hasSmartClone(profile)	1✔
297	if err != nil {	1✔
298	return err	×
299	}	×
300
301	isSNO := false	1✔
302	clusterInfra := &ocpconfigv1.Infrastructure{}	1✔
303	if err := r.client.Get(context.TODO(), types.NamespacedName{Name: "cluster"}, clusterInfra); err != nil {	2✔
304	if !meta.IsNoMatchError(err) && !k8serrors.IsNotFound(err) {	1✔
305	return err	×
306	}	×
307	} else {	1✔
308	isSNO = clusterInfra.Status.ControlPlaneTopology == ocpconfigv1.SingleReplicaTopologyMode &&	1✔
309	clusterInfra.Status.InfrastructureTopology == ocpconfigv1.SingleReplicaTopologyMode	1✔
310	}	1✔
311
312	isDegraded := (!isSNO && !isRWX) \|\| !isSmartClone	1✔
313		1✔
314	// Setting the labeled Gauge to 1 will not delete older metric, so we need to explicitly delete them	1✔
315	scLabels := prometheus.Labels{counterLabelStorageClass: storageClass, counterLabelProvisioner: provisioner}	1✔
316	metricsDeleted := metrics.DeleteStorageProfileStatus(scLabels)	1✔
317	scLabels = createLabels(storageClass, provisioner, isComplete, isDefault, isVirtDefault, isRWX, isSmartClone, isDegraded)	1✔
318	metrics.SetStorageProfileStatus(scLabels, 1)	1✔
319	r.log.Info(fmt.Sprintf("Set metric:%s complete:%t default:%t vdefault:%t rwx:%t smartclone:%t degraded:%t (deleted %d)",	1✔
320	storageClass, isComplete, isDefault, isVirtDefault, isRWX, isSmartClone, isDegraded, metricsDeleted))	1✔
321		1✔
322	return nil	1✔
323	}
324
325	func (r StorageProfileReconciler) hasSmartClone(sp cdiv1.StorageProfile) (bool, error) {	1✔
326	strategy := sp.Status.CloneStrategy	1✔
327	provisioner := sp.Status.Provisioner	1✔
328		1✔
329	if strategy != nil {	2✔
330	if *strategy == cdiv1.CloneStrategyHostAssisted {	2✔
331	return false, nil	1✔
332	}	1✔
333	if *strategy == cdiv1.CloneStrategyCsiClone && provisioner != nil {	2✔
334	driver := &storagev1.CSIDriver{}	1✔
335	if err := r.client.Get(context.TODO(), types.NamespacedName{Name: *provisioner}, driver); err != nil {	2✔
336	return false, cc.IgnoreNotFound(err)	1✔
337	}	1✔
338	return true, nil	1✔
339	}
340	}
341
342	if (strategy == nil \|\| *strategy == cdiv1.CloneStrategySnapshot) && provisioner != nil {	2✔
343	vscs := &snapshotv1.VolumeSnapshotClassList{}	1✔
344	if err := r.client.List(context.TODO(), vscs); err != nil {	1✔
345	return false, err	×
346	}	×
347	return hasDriver(vscs, *provisioner), nil	1✔
348	}
349
350	return false, nil	×
351	}
352
353	func createLabels(storageClass, provisioner string, isComplete, isDefault, isVirtDefault, isRWX, isSmartClone, isDegraded bool) prometheus.Labels {	1✔
354	return prometheus.Labels{	1✔
355	counterLabelStorageClass: storageClass,	1✔
356	counterLabelProvisioner: provisioner,	1✔
357	counterLabelComplete: strconv.FormatBool(isComplete),	1✔
358	counterLabelDefault: strconv.FormatBool(isDefault),	1✔
359	counterLabelVirtDefault: strconv.FormatBool(isVirtDefault),	1✔
360	counterLabelRWX: strconv.FormatBool(isRWX),	1✔
361	counterLabelSmartClone: strconv.FormatBool(isSmartClone),	1✔
362	counterLabelDegraded: strconv.FormatBool(isDegraded),	1✔
363	}	1✔
364	}	1✔
365
366	// MakeEmptyStorageProfileSpec creates StorageProfile manifest
367	func MakeEmptyStorageProfileSpec(name string) *cdiv1.StorageProfile {	1✔
368	return &cdiv1.StorageProfile{	1✔
369	TypeMeta: metav1.TypeMeta{	1✔
370	Kind: "StorageProfile",	1✔
371	APIVersion: "cdi.kubevirt.io/v1beta1",	1✔
372	},	1✔
373	ObjectMeta: metav1.ObjectMeta{	1✔
374	Name: name,	1✔
375	Labels: map[string]string{	1✔
376	common.CDILabelKey: common.CDILabelValue,	1✔
377	common.CDIComponentLabel: "",	1✔
378	},	1✔
379	},	1✔
380	}	1✔
381	}	1✔
382
383	// NewStorageProfileController creates a new instance of the StorageProfile controller.
384	func NewStorageProfileController(mgr manager.Manager, log logr.Logger, installerLabels map[string]string) (controller.Controller, error) {	×
385	uncachedClient, err := client.New(mgr.GetConfig(), client.Options{	×
386	Scheme: mgr.GetScheme(),	×
387	Mapper: mgr.GetRESTMapper(),	×
388	})	×
389	if err != nil {	×
390	return nil, err	×
391	}	×
392
393	reconciler := &StorageProfileReconciler{	×
394	client: mgr.GetClient(),	×
395	uncachedClient: uncachedClient,	×
396	recorder: mgr.GetEventRecorderFor(storageProfileControllerName),	×
397	scheme: mgr.GetScheme(),	×
398	log: log.WithName(storageProfileControllerName),	×
399	installerLabels: installerLabels,	×
400	}	×
401		×
402	storageProfileController, err := controller.New(	×
403	storageProfileControllerName,	×
404	mgr,	×
405	controller.Options{Reconciler: reconciler, MaxConcurrentReconciles: 3})	×
406	if err != nil {	×
407	return nil, err	×
408	}	×
409	if err := addStorageProfileControllerWatches(mgr, storageProfileController, log); err != nil {	×
410	return nil, err	×
411	}	×
412
413	log.Info("Initialized StorageProfile controller")	×
414	return storageProfileController, nil	×
415	}
416
417	func addStorageProfileControllerWatches(mgr manager.Manager, c controller.Controller, log logr.Logger) error {	×
418	if err := c.Watch(source.Kind(mgr.GetCache(), &storagev1.StorageClass{}), &handler.EnqueueRequestForObject{}); err != nil {	×
419	return err	×
420	}	×
421
422	if err := c.Watch(source.Kind(mgr.GetCache(), &cdiv1.StorageProfile{}), &handler.EnqueueRequestForObject{}); err != nil {	×
423	return err	×
424	}	×
425
426	if err := c.Watch(source.Kind(mgr.GetCache(), &v1.PersistentVolume{}), handler.EnqueueRequestsFromMapFunc(	×
427	func(_ context.Context, obj client.Object) []reconcile.Request {	×
428	return []reconcile.Request{{	×
429	NamespacedName: types.NamespacedName{Name: scName(obj)},	×
430	}}	×
431	},	×
432	),
433	predicate.Funcs{
434	CreateFunc: func(e event.CreateEvent) bool { return isNoProvisioner(scName(e.Object), mgr.GetClient()) },	×
435	UpdateFunc: func(e event.UpdateEvent) bool { return isNoProvisioner(scName(e.ObjectNew), mgr.GetClient()) },	×
436	DeleteFunc: func(e event.DeleteEvent) bool { return isNoProvisioner(scName(e.Object), mgr.GetClient()) },	×
437	}); err != nil {	×
438	return err	×
439	}	×
440
441	mapSnapshotClassToProfile := func(ctx context.Context, obj client.Object) []reconcile.Request {	×
442	var scList storagev1.StorageClassList	×
443	if err := mgr.GetClient().List(ctx, &scList); err != nil {	×
444	c.GetLogger().Error(err, "Unable to list StorageClasses")	×
445	return nil	×
446	}	×
447	vsc := obj.(*snapshotv1.VolumeSnapshotClass)	×
448	var reqs []reconcile.Request	×
449	for _, sc := range scList.Items {	×
450	if sc.Provisioner == vsc.Driver {	×
451	reqs = append(reqs, reconcile.Request{NamespacedName: types.NamespacedName{Name: sc.Name}})	×
452	}	×
453	}
454	return reqs	×
455	}
456	if err := mgr.GetClient().List(context.TODO(), &snapshotv1.VolumeSnapshotClassList{}, &client.ListOptions{Limit: 1}); err != nil {	×
457	if meta.IsNoMatchError(err) {	×
458	// Back out if there's no point to attempt watch	×
459	return nil	×
460	}	×
461	if !cc.IsErrCacheNotStarted(err) {	×
462	return err	×
463	}	×
464	}
465	if err := c.Watch(source.Kind(mgr.GetCache(), &snapshotv1.VolumeSnapshotClass{}),	×
466	handler.EnqueueRequestsFromMapFunc(mapSnapshotClassToProfile),	×
467	); err != nil {	×
468	return err	×
469	}	×
470
471	return nil	×
472	}
473
474	func scName(obj client.Object) string {	×
475	return obj.(*v1.PersistentVolume).Spec.StorageClassName	×
476	}	×
477
478	func isIncomplete(sets []cdiv1.ClaimPropertySet) bool {	1✔
479	if len(sets) > 0 {	2✔
480	for _, cps := range sets {	2✔
481	if len(cps.AccessModes) == 0 \|\| cps.VolumeMode == nil {	1✔
482	return true	×
483	}	×
484	}
485	} else {	1✔
486	return true	1✔
487	}	1✔
488
489	return false	1✔
490	}
491
492	func hasRWX(cpSets []cdiv1.ClaimPropertySet) bool {	1✔
493	for _, cpSet := range cpSets {	2✔
494	for _, am := range cpSet.AccessModes {	2✔
495	if am == v1.ReadWriteMany {	2✔
496	return true	1✔
497	}	1✔
498	}
499	}
500	return false	1✔
501	}
502
503	func hasDriver(vscs *snapshotv1.VolumeSnapshotClassList, driver string) bool {	1✔
504	for i := range vscs.Items {	2✔
505	vsc := vscs.Items[i]	1✔
506	if vsc.Driver == driver {	2✔
507	return true	1✔
508	}	1✔
509	}
510	return false	1✔
511	}

kubevirt / containerized-data-importer / #4747

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