fix(vd,vi,cvi): add object ref watchers

Signed-off-by: Isteb4k <[email protected]>

images/virtualization-artifact/pkg/controller/cvi/cvi_reconciler.go

@@ -34,6 +34,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	"github.com/deckhouse/virtualization-controller/pkg/controller/service"
+	"github.com/deckhouse/virtualization-controller/pkg/controller/watchers"
 	virtv2 "github.com/deckhouse/virtualization/api/core/v1alpha2"
 )
 
@@ -143,6 +144,18 @@ func (r *Reconciler) SetupController(_ context.Context, mgr manager.Manager, ctr
 		return fmt.Errorf("error setting watch on VMs: %w", err)
 	}
 
+	cviFromVIEnqueuer := watchers.NewClusterVirtualImageRequestEnqueuer(mgr.GetClient(), &virtv2.VirtualImage{}, virtv2.ClusterVirtualImageObjectRefKindVirtualImage)
+	viWatcher := watchers.NewObjectRefWatcher(watchers.NewVirtualImageFilter(), cviFromVIEnqueuer)
+	if err := viWatcher.Run(mgr, ctr); err != nil {
+		return fmt.Errorf("error setting watch on VIs: %w", err)
+	}
+
+	cviFromCVIEnqueuer := watchers.NewClusterVirtualImageRequestEnqueuer(mgr.GetClient(), &virtv2.ClusterVirtualImage{}, virtv2.ClusterVirtualImageObjectRefKindClusterVirtualImage)
+	cviWatcher := watchers.NewObjectRefWatcher(watchers.NewClusterVirtualImageFilter(), cviFromCVIEnqueuer)
+	if err := cviWatcher.Run(mgr, ctr); err != nil {
+		return fmt.Errorf("error setting watch on CVIs: %w", err)
+	}
+
 	return nil
 }
 

images/virtualization-artifact/pkg/controller/vd/vd_reconciler.go

@@ -37,6 +37,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	"github.com/deckhouse/virtualization-controller/pkg/controller/service"
+	"github.com/deckhouse/virtualization-controller/pkg/controller/watchers"
 	virtv2 "github.com/deckhouse/virtualization/api/core/v1alpha2"
 )
 
@@ -206,6 +207,18 @@ func (r *Reconciler) SetupController(_ context.Context, mgr manager.Manager, ctr
 		return fmt.Errorf("error setting watch on VMs: %w", err)
 	}
 
+	vdFromVIEnqueuer := watchers.NewVirtualDiskRequestEnqueuer(mgr.GetClient(), &virtv2.VirtualImage{}, virtv2.VirtualDiskObjectRefKindVirtualImage)
+	viWatcher := watchers.NewObjectRefWatcher(watchers.NewVirtualImageFilter(), vdFromVIEnqueuer)
+	if err := viWatcher.Run(mgr, ctr); err != nil {
+		return fmt.Errorf("error setting watch on VIs: %w", err)
+	}
+
+	vdFromCVIEnqueuer := watchers.NewVirtualDiskRequestEnqueuer(mgr.GetClient(), &virtv2.ClusterVirtualImage{}, virtv2.VirtualDiskObjectRefKindClusterVirtualImage)
+	cviWatcher := watchers.NewObjectRefWatcher(watchers.NewClusterVirtualImageFilter(), vdFromCVIEnqueuer)
+	if err := cviWatcher.Run(mgr, ctr); err != nil {
+		return fmt.Errorf("error setting watch on CVIs: %w", err)
+	}
+
 	return nil
 }
 

images/virtualization-artifact/pkg/controller/vi/vi_reconciler.go

@@ -19,6 +19,7 @@ package vi
 import (
 	"context"
 	"errors"
+	"fmt"
 	"log/slog"
 	"time"
 
@@ -32,6 +33,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	"github.com/deckhouse/virtualization-controller/pkg/controller/service"
+	"github.com/deckhouse/virtualization-controller/pkg/controller/watchers"
 	virtv2 "github.com/deckhouse/virtualization/api/core/v1alpha2"
 )
 
@@ -120,6 +122,18 @@ func (r *Reconciler) SetupController(_ context.Context, mgr manager.Manager, ctr
 		return err
 	}
 
