fix(cvi): watch vd snapshot

api/core/v1alpha2/cvicondition/condition.go

@@ -60,6 +60,8 @@ const (
 	VirtualDiskNotReadyForUse DatasourceReadyReason = "VirtualDiskNotReadyForUse"
 	// VirtualDiskAttachedToVirtualMachine indicates that the `VirtualDisk` attached to `VirtualMachine`.
 	VirtualDiskAttachedToVirtualMachine DatasourceReadyReason = "VirtualDiskAttachedToVirtualMachine"
+	// VirtualDiskSnapshotNotReady indicates that the `VirtualDiskSnapshot` datasource is not ready, which prevents the import process from starting.
+	VirtualDiskSnapshotNotReady DatasourceReadyReason = "VirtualDiskSnapshotNotReady"
 
 	// WaitForUserUpload indicates that the `ClusterVirtualImage` is waiting for the user to upload a datasource for the import process to continue.
 	WaitForUserUpload ReadyReason = "WaitForUserUpload"

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

@@ -32,12 +32,17 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	"github.com/deckhouse/virtualization-controller/pkg/controller/conditions"
+	"github.com/deckhouse/virtualization-controller/pkg/controller/cvi/internal/watcher"
 	"github.com/deckhouse/virtualization-controller/pkg/controller/reconciler"
 	"github.com/deckhouse/virtualization-controller/pkg/controller/watchers"
 	virtv2 "github.com/deckhouse/virtualization/api/core/v1alpha2"
 	"github.com/deckhouse/virtualization/api/core/v1alpha2/vdcondition"
 )
 
+type Watcher interface {
+	Watch(mgr manager.Manager, ctr controller.Controller) error
+}
+
 type Handler interface {
 	Handle(ctx context.Context, cvi *virtv2.ClusterVirtualImage) (reconcile.Result, error)
 }
@@ -157,6 +162,15 @@ func (r *Reconciler) SetupController(_ context.Context, mgr manager.Manager, ctr
 		return fmt.Errorf("error setting watch on CVIs: %w", err)
 	}
 
+	for _, w := range []Watcher{
+		watcher.NewVirtualDiskSnapshotWatcher(mgr.GetClient()),
+	} {
+		err := w.Watch(mgr, ctr)
+		if err != nil {
+			return fmt.Errorf("error setting watcher: %w", err)
+		}
+	}
+
 	return nil
 }
 

images/virtualization-artifact/pkg/controller/cvi/internal/datasource_ready.go

@@ -101,6 +101,12 @@ func (h DatasourceReadyHandler) Handle(ctx context.Context, cvi *virtv2.ClusterV
 			Reason(cvicondition.VirtualDiskAttachedToVirtualMachine).
 			Message(service.CapitalizeFirstLetter(err.Error() + "."))
 		return reconcile.Result{}, nil
+	case errors.As(err, &source.VirtualDiskSnapshotNotReadyError{}):
+		cb.
+			Status(metav1.ConditionFalse).
+			Reason(cvicondition.VirtualDiskSnapshotNotReady).
+			Message(service.CapitalizeFirstLetter(err.Error()))
+		return reconcile.Result{}, nil
 	default:
 		return reconcile.Result{}, err
 	}

images/virtualization-artifact/pkg/controller/cvi/internal/watcher/vdsnapshot_watcher.go

