filer: dynamically create bucket under /buckets folder

weed/filer2/filer.go

@@ -30,9 +30,11 @@ type Filer struct {
 	MasterClient        *wdclient.MasterClient
 	fileIdDeletionQueue *util.UnboundedQueue
 	GrpcDialOption      grpc.DialOption
+	DirBucketsPath      string
+	buckets             *FilerBuckets
 }
 
-func NewFiler(masters []string, grpcDialOption grpc.DialOption) *Filer {
+func NewFiler(masters []string, grpcDialOption grpc.DialOption, bucketFolder string) *Filer {
 	f := &Filer{
 		directoryCache:      ccache.New(ccache.Configure().MaxSize(1000).ItemsToPrune(100)),
 		MasterClient:        wdclient.NewMasterClient(context.Background(), grpcDialOption, "filer", masters),
@@ -109,11 +111,13 @@ func (f *Filer) CreateEntry(ctx context.Context, entry *Entry, o_excl bool) erro
 			dirEntry = &Entry{
 				FullPath: FullPath(dirPath),
 				Attr: Attr{
-					Mtime:  now,
-					Crtime: now,
-					Mode:   os.ModeDir | 0770,
-					Uid:    entry.Uid,
-					Gid:    entry.Gid,
+					Mtime:       now,
+					Crtime:      now,
+					Mode:        os.ModeDir | 0770,
+					Uid:         entry.Uid,
+					Gid:         entry.Gid,
+					Collection:  entry.Collection,
+					Replication: entry.Replication,
 				},
 			}
 
@@ -125,6 +129,7 @@ func (f *Filer) CreateEntry(ctx context.Context, entry *Entry, o_excl bool) erro
 					return fmt.Errorf("mkdir %s: %v", dirPath, mkdirErr)
 				}
 			} else {
+				f.maybeAddBucket(dirEntry)
 				f.NotifyUpdateEvent(nil, dirEntry, false)
 			}
 
@@ -175,6 +180,7 @@ func (f *Filer) CreateEntry(ctx context.Context, entry *Entry, o_excl bool) erro
 		}
 	}
 
+	f.maybeAddBucket(entry)
 	f.NotifyUpdateEvent(oldEntry, entry, true)
 
 	f.deleteChunksIfNotNew(oldEntry, entry)

weed/filer2/filer_buckets.go

@@ -0,0 +1,113 @@
+package filer2
+
+import (
+	"context"
+	"math"
+	"sync"
+
+	"github.com/chrislusf/seaweedfs/weed/glog"
+)
+
+type BucketName string
+type BucketOption struct {
+	Name        BucketName
+	Replication string
+}
+type FilerBuckets struct {
+	dirBucketsPath string
+	buckets        map[BucketName]*BucketOption
+	sync.RWMutex
+}
+
+func (f *Filer) LoadBuckets(dirBucketsPath string) {
+
+	f.buckets = &FilerBuckets{
+		buckets: make(map[BucketName]*BucketOption),
+	}
+	f.DirBucketsPath = dirBucketsPath
+
+	limit := math.MaxInt32
+
+	entries, err := f.ListDirectoryEntries(context.Background(), FullPath(dirBucketsPath), "", false, limit)
+
+	if err != nil {
+		glog.V(1).Infof("no buckets found: %v", err)
+		return
+	}
+
+	glog.V(1).Infof("buckets found: %d", len(entries))
+
+	f.buckets.Lock()
+	for _, entry := range entries {
+		f.buckets.buckets[BucketName(entry.Name())] = &BucketOption{
+			Name:        BucketName(entry.Name()),
+			Replication: entry.Replication,
+		}
+	}
+	f.buckets.Unlock()
+
+}
+
+func (f *Filer) ReadBucketOption(buketName string) (replication string) {
+
+	f.buckets.RLock()
+	defer f.buckets.RUnlock()
+
+	option, found := f.buckets.buckets[BucketName(buketName)]
+
+	if !found {
+		return ""
+	}
+	return option.Replication
+
+}
+
+func (f *Filer) isBucket(entry *Entry) bool {
+	if !entry.IsDirectory() {
+		return false
+	}
+	parent, dirName := entry.FullPath.DirAndName()
+	if parent != f.DirBucketsPath {
+		return false
+	}
+
+	f.buckets.RLock()
+	defer f.buckets.RUnlock()
+
+	_, found := f.buckets.buckets[BucketName(dirName)]
+
+	return found
+
+}
+
+func (f *Filer) maybeAddBucket(entry *Entry) {
+	if !entry.IsDirectory() {
+		return
+	}
+	parent, dirName := entry.FullPath.DirAndName()
+	if parent != f.DirBucketsPath {
+		return
+	}
+	f.addBucket(dirName, &BucketOption{
+		Name:        BucketName(dirName),
+		Replication: entry.Replication,
+	})
+}
+
+func (f *Filer) addBucket(buketName string, bucketOption *BucketOption) {
+
+	f.buckets.Lock()
+	defer f.buckets.Unlock()
+
+	f.buckets.buckets[BucketName(buketName)] = bucketOption
+
+}
+
+func (f *Filer) deleteBucket(buketName string) {
+
+	f.buckets.Lock()
+	defer f.buckets.Unlock()
+
+	delete(f.buckets.buckets, BucketName(buketName))
+
+}

