#5597

Committed 18 Sep 2025 02:27PM UTC coverage: 59.206% (-0.005%) from 59.211%

Build # #5597

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

datasource-controller: decompose addDataSourceControllerWatches (#3889)

Previously the function was bigger than it needed to be, this decomposes
it to several helper functions for clarity.

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

Run Details

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

175 existing lines in 3 files now uncovered.

17175 of 29009 relevant lines covered (59.21%)

0.65 hits per line

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

33.11

/pkg/controller/datasource-controller.go

/*
Copyright 2022 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.
limitations under the License.
See the License for the specific language governing permissions and
*/

package controller

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

        "github.com/go-logr/logr"
        snapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"

        corev1 "k8s.io/api/core/v1"
        k8serrors "k8s.io/apimachinery/pkg/api/errors"
        "k8s.io/apimachinery/pkg/api/meta"
        "k8s.io/apimachinery/pkg/runtime"
        "k8s.io/apimachinery/pkg/types"
        "k8s.io/client-go/tools/record"

        "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"
        cc "kubevirt.io/containerized-data-importer/pkg/controller/common"
)

// DataSourceReconciler members
type DataSourceReconciler struct {
        client          client.Client
        recorder        record.EventRecorder
        scheme          *runtime.Scheme
        log             logr.Logger
        installerLabels map[string]string
}

const (
        ready                    = "Ready"
        noSource                 = "NoSource"
        dataSourceControllerName = "datasource-controller"
        maxReferenceDepthReached = "MaxReferenceDepthReached"
        selfReference            = "SelfReference"
        crossNamespaceReference  = "CrossNamespaceReference"

        dataSourcePvcField        = "spec.source.pvc"
        dataSourceSnapshotField   = "spec.source.snapshot"
        dataSourceDataSourceField = "spec.source.dataSource"
)

// Reconcile loop for DataSourceReconciler
func (r *DataSourceReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
        dataSource := &cdiv1.DataSource{}
        if err := r.client.Get(ctx, req.NamespacedName, dataSource); err != nil {
                if k8serrors.IsNotFound(err) {
                        return reconcile.Result{}, nil
                }
                return reconcile.Result{}, err
        }
        if err := r.update(ctx, dataSource); err != nil {
                return reconcile.Result{}, err
        }
        return reconcile.Result{}, nil
}

func (r *DataSourceReconciler) update(ctx context.Context, dataSource *cdiv1.DataSource) error {
        dataSourceCopy := dataSource.DeepCopy()
        resolved, err := cc.ResolveDataSourceChain(ctx, r.client, dataSource)
        if err != nil {
                log := r.log.WithValues("datasource", dataSource.Name, "namespace", dataSource.Namespace)
                log.Info(err.Error())
                if err := handleDataSourceRefError(dataSource, err); err != nil {
                        return err
                }
                resolved = dataSource
        } else {
                resolved.Spec.Source.DeepCopyInto(&dataSource.Status.Source)
                dataSource.Status.Conditions = nil
        }

        switch {
        case resolved.Spec.Source.DataSource != nil:
                // Status condition handling already took place, continue to update
        case resolved.Spec.Source.PVC != nil:
                if err := r.handlePvcSource(ctx, resolved.Spec.Source.PVC, dataSource); err != nil {
                        return err
                }
        case resolved.Spec.Source.Snapshot != nil:
                if err := r.handleSnapshotSource(ctx, resolved.Spec.Source.Snapshot, dataSource); err != nil {
                        return err
                }
        default:
                updateDataSourceCondition(dataSource, cdiv1.DataSourceReady, corev1.ConditionFalse, "No source PVC set", noSource)
        }

        if dsCond := FindDataSourceConditionByType(dataSource, cdiv1.DataSourceReady); dsCond != nil && dsCond.Status == corev1.ConditionFalse {
                dataSource.Status.Source = cdiv1.DataSourceSource{}
        }

        if !reflect.DeepEqual(dataSource, dataSourceCopy) {
                if err := r.client.Update(ctx, dataSource); err != nil {
                        return err
                }
        }
        return nil
}

func (r *DataSourceReconciler) handlePvcSource(ctx context.Context, sourcePVC *cdiv1.DataVolumeSourcePVC, dataSource *cdiv1.DataSource) error {
        ns := cc.GetNamespace(sourcePVC.Namespace, dataSource.Namespace)
        isReady := false

        pvc := &corev1.PersistentVolumeClaim{}
        pvcErr := r.client.Get(ctx, types.NamespacedName{Namespace: ns, Name: sourcePVC.Name}, pvc)
        if pvcErr != nil && !k8serrors.IsNotFound(pvcErr) {
                return pvcErr
        }

        dv := &cdiv1.DataVolume{}
        if err := r.client.Get(ctx, types.NamespacedName{Namespace: ns, Name: sourcePVC.Name}, dv); err != nil {
                if !k8serrors.IsNotFound(err) {
                        return err
                }
                if pvcErr != nil {
                        r.log.Info("PVC not found", "name", sourcePVC.Name)
                        updateDataSourceCondition(dataSource, cdiv1.DataSourceReady, corev1.ConditionFalse, "PVC not found", cc.NotFound)
                } else {
                        isReady = true
                }
        } else if dv.Status.Phase == cdiv1.Succeeded {
                isReady = true
        } else {
                updateDataSourceCondition(dataSource, cdiv1.DataSourceReady, corev1.ConditionFalse, fmt.Sprintf("Import DataVolume phase %s", dv.Status.Phase), string(dv.Status.Phase))
        }

        if isReady {
                cc.CopyAllowedLabels(dv.GetLabels(), dataSource, true)
                cc.CopyAllowedLabels(pvc.GetLabels(), dataSource, true)
                updateDataSourceCondition(dataSource, cdiv1.DataSourceReady, corev1.ConditionTrue, "DataSource is ready to be consumed", ready)
        }

        return nil
}

