Added versioning configuration for COS buckets in CSI driver

pkg/driver/controllerserver.go

@@ -48,6 +48,7 @@ func (cs *controllerServer) CreateVolume(_ context.Context, req *csi.CreateVolum
 		kpRootKeyCrn       string
 		pvcName            string
 		pvcNamespace       string
+		BucketVersioning   string
 	)
 	secretMapCustom := make(map[string]string)
 
@@ -164,6 +165,24 @@ func (cs *controllerServer) CreateVolume(_ context.Context, req *csi.CreateVolum
 		bucketName = secretMapCustom["bucketName"]
 	}
 
+	// Check for BucketVersioning parameter
+	if val, ok := secretMap["BucketVersioning"]; ok && val != "" {
+		enable := strings.ToLower(strings.TrimSpace(val))
+		if enable != "true" && enable != "false" {
+			return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("Invalid BucketVersioning value in secret: %s. Must be 'true' or 'false'", val))
+		}
+		BucketVersioning = enable
+		klog.Infof("BucketVersioning set via secret: %s", BucketVersioning)
+	} else if val, ok := params["BucketVersioning"]; ok && val != "" {
+		enable := strings.ToLower(strings.TrimSpace(val))
+		if enable != "true" && enable != "false" {
+			return nil, status.Error(codes.InvalidArgument,
+				fmt.Sprintf("Invalid BucketVersioning value in storage class: %s. Must be 'true' or 'false'", val))
+		}
+		BucketVersioning = enable
+		klog.Infof("BucketVersioning set via storage class: %s", BucketVersioning)
+	}
+
 	creds, err := getCredentials(secretMap)
 	if err != nil {
 		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("Error in getting credentials %v", err))
@@ -184,6 +203,16 @@ func (cs *controllerServer) CreateVolume(_ context.Context, req *csi.CreateVolum
 			params["userProvidedBucket"] = "false"
 			klog.Infof("Created bucket: %s", bucketName)
 		}
