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.74"
+appVersion: "3.76"
 # Dev note: Trigger a helm chart release by `git tag -a helm-<version>`
-version: 4.0.2
+version: 4.0.376

k8s/charts/seaweedfs/templates/s3-deployment.yaml

@@ -14,7 +14,6 @@ spec:
   selector:
     matchLabels:
       app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
-      helm.sh/chart: {{ .Chart.Name }}-{{ .Chart.Version | replace "+" "_" }}
       app.kubernetes.io/instance: {{ .Release.Name }}
       app.kubernetes.io/component: s3
   template:

weed/mount/filer_conf.go

@@ -1,20 +1,17 @@
 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)
@@ -38,7 +35,11 @@ func (wfs *WFS) subscribeFilerConfEvents() (func(), error) {
 		return nil
 	})
 	if err != nil {
-		return nil, err
+		if errors.Is(err, filer_pb.ErrNotFound) {
+			glog.V(0).Infof("fuse filer conf %s not found", confFullName)
+		} else {
+			return nil, err
+		}
 	}
 
 	processEventFn := func(resp *filer_pb.SubscribeMetadataResponse) error {
@@ -66,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/mq/schema/schema.go

@@ -24,3 +24,30 @@ func (s *Schema) GetField(name string) (*schema_pb.Field, bool) {
 	field, ok := s.fieldMap[name]
 	return field, ok
 }
+
+func TypeToString(t *schema_pb.Type) string {
+	switch t.Kind.(type) {
+	case *schema_pb.Type_ScalarType:
+		switch t.GetScalarType() {
+		case schema_pb.ScalarType_BOOL:
+			return "bool"
+		case schema_pb.ScalarType_INT32:
+			return "int32"
+		case schema_pb.ScalarType_INT64:
+			return "int64"
+		case schema_pb.ScalarType_FLOAT:
+			return "float"
+		case schema_pb.ScalarType_DOUBLE:
+			return "double"
+		case schema_pb.ScalarType_BYTES:
+			return "bytes"
+		case schema_pb.ScalarType_STRING:
+			return "string"
+		}
+	case *schema_pb.Type_ListType:
+		return "list"
+	case *schema_pb.Type_RecordType:
+		return "record"
+	}
+	return "unknown"
+}

weed/mq/topic/partition.go

@@ -1,6 +1,9 @@
 package topic
 
-import "github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
+import (
+	"fmt"
+	"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
+)
 
 const PartitionCount = 4096
 
@@ -81,3 +84,11 @@ func (partition Partition) Overlaps(partition2 Partition) bool {
 	}
 	return true
 }
+
+func (partition Partition) String() string {
+	return fmt.Sprintf("%04d-%04d", partition.RangeStart, partition.RangeStop)
+}
+
+func ToString(partition *mq_pb.Partition) string {
+	return fmt.Sprintf("%04d-%04d", partition.RangeStart, partition.RangeStop)
+}

weed/s3api/auth_signature_v4.go

@@ -148,15 +148,38 @@ func (iam *IdentityAccessManagement) doesSignatureMatch(hashedPayload string, r
 		}
 	}
 
+	if forwardedPrefix := r.Header.Get("X-Forwarded-Prefix"); forwardedPrefix != "" {
+		// 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)
+
+		errCode = iam.genAndCompareSignatureV4(canonicalRequest, cred.SecretKey, t, signV4Values)
+		if errCode == s3err.ErrNone {
+			return identity, errCode
+		}
+	}
+
 	// Get canonical request.
 	canonicalRequest := getCanonicalRequest(extractedSignedHeaders, hashedPayload, queryStr, req.URL.Path, req.Method)
 
+	errCode = iam.genAndCompareSignatureV4(canonicalRequest, cred.SecretKey, t, signV4Values)
+
+	if errCode == s3err.ErrNone {
+		return identity, errCode
+	}
+	return nil, errCode
+}
+
+// Generate and compare signature for request.
+func (iam *IdentityAccessManagement) genAndCompareSignatureV4(canonicalRequest, secretKey string, t time.Time, signV4Values signValues) s3err.ErrorCode {
 	// Get string to sign from canonical request.
 	stringToSign := getStringToSign(canonicalRequest, t, signV4Values.Credential.getScope())
 
 	// Calculate signature.
 	newSignature := iam.getSignature(
-		cred.SecretKey,
+		secretKey,
 		signV4Values.Credential.scope.date,
 		signV4Values.Credential.scope.region,
 		signV4Values.Credential.scope.service,
@@ -165,11 +188,9 @@ func (iam *IdentityAccessManagement) doesSignatureMatch(hashedPayload string, r
 
 	// Verify if signature match.
 	if !compareSignatureV4(newSignature, signV4Values.Signature) {
-		return nil, s3err.ErrSignatureDoesNotMatch
+		return s3err.ErrSignatureDoesNotMatch
 	}
-
-	// Return error none.
-	return identity, s3err.ErrNone
+	return s3err.ErrNone
 }
 
 // credentialHeader data type represents structured form of Credential
@@ -664,7 +685,13 @@ func extractSignedHeaders(signedHeaders []string, r *http.Request) (http.Header,
 			extractedSignedHeaders.Set(header, "100-continue")
 		case "host":
 			// Go http server removes "host" from Request.Header
-			extractedSignedHeaders.Set(header, r.Host)
+			if forwardedHost := r.Header.Get("X-Forwarded-Host"); forwardedHost != "" {
+				extractedSignedHeaders.Set(header, forwardedHost)
+			} else if forwardedFor := r.Header.Get("X-Forwarded-For"); forwardedFor != "" {
+				extractedSignedHeaders.Set(header, forwardedFor)
+			} else {
+				extractedSignedHeaders.Set(header, r.Host)
+			}
 		case "transfer-encoding":
 			for _, enc := range r.TransferEncoding {
 				extractedSignedHeaders.Add(header, enc)

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"
@@ -47,3 +49,53 @@ func (s3a *S3ApiServer) DeleteBucketPolicyHandler(w http.ResponseWriter, r *http
 func (s3a *S3ApiServer) PutBucketVersioningHandler(w http.ResponseWriter, r *http.Request) {
 	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) {
+	s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
+}
+
+func (s3a *S3ApiServer) PutBucketEncryptionHandler(w http.ResponseWriter, r *http.Request) {
+	s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
+}
+
+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,6 +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/storage/super_block/replica_placement.go

@@ -12,6 +12,14 @@ type ReplicaPlacement struct {
 
 func NewReplicaPlacementFromString(t string) (*ReplicaPlacement, error) {
 	rp := &ReplicaPlacement{}
+	switch len(t) {
+	case 0:
+		t = "000"
+	case 1:
+		t = "00" + t
+	case 2:
+		t = "0" + t
+	}
 	for i, c := range t {
 		count := int(c - '0')
 		if count < 0 {

weed/util/constants.go

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