func (r *DataSourceReconciler) handleSnapshotSource(ctx context.Context, sourceSnapshot *cdiv1.DataVolumeSourceSnapshot, dataSource *cdiv1.DataSource) error {
        snapshot := &snapshotv1.VolumeSnapshot{}
        ns := cc.GetNamespace(sourceSnapshot.Namespace, dataSource.Namespace)
        if err := r.client.Get(ctx, types.NamespacedName{Namespace: ns, Name: sourceSnapshot.Name}, snapshot); err != nil {
                if !k8serrors.IsNotFound(err) {
                        return err
                }
                r.log.Info("Snapshot not found", "name", sourceSnapshot.Name)
                updateDataSourceCondition(dataSource, cdiv1.DataSourceReady, corev1.ConditionFalse, "Snapshot not found", cc.NotFound)
        } else if cc.IsSnapshotReady(snapshot) {
                cc.CopyAllowedLabels(snapshot.GetLabels(), dataSource, true)
                updateDataSourceCondition(dataSource, cdiv1.DataSourceReady, corev1.ConditionTrue, "DataSource is ready to be consumed", ready)
        } else {
                updateDataSourceCondition(dataSource, cdiv1.DataSourceReady, corev1.ConditionFalse, "Snapshot phase is not ready", "SnapshotNotReady")
        }

        return nil
}

func handleDataSourceRefError(dataSource *cdiv1.DataSource, err error) error {
        reason := ""
        switch {
        case errors.Is(err, cc.ErrDataSourceMaxDepthReached):
                reason = maxReferenceDepthReached
        case errors.Is(err, cc.ErrDataSourceSelfReference):
                reason = selfReference
        case errors.Is(err, cc.ErrDataSourceCrossNamespace):
                reason = crossNamespaceReference
        case k8serrors.IsNotFound(err):
                reason = cc.NotFound
        default:
                return err
        }
        updateDataSourceCondition(dataSource, cdiv1.DataSourceReady, corev1.ConditionFalse, err.Error(), reason)
        return nil
}

func updateDataSourceCondition(ds *cdiv1.DataSource, conditionType cdiv1.DataSourceConditionType, status corev1.ConditionStatus, message, reason string) {
        if condition := FindDataSourceConditionByType(ds, conditionType); condition != nil {
                updateConditionState(&condition.ConditionState, status, message, reason)
        } else {
                condition = &cdiv1.DataSourceCondition{Type: conditionType}
                updateConditionState(&condition.ConditionState, status, message, reason)
                ds.Status.Conditions = append(ds.Status.Conditions, *condition)
        }
}

// FindDataSourceConditionByType finds DataSourceCondition by condition type
func FindDataSourceConditionByType(ds *cdiv1.DataSource, conditionType cdiv1.DataSourceConditionType) *cdiv1.DataSourceCondition {
        for i, condition := range ds.Status.Conditions {
                if condition.Type == conditionType {
                        return &ds.Status.Conditions[i]
                }
        }
        return nil
}

// NewDataSourceController creates a new instance of the DataSource controller
func NewDataSourceController(mgr manager.Manager, log logr.Logger, installerLabels map[string]string) (controller.Controller, error) {
        reconciler := &DataSourceReconciler{
                client:          mgr.GetClient(),
                recorder:        mgr.GetEventRecorderFor(dataSourceControllerName),
                scheme:          mgr.GetScheme(),
                log:             log.WithName(dataSourceControllerName),
                installerLabels: installerLabels,
        }
        DataSourceController, err := controller.New(dataSourceControllerName, mgr, controller.Options{
                MaxConcurrentReconciles: 3,
                Reconciler:              reconciler,
        })
        if err != nil {
                return nil, err
        }
        if err := addDataSourceControllerWatches(mgr, DataSourceController, log); err != nil {
                return nil, err
        }
        log.Info("Initialized DataSource controller")
        return DataSourceController, nil
}

func addDataSourceControllerWatches(mgr manager.Manager, c controller.Controller, log logr.Logger) error {
        if err := setupIndexers(mgr); err != nil {
                return err
        }
        if err := setupWatches(mgr, c, log); err != nil {
                return err
        }
        return nil
}

