#5557

Committed 01 Sep 2025 01:14PM UTC coverage: 59.225% (+0.02%) from 59.201%

Build # #5557

Build Type

Pull #3889

travis-ci

Committed by

Acedus

Commit Message

datasource-controller: decompose addDataSourceControllerWatches

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>

Pull Request Pull Request #3889: datasource-controller: decompose addDataSourceControllerWatches

Run Details

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

5 existing lines in 1 file now uncovered.

17173 of 28996 relevant lines covered (59.23%)

0.65 hits per line

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

33.33

/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 {
                        return []string{client.ObjectKey(*pvc).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 {
                        return []string{client.ObjectKey(*snapshot).String()}
                }
                return nil
        }); err != nil {
                return err
        }

        if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &cdiv1.DataSource{}, dataSourceDataSourceField, func(obj client.Object) []string {
                ds := obj.(*cdiv1.DataSource)
                if sourceDS := ds.Spec.Source.DataSource; sourceDS != nil {
                        return []string{client.ObjectKey(*sourceDS).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 {	×
NEW 256	return []string{client.ObjectKey(*pvc).String()}	×
257	}	×
258	return nil	×
259	}); err != nil {	×
260	return err	×
261	}	×
262
263	if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &cdiv1.DataSource{}, dataSourceSnapshotField, func(obj client.Object) []string {	×
264	if snapshot := obj.(*cdiv1.DataSource).Spec.Source.Snapshot; snapshot != nil {	×
NEW 265	return []string{client.ObjectKey(*snapshot).String()}	×
266	}	×
267	return nil	×
268	}); err != nil {	×
269	return err	×
270	}	×
271
272	if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &cdiv1.DataSource{}, dataSourceDataSourceField, func(obj client.Object) []string {	×
273	ds := obj.(*cdiv1.DataSource)	×
274	if sourceDS := ds.Spec.Source.DataSource; sourceDS != nil {	×
NEW 275	return []string{client.ObjectKey(*sourceDS).String()}	×
276	}	×
277	return nil	×
278	}); err != nil {	×
279	return err	×
280	}	×
281
NEW 282	return nil	×
283	}
284
NEW 285	func setupWatches(mgr manager.Manager, c controller.Controller, log logr.Logger) error {	×
NEW 286	if err := c.Watch(source.Kind(mgr.GetCache(), &cdiv1.DataSource{},	×
NEW 287	handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, obj *cdiv1.DataSource) []reconcile.Request {	×
NEW 288	reqs := []reconcile.Request{	×
NEW 289	{	×
NEW 290	NamespacedName: types.NamespacedName{	×
NEW 291	Name: obj.Name,	×
NEW 292	Namespace: obj.Namespace,	×
NEW 293	},	×
NEW 294	},	×
NEW 295	}	×
NEW 296	return appendMatchingDataSourceRequests(ctx, mgr, dataSourceDataSourceField, obj, reqs, log)	×
NEW 297	}),	×
298	predicate.TypedFuncs[*cdiv1.DataSource]{
NEW 299	CreateFunc: func(e event.TypedCreateEvent[*cdiv1.DataSource]) bool { return true },	×
NEW 300	DeleteFunc: func(e event.TypedDeleteEvent[*cdiv1.DataSource]) bool { return true },	×
NEW 301	UpdateFunc: func(e event.TypedUpdateEvent[*cdiv1.DataSource]) bool {	×
NEW 302	return !sameSourceSpec(e.ObjectOld, e.ObjectNew) \|\|	×
NEW 303	!sameConditions(e.ObjectOld, e.ObjectNew)	×
NEW 304	},	×
305	},
NEW 306	)); err != nil {	×
NEW 307	return err	×
UNCOV 308	}	×
309
310	if err := c.Watch(source.Kind(mgr.GetCache(), &cdiv1.DataVolume{},	×
NEW 311	handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, obj *cdiv1.DataVolume) []reconcile.Request {	×
NEW 312	return mapToDataSource(ctx, mgr, obj, log)	×
UNCOV 313	}),	×
314	predicate.TypedFuncs[*cdiv1.DataVolume]{
315	CreateFunc: func(e event.TypedCreateEvent[*cdiv1.DataVolume]) bool { return true },	×
316	DeleteFunc: func(e event.TypedDeleteEvent[*cdiv1.DataVolume]) bool { return true },	×
317	// Only DV status phase update is interesting to reconcile
318	UpdateFunc: func(e event.TypedUpdateEvent[*cdiv1.DataVolume]) bool {	×
319	return e.ObjectOld.Status.Phase != e.ObjectNew.Status.Phase \|\|	×
320	!reflect.DeepEqual(e.ObjectOld.Labels, e.ObjectNew.Labels)	×
321	},	×
322	},
323	)); err != nil {	×
324	return err	×
325	}	×
326
327	if err := c.Watch(source.Kind(mgr.GetCache(), &corev1.PersistentVolumeClaim{},	×
NEW 328	handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, obj *corev1.PersistentVolumeClaim) []reconcile.Request {	×
NEW 329	return mapToDataSource(ctx, mgr, obj, log)	×
UNCOV 330	}),	×
331	predicate.TypedFuncs[*corev1.PersistentVolumeClaim]{
332	CreateFunc: func(e event.TypedCreateEvent[*corev1.PersistentVolumeClaim]) bool { return true },	×
333	DeleteFunc: func(e event.TypedDeleteEvent[*corev1.PersistentVolumeClaim]) bool { return true },	×
334	UpdateFunc: func(e event.TypedUpdateEvent[*corev1.PersistentVolumeClaim]) bool {	×
335	return e.ObjectOld.Status.Phase != e.ObjectNew.Status.Phase \|\|	×
336	!reflect.DeepEqual(e.ObjectOld.Labels, e.ObjectNew.Labels)	×
337	},	×
338	},
339	)); err != nil {	×
340	return err	×
341	}	×
342
343	if err := mgr.GetClient().List(context.TODO(), &snapshotv1.VolumeSnapshotList{}); err != nil {	×
344	if meta.IsNoMatchError(err) {	×
345	// Back out if there's no point to attempt watch	×
346	return nil	×
347	}	×
348	if !cc.IsErrCacheNotStarted(err) {	×
349	return err	×
350	}	×
351	}
352	if err := c.Watch(source.Kind(mgr.GetCache(), &snapshotv1.VolumeSnapshot{},	×
NEW 353	handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, obj *snapshotv1.VolumeSnapshot) []reconcile.Request {	×
NEW 354	return mapToDataSource(ctx, mgr, obj, log)	×
UNCOV 355	}),	×
356	predicate.TypedFuncs[*snapshotv1.VolumeSnapshot]{
357	CreateFunc: func(e event.TypedCreateEvent[*snapshotv1.VolumeSnapshot]) bool { return true },	×
358	DeleteFunc: func(e event.TypedDeleteEvent[*snapshotv1.VolumeSnapshot]) bool { return true },	×
359	UpdateFunc: func(e event.TypedUpdateEvent[*snapshotv1.VolumeSnapshot]) bool {	×
360	return !reflect.DeepEqual(e.ObjectOld.Status, e.ObjectNew.Status) \|\|	×
361	!reflect.DeepEqual(e.ObjectOld.Labels, e.ObjectNew.Labels)	×
362	},	×
363	},
364	)); err != nil {	×
365	return err	×
366	}	×
367
368	return nil	×
369	}
370
NEW 371	func appendMatchingDataSourceRequests(ctx context.Context, mgr manager.Manager, indexingKey string, obj client.Object, reqs []reconcile.Request, log logr.Logger) []reconcile.Request {	×
NEW 372	var dataSources cdiv1.DataSourceList	×
NEW 373	matchingFields := client.MatchingFields{indexingKey: client.ObjectKeyFromObject(obj).String()}	×
NEW 374	if err := mgr.GetClient().List(ctx, &dataSources, matchingFields); err != nil {	×
NEW 375	log.Error(err, "Unable to list DataSources", "matchingFields", matchingFields)	×
NEW 376	return reqs	×
NEW 377	}	×
NEW 378	for _, ds := range dataSources.Items {	×
NEW 379	reqs = append(reqs, reconcile.Request{NamespacedName: types.NamespacedName{Namespace: ds.Namespace, Name: ds.Name}})	×
NEW 380	}	×
NEW 381	return reqs	×
382	}
383
NEW 384	func mapToDataSource(ctx context.Context, mgr manager.Manager, obj client.Object, log logr.Logger) []reconcile.Request {	×
NEW 385	reqs := appendMatchingDataSourceRequests(ctx, mgr, dataSourcePvcField, obj, nil, log)	×
NEW 386	return appendMatchingDataSourceRequests(ctx, mgr, dataSourceSnapshotField, obj, reqs, log)	×
NEW 387	}	×
388
389	func sameSourceSpec(objOld, objNew client.Object) bool {	×
390	dsOld, okOld := objOld.(*cdiv1.DataSource)	×
391	dsNew, okNew := objNew.(*cdiv1.DataSource)	×
392		×
393	if !okOld \|\| !okNew {	×
394	return false	×
395	}	×
396	if dsOld.Spec.Source.PVC != nil {	×
397	return reflect.DeepEqual(dsOld.Spec.Source.PVC, dsNew.Spec.Source.PVC)	×
398	}	×
399	if dsOld.Spec.Source.Snapshot != nil {	×
400	return reflect.DeepEqual(dsOld.Spec.Source.Snapshot, dsNew.Spec.Source.Snapshot)	×
401	}	×
402	if dsOld.Spec.Source.DataSource != nil {	×
403	return reflect.DeepEqual(dsOld.Spec.Source.DataSource, dsNew.Spec.Source.DataSource)	×
404	}	×
405
406	return false	×
407	}
408
409	func sameConditions(objOld, objNew client.Object) bool {	×
410	dsOld, okOld := objOld.(*cdiv1.DataSource)	×
411	dsNew, okNew := objNew.(*cdiv1.DataSource)	×
412		×
413	if !okOld \|\| !okNew {	×
414	return false	×
415	}	×
416
417	oldConditions := dsOld.Status.Conditions	×
418	newConditions := dsNew.Status.Conditions	×
419		×
420	if len(oldConditions) != len(newConditions) {	×
421	return false	×
422	}	×
423
424	condMap := make(map[cdiv1.DataSourceConditionType]cdiv1.DataSourceCondition, len(oldConditions))	×
425	for _, c := range oldConditions {	×
426	condMap[c.Type] = c	×
427	}	×
428
429	for _, c := range newConditions {	×
430	if oldC, ok := condMap[c.Type]; !ok \|\|	×
431	oldC.Reason != c.Reason \|\|	×
432	oldC.Message != c.Message \|\|	×
433	oldC.Status != c.Status {	×
434	return false	×
435	}	×
436	}
437
438	return true	×
439	}

kubevirt / containerized-data-importer / #5557

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