+	viFromVIEnqueuer := watchers.NewVirtualImageRequestEnqueuer(mgr.GetClient(), &virtv2.VirtualImage{}, virtv2.VirtualImageObjectRefKindVirtualImage)
+	viWatcher := watchers.NewObjectRefWatcher(watchers.NewVirtualImageFilter(), viFromVIEnqueuer)
+	if err := viWatcher.Run(mgr, ctr); err != nil {
+		return fmt.Errorf("error setting watch on VIs: %w", err)
+	}
+
+	viFromCVIEnqueuer := watchers.NewVirtualImageRequestEnqueuer(mgr.GetClient(), &virtv2.ClusterVirtualImage{}, virtv2.VirtualImageObjectRefKindClusterVirtualImage)
+	cviWatcher := watchers.NewObjectRefWatcher(watchers.NewClusterVirtualImageFilter(), viFromCVIEnqueuer)
+	if err := cviWatcher.Run(mgr, ctr); err != nil {
+		return fmt.Errorf("error setting watch on CVIs: %w", err)
+	}
+
 	return nil
 }
 

images/virtualization-artifact/pkg/controller/watchers/cvi_enqueuer.go

@@ -0,0 +1,88 @@
+/*
+Copyright 2024 Flant JSC
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package watchers
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	"github.com/deckhouse/virtualization-controller/pkg/controller/service"
+	virtv2 "github.com/deckhouse/virtualization/api/core/v1alpha2"
+	"github.com/deckhouse/virtualization/api/core/v1alpha2/cvicondition"
+)
+
+type ClusterVirtualImageRequestEnqueuer struct {
+	enqueueFromObj  client.Object
+	enqueueFromKind virtv2.ClusterVirtualImageObjectRefKind
+	client          client.Client
+	logger          *slog.Logger
+}
+
+func NewClusterVirtualImageRequestEnqueuer(client client.Client, enqueueFromObj client.Object, enqueueFromKind virtv2.ClusterVirtualImageObjectRefKind) *ClusterVirtualImageRequestEnqueuer {
+	return &ClusterVirtualImageRequestEnqueuer{
+		enqueueFromObj:  enqueueFromObj,
+		enqueueFromKind: enqueueFromKind,
+		client:          client,
+		logger:          slog.Default().With("enqueuer", "cvi"),
+	}
+}
+
+func (w ClusterVirtualImageRequestEnqueuer) GetEnqueueFrom() client.Object {
+	return w.enqueueFromObj
+}
+
+func (w ClusterVirtualImageRequestEnqueuer) EnqueueRequests(ctx context.Context, obj client.Object) (requests []reconcile.Request) {
+	var cvis virtv2.ClusterVirtualImageList
+	err := w.client.List(ctx, &cvis)
+	if err != nil {
+		w.logger.Error(fmt.Sprintf("failed to list cvi: %s", err))
+		return
+	}
+
+	for _, cvi := range cvis.Items {
+		dsReady, _ := service.GetCondition(cvicondition.DatasourceReadyType, cvi.Status.Conditions)
+		if dsReady.Status == metav1.ConditionTrue {
+			continue
+		}
+
+		if cvi.Spec.DataSource.Type != virtv2.DataSourceTypeObjectRef {
+			continue
+		}
+
+		ref := cvi.Spec.DataSource.ObjectRef
+
+		if ref == nil || ref.Kind != w.enqueueFromKind {
+			continue
+		}
+
+		if ref.Name == obj.GetName() {
+			requests = append(requests, reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name: cvi.Name,
+				},
+			})
+		}
+	}
+
+	return
+}

images/virtualization-artifact/pkg/controller/watchers/cvi_filter.go

@@ -0,0 +1,57 @@
+/*
+Copyright 2024 Flant JSC
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package watchers
+
+import (
+	"fmt"
+	"log/slog"
+
+	"sigs.k8s.io/controller-runtime/pkg/event"
+
+	virtv2 "github.com/deckhouse/virtualization/api/core/v1alpha2"
+)
+
+type ClusterVirtualImageFilter struct {
+	logger *slog.Logger
+}
+
+func NewClusterVirtualImageFilter() *ClusterVirtualImageFilter {
+	return &ClusterVirtualImageFilter{
+		logger: slog.Default().With("filter", "cvi"),
+	}
+}
+
+func (f ClusterVirtualImageFilter) FilterUpdateEvents(e event.UpdateEvent) bool {
+	oldCVI, ok := e.ObjectOld.(*virtv2.ClusterVirtualImage)
+	if !ok {
+		f.logger.Error(fmt.Sprintf("expected an old ClusterVirtualImage but got a %T", e.ObjectOld))
+		return false
+	}
+
+	newCVI, ok := e.ObjectNew.(*virtv2.ClusterVirtualImage)
+	if !ok {
+		f.logger.Error(fmt.Sprintf("expected a new ClusterVirtualImage but got a %T", e.ObjectNew))
+		return false
+	}
+
+	if newCVI.Generation != newCVI.Status.ObservedGeneration {
+		return false
+	}
+
+	// Triggered only if the resource phase changed to Ready.
+	return oldCVI.Status.Phase != newCVI.Status.Phase && newCVI.Status.Phase == virtv2.ImageReady
+}

images/virtualization-artifact/pkg/controller/watchers/object_ref_watcher.go

@@ -0,0 +1,69 @@
+/*
+Copyright 2024 Flant JSC
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package watchers
+
+import (
+	"context"
+	"log/slog"
+
+	"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"
+)
+
+type ObjectRefWatcher struct {
+	filter   UpdateEventsFilter
+	enqueuer RequestEnqueuer
+	logger   *slog.Logger
+}
+
+type RequestEnqueuer interface {
+	EnqueueRequests(context.Context, client.Object) []reconcile.Request
+	GetEnqueueFrom() client.Object
+}
+
+type UpdateEventsFilter interface {
+	FilterUpdateEvents(event.UpdateEvent) bool
+}
+
+func NewObjectRefWatcher(
+	filter UpdateEventsFilter,
+	enqueuer RequestEnqueuer,
+) *ObjectRefWatcher {
+	return &ObjectRefWatcher{
+		filter:   filter,
+		enqueuer: enqueuer,
+		logger:   slog.Default().With("watcher", "cvi"),
+	}
+}
+
+func (w ObjectRefWatcher) Run(mgr manager.Manager, ctr controller.Controller) error {
+	return ctr.Watch(
+		source.Kind(mgr.GetCache(), w.enqueuer.GetEnqueueFrom()),
+		handler.EnqueueRequestsFromMapFunc(w.enqueuer.EnqueueRequests),
+		predicate.Funcs{
+			CreateFunc: func(e event.CreateEvent) bool { return false },
+			DeleteFunc: func(e event.DeleteEvent) bool { return false },
+			UpdateFunc: w.filter.FilterUpdateEvents,
+		},
+	)
+}

images/virtualization-artifact/pkg/controller/watchers/vd_enqueuer.go

@@ -0,0 +1,89 @@
+/*
+Copyright 2024 Flant JSC
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package watchers
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	"github.com/deckhouse/virtualization-controller/pkg/controller/service"
+	virtv2 "github.com/deckhouse/virtualization/api/core/v1alpha2"
+	"github.com/deckhouse/virtualization/api/core/v1alpha2/vdcondition"
+)
+
+type VirtualDiskRequestEnqueuer struct {
+	enqueueFromObj  client.Object
+	enqueueFromKind virtv2.VirtualDiskObjectRefKind
+	client          client.Client
+	logger          *slog.Logger
+}
+
+func NewVirtualDiskRequestEnqueuer(client client.Client, enqueueFromObj client.Object, enqueueFromKind virtv2.VirtualDiskObjectRefKind) *VirtualDiskRequestEnqueuer {
+	return &VirtualDiskRequestEnqueuer{
+		enqueueFromObj:  enqueueFromObj,
+		enqueueFromKind: enqueueFromKind,
+		client:          client,
+		logger:          slog.Default().With("enqueuer", "vd"),
+	}
+}
+
+func (w VirtualDiskRequestEnqueuer) GetEnqueueFrom() client.Object {
+	return w.enqueueFromObj
+}
+
+func (w VirtualDiskRequestEnqueuer) EnqueueRequests(ctx context.Context, obj client.Object) (requests []reconcile.Request) {
+	var vds virtv2.VirtualDiskList
+	err := w.client.List(ctx, &vds)
+	if err != nil {
+		w.logger.Error(fmt.Sprintf("failed to list vd: %s", err))
+		return
+	}
+
+	for _, vd := range vds.Items {
+		dsReady, _ := service.GetCondition(vdcondition.DatasourceReadyType, vd.Status.Conditions)
+		if dsReady.Status == metav1.ConditionTrue {
+			continue
+		}
+
+		if vd.Spec.DataSource == nil || vd.Spec.DataSource.Type != virtv2.DataSourceTypeObjectRef {
+			continue
+		}
+
+		ref := vd.Spec.DataSource.ObjectRef
+
+		if ref == nil || ref.Kind != w.enqueueFromKind {
+			continue
+		}
+
+		if ref.Name == obj.GetName() {
+			requests = append(requests, reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Namespace: vd.Namespace,
+					Name:      vd.Name,
+				},
+			})
+		}
+	}
+
+	return
+}