func setupIndexers(mgr manager.Manager) error {
        if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &cdiv1.DataSource{}, dataSourcePvcField, func(obj client.Object) []string {
                if pvc := obj.(*cdiv1.DataSource).Spec.Source.PVC; pvc != nil {
                        ns := cc.GetNamespace(pvc.Namespace, obj.GetNamespace())
                        return []string{types.NamespacedName{Name: pvc.Name, Namespace: ns}.String()}
                }
                return nil
        }); err != nil {
                return err
        }

        if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &cdiv1.DataSource{}, dataSourceSnapshotField, func(obj client.Object) []string {
                if snapshot := obj.(*cdiv1.DataSource).Spec.Source.Snapshot; snapshot != nil {
                        ns := cc.GetNamespace(snapshot.Namespace, obj.GetNamespace())
                        return []string{types.NamespacedName{Name: snapshot.Name, Namespace: ns}.String()}
                }
                return nil
        }); err != nil {
                return err
        }

        if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &cdiv1.DataSource{}, dataSourceDataSourceField, func(obj client.Object) []string {
                if sourceDS := obj.(*cdiv1.DataSource).Spec.Source.DataSource; sourceDS != nil {
                        ns := cc.GetNamespace(sourceDS.Namespace, obj.GetNamespace())
                        return []string{types.NamespacedName{Name: sourceDS.Name, Namespace: ns}.String()}
                }
                return nil
        }); err != nil {
                return err
        }

        return nil
}

func setupWatches(mgr manager.Manager, c controller.Controller, log logr.Logger) error {
        if err := c.Watch(source.Kind(mgr.GetCache(), &cdiv1.DataSource{},
                handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, obj *cdiv1.DataSource) []reconcile.Request {
                        reqs := []reconcile.Request{
                                {
                                        NamespacedName: types.NamespacedName{
                                                Name:      obj.Name,
                                                Namespace: obj.Namespace,
                                        },
                                },
                        }
                        return appendMatchingDataSourceRequests(ctx, mgr, dataSourceDataSourceField, obj, reqs, log)
                }),
                predicate.TypedFuncs[*cdiv1.DataSource]{
                        CreateFunc: func(e event.TypedCreateEvent[*cdiv1.DataSource]) bool { return true },
                        DeleteFunc: func(e event.TypedDeleteEvent[*cdiv1.DataSource]) bool { return true },
                        UpdateFunc: func(e event.TypedUpdateEvent[*cdiv1.DataSource]) bool {
                                return !sameSourceSpec(e.ObjectOld, e.ObjectNew) ||
                                        !sameConditions(e.ObjectOld, e.ObjectNew)
                        },
                },
        )); err != nil {
                return err
        }

        if err := c.Watch(source.Kind(mgr.GetCache(), &cdiv1.DataVolume{},
                handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, obj *cdiv1.DataVolume) []reconcile.Request {
                        return mapToDataSource(ctx, mgr, obj, log)
                }),
                predicate.TypedFuncs[*cdiv1.DataVolume]{
                        CreateFunc: func(e event.TypedCreateEvent[*cdiv1.DataVolume]) bool { return true },
                        DeleteFunc: func(e event.TypedDeleteEvent[*cdiv1.DataVolume]) bool { return true },
                        // Only DV status phase update is interesting to reconcile
                        UpdateFunc: func(e event.TypedUpdateEvent[*cdiv1.DataVolume]) bool {
                                return e.ObjectOld.Status.Phase != e.ObjectNew.Status.Phase ||
                                        !reflect.DeepEqual(e.ObjectOld.Labels, e.ObjectNew.Labels)
                        },
                },
        )); err != nil {
                return err
        }

        if err := c.Watch(source.Kind(mgr.GetCache(), &corev1.PersistentVolumeClaim{},
                handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, obj *corev1.PersistentVolumeClaim) []reconcile.Request {
                        return mapToDataSource(ctx, mgr, obj, log)
                }),
                predicate.TypedFuncs[*corev1.PersistentVolumeClaim]{
                        CreateFunc: func(e event.TypedCreateEvent[*corev1.PersistentVolumeClaim]) bool { return true },
                        DeleteFunc: func(e event.TypedDeleteEvent[*corev1.PersistentVolumeClaim]) bool { return true },
                        UpdateFunc: func(e event.TypedUpdateEvent[*corev1.PersistentVolumeClaim]) bool {
                                return e.ObjectOld.Status.Phase != e.ObjectNew.Status.Phase ||
                                        !reflect.DeepEqual(e.ObjectOld.Labels, e.ObjectNew.Labels)
                        },
                },
        )); err != nil {
                return err
        }

        if err := mgr.GetClient().List(context.TODO(), &snapshotv1.VolumeSnapshotList{}); 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.VolumeSnapshot{},
                handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, obj *snapshotv1.VolumeSnapshot) []reconcile.Request {
                        return mapToDataSource(ctx, mgr, obj, log)
                }),
                predicate.TypedFuncs[*snapshotv1.VolumeSnapshot]{
                        CreateFunc: func(e event.TypedCreateEvent[*snapshotv1.VolumeSnapshot]) bool { return true },
                        DeleteFunc: func(e event.TypedDeleteEvent[*snapshotv1.VolumeSnapshot]) bool { return true },
                        UpdateFunc: func(e event.TypedUpdateEvent[*snapshotv1.VolumeSnapshot]) bool {
                                return !reflect.DeepEqual(e.ObjectOld.Status, e.ObjectNew.Status) ||
                                        !reflect.DeepEqual(e.ObjectOld.Labels, e.ObjectNew.Labels)
                        },
                },
        )); err != nil {
                return err
        }

        return nil
}

