3.76

* add {Get,Put,Delete}BucketTagging Handlers

* s3 add skip bucket PublicAccessBlock handlers

---------

Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>

k8s/charts/seaweedfs/Chart.yaml

@@ -1,6 +1,6 @@
 apiVersion: v1
 description: SeaweedFS
 name: seaweedfs
-appVersion: "3.75"
+appVersion: "3.76"
 # Dev note: Trigger a helm chart release by `git tag -a helm-<version>`
-version: 4.0.375
+version: 4.0.376

weed/mount/filer_conf.go

@@ -3,19 +3,15 @@ package mount
 import (
 	"errors"
 	"fmt"
-	"path/filepath"
-	"sync/atomic"
-	"time"
-
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
-	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/mount/meta_cache"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/util"
+	"path/filepath"
 )
 
-func (wfs *WFS) subscribeFilerConfEvents() (func(), error) {
-	now := time.Now()
+func (wfs *WFS) subscribeFilerConfEvents() (*meta_cache.MetadataFollower, error) {
 	confDir := filer.DirectoryEtcSeaweedFS
 	confName := filer.FilerConfName
 	confFullName := filepath.Join(filer.DirectoryEtcSeaweedFS, filer.FilerConfName)
@@ -71,41 +67,9 @@ func (wfs *WFS) subscribeFilerConfEvents() (func(), error) {
 
 		return nil
 	}
-
-	metadataFollowOption := &pb.MetadataFollowOption{
-		ClientName:             "fuse",
-		ClientId:               wfs.signature,
-		ClientEpoch:            1,
-		SelfSignature:          0,
-		PathPrefix:             confFullName,
-		AdditionalPathPrefixes: nil,
-		StartTsNs:              now.UnixNano(),
-		StopTsNs:               0,
-		EventErrorType:         pb.FatalOnError,
-	}
-
-	return func() {
-		// sync new conf changes
-		util.RetryUntil("followFilerConfChanges", func() error {
-			metadataFollowOption.ClientEpoch++
-			i := atomic.LoadInt32(&wfs.option.filerIndex)
-			n := len(wfs.option.FilerAddresses)
-			err = pb.FollowMetadata(wfs.option.FilerAddresses[i], wfs.option.GrpcDialOption, metadataFollowOption, processEventFn)
-			if err == nil {
-				atomic.StoreInt32(&wfs.option.filerIndex, i)
-				return nil
-			}
-
-			i++
-			if i >= int32(n) {
-				i = 0
-			}
-
-			return err
-		}, func(err error) bool {
-			glog.V(0).Infof("fuse follow filer conf changes: %v", err)
-			return true
-		})
+	return &meta_cache.MetadataFollower{
+		PathPrefixToWatch: confFullName,
+		ProcessEventFn:    processEventFn,
 	}, nil
 }
 

weed/mount/meta_cache/meta_cache_subscribe.go

@@ -10,7 +10,42 @@ import (
 	"strings"
 )
 
-func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.FilerClient, dir string, lastTsNs int64) error {
+type MetadataFollower struct {
+	PathPrefixToWatch string
+	ProcessEventFn    func(resp *filer_pb.SubscribeMetadataResponse) error
+}
+
+func mergeProceesors(mainProcessor func(resp *filer_pb.SubscribeMetadataResponse) error, followers ...*MetadataFollower) func(resp *filer_pb.SubscribeMetadataResponse) error {
+	return func(resp *filer_pb.SubscribeMetadataResponse) error {
+
+		// build the full path
+		entry := resp.EventNotification.NewEntry
+		if entry == nil {
+			entry = resp.EventNotification.OldEntry
+		}
+		dir := resp.Directory
+		if resp.EventNotification.NewParentPath != "" {
+			dir = resp.EventNotification.NewParentPath
+		}
+		fp := util.NewFullPath(dir, entry.Name)
+
+		for _, follower := range followers {
+			if strings.HasPrefix(string(fp), follower.PathPrefixToWatch) {
+				if err := follower.ProcessEventFn(resp); err != nil {
+					return err
+				}
+			}
+		}
+		return mainProcessor(resp)
+	}
+}
+
+func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.FilerClient, dir string, lastTsNs int64, followers ...*MetadataFollower) error {
+
+	var prefixes []string
+	for _, follower := range followers {
+		prefixes = append(prefixes, follower.PathPrefixToWatch)
+	}
 
 	processEventFn := func(resp *filer_pb.SubscribeMetadataResponse) error {
 		message := resp.EventNotification
@@ -69,7 +104,7 @@ func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.Fil
 		ClientEpoch:            1,
 		SelfSignature:          selfSignature,
 		PathPrefix:             prefix,
-		AdditionalPathPrefixes: nil,
+		AdditionalPathPrefixes: prefixes,
 		DirectoriesToWatch:     nil,
 		StartTsNs:              lastTsNs,
 		StopTsNs:               0,
@@ -77,7 +112,7 @@ func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.Fil
 	}
 	util.RetryUntil("followMetaUpdates", func() error {
 		metadataFollowOption.ClientEpoch++
-		return pb.WithFilerClientFollowMetadata(client, metadataFollowOption, processEventFn)
+		return pb.WithFilerClientFollowMetadata(client, metadataFollowOption, mergeProceesors(processEventFn, followers...))
 	}, func(err error) bool {
 		glog.Errorf("follow metadata updates: %v", err)
 		return true

weed/mount/weedfs.go

@@ -141,15 +141,13 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 }
 
 func (wfs *WFS) StartBackgroundTasks() error {
-	fn, err := wfs.subscribeFilerConfEvents()
+	follower, err := wfs.subscribeFilerConfEvents()
 	if err != nil {
 		return err
 	}
 
-	go fn()
-
 	startTime := time.Now()
-	go meta_cache.SubscribeMetaEvents(wfs.metaCache, wfs.signature, wfs, wfs.option.FilerMountRootPath, startTime.UnixNano())
+	go meta_cache.SubscribeMetaEvents(wfs.metaCache, wfs.signature, wfs, wfs.option.FilerMountRootPath, startTime.UnixNano(), follower)
 	go wfs.loopCheckQuota()
 
 	return nil

weed/s3api/auth_signature_v4.go

@@ -149,11 +149,11 @@ func (iam *IdentityAccessManagement) doesSignatureMatch(hashedPayload string, r
 	}
 
 	if forwardedPrefix := r.Header.Get("X-Forwarded-Prefix"); forwardedPrefix != "" {
-		// Handling usage of reverse proxy at prefix. Note that it's an undefined behavior for AWS S3 and not supported in MinIO.
+		// Handling usage of reverse proxy at prefix.
 		// Trying with prefix before main path.
 
 		// Get canonical request.
-		canonicalRequest := getCanonicalRequest(extractedSignedHeaders, hashedPayload, queryStr, forwardedPrefix + req.URL.Path, req.Method)
+		canonicalRequest := getCanonicalRequest(extractedSignedHeaders, hashedPayload, queryStr, forwardedPrefix+req.URL.Path, req.Method)
 
 		errCode = iam.genAndCompareSignatureV4(canonicalRequest, cred.SecretKey, t, signV4Values)
 		if errCode == s3err.ErrNone {
@@ -686,7 +686,6 @@ func extractSignedHeaders(signedHeaders []string, r *http.Request) (http.Header,
 		case "host":
 			// Go http server removes "host" from Request.Header
 			if forwardedHost := r.Header.Get("X-Forwarded-Host"); forwardedHost != "" {
-				// Trying to use reverse proxy at prefix. Note that it's an undefined behavior for AWS S3 and not supported in MinIO.
 				extractedSignedHeaders.Set(header, forwardedHost)
 			} else if forwardedFor := r.Header.Get("X-Forwarded-For"); forwardedFor != "" {
 				extractedSignedHeaders.Set(header, forwardedFor)

weed/s3api/policy/post-policy.go

@@ -315,6 +315,6 @@ func errInvalidArgument(message string) error {
 		StatusCode: http.StatusBadRequest,
 		Code:       "InvalidArgument",
 		Message:    message,
-		RequestID:  "minio",
+		RequestID:  "client",
 	}
 }

weed/s3api/s3api_bucket_skip_handlers.go

@@ -1,6 +1,8 @@
 package s3api
 
 import (
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"net/http"
 
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
@@ -48,6 +50,28 @@ func (s3a *S3ApiServer) PutBucketVersioningHandler(w http.ResponseWriter, r *htt
 	s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
 }
 
+// GetBucketTaggingHandler Returns the tag set associated with the bucket
+// https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetBucketTagging.html
+func (s3a *S3ApiServer) GetBucketTaggingHandler(w http.ResponseWriter, r *http.Request) {
+	bucket, _ := s3_constants.GetBucketAndObject(r)
+	glog.V(3).Infof("GetBucketTagging %s", bucket)
+
+	if err := s3a.checkBucket(r, bucket); err != s3err.ErrNone {
+		s3err.WriteErrorResponse(w, r, err)
+		return
+	}
+
+	s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchTagSet)
+}
+
+func (s3a *S3ApiServer) PutBucketTaggingHandler(w http.ResponseWriter, r *http.Request) {
+	s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
+}
+
+func (s3a *S3ApiServer) DeleteBucketTaggingHandler(w http.ResponseWriter, r *http.Request) {
+	s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
+}
+
 // GetBucketEncryptionHandler Returns the default encryption configuration
 // https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetBucketEncryption.html
 func (s3a *S3ApiServer) GetBucketEncryptionHandler(w http.ResponseWriter, r *http.Request) {
@@ -61,3 +85,17 @@ func (s3a *S3ApiServer) PutBucketEncryptionHandler(w http.ResponseWriter, r *htt
 func (s3a *S3ApiServer) DeleteBucketEncryptionHandler(w http.ResponseWriter, r *http.Request) {
 	s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
 }
+
+// GetPublicAccessBlockHandler Retrieves the PublicAccessBlock configuration for an S3 bucket
+// https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetPublicAccessBlock.html
+func (s3a *S3ApiServer) GetPublicAccessBlockHandler(w http.ResponseWriter, r *http.Request) {
+	s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
+}
+
+func (s3a *S3ApiServer) PutPublicAccessBlockHandler(w http.ResponseWriter, r *http.Request) {
+	s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
+}
+
+func (s3a *S3ApiServer) DeletePublicAccessBlockHandler(w http.ResponseWriter, r *http.Request) {
+	s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
+}

weed/s3api/s3api_server.go

@@ -258,11 +258,21 @@ func (s3a *S3ApiServer) registerRouter(router *mux.Router) {
 		bucket.Methods(http.MethodGet).HandlerFunc(track(s3a.iam.Auth(s3a.cb.Limit(s3a.GetBucketVersioningHandler, ACTION_READ)), "GET")).Queries("versioning", "")
 		bucket.Methods(http.MethodPut).HandlerFunc(track(s3a.iam.Auth(s3a.cb.Limit(s3a.PutBucketVersioningHandler, ACTION_WRITE)), "PUT")).Queries("versioning", "")
 
+		// GetBucketTagging
+		bucket.Methods(http.MethodGet).HandlerFunc(track(s3a.iam.Auth(s3a.cb.Limit(s3a.GetBucketTaggingHandler, ACTION_TAGGING)), "GET")).Queries("tagging", "")
+		bucket.Methods(http.MethodPut).HandlerFunc(track(s3a.iam.Auth(s3a.cb.Limit(s3a.PutBucketTaggingHandler, ACTION_TAGGING)), "PUT")).Queries("tagging", "")
+		bucket.Methods(http.MethodDelete).HandlerFunc(track(s3a.iam.Auth(s3a.cb.Limit(s3a.DeleteBucketTaggingHandler, ACTION_TAGGING)), "DELETE")).Queries("tagging", "")
+
 		// GetBucketEncryption
 		bucket.Methods(http.MethodGet).HandlerFunc(track(s3a.iam.Auth(s3a.cb.Limit(s3a.GetBucketEncryptionHandler, ACTION_ADMIN)), "GET")).Queries("encryption", "")
 		bucket.Methods(http.MethodPut).HandlerFunc(track(s3a.iam.Auth(s3a.cb.Limit(s3a.PutBucketEncryptionHandler, ACTION_ADMIN)), "PUT")).Queries("encryption", "")
 		bucket.Methods(http.MethodDelete).HandlerFunc(track(s3a.iam.Auth(s3a.cb.Limit(s3a.DeleteBucketEncryptionHandler, ACTION_ADMIN)), "DELETE")).Queries("encryption", "")
 
+		// GetPublicAccessBlockHandler
+		bucket.Methods(http.MethodGet).HandlerFunc(track(s3a.iam.Auth(s3a.cb.Limit(s3a.GetPublicAccessBlockHandler, ACTION_ADMIN)), "GET")).Queries("publicAccessBlock", "")
+		bucket.Methods(http.MethodPut).HandlerFunc(track(s3a.iam.Auth(s3a.cb.Limit(s3a.PutPublicAccessBlockHandler, ACTION_ADMIN)), "PUT")).Queries("publicAccessBlock", "")
+		bucket.Methods(http.MethodDelete).HandlerFunc(track(s3a.iam.Auth(s3a.cb.Limit(s3a.DeletePublicAccessBlockHandler, ACTION_ADMIN)), "DELETE")).Queries("publicAccessBlock", "")
+
 		// ListObjectsV2
 		bucket.Methods(http.MethodGet).HandlerFunc(track(s3a.iam.Auth(s3a.cb.Limit(s3a.ListObjectsV2Handler, ACTION_LIST)), "LIST")).Queries("list-type", "2")
 

weed/s3api/s3err/s3api_errors.go

@@ -109,6 +109,7 @@ const (
 	ErrRequestBytesExceed
 
 	OwnershipControlsNotFoundError
+	ErrNoSuchTagSet
 )
 
 // error code to APIError structure, these fields carry respective
@@ -184,6 +185,11 @@ var errorCodeResponse = map[ErrorCode]APIError{
 		Description:    "The bucket policy does not exist",
 		HTTPStatusCode: http.StatusNotFound,
 	},
+	ErrNoSuchTagSet: {
+		Code:           "NoSuchTagSet",
+		Description:    "The TagSet does not exist",
+		HTTPStatusCode: http.StatusNotFound,
+	},
 	ErrNoSuchCORSConfiguration: {
 		Code:           "NoSuchCORSConfiguration",
 		Description:    "The CORS configuration does not exist",

weed/util/constants.go

@@ -8,7 +8,7 @@ const HttpStatusCancelled = 499
 
 var (
 	MAJOR_VERSION  = int32(3)
-	MINOR_VERSION  = int32(75)
+	MINOR_VERSION  = int32(76)
 	VERSION_NUMBER = fmt.Sprintf("%d.%02d", MAJOR_VERSION, MINOR_VERSION)
 	VERSION        = sizeLimit + " " + VERSION_NUMBER
 	COMMIT         = ""