s3: sync bucket info from filer (#3759)

1 files changed, 213 insertions, 0 deletions

diff --git a/weed/s3api/bucket_metadata.go b/weed/s3api/bucket_metadata.go
new file mode 100644
index 000000000..1b9b09981
--- /dev/null
+++ b/weed/s3api/bucket_metadata.go
@@ -0,0 +1,213 @@
+package s3api
+
+import (
+	"bytes"
+	"encoding/json"
+	"github.com/aws/aws-sdk-go/private/protocol/json/jsonutil"
+	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+
+	//"github.com/seaweedfs/seaweedfs/weed/s3api"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"math"
+	"sync"
+)
+
+var loadBucketMetadataFromFiler = func(r *BucketRegistry, bucketName string) (*BucketMetaData, error) {
+	entry, err := filer_pb.GetEntry(r.s3a, util.NewFullPath(r.s3a.option.BucketsPath, bucketName))
+	if err != nil {
+		return nil, err
+	}
+
+	return buildBucketMetadata(entry), nil
+}
+
+type BucketMetaData struct {
+	_ struct{} `type:"structure"`
+
+	Name string
+
+	//By default, when another AWS account uploads an object to S3 bucket,
+	//that account (the object writer) owns the object, has access to it, and
+	//can grant other users access to it through ACLs. You can use Object Ownership
+	//to change this default behavior so that ACLs are disabled and you, as the
+	//bucket owner, automatically own every object in your bucket.
+	ObjectOwnership string
+
+	// Container for the bucket owner's display name and ID.
+	Owner *s3.Owner `type:"structure"`
+
+	// A list of grants for access controls.
+	Acl []*s3.Grant `locationName:"AccessControlList" locationNameList:"Grant" type:"list"`
+}
+
+type BucketRegistry struct {
+	metadataCache     map[string]*BucketMetaData
+	metadataCacheLock sync.RWMutex
+
+	notFound     map[string]struct{}
+	notFoundLock sync.RWMutex
+	s3a          *S3ApiServer
+}
+
+func NewBucketRegistry(s3a *S3ApiServer) *BucketRegistry {
+	br := &BucketRegistry{
+		metadataCache: make(map[string]*BucketMetaData),
+		notFound:      make(map[string]struct{}),
+		s3a:           s3a,
+	}
+	err := br.init()
+	if err != nil {
+		glog.Fatal("init bucket registry failed", err)
+		return nil
+	}
+	return br
+}
+
+func (r *BucketRegistry) init() error {
+	err := filer_pb.List(r.s3a, r.s3a.option.BucketsPath, "", func(entry *filer_pb.Entry, isLast bool) error {
+		r.LoadBucketMetadata(entry)
+		return nil
+	}, "", false, math.MaxUint32)
+	return err
+}
+
+func (r *BucketRegistry) LoadBucketMetadata(entry *filer_pb.Entry) {
+	bucketMetadata := buildBucketMetadata(entry)
+	r.metadataCacheLock.Lock()
+	defer r.metadataCacheLock.Unlock()
+	r.metadataCache[entry.Name] = bucketMetadata
+}
+
+func buildBucketMetadata(entry *filer_pb.Entry) *BucketMetaData {
+	entryJson, _ := json.Marshal(entry)
+	glog.V(3).Infof("build bucket metadata,entry=%s", entryJson)
+	bucketMetadata := &BucketMetaData{
+		Name: entry.Name,
+
+		//Default ownership: OwnershipBucketOwnerEnforced, which means Acl is disabled
+		ObjectOwnership: s3_constants.OwnershipBucketOwnerEnforced,
+
+		// Default owner: `AccountAdmin`
+		Owner: &s3.Owner{
+			ID:          &AccountAdmin.Id,
+			DisplayName: &AccountAdmin.Name,
+		},
+	}
+	if entry.Extended != nil {
+		//ownership control
+		ownership, ok := entry.Extended[s3_constants.ExtOwnershipKey]
+		if ok {
+			ownership := string(ownership)
+			valid := s3_constants.ValidateOwnership(ownership)
+			if valid {
+				bucketMetadata.ObjectOwnership = ownership
+			} else {
+				glog.Warningf("Invalid ownership: %s, bucket: %s", ownership, bucketMetadata.Name)
+			}
+		}
+
+		//access control policy
+		acpBytes, ok := entry.Extended[s3_constants.ExtAcpKey]
+		if ok {
+			var acp s3.AccessControlPolicy
+			err := jsonutil.UnmarshalJSON(&acp, bytes.NewReader(acpBytes))
+			if err == nil {
+				//validate owner
+				if acp.Owner != nil && acp.Owner.ID != nil {
+					bucketMetadata.Owner = acp.Owner
+				} else {
+					glog.Warningf("bucket ownerId is empty! bucket: %s", bucketMetadata.Name)
+				}
+
+				//acl
+				bucketMetadata.Acl = acp.Grants
+			} else {
+				glog.Warningf("Unmarshal ACP: %s(%v), bucket: %s", string(acpBytes), err, bucketMetadata.Name)
+			}
+		}
+	}
+	return bucketMetadata
+}
+
+func (r *BucketRegistry) RemoveBucketMetadata(entry *filer_pb.Entry) {
+	r.removeMetadataCache(entry.Name)
+	r.unMarkNotFound(entry.Name)
+}
+
+func (r *BucketRegistry) GetBucketMetadata(bucketName string) (*BucketMetaData, s3err.ErrorCode) {
+	r.metadataCacheLock.RLock()
+	bucketMetadata, ok := r.metadataCache[bucketName]
+	r.metadataCacheLock.RUnlock()
+	if ok {
+		return bucketMetadata, s3err.ErrNone
+	}
+
+	r.notFoundLock.RLock()
+	_, ok = r.notFound[bucketName]
+	r.notFoundLock.RUnlock()
+	if ok {
+		return nil, s3err.ErrNoSuchBucket
+	}
+
+	bucketMetadata, errCode := r.LoadBucketMetadataFromFiler(bucketName)
+	if errCode != s3err.ErrNone {
+		return nil, errCode
+	}
+
+	r.setMetadataCache(bucketMetadata)
+	r.unMarkNotFound(bucketName)
+	return bucketMetadata, s3err.ErrNone
+}
+
+func (r *BucketRegistry) LoadBucketMetadataFromFiler(bucketName string) (*BucketMetaData, s3err.ErrorCode) {
+	r.notFoundLock.Lock()
+	defer r.notFoundLock.Unlock()
+
+	//check if already exists
+	r.metadataCacheLock.RLock()
+	bucketMetaData, ok := r.metadataCache[bucketName]
+	r.metadataCacheLock.RUnlock()
+	if ok {
+		return bucketMetaData, s3err.ErrNone
+	}
+
+	//if not exists, load from filer
+	bucketMetadata, err := loadBucketMetadataFromFiler(r, bucketName)
+	if err != nil {
+		if err == filer_pb.ErrNotFound {
+			// The bucket doesn't actually exist and should no longer loaded from the filer
+			r.notFound[bucketName] = struct{}{}
+			return nil, s3err.ErrNoSuchBucket
+		}
+		return nil, s3err.ErrInternalError
+	}
+	return bucketMetadata, s3err.ErrNone
+}
+
+func (r *BucketRegistry) setMetadataCache(metadata *BucketMetaData) {
+	r.metadataCacheLock.Lock()
+	defer r.metadataCacheLock.Unlock()
+	r.metadataCache[metadata.Name] = metadata
+}
+
+func (r *BucketRegistry) removeMetadataCache(bucket string) {
+	r.metadataCacheLock.Lock()
+	defer r.metadataCacheLock.Unlock()
+	delete(r.metadataCache, bucket)
+}
+
+func (r *BucketRegistry) markNotFound(bucket string) {
+	r.notFoundLock.Lock()
+	defer r.notFoundLock.Unlock()
+	r.notFound[bucket] = struct{}{}
+}
+
+func (r *BucketRegistry) unMarkNotFound(bucket string) {
+	r.notFoundLock.Lock()
+	defer r.notFoundLock.Unlock()
+	delete(r.notFound, bucket)
+}