func appendMatchingDataSourceRequests(ctx context.Context, mgr manager.Manager, indexingKey string, obj client.Object, reqs []reconcile.Request, log logr.Logger) []reconcile.Request {
        var dataSources cdiv1.DataSourceList
        matchingFields := client.MatchingFields{indexingKey: client.ObjectKeyFromObject(obj).String()}
        if err := mgr.GetClient().List(ctx, &dataSources, matchingFields); err != nil {
                log.Error(err, "Unable to list DataSources", "matchingFields", matchingFields)
                return reqs
        }
        for _, ds := range dataSources.Items {
                reqs = append(reqs, reconcile.Request{NamespacedName: types.NamespacedName{Namespace: ds.Namespace, Name: ds.Name}})
        }
        return reqs
}

func mapToDataSource(ctx context.Context, mgr manager.Manager, obj client.Object, log logr.Logger) []reconcile.Request {
        reqs := appendMatchingDataSourceRequests(ctx, mgr, dataSourcePvcField, obj, nil, log)
        return appendMatchingDataSourceRequests(ctx, mgr, dataSourceSnapshotField, obj, reqs, log)
}

func sameSourceSpec(objOld, objNew client.Object) bool {
        dsOld, okOld := objOld.(*cdiv1.DataSource)
        dsNew, okNew := objNew.(*cdiv1.DataSource)

        if !okOld || !okNew {
                return false
        }
        if dsOld.Spec.Source.PVC != nil {
                return reflect.DeepEqual(dsOld.Spec.Source.PVC, dsNew.Spec.Source.PVC)
        }
        if dsOld.Spec.Source.Snapshot != nil {
                return reflect.DeepEqual(dsOld.Spec.Source.Snapshot, dsNew.Spec.Source.Snapshot)
        }
        if dsOld.Spec.Source.DataSource != nil {
                return reflect.DeepEqual(dsOld.Spec.Source.DataSource, dsNew.Spec.Source.DataSource)
        }

        return false
}

func sameConditions(objOld, objNew client.Object) bool {
        dsOld, okOld := objOld.(*cdiv1.DataSource)
        dsNew, okNew := objNew.(*cdiv1.DataSource)

        if !okOld || !okNew {
                return false
        }

        oldConditions := dsOld.Status.Conditions
        newConditions := dsNew.Status.Conditions

        if len(oldConditions) != len(newConditions) {
                return false
        }

        condMap := make(map[cdiv1.DataSourceConditionType]cdiv1.DataSourceCondition, len(oldConditions))
        for _, c := range oldConditions {
                condMap[c.Type] = c
        }

        for _, c := range newConditions {
                if oldC, ok := condMap[c.Type]; !ok ||
                        oldC.Reason != c.Reason ||
                        oldC.Message != c.Message ||
                        oldC.Status != c.Status {
                        return false
                }
        }

        return true
}

