fix csi

.gitignore

@@ -36,4 +36,5 @@ __pycache__/
 # dev
 hack.sh
 **/Dockerfile-dev
+images/**/Makefile
 .secret

images/csi-driver/cmd/main.go

@@ -33,7 +33,7 @@ import (
 	"github.com/deckhouse/sds-common-lib/kubeclient"
 	snc "github.com/deckhouse/sds-node-configurator/api/v1alpha1"
 	"github.com/deckhouse/sds-replicated-volume/api/v1alpha1"
-	v1alpha2 "github.com/deckhouse/sds-replicated-volume/api/v1alpha2old"
+	"github.com/deckhouse/sds-replicated-volume/api/v1alpha2"
 	"github.com/deckhouse/sds-replicated-volume/images/csi-driver/config"
 	"github.com/deckhouse/sds-replicated-volume/images/csi-driver/driver"
 	"github.com/deckhouse/sds-replicated-volume/lib/go/common/logger"

images/csi-driver/driver/controller.go

@@ -30,7 +30,7 @@ import (
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 
-	v1alpha2 "github.com/deckhouse/sds-replicated-volume/api/v1alpha2old"
+	"github.com/deckhouse/sds-replicated-volume/api/v1alpha2"
 	"github.com/deckhouse/sds-replicated-volume/images/csi-driver/internal"
 	"github.com/deckhouse/sds-replicated-volume/images/csi-driver/pkg/utils"
 )
@@ -449,7 +449,16 @@ func (d *Driver) ControllerExpandVolume(ctx context.Context, request *csi.Contro
 		d.log.Error(err, fmt.Sprintf("[ControllerExpandVolume][traceID:%s][volumeID:%s] error WaitForReplicatedVolumeReady", traceID, volumeID))
 		return nil, err
 	}
-	d.log.Info(fmt.Sprintf("[ControllerExpandVolume][traceID:%s][volumeID:%s] finish resize ReplicatedVolume, attempt counter = %d", traceID, volumeID, attemptCounter))
+	d.log.Info(fmt.Sprintf("[ControllerExpandVolume][traceID:%s][volumeID:%s] finish wait for ready, attempt counter = %d", traceID, volumeID, attemptCounter))
+
+	// Wait for actualSize to match requested size (within delta)
+	d.log.Info(fmt.Sprintf("[ControllerExpandVolume][traceID:%s][volumeID:%s] waiting for actualSize to match requested size", traceID, volumeID))
+	attemptCounter, err = utils.WaitForActualSize(ctx, d.cl, d.log, traceID, volumeID, *requestCapacity, resizeDelta)
+	if err != nil {
+		d.log.Error(err, fmt.Sprintf("[ControllerExpandVolume][traceID:%s][volumeID:%s] error WaitForActualSize", traceID, volumeID))
+		return nil, err
+	}
+	d.log.Info(fmt.Sprintf("[ControllerExpandVolume][traceID:%s][volumeID:%s] finish wait for actualSize, attempt counter = %d", traceID, volumeID, attemptCounter))
 
 	d.log.Info(fmt.Sprintf("[ControllerExpandVolume][traceID:%s][volumeID:%s] Volume expanded successfully", traceID, volumeID))
 

images/csi-driver/driver/controller_publish_test.go

@@ -32,7 +32,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
-	v1alpha2 "github.com/deckhouse/sds-replicated-volume/api/v1alpha2old"
+	"github.com/deckhouse/sds-replicated-volume/api/v1alpha2"
 	"github.com/deckhouse/sds-replicated-volume/images/csi-driver/internal"
 	"github.com/deckhouse/sds-replicated-volume/lib/go/common/logger"
 )

images/csi-driver/driver/controller_test.go