@@ -0,0 +1,126 @@
+/*
+Copyright 2025 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 watcher
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/types"
+	"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"
+
+	"github.com/deckhouse/deckhouse/pkg/log"
+	"github.com/deckhouse/virtualization-controller/pkg/controller/indexer"
+	virtv2 "github.com/deckhouse/virtualization/api/core/v1alpha2"
+)
+
+type VirtualDiskSnapshotWatcher struct {
+	logger *log.Logger
+	client client.Client
+}
+
+func NewVirtualDiskSnapshotWatcher(client client.Client) *VirtualDiskSnapshotWatcher {
+	return &VirtualDiskSnapshotWatcher{
+		logger: log.Default().With("watcher", strings.ToLower(virtv2.VirtualDiskSnapshotKind)),
+		client: client,
+	}
+}
+
+func (w VirtualDiskSnapshotWatcher) Watch(mgr manager.Manager, ctr controller.Controller) error {
+	return ctr.Watch(
+		source.Kind(mgr.GetCache(), &virtv2.VirtualDiskSnapshot{}),
+		handler.EnqueueRequestsFromMapFunc(w.enqueueRequests),
+		predicate.Funcs{
+			CreateFunc: func(e event.CreateEvent) bool { return true },
+			DeleteFunc: func(e event.DeleteEvent) bool { return true },
+			UpdateFunc: w.filterUpdateEvents,
+		},
+	)
+}
+
+func (w VirtualDiskSnapshotWatcher) enqueueRequests(ctx context.Context, obj client.Object) (requests []reconcile.Request) {
+	vdSnapshot, ok := obj.(*virtv2.VirtualDiskSnapshot)
+	if !ok {
+		w.logger.Error(fmt.Sprintf("expected a VirtualDiskSnapshot but got a %T", obj))
+		return
+	}
+
+	var cvis virtv2.ClusterVirtualImageList
+	err := w.client.List(ctx, &cvis, &client.ListOptions{
+		FieldSelector: fields.OneTermEqualSelector(indexer.IndexFieldCVIByVDSnapshot, types.NamespacedName{
+			Namespace: vdSnapshot.Namespace,
+			Name:      vdSnapshot.Name,
+		}.String()),
+	})
+	if err != nil {
+		w.logger.Error(fmt.Sprintf("failed to list cluster virtual images: %s", err))
+		return
+	}
+
+	for _, cvi := range cvis.Items {
+		if !isSnapshotDataSource(cvi.Spec.DataSource, vdSnapshot) {
+			w.logger.Error("cvi list by vd snapshot returns unexpected resources, please report a bug")
+			continue
+		}
+
+		requests = append(requests, reconcile.Request{
+			NamespacedName: types.NamespacedName{
+				Name: cvi.Name,
+			},
+		})
+	}
+
+	return
+}
+
+func (w VirtualDiskSnapshotWatcher) filterUpdateEvents(e event.UpdateEvent) bool {
+	oldVDSnapshot, ok := e.ObjectOld.(*virtv2.VirtualDiskSnapshot)
+	if !ok {
+		w.logger.Error(fmt.Sprintf("expected an old VirtualDiskSnapshot but got a %T", e.ObjectOld))
+		return false
+	}
+
+	newVDSnapshot, ok := e.ObjectNew.(*virtv2.VirtualDiskSnapshot)
+	if !ok {
+		w.logger.Error(fmt.Sprintf("expected a new VirtualDiskSnapshot but got a %T", e.ObjectNew))
+		return false
+	}
+
+	return oldVDSnapshot.Status.Phase != newVDSnapshot.Status.Phase
+}
+
+func isSnapshotDataSource(ds virtv2.ClusterVirtualImageDataSource, vdSnapshot metav1.Object) bool {
+	if ds.Type != virtv2.DataSourceTypeObjectRef {
+		return false
+	}
+
+	if ds.ObjectRef == nil || ds.ObjectRef.Kind != virtv2.ClusterVirtualImageObjectRefKindVirtualDiskSnapshot {
+		return false
+	}
+
+	return ds.ObjectRef.Name == vdSnapshot.GetName() && ds.ObjectRef.Namespace == vdSnapshot.GetNamespace()
+}

images/virtualization-artifact/pkg/controller/indexer/cvi_indexer.go

@@ -0,0 +1,51 @@
+/*
+Copyright 2025 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 indexer
+
+import (
+	"context"
+
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+
+	virtv2 "github.com/deckhouse/virtualization/api/core/v1alpha2"
+)
+
+func IndexCVIByVDSnapshot(ctx context.Context, mgr manager.Manager) error {
+	return mgr.GetFieldIndexer().IndexField(ctx, &virtv2.ClusterVirtualImage{}, IndexFieldCVIByVDSnapshot, func(object client.Object) []string {
+		cvi, ok := object.(*virtv2.ClusterVirtualImage)
+		if !ok || cvi == nil {
+			return nil
+		}
+
+		if cvi.Spec.DataSource.Type != virtv2.DataSourceTypeObjectRef {
+			return nil
+		}
+
+		if cvi.Spec.DataSource.ObjectRef == nil || cvi.Spec.DataSource.ObjectRef.Kind != virtv2.ClusterVirtualImageObjectRefKindVirtualDiskSnapshot {
+			return nil
+		}
+
+		key := types.NamespacedName{
+			Namespace: cvi.Spec.DataSource.ObjectRef.Namespace,
+			Name:      cvi.Spec.DataSource.ObjectRef.Name,
+		}
+
+		return []string{key.String()}
+	})
+}

images/virtualization-artifact/pkg/controller/indexer/indexer.go

@@ -37,8 +37,9 @@ const (
 
 	IndexFieldVMIPLeaseByVMIP = "spec.virtualMachineIPAddressRef.Name"
 
-	IndexFieldVDByVDSnapshot = "vd,spec.DataSource.ObjectRef.Name,.Kind=VirtualDiskSnapshot"
-	IndexFieldVIByVDSnapshot = "vi,spec.DataSource.ObjectRef.Name,.Kind=VirtualDiskSnapshot"
+	IndexFieldVDByVDSnapshot  = "vd,spec.DataSource.ObjectRef.Name,.Kind=VirtualDiskSnapshot"
+	IndexFieldVIByVDSnapshot  = "vi,spec.DataSource.ObjectRef.Name,.Kind=VirtualDiskSnapshot"
+	IndexFieldCVIByVDSnapshot = "cvi,spec.DataSource.ObjectRef.Name,.Kind=VirtualDiskSnapshot"
 
 	IndexFieldVDByStorageClass = "vd.spec.PersistentVolumeClaim.StorageClass"
 	IndexFieldVIByStorageClass = "vi.spec.PersistentVolumeClaim.StorageClass"
@@ -71,6 +72,7 @@ func IndexALL(ctx context.Context, mgr manager.Manager) error {
 		IndexVDByStorageClass,
 		IndexVIByVDSnapshot,
 		IndexVIByStorageClass,
+		IndexCVIByVDSnapshot,
 		IndexVMIPByAddress,
 		IndexVMBDAByVM,
 	} {

images/virtualization-artifact/pkg/controller/vdsnapshot/internal/watcher/kvvmi_watcher.go → images/virtualization-artifact/pkg/controller/vdsnapshot/internal/watcher/vm_watcher.go

@@ -105,20 +105,31 @@ func (w VirtualMachineWatcher) enqueueRequests(ctx context.Context, obj client.O
 }
 
 func (w VirtualMachineWatcher) filterUpdateEvents(e event.UpdateEvent) bool {
-	oldKVVMI, ok := e.ObjectOld.(*virtv2.VirtualMachine)
+	oldVM, ok := e.ObjectOld.(*virtv2.VirtualMachine)
 	if !ok {
 		slog.Default().Error(fmt.Sprintf("expected an old VirtualMachine but got a %T", e.ObjectOld))
 		return false
 	}
 
-	newKVVMI, ok := e.ObjectNew.(*virtv2.VirtualMachine)
+	newVM, ok := e.ObjectNew.(*virtv2.VirtualMachine)
 	if !ok {
 		slog.Default().Error(fmt.Sprintf("expected a new VirtualMachine but got a %T", e.ObjectNew))
 		return false
 	}
 
-	oldFSFrozen, _ := conditions.GetCondition(vmcondition.TypeFilesystemFrozen, oldKVVMI.Status.Conditions)
-	newFSFrozen, _ := conditions.GetCondition(vmcondition.TypeFilesystemFrozen, newKVVMI.Status.Conditions)
+	if oldVM.Status.Phase != newVM.Status.Phase {
+		return true
+	}
+
+	oldFSFrozen, _ := conditions.GetCondition(vmcondition.TypeFilesystemFrozen, oldVM.Status.Conditions)
+	newFSFrozen, _ := conditions.GetCondition(vmcondition.TypeFilesystemFrozen, newVM.Status.Conditions)
+
+	if oldFSFrozen.Status != newFSFrozen.Status {
+		return true
+	}
+
+	oldAgentReady, _ := conditions.GetCondition(vmcondition.TypeAgentReady, oldVM.Status.Conditions)
+	newAgentReady, _ := conditions.GetCondition(vmcondition.TypeAgentReady, newVM.Status.Conditions)
 
-	return oldFSFrozen.Status != newFSFrozen.Status
+	return oldAgentReady.Status != newAgentReady.Status
 }

images/virtualization-artifact/pkg/controller/vi/internal/watcher/vdsnapshot_watcher.go

@@ -80,7 +80,7 @@ func (w VirtualDiskSnapshotWatcher) enqueueRequests(ctx context.Context, obj cli
 
 	for _, vi := range vis.Items {
 		if !isSnapshotDataSource(vi.Spec.DataSource, vdSnapshot.Name) {
-			w.logger.Error("vd list by vd snapshot returns unexpected resources, please report a bug")
+			w.logger.Error("vi list by vd snapshot returns unexpected resources, please report a bug")
 			continue
 		}