weed/filer2/filer_delete_entry.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 )
 
 func (f *Filer) DeleteEntryMetaAndData(ctx context.Context, p FullPath, isRecursive bool, ignoreRecursiveError, shouldDeleteChunks bool) (err error) {
@@ -18,27 +19,35 @@ func (f *Filer) DeleteEntryMetaAndData(ctx context.Context, p FullPath, isRecurs
 		return findErr
 	}
 
+	isCollection := f.isBucket(entry)
+
 	var chunks []*filer_pb.FileChunk
 	chunks = append(chunks, entry.Chunks...)
 	if entry.IsDirectory() {
 		// delete the folder children, not including the folder itself
 		var dirChunks []*filer_pb.FileChunk
-		dirChunks, err = f.doBatchDeleteFolderMetaAndData(ctx, entry, isRecursive, ignoreRecursiveError, shouldDeleteChunks)
+		dirChunks, err = f.doBatchDeleteFolderMetaAndData(ctx, entry, isRecursive, ignoreRecursiveError, shouldDeleteChunks && !isCollection)
 		if err != nil {
+			glog.V(0).Infof("delete directory %s: %v", p, err)
 			return fmt.Errorf("delete directory %s: %v", p, err)
 		}
 		chunks = append(chunks, dirChunks...)
-		f.cacheDelDirectory(string(p))
 	}
+
 	// delete the file or folder
 	err = f.doDeleteEntryMetaAndData(ctx, entry, shouldDeleteChunks)
 	if err != nil {
 		return fmt.Errorf("delete file %s: %v", p, err)
 	}
 
-	if shouldDeleteChunks {
+	if shouldDeleteChunks && !isCollection {
 		go f.DeleteChunks(chunks)
 	}
+	if isCollection {
+		collectionName := entry.Name()
+		f.doDeleteCollection(ctx, collectionName)
+		f.deleteBucket(collectionName)
+	}
 
 	return nil
 }
@@ -55,6 +64,9 @@ func (f *Filer) doBatchDeleteFolderMetaAndData(ctx context.Context, entry *Entry
 		}
 		if lastFileName == "" && !isRecursive && len(entries) > 0 {
 			// only for first iteration in the loop
+			for _, child := range entries {
+				println("existing children", child.Name())
+			}
 			return nil, fmt.Errorf("fail to delete non-empty folder: %s", entry.FullPath)
 		}
 
@@ -100,3 +112,17 @@ func (f *Filer) doDeleteEntryMetaAndData(ctx context.Context, entry *Entry, shou
 
 	return nil
 }
+
+func (f *Filer) doDeleteCollection(ctx context.Context, collectionName string) (err error) {
+
+	return f.MasterClient.WithClient(ctx, func(client master_pb.SeaweedClient) error {
+		_, err := client.CollectionDelete(ctx, &master_pb.CollectionDeleteRequest{
+			Name: collectionName,
+		})
+		if err != nil {
+			glog.Infof("delete collection %s: %v", collectionName, err)
+		}
+		return err
+	})
+
+}

weed/filer2/leveldb/leveldb_store_test.go

@@ -9,7 +9,7 @@ import (
 )
 
 func TestCreateAndFind(t *testing.T) {
-	filer := filer2.NewFiler(nil, nil)
+	filer := filer2.NewFiler(nil, nil, "")
 	dir, _ := ioutil.TempDir("", "seaweedfs_filer_test")
 	defer os.RemoveAll(dir)
 	store := &LevelDBStore{}
@@ -64,7 +64,7 @@ func TestCreateAndFind(t *testing.T) {
 }
 
 func TestEmptyRoot(t *testing.T) {
-	filer := filer2.NewFiler(nil, nil)
+	filer := filer2.NewFiler(nil, nil, "")
 	dir, _ := ioutil.TempDir("", "seaweedfs_filer_test2")
 	defer os.RemoveAll(dir)
 	store := &LevelDBStore{}

weed/filer2/leveldb2/leveldb2_store_test.go

@@ -9,7 +9,7 @@ import (
 )
 
 func TestCreateAndFind(t *testing.T) {
-	filer := filer2.NewFiler(nil, nil)
+	filer := filer2.NewFiler(nil, nil, "")
 	dir, _ := ioutil.TempDir("", "seaweedfs_filer_test")
 	defer os.RemoveAll(dir)
 	store := &LevelDB2Store{}
@@ -64,7 +64,7 @@ func TestCreateAndFind(t *testing.T) {
 }
 
 func TestEmptyRoot(t *testing.T) {
-	filer := filer2.NewFiler(nil, nil)
+	filer := filer2.NewFiler(nil, nil, "")
 	dir, _ := ioutil.TempDir("", "seaweedfs_filer_test2")
 	defer os.RemoveAll(dir)
 	store := &LevelDB2Store{}