Merge pull request kubernetes-sigs#453 from Trickybrain/Idempotency

Fix CSI Driver creating multiple filesystems per PVC

Author: k8s-ci-robot

cmd/hooks/prestop_test.go

@@ -21,9 +21,9 @@ import (
 	"sigs.k8s.io/aws-fsx-csi-driver/pkg/driver"
 	"testing"
 
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 	v1 "k8s.io/api/core/v1"
 	driverMocks "sigs.k8s.io/aws-fsx-csi-driver/pkg/driver/mocks"
 )

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/onsi/gomega v1.35.1
 	github.com/spf13/pflag v1.0.6
 	github.com/stretchr/testify v1.10.0
+	go.uber.org/mock v0.6.0
 	google.golang.org/grpc v1.72.0
 	k8s.io/api v0.33.0
 	k8s.io/apimachinery v0.33.0

go.sum

@@ -177,6 +177,8 @@ go.opentelemetry.io/otel/trace v1.35.0 h1:dPpEfJu1sDIqruz7BHFG3c7528f6ddfSWfFDVt
 go.opentelemetry.io/otel/trace v1.35.0/go.mod h1:WUk7DtFp1Aw2MkvqGdwiXYDZZNvA/1J8o6xRXLrIkyc=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y=
+go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
 go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
 go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
@@ -234,8 +236,8 @@ golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roY
 golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.1.1/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
-golang.org/x/tools v0.26.0 h1:v/60pFQmzmT9ExmjDv2gGIfi3OqfKoEP6I5+umXlbnQ=
-golang.org/x/tools v0.26.0/go.mod h1:TPVVj70c7JJ3WCazhD8OdXcZg/og+b9+tH/KxylGwH0=
+golang.org/x/tools v0.36.0 h1:kWS0uv/zsvHEle1LbV5LE8QujrxB3wfQyxHfhOk0Qkg=
+golang.org/x/tools v0.36.0/go.mod h1:WBDiHKJK8YgLHlcQPYQzNCkUxUypCaa5ZegCVutKm+s=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

pkg/cloud/cloud.go

@@ -120,6 +120,7 @@ type Cloud interface {
 	DescribeFileSystem(ctx context.Context, fileSystemId string) (fs *FileSystem, err error)
 	WaitForFileSystemAvailable(ctx context.Context, fileSystemId string) error
 	WaitForFileSystemResize(ctx context.Context, fileSystemId string, resizeGiB int32) error
+	FindFileSystemByVolumeName(ctx context.Context, volumeName string) (*FileSystem, error)
 }
 
 type cloud struct {
@@ -445,3 +446,74 @@ func isBadRequestUpdateInProgress(err error) bool {
 	var badRequest *types.BadRequest
 	return errors.As(err, &badRequest) && strings.Contains(err.Error(), "Unable to perform the storage capacity update. There is an update already in progress.")
 }
+
+func (c *cloud) FindFileSystemByVolumeName(ctx context.Context, volumeName string) (*FileSystem, error) {
+	var nextToken *string
+	const maxResults = 100
+
+	klog.V(4).InfoS("Searching for existing filesystem", "volumeName", volumeName)
+
+	// AWS FSx DescribeFileSystems API doesn't support filtering by tags,
+	// so we paginate through all filesystems and filter client-side
+	for {
+		input := &fsx.DescribeFileSystemsInput{
+			MaxResults: aws.Int32(maxResults),
+			NextToken:  nextToken,
+		}
+
+		output, err := c.fsx.DescribeFileSystems(ctx, input)
+		if err != nil {
+			return nil, fmt.Errorf("failed to describe filesystems: %v", err)
+		}
+
+		klog.V(5).InfoS("Checking batch of filesystems", "count", len(output.FileSystems))
+
+		// Search current batch
+		for _, fs := range output.FileSystems {
+			// Skip if filesystem is being deleted
+			if fs.Lifecycle != types.FileSystemLifecycleAvailable &&
+				fs.Lifecycle != types.FileSystemLifecycleCreating {
+				continue
+			}
+
+			// Check tags for volume name match
+			for _, tag := range fs.Tags {
+				if *tag.Key == VolumeNameTagKey && *tag.Value == volumeName {
+					klog.V(2).InfoS("Found existing filesystem",
+						"volumeName", volumeName,
+						"fileSystemId", *fs.FileSystemId,
+						"lifecycle", string(fs.Lifecycle))
+
+					mountName := "fsx"
+					if fs.LustreConfiguration.MountName != nil {
+						mountName = *fs.LustreConfiguration.MountName
+					}
+
+					perUnitStorageThroughput := int32(0)
+					if fs.LustreConfiguration.PerUnitStorageThroughput != nil {
+						perUnitStorageThroughput = *fs.LustreConfiguration.PerUnitStorageThroughput
+					}
+
+					return &FileSystem{
+						FileSystemId:             *fs.FileSystemId,
+						CapacityGiB:              *fs.StorageCapacity,
+						DnsName:                  *fs.DNSName,
+						MountName:                mountName,
+						StorageType:              string(fs.StorageType),
+						DeploymentType:           string(fs.LustreConfiguration.DeploymentType),
+						PerUnitStorageThroughput: perUnitStorageThroughput,
+					}, nil
+				}
+			}
+		}
+
+		// Check if more results exist
+		if output.NextToken == nil {
+			break
+		}
+		nextToken = output.NextToken
+	}
+
+	klog.V(2).InfoS("No existing filesystem found", "volumeName", volumeName)
+	return nil, ErrNotFound
+}

pkg/cloud/cloud_test.go

@@ -25,7 +25,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/fsx"
 	"github.com/aws/aws-sdk-go-v2/service/fsx/types"
-	"github.com/golang/mock/gomock"
+	"go.uber.org/mock/gomock"
 	"sigs.k8s.io/aws-fsx-csi-driver/pkg/cloud/mocks"
 )
 

pkg/cloud/fakes.go

@@ -105,3 +105,12 @@ func (c *FakeCloudProvider) WaitForFileSystemAvailable(ctx context.Context, file
 func (c *FakeCloudProvider) WaitForFileSystemResize(ctx context.Context, fileSystemId string, resizeGiB int32) error {
 	return nil
 }
+
+func (c *FakeCloudProvider) FindFileSystemByVolumeName(ctx context.Context, volumeName string) (*FileSystem, error) {
+	// Check if filesystem exists for this volume name
+	fs, exists := c.fileSystems[volumeName]
+	if exists {
+		return fs, nil
+	}
+	return nil, ErrNotFound
+}

pkg/cloud/metadata_test.go

@@ -23,7 +23,7 @@ import (
 	"testing"
 
 	"github.com/aws/aws-sdk-go-v2/feature/ec2/imds"
-	"github.com/golang/mock/gomock"
+	"go.uber.org/mock/gomock"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes"