Delete folder api implementation

internal/fs/inode/dir.go

@@ -854,6 +854,16 @@ func (d *dirInode) DeleteChildDir(
 		err = fmt.Errorf("DeleteObject: %w", err)
 		return
 	}
+
+	if d.bucket.BucketType() == gcs.Hierarchical {
+		// Delete Folder deletes folder (in case of Hierarchical Bucket).
+		err = d.bucket.DeleteFolder(ctx, childName.GcsObjectName())
+	}
+
+	if err != nil {
+		return
+	}
+
 	d.cache.Erase(name)
 
 	return

internal/gcsx/prefix_bucket.go

@@ -211,3 +211,8 @@ func (b *prefixBucket) DeleteObject(
 	err = b.wrapped.DeleteObject(ctx, mReq)
 	return
 }
+
+func (b *prefixBucket) DeleteFolder(ctx context.Context, folderName string) (err error) {
+	mFolderName := b.wrappedName(folderName)
+	return b.wrapped.DeleteFolder(ctx, mFolderName)
+}

internal/monitor/bucket.go

@@ -195,6 +195,13 @@ func (mb *monitoringBucket) DeleteObject(
 	return err
 }
 
+func (mb *monitoringBucket) DeleteFolder(ctx context.Context, folderName string) error {
+	startTime := time.Now()
+	err := mb.wrapped.DeleteFolder(ctx, folderName)
+	recordRequest(ctx, "DeleteFolder", startTime)
+	return err
+}
+
 // recordReader increments the reader count when it's opened or closed.
 func recordReader(ctx context.Context, ioMethod string) {
 	if err := stats.RecordWithTags(

internal/ratelimit/throttled_bucket.go

@@ -184,6 +184,19 @@ func (b *throttledBucket) DeleteObject(
 	return
 }
 
+func (b *throttledBucket) DeleteFolder(ctx context.Context, folderName string) (err error) {
+	// Wait for permission to call through.
+	err = b.opThrottle.Wait(ctx, 1)
+	if err != nil {
+		return
+	}
+
+	// Call through.
+	err = b.wrapped.DeleteFolder(ctx, folderName)
+
+	return
+}
+
 ////////////////////////////////////////////////////////////////////////
 // readerCloser
 ////////////////////////////////////////////////////////////////////////

internal/storage/bucket_handle.go

@@ -33,7 +33,6 @@ import (
 	"github.com/googlecloudplatform/gcsfuse/v2/internal/logger"
 	"github.com/googlecloudplatform/gcsfuse/v2/internal/storage/gcs"
 	"github.com/googlecloudplatform/gcsfuse/v2/internal/storage/storageutil"
-
 	"google.golang.org/api/googleapi"
 	"google.golang.org/api/iterator"
 )
@@ -468,6 +467,35 @@ func (b *bucketHandle) ComposeObjects(ctx context.Context, req *gcs.ComposeObjec
 	return
 }
 
+func (b *bucketHandle) DeleteFolder(ctx context.Context, folderName string) (err error) {
+	var notfound *gcs.NotFoundError
+	var callOptions []gax.CallOption
+
+	err = b.DeleteObject(
+		ctx,
+		&gcs.DeleteObjectRequest{
+			Name:       folderName,
+			Generation: 0, // Delete the latest version of object named after dir.
+		})
+
+	if err != nil {
+		// Ignore err if it is object not found as in HNS bucket object may exist as a folder.
+		if !errors.As(err, &notfound) {
+			return err
+		}
+	}
+
+	err = b.controlClient.DeleteFolder(ctx, &controlpb.DeleteFolderRequest{
+		Name: "projects/_/buckets/" + b.bucketName + "/folders/" + folderName,
+	}, callOptions...)
+
+	if err != nil {
+		err = fmt.Errorf("DeleteFolder: %w", err)
+	}
+
+	return err
+}
+
 // TODO: Consider adding this method to the bucket interface if additional
 // layout options are needed in the future.
 func (b *bucketHandle) getStorageLayout() (*controlpb.StorageLayout, error) {

internal/storage/bucket_handle_test.go

@@ -1249,3 +1249,45 @@ func (testSuite *BucketHandleTest) TestDefaultBucketTypeWithControlClientNil() {
 
 	assert.Equal(testSuite.T(), gcs.NonHierarchical, testSuite.bucketHandle.bucketType, "Expected Hierarchical bucket type")
 }
+
+func (testSuite *BucketHandleTest) TestDeleteFolderWhenFolderExitForHierarchicalBucket() {
+	ctx := context.Background()
+	mockClient := new(MockStorageControlClient)
+	mockClient.On("DeleteFolder", ctx, &controlpb.DeleteFolderRequest{Name: "projects/_/buckets/" + TestBucketName + "/folders/" + TestObjectName}, mock.Anything).
+		Return(nil)
+	testSuite.bucketHandle.controlClient = mockClient
+	testSuite.bucketHandle.bucketType = gcs.Hierarchical
+
+	err := testSuite.bucketHandle.DeleteFolder(ctx, TestObjectName)
+
+	mockClient.AssertExpectations(testSuite.T())
+	assert.Nil(testSuite.T(), err)
+}
+
+func (testSuite *BucketHandleTest) TestDeleteFolderWhenFolderExistButSameObjectNotExistInHierarchicalBucket() {
+	ctx := context.Background()
+	mockClient := new(MockStorageControlClient)
+	mockClient.On("DeleteFolder", ctx, &controlpb.DeleteFolderRequest{Name: "projects/_/buckets/" + TestBucketName + "/folders/" + missingObjectName}, mock.Anything).
+		Return(nil)
+	testSuite.bucketHandle.controlClient = mockClient
+	testSuite.bucketHandle.bucketType = gcs.Hierarchical
+
+	err := testSuite.bucketHandle.DeleteFolder(ctx, missingObjectName)
+
+	mockClient.AssertExpectations(testSuite.T())
+	assert.Nil(testSuite.T(), err)
+}
+
+func (testSuite *BucketHandleTest) TestDeleteFolderWhenFolderNotExistForHierarchicalBucket() {
+	ctx := context.Background()
+	mockClient := new(MockStorageControlClient)
+	mockClient.On("DeleteFolder", mock.Anything, &controlpb.DeleteFolderRequest{Name: "projects/_/buckets/" + TestBucketName + "/folders/" + missingObjectName}, mock.Anything).
+		Return(errors.New("mock error"))
+	testSuite.bucketHandle.controlClient = mockClient
+	testSuite.bucketHandle.bucketType = gcs.Hierarchical
+
+	err := testSuite.bucketHandle.DeleteFolder(ctx, missingObjectName)
+
+	mockClient.AssertExpectations(testSuite.T())
+	assert.Equal(testSuite.T(), "DeleteFolder: mock error", err.Error())
+}

internal/storage/caching/fast_stat_bucket.go

@@ -270,6 +270,12 @@ func (b *fastStatBucket) DeleteObject(
 	return
 }
 
+func (b *fastStatBucket) DeleteFolder(ctx context.Context, folderName string) error {
+	b.invalidate(folderName)
+	err := b.wrapped.DeleteFolder(ctx, folderName)
+	return err
+}
+
 func (b *fastStatBucket) StatObjectFromGcs(ctx context.Context,
 	req *gcs.StatObjectRequest) (m *gcs.MinObject, e *gcs.ExtendedObjectAttributes, err error) {
 	m, e, err = b.wrapped.StatObject(ctx, req)

internal/storage/control_client_wrapper.go

@@ -25,4 +25,8 @@ type StorageControlClient interface {
 	GetStorageLayout(ctx context.Context,
 		req *controlpb.GetStorageLayoutRequest,
 		opts ...gax.CallOption) (*controlpb.StorageLayout, error)
+
+	DeleteFolder(ctx context.Context,
+		req *controlpb.DeleteFolderRequest,
+		opts ...gax.CallOption) error
 }

internal/storage/debug_bucket.go

@@ -229,3 +229,11 @@ func (b *debugBucket) DeleteObject(
 	err = b.wrapped.DeleteObject(ctx, req)
 	return
 }
+
+func (b *debugBucket) DeleteFolder(ctx context.Context, folderName string) (err error) {
+	id, desc, start := b.startRequest("DeleteFolder(%q)", folderName)
+	defer b.finishRequest(id, desc, start, &err)
+
+	err = b.wrapped.DeleteFolder(ctx, folderName)
+	return err
+}

internal/storage/fake/bucket.go

@@ -866,3 +866,19 @@ func (b *bucket) DeleteObject(
 
 	return
 }
+
+func (b *bucket) DeleteFolder(ctx context.Context, folderName string) (err error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	// Do we possess the object with the given name?
+	index := b.objects.find(folderName)
+	if index == len(b.objects) {
+		return
+	}
+
+	// Remove the object.
+	b.objects = append(b.objects[:index], b.objects[index+1:]...)
+
+	return
+}

internal/storage/gcs/bucket.go

@@ -140,4 +140,6 @@ type Bucket interface {
 	DeleteObject(
 		ctx context.Context,
 		req *DeleteObjectRequest) error
+
+	DeleteFolder(ctx context.Context, folderName string) error
 }

internal/storage/mock_bucket.go

@@ -155,6 +155,27 @@ func (m *mockBucket) DeleteObject(p0 context.Context, p1 *gcs.DeleteObjectReques
 	return
 }
 
+func (m *mockBucket) DeleteFolder(ctx context.Context, folderName string) (o0 error) {
+	// Get a file name and line number for the caller.
+	_, file, line, _ := runtime.Caller(1)
+
+	// Hand the call off to the controller, which does most of the work.
+	retVals := m.controller.HandleMethodCall(
+		m,
+		"DeleteFolder",
+		file,
+		line,
+		[]interface{}{})
+	if len(retVals) != 1 {
+		panic(fmt.Sprintf("mockBucket.DeleteFolder: invalid return values: %v", retVals))
+	}
+	// o0 string
+	if retVals[0] != nil {
+		o0 = retVals[0].(error)
+	}
+	return
+}
+
 func (m *mockBucket) ListObjects(p0 context.Context, p1 *gcs.ListObjectsRequest) (o0 *gcs.Listing, o1 error) {
 	// Get a file name and line number for the caller.
 	_, file, line, _ := runtime.Caller(1)

internal/storage/mock_control_client.go

@@ -35,3 +35,11 @@ func (m *MockStorageControlClient) GetStorageLayout(ctx context.Context,
 	args := m.Called(ctx, req, opts)
 	return args.Get(0).(*controlpb.StorageLayout), args.Error(1)
 }
+
+// Implement the DeleteFolder method for the mock.
+func (m *MockStorageControlClient) DeleteFolder(ctx context.Context,
+	req *controlpb.DeleteFolderRequest,
+	opts ...gax.CallOption) error {
+	args := m.Called(ctx, req, opts)
+	return args.Error(0)
+}

tools/integration_tests/operations/operations_test.go

@@ -132,6 +132,16 @@ func createMountConfigsAndEquivalentFlags() (flags [][]string) {
 	filePath3 := setup.YAMLConfigFile(mountConfig3, "config3.yaml")
 	flags = append(flags, []string{"--config-file=" + filePath3})
 
+	mountConfig4 := config.MountConfig{
+		EnableHNS: true,
+		LogConfig: config.LogConfig{
+			Severity:        config.TRACE,
+			LogRotateConfig: config.DefaultLogRotateConfig(),
+		},
+	}
+	filePath4 := setup.YAMLConfigFile(mountConfig4, "config4.yaml")
+	flags = append(flags, []string{"--config-file=" + filePath4})
+
 	return flags
 }