+
+		if BucketVersioning != "" {
+			enable := BucketVersioning == "true"
+			if err := sess.SetBucketVersioning(bucketName, enable); err != nil {
+				klog.Errorf("Failed to set versioning for bucket: %s, error: %v", bucketName, err)
+				return nil, status.Error(codes.Internal, fmt.Sprintf("Failed to set versioning for bucket %s: %v", bucketName, err))
+			}
+			klog.Infof("Bucket versioning set to %t for bucket %s", enable, bucketName)
+		}
+
 		params["bucketName"] = bucketName
 	} else {
 		// Generate random temp bucket name based on volume id
@@ -197,6 +226,15 @@ func (cs *controllerServer) CreateVolume(_ context.Context, req *csi.CreateVolum
 		if err != nil {
 			return nil, status.Error(codes.PermissionDenied, fmt.Sprintf("%v: %v", err, tempBucketName))
 		}
+		// Enable versioning for new temp bucket
+		if BucketVersioning != "" {
+			enable := BucketVersioning == "true"
+			if err := sess.SetBucketVersioning(tempBucketName, enable); err != nil {
+				klog.Errorf("Failed to set versioning for temp bucket: %s, error: %v", tempBucketName, err)
+				return nil, status.Error(codes.Internal, fmt.Sprintf("Failed to set versioning for bucket %s: %v", tempBucketName, err))
+			}
+			klog.Infof("Bucket versioning set to %t for temp bucket %s", enable, tempBucketName)
+		}
 		klog.Infof("Created temp bucket: %s", tempBucketName)
 		params["userProvidedBucket"] = "false"
 		params["bucketName"] = tempBucketName
@@ -441,6 +479,7 @@ func parseCustomSecret(secret *v1.Secret) map[string]string {
 		iamEndpoint        string
 		cosEndpoint        string
 		locationConstraint string
+		BucketVersioning   string
 	)
 
 	if bytesVal, ok := secret.Data["accessKey"]; ok {
@@ -479,6 +518,10 @@ func parseCustomSecret(secret *v1.Secret) map[string]string {
 		locationConstraint = string(bytesVal)
 	}
 
+	if bytesVal, ok := secret.Data["BucketVersioning"]; ok {
+		BucketVersioning = string(bytesVal)
+	}
+
 	secretMapCustom["accessKey"] = accessKey
 	secretMapCustom["secretKey"] = secretKey
 	secretMapCustom["apiKey"] = apiKey
@@ -488,6 +531,7 @@ func parseCustomSecret(secret *v1.Secret) map[string]string {
 	secretMapCustom["iamEndpoint"] = iamEndpoint
 	secretMapCustom["cosEndpoint"] = cosEndpoint
 	secretMapCustom["locationConstraint"] = locationConstraint
+	secretMapCustom["BucketVersioning"] = BucketVersioning
 
 	return secretMapCustom
 }

pkg/s3client/fake_s3client.go

@@ -11,6 +11,7 @@ type FakeCOSSessionFactory struct {
 	FailCheckBucketAccess bool
 	FailCreateBucket      bool
 	FailDeleteBucket      bool
+	FailBucketVersioning  bool
 }
 
 type fakeCOSSession struct {
@@ -31,6 +32,13 @@ func (s *fakeCOSSession) CheckBucketAccess(bucket string) error {
 	return nil
 }
 
+func (s *fakeCOSSession) SetBucketVersioning(bucket string, enable bool) error {
+	if s.factory.FailBucketVersioning {
+		return errors.New("failed to set bucket versioning")
+	}
+	return nil
+}
+
 func (s *fakeCOSSession) CheckObjectPathExistence(bucket, objectpath string) (bool, error) {
 	return true, nil
 }

pkg/s3client/s3client.go

@@ -61,6 +61,8 @@ type ObjectStorageSession interface {
 
 	// DeleteBucket methods deletes a bucket (with all of its objects)
 	DeleteBucket(bucket string) error
+
+	SetBucketVersioning(bucket string, enable bool) error
 }
 
 // COSSessionFactory represents a COS (S3) session factory
@@ -91,6 +93,7 @@ type s3API interface {
 	ListObjectsV2(input *s3.ListObjectsV2Input) (*s3.ListObjectsV2Output, error)
 	DeleteObject(input *s3.DeleteObjectInput) (*s3.DeleteObjectOutput, error)
 	DeleteBucket(input *s3.DeleteBucketInput) (*s3.DeleteBucketOutput, error)
+	PutBucketVersioning(input *s3.PutBucketVersioningInput) (*s3.PutBucketVersioningOutput, error)
 }
 
 func (s *COSSession) CheckBucketAccess(bucket string) error {
@@ -182,6 +185,26 @@ func (s *COSSession) DeleteBucket(bucket string) error {
 	return err
 }
 
+func (s *COSSession) SetBucketVersioning(bucket string, enable bool) error {
+	status := "Suspended"
+	if enable {
+		status = "Enabled"
+	}
+	s.logger.Info("Setting versioning for bucket", zap.String("bucket", bucket), zap.Bool("enable", enable))
+	_, err := s.svc.PutBucketVersioning(&s3.PutBucketVersioningInput{
+		Bucket: aws.String(bucket),
+		VersioningConfiguration: &s3.VersioningConfiguration{
+			Status: aws.String(status),
+		},
+	})
+	if err != nil {
+		s.logger.Error("Failed to set versioning", zap.String("bucket", bucket), zap.Bool("enable", enable), zap.Error(err))
+		return fmt.Errorf("failed to set versioning to %v for bucket '%s': %v", enable, bucket, err)
+	}
+	s.logger.Info("Versioning set successfully for bucket", zap.String("bucket", bucket), zap.Bool("enable", enable))
+	return nil
+}
+
 func NewS3Client(lgr *zap.Logger) (ObjectStorageSession, error) {
 	cosSession := new(COSSession)
 	cosSession.logger = lgr

pkg/s3client/s3client_test.go

@@ -12,13 +12,14 @@ import (
 )
 
 type fakeS3API struct {
-	ErrHeadBucket    error
-	ErrCreateBucket  error
-	ErrListObjects   error
-	ErrListObjectsV2 error
-	ErrDeleteObject  error
-	ErrDeleteBucket  error
-	ObjectPath       string
+	ErrHeadBucket          error
+	ErrCreateBucket        error
+	ErrListObjects         error
+	ErrListObjectsV2       error
+	ErrDeleteObject        error
+	ErrDeleteBucket        error
+	ObjectPath             string
+	ErrPutBucketVersioning error
 }
 
 const (
@@ -49,6 +50,10 @@ func (a *fakeS3API) CreateBucket(input *s3.CreateBucketInput) (*s3.CreateBucketO
 	return nil, a.ErrCreateBucket
 }
 
+func (a *fakeS3API) PutBucketVersioning(input *s3.PutBucketVersioningInput) (*s3.PutBucketVersioningOutput, error) {
+	return &s3.PutBucketVersioningOutput{}, a.ErrPutBucketVersioning
+}
+
 func (a *fakeS3API) ListObjects(input *s3.ListObjectsInput) (*s3.ListObjectsOutput, error) {
 	return &s3.ListObjectsOutput{
 		Contents: []*s3.Object{{Key: &testObject}},
@@ -156,6 +161,26 @@ func Test_CreateBucket_Positive(t *testing.T) {
 	assert.NoError(t, err)
 }
 
+func Test_SetBucketVersioning_True_Positive(t *testing.T) {
+	sess := getSession(&fakeS3API{})
+	err := sess.SetBucketVersioning(testBucket, true)
+	assert.NoError(t, err)
+}
+
+func Test_SetBucketVersioning_False_Positive(t *testing.T) {
+	sess := getSession(&fakeS3API{})
+	err := sess.SetBucketVersioning(testBucket, false)
+	assert.NoError(t, err)
+}
+
+func Test_SetBucketVersioning_Error(t *testing.T) {
+	sess := getSession(&fakeS3API{ErrPutBucketVersioning: errFoo})
+	err := sess.SetBucketVersioning(testBucket, true)
+	if assert.Error(t, err) {
+		assert.EqualError(t, err, "failed to set versioning to true for bucket 'test-bucket': foo")
+	}
+}
+
 func Test_DeleteBucket_BucketAlreadyDeleted_Positive(t *testing.T) {
 	sess := getSession(&fakeS3API{ErrListObjects: awserr.New("NoSuchBucket", "", errFoo)})
 	err := sess.DeleteBucket(testBucket)

tests/sanity/sanity_test.go

@@ -151,6 +151,11 @@ func (s *fakeObjectStorageSession) CheckBucketAccess(bucket string) error {
 	return nil
 }
 
+func (s *fakeObjectStorageSession) SetBucketVersioning(bucketName string, enable bool) error {
+	s.logger.Info(fmt.Sprintf("Fake SetBucketVersioning called for bucket %s with enable=%t", bucketName, enable))
+	return nil
+}
+
 func (s *fakeObjectStorageSession) CheckObjectPathExistence(bucket, objectpath string) (bool, error) {
 	return true, nil
 }