@@ -33,7 +33,7 @@ import (
 
 	snc "github.com/deckhouse/sds-node-configurator/api/v1alpha1"
 	srv "github.com/deckhouse/sds-replicated-volume/api/v1alpha1"
-	v1alpha2 "github.com/deckhouse/sds-replicated-volume/api/v1alpha2old"
+	"github.com/deckhouse/sds-replicated-volume/api/v1alpha2"
 	"github.com/deckhouse/sds-replicated-volume/images/csi-driver/internal"
 	"github.com/deckhouse/sds-replicated-volume/lib/go/common/logger"
 )
@@ -590,6 +590,13 @@ var _ = Describe("ControllerExpandVolume", func() {
 					},
 				}
 				Expect(cl.Update(ctx, updatedRV)).To(Succeed())
+
+				// Update actualSize after Ready to simulate controller updating actualSize
+				time.Sleep(200 * time.Millisecond)
+				updatedRV2 := &v1alpha2.ReplicatedVolume{}
+				Expect(cl.Get(ctx, client.ObjectKey{Name: volumeID}, updatedRV2)).To(Succeed())
+				updatedRV2.Status.ActualSize = resource.MustParse("2Gi")
+				Expect(cl.Update(ctx, updatedRV2)).To(Succeed())
 			}()
 
 			timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
@@ -622,6 +629,9 @@ var _ = Describe("ControllerExpandVolume", func() {
 			updatedRV := &v1alpha2.ReplicatedVolume{}
 			Expect(cl.Get(ctx, client.ObjectKey{Name: volumeID}, updatedRV)).To(Succeed())
 			Expect(updatedRV.Spec.Size.Value()).To(Equal(int64(2147483648)))
+			// Verify that actualSize was set
+			Expect(updatedRV.Status).NotTo(BeNil())
+			Expect(updatedRV.Status.ActualSize.Value()).To(Equal(int64(2147483648)))
 		})
 
 		It("should return success without resize when requested size is less than current size", func() {
@@ -672,6 +682,13 @@ var _ = Describe("ControllerExpandVolume", func() {
 					},
 				}
 				Expect(cl.Update(ctx, updatedRV)).To(Succeed())
+
+				// Update actualSize after Ready to simulate controller updating actualSize
+				time.Sleep(200 * time.Millisecond)
+				updatedRV2 := &v1alpha2.ReplicatedVolume{}
+				Expect(cl.Get(ctx, client.ObjectKey{Name: volumeID}, updatedRV2)).To(Succeed())
+				updatedRV2.Status.ActualSize = resource.MustParse("2Gi")
+				Expect(cl.Update(ctx, updatedRV2)).To(Succeed())
 			}()
 
 			timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second)

images/csi-driver/pkg/utils/func.go

@@ -31,7 +31,7 @@ import (
 
 	snc "github.com/deckhouse/sds-node-configurator/api/v1alpha1"
 	srv "github.com/deckhouse/sds-replicated-volume/api/v1alpha1"
-	v1alpha2 "github.com/deckhouse/sds-replicated-volume/api/v1alpha2old"
+	"github.com/deckhouse/sds-replicated-volume/api/v1alpha2"
 	"github.com/deckhouse/sds-replicated-volume/lib/go/common/logger"
 	"k8s.io/apimachinery/pkg/api/meta"
 )
@@ -264,6 +264,56 @@ func WaitForReplicatedVolumeReady(
 	}
 }
 
+// WaitForActualSize waits for ReplicatedVolume actualSize to match requested size (within delta)
+func WaitForActualSize(
+	ctx context.Context,
+	kc client.Client,
+	log *logger.Logger,
+	traceID, name string,
+	requestedSize, delta resource.Quantity,
+) (int, error) {
+	var attemptCounter int
+	log.Info(fmt.Sprintf("[WaitForActualSize][traceID:%s][volumeID:%s] Waiting for actualSize to match requested size %s (delta: %s)", traceID, name, requestedSize.String(), delta.String()))
+	for {
+		attemptCounter++
+		select {
+		case <-ctx.Done():
+			log.Warning(fmt.Sprintf("[WaitForActualSize][traceID:%s][volumeID:%s] context done. Failed to wait for actualSize", traceID, name))
+			return attemptCounter, ctx.Err()
+		default:
+			time.Sleep(500 * time.Millisecond)
+		}
+
+		rv, err := GetReplicatedVolume(ctx, kc, name)
+		if err != nil {
+			return attemptCounter, err
+		}
+
+		if attemptCounter%10 == 0 {
+			actualSizeStr := "not set"
+			if rv.Status != nil && !rv.Status.ActualSize.IsZero() {
+				actualSizeStr = rv.Status.ActualSize.String()
+			}
+			log.Info(fmt.Sprintf("[WaitForActualSize][traceID:%s][volumeID:%s] Attempt: %d, requestedSize: %s, actualSize: %s", traceID, name, attemptCounter, requestedSize.String(), actualSizeStr))
+		}
+
+		if rv.DeletionTimestamp != nil {
+			return attemptCounter, fmt.Errorf("failed to wait for actualSize for ReplicatedVolume %s, reason: ReplicatedVolume is being deleted", name)
+		}
+
+		if rv.Status != nil && !rv.Status.ActualSize.IsZero() {
+			// Check if actualSize matches requested size within delta
+			if AreSizesEqualWithinDelta(requestedSize, rv.Status.ActualSize, delta) || rv.Status.ActualSize.Value() >= requestedSize.Value() {
+				log.Info(fmt.Sprintf("[WaitForActualSize][traceID:%s][volumeID:%s] actualSize matches requested size: %s", traceID, name, rv.Status.ActualSize.String()))
+				return attemptCounter, nil
+			}
+			log.Trace(fmt.Sprintf("[WaitForActualSize][traceID:%s][volumeID:%s] Attempt %d, actualSize %s does not match requested size %s yet. Waiting...", traceID, name, attemptCounter, rv.Status.ActualSize.String(), requestedSize.String()))
+		} else {
+			log.Trace(fmt.Sprintf("[WaitForActualSize][traceID:%s][volumeID:%s] Attempt %d, actualSize not set yet. Waiting...", traceID, name, attemptCounter))
+		}
+	}
+}
+
 // DeleteReplicatedVolume deletes a ReplicatedVolume resource
 func DeleteReplicatedVolume(ctx context.Context, kc client.Client, log *logger.Logger, traceID, name string) error {
 	log.Trace(fmt.Sprintf("[DeleteReplicatedVolume][traceID:%s][volumeID:%s] Trying to find ReplicatedVolume", traceID, name))
@@ -341,16 +391,18 @@ func removeRVFinalizerIfExist(ctx context.Context, kc client.Client, log *logger
 // GetReplicatedVolumeReplicaForNode gets ReplicatedVolumeReplica for a specific node
 func GetReplicatedVolumeReplicaForNode(ctx context.Context, kc client.Client, volumeName, nodeName string) (*v1alpha2.ReplicatedVolumeReplica, error) {
 	rvrList := &v1alpha2.ReplicatedVolumeReplicaList{}
+	// Use MatchingFields with only one field selector (spec.replicatedVolumeName)
+	// Then filter by nodeName manually, as client.MatchingFields doesn't support multiple fields
 	err := kc.List(
 		ctx,
 		rvrList,
 		client.MatchingFields{"spec.replicatedVolumeName": volumeName},
-		client.MatchingFields{"spec.nodeName": nodeName},
 	)
 	if err != nil {
 		return nil, err
 	}
 
+	// Filter by nodeName manually
 	for i := range rvrList.Items {
 		if rvrList.Items[i].Spec.NodeName == nodeName {
 			return &rvrList.Items[i], nil

images/csi-driver/pkg/utils/func_publish_test.go

@@ -29,7 +29,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
-	v1alpha2 "github.com/deckhouse/sds-replicated-volume/api/v1alpha2old"
+	"github.com/deckhouse/sds-replicated-volume/api/v1alpha2"
 	"github.com/deckhouse/sds-replicated-volume/lib/go/common/logger"
 )