1	/*
2	Copyright 2022 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	limitations under the License.
14	See the License for the specific language governing permissions and
15	*/
16
17	package controller
18
19	import (
20	"context"
21	"errors"
22	"fmt"
23	"reflect"
24
25	"github.com/go-logr/logr"
26	snapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"
27
28	corev1 "k8s.io/api/core/v1"
29	k8serrors "k8s.io/apimachinery/pkg/api/errors"
30	"k8s.io/apimachinery/pkg/api/meta"
31	"k8s.io/apimachinery/pkg/runtime"
32	"k8s.io/apimachinery/pkg/types"
33	"k8s.io/client-go/tools/record"
34
35	"sigs.k8s.io/controller-runtime/pkg/client"
36	"sigs.k8s.io/controller-runtime/pkg/controller"
37	"sigs.k8s.io/controller-runtime/pkg/event"
38	"sigs.k8s.io/controller-runtime/pkg/handler"
39	"sigs.k8s.io/controller-runtime/pkg/manager"
40	"sigs.k8s.io/controller-runtime/pkg/predicate"
41	"sigs.k8s.io/controller-runtime/pkg/reconcile"
42	"sigs.k8s.io/controller-runtime/pkg/source"
43
44	cdiv1 "kubevirt.io/containerized-data-importer-api/pkg/apis/core/v1beta1"
45	cc "kubevirt.io/containerized-data-importer/pkg/controller/common"
46	)
47
48	// DataSourceReconciler members
49	type DataSourceReconciler struct {
50	client client.Client
51	recorder record.EventRecorder
52	scheme *runtime.Scheme
53	log logr.Logger
54	installerLabels map[string]string
55	}
56
57	const (
58	ready = "Ready"
59	noSource = "NoSource"
60	dataSourceControllerName = "datasource-controller"
61	maxReferenceDepthReached = "MaxReferenceDepthReached"
62	selfReference = "SelfReference"
63	crossNamespaceReference = "CrossNamespaceReference"
64
65	dataSourcePvcField = "spec.source.pvc"
66	dataSourceSnapshotField = "spec.source.snapshot"
67	dataSourceDataSourceField = "spec.source.dataSource"
68	)
69
70	// Reconcile loop for DataSourceReconciler
71	func (r *DataSourceReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {	1✔
72	dataSource := &cdiv1.DataSource{}	1✔
73	if err := r.client.Get(ctx, req.NamespacedName, dataSource); err != nil {	2✔
74	if k8serrors.IsNotFound(err) {	2✔
75	return reconcile.Result{}, nil	1✔
76	}	1✔
77	return reconcile.Result{}, err	×
78	}
79	if err := r.update(ctx, dataSource); err != nil {	1✔
80	return reconcile.Result{}, err	×
81	}	×
82	return reconcile.Result{}, nil	1✔
83	}
84
85	func (r DataSourceReconciler) update(ctx context.Context, dataSource cdiv1.DataSource) error {	1✔
86	dataSourceCopy := dataSource.DeepCopy()	1✔
87	resolved, err := cc.ResolveDataSourceChain(ctx, r.client, dataSource)	1✔
88	if err != nil {	2✔
89	log := r.log.WithValues("datasource", dataSource.Name, "namespace", dataSource.Namespace)	1✔
90	log.Info(err.Error())	1✔
91	if err := handleDataSourceRefError(dataSource, err); err != nil {	1✔
92	return err	×
93	}	×
94	resolved = dataSource	1✔
95	} else {	1✔
96	resolved.Spec.Source.DeepCopyInto(&dataSource.Status.Source)	1✔
97	dataSource.Status.Conditions = nil	1✔
98	}	1✔
99
100	switch {	1✔
101	case resolved.Spec.Source.DataSource != nil:	1✔
102	// Status condition handling already took place, continue to update
103	case resolved.Spec.Source.PVC != nil:	1✔
104	if err := r.handlePvcSource(ctx, resolved.Spec.Source.PVC, dataSource); err != nil {	1✔
105	return err	×
106	}	×
107	case resolved.Spec.Source.Snapshot != nil:	1✔
108	if err := r.handleSnapshotSource(ctx, resolved.Spec.Source.Snapshot, dataSource); err != nil {	1✔
109	return err	×
110	}	×
111	default:	1✔
112	updateDataSourceCondition(dataSource, cdiv1.DataSourceReady, corev1.ConditionFalse, "No source PVC set", noSource)	1✔
113	}
114
115	if dsCond := FindDataSourceConditionByType(dataSource, cdiv1.DataSourceReady); dsCond != nil && dsCond.Status == corev1.ConditionFalse {	2✔
116	dataSource.Status.Source = cdiv1.DataSourceSource{}	1✔
117	}	1✔
118
119	if !reflect.DeepEqual(dataSource, dataSourceCopy) {	2✔
120	if err := r.client.Update(ctx, dataSource); err != nil {	1✔
121	return err	×
122	}	×
123	}
124	return nil	1✔
125	}
126
127	func (r DataSourceReconciler) handlePvcSource(ctx context.Context, sourcePVC cdiv1.DataVolumeSourcePVC, dataSource *cdiv1.DataSource) error {	1✔
128	ns := cc.GetNamespace(sourcePVC.Namespace, dataSource.Namespace)	1✔
129	isReady := false	1✔
130		1✔
131	pvc := &corev1.PersistentVolumeClaim{}	1✔
132	pvcErr := r.client.Get(ctx, types.NamespacedName{Namespace: ns, Name: sourcePVC.Name}, pvc)	1✔
133	if pvcErr != nil && !k8serrors.IsNotFound(pvcErr) {	1✔
134	return pvcErr	×
135	}	×
136
137	dv := &cdiv1.DataVolume{}	1✔
138	if err := r.client.Get(ctx, types.NamespacedName{Namespace: ns, Name: sourcePVC.Name}, dv); err != nil {	2✔
139	if !k8serrors.IsNotFound(err) {	1✔
140	return err	×
141	}	×
142	if pvcErr != nil {	2✔
143	r.log.Info("PVC not found", "name", sourcePVC.Name)	1✔
144	updateDataSourceCondition(dataSource, cdiv1.DataSourceReady, corev1.ConditionFalse, "PVC not found", cc.NotFound)	1✔
145	} else {	2✔
146	isReady = true	1✔
147	}	1✔
148	} else if dv.Status.Phase == cdiv1.Succeeded {	2✔
149	isReady = true	1✔
150	} else {	2✔
151	updateDataSourceCondition(dataSource, cdiv1.DataSourceReady, corev1.ConditionFalse, fmt.Sprintf("Import DataVolume phase %s", dv.Status.Phase), string(dv.Status.Phase))	1✔
152	}	1✔
153
154	if isReady {	2✔
155	cc.CopyAllowedLabels(dv.GetLabels(), dataSource, true)	1✔
156	cc.CopyAllowedLabels(pvc.GetLabels(), dataSource, true)	1✔
157	updateDataSourceCondition(dataSource, cdiv1.DataSourceReady, corev1.ConditionTrue, "DataSource is ready to be consumed", ready)	1✔
158	}	1✔
159
160	return nil	1✔
161	}
162
163	func (r DataSourceReconciler) handleSnapshotSource(ctx context.Context, sourceSnapshot cdiv1.DataVolumeSourceSnapshot, dataSource *cdiv1.DataSource) error {	1✔
164	snapshot := &snapshotv1.VolumeSnapshot{}	1✔
165	ns := cc.GetNamespace(sourceSnapshot.Namespace, dataSource.Namespace)	1✔
166	if err := r.client.Get(ctx, types.NamespacedName{Namespace: ns, Name: sourceSnapshot.Name}, snapshot); err != nil {	2✔
167	if !k8serrors.IsNotFound(err) {	1✔
168	return err	×
169	}	×
170	r.log.Info("Snapshot not found", "name", sourceSnapshot.Name)	1✔
171	updateDataSourceCondition(dataSource, cdiv1.DataSourceReady, corev1.ConditionFalse, "Snapshot not found", cc.NotFound)	1✔
172	} else if cc.IsSnapshotReady(snapshot) {	2✔
173	cc.CopyAllowedLabels(snapshot.GetLabels(), dataSource, true)	1✔
174	updateDataSourceCondition(dataSource, cdiv1.DataSourceReady, corev1.ConditionTrue, "DataSource is ready to be consumed", ready)	1✔
175	} else {	2✔
176	updateDataSourceCondition(dataSource, cdiv1.DataSourceReady, corev1.ConditionFalse, "Snapshot phase is not ready", "SnapshotNotReady")	1✔
177	}	1✔
178
179	return nil	1✔
180	}
181
182	func handleDataSourceRefError(dataSource *cdiv1.DataSource, err error) error {	1✔
183	reason := ""	1✔
184	switch {	1✔
185	case errors.Is(err, cc.ErrDataSourceMaxDepthReached):	1✔
186	reason = maxReferenceDepthReached	1✔
187	case errors.Is(err, cc.ErrDataSourceSelfReference):	1✔
188	reason = selfReference	1✔
189	case errors.Is(err, cc.ErrDataSourceCrossNamespace):	1✔
190	reason = crossNamespaceReference	1✔
191	case k8serrors.IsNotFound(err):	1✔
192	reason = cc.NotFound	1✔
193	default:	×
194	return err	×
195	}
196	updateDataSourceCondition(dataSource, cdiv1.DataSourceReady, corev1.ConditionFalse, err.Error(), reason)	1✔
197	return nil	1✔
198	}
199
200	func updateDataSourceCondition(ds *cdiv1.DataSource, conditionType cdiv1.DataSourceConditionType, status corev1.ConditionStatus, message, reason string) {	1✔
201	if condition := FindDataSourceConditionByType(ds, conditionType); condition != nil {	1✔
202	updateConditionState(&condition.ConditionState, status, message, reason)	×
203	} else {	1✔
204	condition = &cdiv1.DataSourceCondition{Type: conditionType}	1✔
205	updateConditionState(&condition.ConditionState, status, message, reason)	1✔
206	ds.Status.Conditions = append(ds.Status.Conditions, *condition)	1✔
207	}	1✔
208	}
209
210	// FindDataSourceConditionByType finds DataSourceCondition by condition type
211	func FindDataSourceConditionByType(ds cdiv1.DataSource, conditionType cdiv1.DataSourceConditionType) cdiv1.DataSourceCondition {	1✔
212	for i, condition := range ds.Status.Conditions {	2✔
213	if condition.Type == conditionType {	2✔
214	return &ds.Status.Conditions[i]	1✔
215	}	1✔
216	}
217	return nil	1✔
218	}
219
220	// NewDataSourceController creates a new instance of the DataSource controller
221	func NewDataSourceController(mgr manager.Manager, log logr.Logger, installerLabels map[string]string) (controller.Controller, error) {	×
222	reconciler := &DataSourceReconciler{	×
223	client: mgr.GetClient(),	×
224	recorder: mgr.GetEventRecorderFor(dataSourceControllerName),	×
225	scheme: mgr.GetScheme(),	×
226	log: log.WithName(dataSourceControllerName),	×
227	installerLabels: installerLabels,	×
228	}	×
229	DataSourceController, err := controller.New(dataSourceControllerName, mgr, controller.Options{	×
230	MaxConcurrentReconciles: 3,	×
231	Reconciler: reconciler,	×
232	})	×
233	if err != nil {	×
234	return nil, err	×
235	}	×
236	if err := addDataSourceControllerWatches(mgr, DataSourceController, log); err != nil {	×
237	return nil, err	×
238	}	×
239	log.Info("Initialized DataSource controller")	×
240	return DataSourceController, nil	×
241	}
242
243	func addDataSourceControllerWatches(mgr manager.Manager, c controller.Controller, log logr.Logger) error {	×
NEW 244	if err := setupIndexers(mgr); err != nil {	×
NEW 245	return err	×
UNCOV 246	}	×
NEW 247	if err := setupWatches(mgr, c, log); err != nil {	×
248	return err	×
249	}	×
NEW 250	return nil	×
251	}
252
NEW 253	func setupIndexers(mgr manager.Manager) error {	×
254	if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &cdiv1.DataSource{}, dataSourcePvcField, func(obj client.Object) []string {	×
255	if pvc := obj.(*cdiv1.DataSource).Spec.Source.PVC; pvc != nil {	×
256	ns := cc.GetNamespace(pvc.Namespace, obj.GetNamespace())	×
NEW 257	return []string{types.NamespacedName{Name: pvc.Name, Namespace: ns}.String()}	×
258	}	×
259	return nil	×
260	}); err != nil {	×
261	return err	×
262	}	×
263
264	if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &cdiv1.DataSource{}, dataSourceSnapshotField, func(obj client.Object) []string {	×
265	if snapshot := obj.(*cdiv1.DataSource).Spec.Source.Snapshot; snapshot != nil {	×
266	ns := cc.GetNamespace(snapshot.Namespace, obj.GetNamespace())	×
NEW 267	return []string{types.NamespacedName{Name: snapshot.Name, Namespace: ns}.String()}	×
268	}	×
269	return nil	×
270	}); err != nil {	×
271	return err	×
272	}	×
273
274	if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &cdiv1.DataSource{}, dataSourceDataSourceField, func(obj client.Object) []string {	×
NEW 275	if sourceDS := obj.(*cdiv1.DataSource).Spec.Source.DataSource; sourceDS != nil {	×
NEW 276	ns := cc.GetNamespace(sourceDS.Namespace, obj.GetNamespace())	×
NEW 277	return []string{types.NamespacedName{Name: sourceDS.Name, Namespace: ns}.String()}	×
278	}	×
279	return nil	×
280	}); err != nil {	×
281	return err	×
282	}	×
283
NEW 284	return nil	×
285	}
286
NEW 287	func setupWatches(mgr manager.Manager, c controller.Controller, log logr.Logger) error {	×
NEW 288	if err := c.Watch(source.Kind(mgr.GetCache(), &cdiv1.DataSource{},	×
NEW 289	handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, obj *cdiv1.DataSource) []reconcile.Request {	×
NEW 290	reqs := []reconcile.Request{	×
NEW 291	{	×
NEW 292	NamespacedName: types.NamespacedName{	×
NEW 293	Name: obj.Name,	×
NEW 294	Namespace: obj.Namespace,	×
NEW 295	},	×
NEW 296	},	×
NEW 297	}	×
NEW 298	return appendMatchingDataSourceRequests(ctx, mgr, dataSourceDataSourceField, obj, reqs, log)	×
NEW 299	}),	×
300	predicate.TypedFuncs[*cdiv1.DataSource]{
NEW 301	CreateFunc: func(e event.TypedCreateEvent[*cdiv1.DataSource]) bool { return true },	×
NEW 302	DeleteFunc: func(e event.TypedDeleteEvent[*cdiv1.DataSource]) bool { return true },	×
NEW 303	UpdateFunc: func(e event.TypedUpdateEvent[*cdiv1.DataSource]) bool {	×
NEW 304	return !sameSourceSpec(e.ObjectOld, e.ObjectNew) \|\|	×
NEW 305	!sameConditions(e.ObjectOld, e.ObjectNew)	×
NEW 306	},	×
307	},
NEW 308	)); err != nil {	×
NEW 309	return err	×
UNCOV 310	}	×
311
312	if err := c.Watch(source.Kind(mgr.GetCache(), &cdiv1.DataVolume{},	×
NEW 313	handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, obj *cdiv1.DataVolume) []reconcile.Request {	×
NEW 314	return mapToDataSource(ctx, mgr, obj, log)	×
UNCOV 315	}),	×
316	predicate.TypedFuncs[*cdiv1.DataVolume]{
317	CreateFunc: func(e event.TypedCreateEvent[*cdiv1.DataVolume]) bool { return true },	×
318	DeleteFunc: func(e event.TypedDeleteEvent[*cdiv1.DataVolume]) bool { return true },	×
319	// Only DV status phase update is interesting to reconcile
320	UpdateFunc: func(e event.TypedUpdateEvent[*cdiv1.DataVolume]) bool {	×
321	return e.ObjectOld.Status.Phase != e.ObjectNew.Status.Phase \|\|	×
322	!reflect.DeepEqual(e.ObjectOld.Labels, e.ObjectNew.Labels)	×
323	},	×
324	},
325	)); err != nil {	×
326	return err	×
327	}	×
328
329	if err := c.Watch(source.Kind(mgr.GetCache(), &corev1.PersistentVolumeClaim{},	×
NEW 330	handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, obj *corev1.PersistentVolumeClaim) []reconcile.Request {	×
NEW 331	return mapToDataSource(ctx, mgr, obj, log)	×
UNCOV 332	}),	×
333	predicate.TypedFuncs[*corev1.PersistentVolumeClaim]{
334	CreateFunc: func(e event.TypedCreateEvent[*corev1.PersistentVolumeClaim]) bool { return true },	×
335	DeleteFunc: func(e event.TypedDeleteEvent[*corev1.PersistentVolumeClaim]) bool { return true },	×
336	UpdateFunc: func(e event.TypedUpdateEvent[*corev1.PersistentVolumeClaim]) bool {	×
337	return e.ObjectOld.Status.Phase != e.ObjectNew.Status.Phase \|\|	×
338	!reflect.DeepEqual(e.ObjectOld.Labels, e.ObjectNew.Labels)	×
339	},	×
340	},
341	)); err != nil {	×
342	return err	×
343	}	×
344
345	if err := mgr.GetClient().List(context.TODO(), &snapshotv1.VolumeSnapshotList{}); err != nil {	×
346	if meta.IsNoMatchError(err) {	×
347	// Back out if there's no point to attempt watch	×
348	return nil	×
349	}	×
350	if !cc.IsErrCacheNotStarted(err) {	×
351	return err	×
352	}	×
353	}
354	if err := c.Watch(source.Kind(mgr.GetCache(), &snapshotv1.VolumeSnapshot{},	×
NEW 355	handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, obj *snapshotv1.VolumeSnapshot) []reconcile.Request {	×
NEW 356	return mapToDataSource(ctx, mgr, obj, log)	×
UNCOV 357	}),	×
358	predicate.TypedFuncs[*snapshotv1.VolumeSnapshot]{
359	CreateFunc: func(e event.TypedCreateEvent[*snapshotv1.VolumeSnapshot]) bool { return true },	×
360	DeleteFunc: func(e event.TypedDeleteEvent[*snapshotv1.VolumeSnapshot]) bool { return true },	×
361	UpdateFunc: func(e event.TypedUpdateEvent[*snapshotv1.VolumeSnapshot]) bool {	×
362	return !reflect.DeepEqual(e.ObjectOld.Status, e.ObjectNew.Status) \|\|	×
363	!reflect.DeepEqual(e.ObjectOld.Labels, e.ObjectNew.Labels)	×
364	},	×
365	},
366	)); err != nil {	×
367	return err	×
368	}	×
369
370	return nil	×
371	}
372
NEW 373	func appendMatchingDataSourceRequests(ctx context.Context, mgr manager.Manager, indexingKey string, obj client.Object, reqs []reconcile.Request, log logr.Logger) []reconcile.Request {	×
NEW 374	var dataSources cdiv1.DataSourceList	×
NEW 375	matchingFields := client.MatchingFields{indexingKey: client.ObjectKeyFromObject(obj).String()}	×
NEW 376	if err := mgr.GetClient().List(ctx, &dataSources, matchingFields); err != nil {	×
NEW 377	log.Error(err, "Unable to list DataSources", "matchingFields", matchingFields)	×
NEW 378	return reqs	×
NEW 379	}	×
NEW 380	for _, ds := range dataSources.Items {	×
NEW 381	reqs = append(reqs, reconcile.Request{NamespacedName: types.NamespacedName{Namespace: ds.Namespace, Name: ds.Name}})	×
NEW 382	}	×
NEW 383	return reqs	×
384	}
385
NEW 386	func mapToDataSource(ctx context.Context, mgr manager.Manager, obj client.Object, log logr.Logger) []reconcile.Request {	×
NEW 387	reqs := appendMatchingDataSourceRequests(ctx, mgr, dataSourcePvcField, obj, nil, log)	×
NEW 388	return appendMatchingDataSourceRequests(ctx, mgr, dataSourceSnapshotField, obj, reqs, log)	×
NEW 389	}	×
390
391	func sameSourceSpec(objOld, objNew client.Object) bool {	×
392	dsOld, okOld := objOld.(*cdiv1.DataSource)	×
393	dsNew, okNew := objNew.(*cdiv1.DataSource)	×
394		×
395	if !okOld \|\| !okNew {	×
396	return false	×
397	}	×
398	if dsOld.Spec.Source.PVC != nil {	×
399	return reflect.DeepEqual(dsOld.Spec.Source.PVC, dsNew.Spec.Source.PVC)	×
400	}	×
401	if dsOld.Spec.Source.Snapshot != nil {	×
402	return reflect.DeepEqual(dsOld.Spec.Source.Snapshot, dsNew.Spec.Source.Snapshot)	×
403	}	×
404	if dsOld.Spec.Source.DataSource != nil {	×
405	return reflect.DeepEqual(dsOld.Spec.Source.DataSource, dsNew.Spec.Source.DataSource)	×
406	}	×
407
408	return false	×
409	}
410
411	func sameConditions(objOld, objNew client.Object) bool {	×
412	dsOld, okOld := objOld.(*cdiv1.DataSource)	×
413	dsNew, okNew := objNew.(*cdiv1.DataSource)	×
414		×
415	if !okOld \|\| !okNew {	×
416	return false	×
417	}	×
418
419	oldConditions := dsOld.Status.Conditions	×
420	newConditions := dsNew.Status.Conditions	×
421		×
422	if len(oldConditions) != len(newConditions) {	×
423	return false	×
424	}	×
425
426	condMap := make(map[cdiv1.DataSourceConditionType]cdiv1.DataSourceCondition, len(oldConditions))	×
427	for _, c := range oldConditions {	×
428	condMap[c.Type] = c	×
429	}	×
430
431	for _, c := range newConditions {	×
432	if oldC, ok := condMap[c.Type]; !ok \|\|	×
433	oldC.Reason != c.Reason \|\|	×
434	oldC.Message != c.Message \|\|	×
435	oldC.Status != c.Status {	×
436	return false	×
437	}	×
438	}
439
440	return true	×
441	}

kubevirt / containerized-data-importer / #5597

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