3.16

fix https://github.com/chrislusf/seaweedfs/issues/3315

.github/workflows/go.yml

@@ -21,7 +21,7 @@ jobs:
     steps:
 
     - name: Set up Go 1.x
-      uses: actions/setup-go@b22fbbc2921299758641fab08929b4ac52b32923 # v2
+      uses: actions/setup-go@84cbf8094393cdc5fe1fe1671ff2647332956b1a # v2
       with:
         go-version: ^1.13
       id: go

go.mod

@@ -10,7 +10,7 @@ require (
 	github.com/Azure/azure-storage-blob-go v0.15.0
 	github.com/OneOfOne/xxhash v1.2.8
 	github.com/Shopify/sarama v1.34.1
-	github.com/aws/aws-sdk-go v1.44.47
+	github.com/aws/aws-sdk-go v1.44.51
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/buraksezer/consistent v0.0.0-20191006190839-693edf70fd72
 	github.com/bwmarrin/snowflake v0.3.0
@@ -154,7 +154,7 @@ require (
 	github.com/hashicorp/raft-boltdb v0.0.0-20220329195025-15018e9b97e0
 	github.com/tikv/client-go/v2 v2.0.1
 	github.com/ydb-platform/ydb-go-sdk-auth-environ v0.1.2
-	github.com/ydb-platform/ydb-go-sdk/v3 v3.27.0
+	github.com/ydb-platform/ydb-go-sdk/v3 v3.28.0
 	google.golang.org/grpc/security/advancedtls v0.0.0-20220622233350-5cdb09fa29c1
 )
 

go.sum

@@ -157,8 +157,8 @@ github.com/armon/go-metrics v0.3.10/go.mod h1:4O98XIr/9W0sxpJ8UaYkvjk10Iff7SnFrb
 github.com/aws/aws-sdk-go v1.15.27/go.mod h1:mFuSZ37Z9YOHbQEwBWztmVzqXrEkub65tZoCYDt7FT0=
 github.com/aws/aws-sdk-go v1.37.0/go.mod h1:hcU610XS61/+aQV88ixoOzUoG7v3b31pl2zKMmprdro=
 github.com/aws/aws-sdk-go v1.43.31/go.mod h1:y4AeaBuwd2Lk+GepC1E9v0qOiTws0MIWAX4oIKwKHZo=
-github.com/aws/aws-sdk-go v1.44.47 h1:uyiNvoR4wfZ8Bp4ghgbyzGFIg5knjZMUAd5S9ba9qNU=
-github.com/aws/aws-sdk-go v1.44.47/go.mod h1:y4AeaBuwd2Lk+GepC1E9v0qOiTws0MIWAX4oIKwKHZo=
+github.com/aws/aws-sdk-go v1.44.51 h1:jO9hoLynZOrMM4dj0KjeKIK+c6PA+HQbKoHOkAEye2Y=
+github.com/aws/aws-sdk-go v1.44.51/go.mod h1:y4AeaBuwd2Lk+GepC1E9v0qOiTws0MIWAX4oIKwKHZo=
 github.com/aws/aws-sdk-go-v2 v1.16.2 h1:fqlCk6Iy3bnCumtrLz9r3mJ/2gUT0pJ0wLFVIdWh+JA=
 github.com/aws/aws-sdk-go-v2 v1.16.2/go.mod h1:ytwTPBG6fXTZLxxeeCCWj2/EMYp/xDUgX+OET6TLNNU=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.4.1/go.mod h1:n8Bs1ElDD2wJ9kCRTczA83gYbBmjSwZp3umc6zF4EeM=
@@ -924,8 +924,8 @@ github.com/ydb-platform/ydb-go-genproto v0.0.0-20220531094121-36ca6bddb9f7/go.mo
 github.com/ydb-platform/ydb-go-sdk-auth-environ v0.1.2 h1:EYSI1kulnHb0H0zt3yOw4cRj4ABMSMGwNe43D+fX7e4=
 github.com/ydb-platform/ydb-go-sdk-auth-environ v0.1.2/go.mod h1:Xfjce+VMU9yJVr1lj60yK2fFPWjB4jr/4cp3K7cjzi4=
 github.com/ydb-platform/ydb-go-sdk/v3 v3.25.3/go.mod h1:PFizF/vJsdAgEwjK3DVSBD52kdmRkWfSIS2q2pA+e88=
-github.com/ydb-platform/ydb-go-sdk/v3 v3.27.0 h1:paSdC12yRI19Vv9ej6qAjUu2/r/WaW/rzsUogd+lg34=
-github.com/ydb-platform/ydb-go-sdk/v3 v3.27.0/go.mod h1:vXjmbeEAWlkVE5/ym3XHhtnWk7aDGGqFMKrfgwbRUkQ=
+github.com/ydb-platform/ydb-go-sdk/v3 v3.28.0 h1:F394Kkl+QPLrl0+fWpoSafdfgGqQCZOyzvXLxzVUWfs=
+github.com/ydb-platform/ydb-go-sdk/v3 v3.28.0/go.mod h1:vXjmbeEAWlkVE5/ym3XHhtnWk7aDGGqFMKrfgwbRUkQ=
 github.com/ydb-platform/ydb-go-yc v0.8.3 h1:92UUUMsfvtMl6mho8eQ9lbkiPrF3a9CT+RrVRAKNRwo=
 github.com/ydb-platform/ydb-go-yc v0.8.3/go.mod h1:zUolAFGzJ5XG8uwiseTLr9Lapm7L7hdVdZgLSuv9FXE=
 github.com/ydb-platform/ydb-go-yc-metadata v0.5.2 h1:nMtixUijP0Z7iHJNT9fOL+dbmEzZxqU6Xk87ll7hqXg=

k8s/helm_charts2/Chart.yaml

@@ -1,5 +1,5 @@
 apiVersion: v1
 description: SeaweedFS
 name: seaweedfs
-appVersion: "3.14"
-version: "3.14"
+appVersion: "3.16"
+version: "3.16"

k8s/helm_charts2/templates/filer-statefulset.yaml

@@ -46,7 +46,7 @@ spec:
       imagePullSecrets:
         - name: {{ .Values.global.imagePullSecrets }}
       {{- end }}
-      serviceAccountName: seaweefds-rw-sa #hack for delete pod master after migration
+      serviceAccountName: seaweedfs-rw-sa #hack for delete pod master after migration
       terminationGracePeriodSeconds: 60
       {{- if .Values.filer.priorityClassName }}
       priorityClassName: {{ .Values.filer.priorityClassName | quote }}

k8s/helm_charts2/templates/service-account.yaml

@@ -3,7 +3,7 @@
 kind: ClusterRole
 apiVersion: rbac.authorization.k8s.io/v1
 metadata:
-  name: seaweefds-rw-cr
+  name: seaweedfs-rw-cr
 rules:
   - apiGroups: [""]
     resources: ["pods"]
@@ -12,18 +12,18 @@ rules:
 apiVersion: v1
 kind: ServiceAccount
 metadata:
-  name: seaweefds-rw-sa
+  name: seaweedfs-rw-sa
   namespace: {{ .Release.Namespace }}
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
 metadata:
-  name: system:serviceaccount:seaweefds-rw-sa:default
+  name: system:serviceaccount:seaweedfs-rw-sa:default
 subjects:
 - kind: ServiceAccount
-  name: seaweefds-rw-sa
+  name: seaweedfs-rw-sa
   namespace: {{ .Release.Namespace }}
 roleRef:
   apiGroup: rbac.authorization.k8s.io
   kind: ClusterRole
-  name: seaweefds-rw-cr
+  name: seaweedfs-rw-cr

weed/command/scaffold/filer.toml

@@ -337,3 +337,11 @@ pdaddrs = "localhost:2379"
 deleterange_concurrency = 1
 # Enable 1PC
 enable_1pc = false
+# Set the CA certificate path
+ca_path=""
+# Set the certificate path
+cert_path=""
+# Set the private key path
+key_path=""
+# The name list used to verify the cn name
+verify_cn=""

weed/filer/filer.go

@@ -79,7 +79,7 @@ func (f *Filer) MaybeBootstrapFromPeers(self pb.ServerAddress, existingNodes []*
 		return
 	}
 
-	glog.V(0).Infof("bootstrap from %v", earliestNode.Address)
+	glog.V(0).Infof("bootstrap from %v clientId:%d", earliestNode.Address, f.UniqueFileId)
 	err = pb.FollowMetadata(pb.ServerAddress(earliestNode.Address), f.GrpcDialOption, "bootstrap", int32(f.UniqueFileId), "/", nil,
 		0, snapshotTime.UnixNano(), f.Signature, func(resp *filer_pb.SubscribeMetadataResponse) error {
 			return Replay(f.Store, resp)

weed/filer/meta_aggregator.go

@@ -57,7 +57,7 @@ func (ma *MetaAggregator) OnPeerUpdate(update *master_pb.ClusterNodeUpdate, star
 	if update.IsAdd {
 		// every filer should subscribe to a new filer
 		if ma.setActive(address, true) {
-			go ma.loopSubscribeToOnefiler(ma.filer, ma.self, address, startFrom)
+			go ma.loopSubscribeToOneFiler(ma.filer, ma.self, address, startFrom)
 		}
 	} else {
 		ma.setActive(address, false)
@@ -89,10 +89,10 @@ func (ma *MetaAggregator) isActive(address pb.ServerAddress) (isActive bool) {
 	return count > 0 && isActive
 }
 
-func (ma *MetaAggregator) loopSubscribeToOnefiler(f *Filer, self pb.ServerAddress, peer pb.ServerAddress, startFrom time.Time) {
+func (ma *MetaAggregator) loopSubscribeToOneFiler(f *Filer, self pb.ServerAddress, peer pb.ServerAddress, startFrom time.Time) {
 	lastTsNs := startFrom.UnixNano()
 	for {
-		glog.V(0).Infof("loopSubscribeToOnefiler read %s start from %v %d", peer, time.Unix(0, lastTsNs), lastTsNs)
+		glog.V(0).Infof("loopSubscribeToOneFiler read %s start from %v %d", peer, time.Unix(0, lastTsNs), lastTsNs)
 		nextLastTsNs, err := ma.doSubscribeToOneFiler(f, self, peer, lastTsNs)
 		if !ma.isActive(peer) {
 			glog.V(0).Infof("stop subscribing remote %s meta change", peer)
@@ -185,7 +185,7 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 		return nil
 	}
 
-	glog.V(4).Infof("subscribing remote %s meta change: %v", peer, time.Unix(0, lastTsNs))
+	glog.V(0).Infof("subscribing remote %s meta change: %v, clientId:%d", peer, time.Unix(0, lastTsNs), ma.filer.UniqueFileId)
 	err = pb.WithFilerClient(true, peer, ma.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()

weed/filer/read_write.go

@@ -2,9 +2,10 @@ package filer
 
 import (
 	"bytes"
+	"time"
+
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/wdclient"
-	"time"
 )
 
 func ReadEntry(masterClient *wdclient.MasterClient, filerClient filer_pb.SeaweedFilerClient, dir, name string, byteBuffer *bytes.Buffer) error {
@@ -60,7 +61,7 @@ func SaveInsideFiler(client filer_pb.SeaweedFilerClient, dir, name string, conte
 				},
 				Content: content,
 			},
-			SkipCheckParentDirectory: true,
+			SkipCheckParentDirectory: false,
 		})
 	} else if err == nil {
 		entry := resp.Entry

weed/filer/reader_cache.go

@@ -18,7 +18,8 @@ type ReaderCache struct {
 }
 
 type SingleChunkCacher struct {
-	sync.RWMutex
+	sync.Mutex
+	cond          *sync.Cond
 	parent        *ReaderCache
 	chunkFileId   string
 	data          []byte
@@ -75,7 +76,9 @@ func (rc *ReaderCache) ReadChunkAt(buffer []byte, fileId string, cipherKey []byt
 	rc.Lock()
 	defer rc.Unlock()
 	if cacher, found := rc.downloaders[fileId]; found {
-		return cacher.readChunkAt(buffer, offset)
+		if n, err := cacher.readChunkAt(buffer, offset); n != 0 && err == nil {
+			return n, err
+		}
 	}
 	if shouldCache || rc.lookupFileIdFn == nil {
 		n, err := rc.chunkCache.ReadChunkAt(buffer, fileId, uint64(offset))
@@ -140,6 +143,7 @@ func newSingleChunkCacher(parent *ReaderCache, fileId string, cipherKey []byte,
 		chunkSize:   chunkSize,
 		shouldCache: shouldCache,
 	}
+	t.cond = sync.NewCond(t)
 	return t
 }
 
@@ -168,11 +172,15 @@ func (s *SingleChunkCacher) startCaching() {
 	if s.shouldCache {
 		s.parent.chunkCache.SetChunk(s.chunkFileId, s.data)
 	}
+	s.cond.Broadcast()
 
 	return
 }
 
 func (s *SingleChunkCacher) destroy() {
+	s.Lock()
+	defer s.Unlock()
+
 	if s.data != nil {
 		mem.Free(s.data)
 		s.data = nil
@@ -180,13 +188,21 @@ func (s *SingleChunkCacher) destroy() {
 }
 
 func (s *SingleChunkCacher) readChunkAt(buf []byte, offset int64) (int, error) {
-	s.RLock()
-	defer s.RUnlock()
+	s.Lock()
+	defer s.Unlock()
+
+	for s.completedTime.IsZero() {
+		s.cond.Wait()
+	}
 
 	if s.err != nil {
 		return 0, s.err
 	}
 
+	if len(s.data) == 0 {
+		return 0, nil
+	}
+
 	return copy(buf, s.data[offset:]), nil
 
 }

weed/filer/reader_pattern.go

@@ -1,8 +1,8 @@
 package filer
 
 type ReaderPattern struct {
-	isStreaming    bool
-	lastReadOffset int64
+	isSequentialCounter int64
+	lastReadStopOffset  int64
 }
 
 // For streaming read: only cache the first chunk
@@ -10,29 +10,20 @@ type ReaderPattern struct {
 
 func NewReaderPattern() *ReaderPattern {
 	return &ReaderPattern{
-		isStreaming:    true,
-		lastReadOffset: -1,
+		isSequentialCounter: 0,
+		lastReadStopOffset:  0,
 	}
 }
 
 func (rp *ReaderPattern) MonitorReadAt(offset int64, size int) {
-	isStreaming := true
-	if rp.lastReadOffset > offset {
-		isStreaming = false
+	if rp.lastReadStopOffset == offset {
+		rp.isSequentialCounter++
+	} else {
+		rp.isSequentialCounter--
 	}
-	if rp.lastReadOffset == -1 {
-		if offset != 0 {
-			isStreaming = false
-		}
-	}
-	rp.lastReadOffset = offset
-	rp.isStreaming = isStreaming
-}
-
-func (rp *ReaderPattern) IsStreamingMode() bool {
-	return rp.isStreaming
+	rp.lastReadStopOffset = offset + int64(size)
 }
 
 func (rp *ReaderPattern) IsRandomMode() bool {
-	return !rp.isStreaming
+	return rp.isSequentialCounter >= 0
 }

weed/filer/s3iam_conf.go

@@ -2,9 +2,12 @@ package filer
 
 import (
 	"bytes"
+	"fmt"
+	"io"
+
+	"github.com/chrislusf/seaweedfs/weed/pb/iam_pb"
 	"github.com/golang/protobuf/jsonpb"
 	"github.com/golang/protobuf/proto"
-	"io"
 )
 
 func ParseS3ConfigurationFromBytes[T proto.Message](content []byte, config T) error {
@@ -23,3 +26,18 @@ func ProtoToText(writer io.Writer, config proto.Message) error {
 
 	return m.Marshal(writer, config)
 }
+
+// CheckDuplicateAccessKey returns an error message when s3cfg has duplicate access keys
+func CheckDuplicateAccessKey(s3cfg *iam_pb.S3ApiConfiguration) error {
+	accessKeySet := make(map[string]string)
+	for _, ident := range s3cfg.Identities {
+		for _, cred := range ident.Credentials {
+			if userName, found := accessKeySet[cred.AccessKey]; !found {
+				accessKeySet[cred.AccessKey] = ident.Name
+			} else {
+				return fmt.Errorf("duplicate accessKey[%s], already configured in user[%s]", cred.AccessKey, userName)
+			}
+		}
+	}
+	return nil
+}

weed/filer/s3iam_conf_test.go

@@ -2,9 +2,10 @@ package filer
 
 import (
 	"bytes"
-	. "github.com/chrislusf/seaweedfs/weed/s3api/s3_constants"
 	"testing"
 
+	. "github.com/chrislusf/seaweedfs/weed/s3api/s3_constants"
+
 	"github.com/chrislusf/seaweedfs/weed/pb/iam_pb"
 
 	"github.com/stretchr/testify/assert"
@@ -55,3 +56,93 @@ func TestS3Conf(t *testing.T) {
 	assert.Equal(t, "some_access_key1", s3ConfSaved.Identities[0].Credentials[0].AccessKey)
 	assert.Equal(t, "some_secret_key2", s3ConfSaved.Identities[1].Credentials[0].SecretKey)
 }
+
+func TestCheckDuplicateAccessKey(t *testing.T) {
+	var tests = []struct {
+		s3cfg *iam_pb.S3ApiConfiguration
+		err   string
+	}{
+		{
+			&iam_pb.S3ApiConfiguration{
+				Identities: []*iam_pb.Identity{
+					{
+						Name: "some_name",
+						Credentials: []*iam_pb.Credential{
+							{
+								AccessKey: "some_access_key1",
+								SecretKey: "some_secret_key1",
+							},
+						},
+						Actions: []string{
+							ACTION_ADMIN,
+							ACTION_READ,
+							ACTION_WRITE,
+						},
+					},
+					{
+						Name: "some_read_only_user",
+						Credentials: []*iam_pb.Credential{
+							{
+								AccessKey: "some_access_key2",
+								SecretKey: "some_secret_key2",
+							},
+						},
+						Actions: []string{
+							ACTION_READ,
+							ACTION_TAGGING,
+							ACTION_LIST,
+						},
+					},
+				},
+			},
+			"",
+		},
+		{
+			&iam_pb.S3ApiConfiguration{
+				Identities: []*iam_pb.Identity{
+					{
+						Name: "some_name",
+						Credentials: []*iam_pb.Credential{
+							{
+								AccessKey: "some_access_key1",
+								SecretKey: "some_secret_key1",
+							},
+						},
+						Actions: []string{
+							ACTION_ADMIN,
+							ACTION_READ,
+							ACTION_WRITE,
+						},
+					},
+					{
+						Name: "some_read_only_user",
+						Credentials: []*iam_pb.Credential{
+							{
+								AccessKey: "some_access_key1",
+								SecretKey: "some_secret_key1",
+							},
+						},
+						Actions: []string{
+							ACTION_READ,
+							ACTION_TAGGING,
+							ACTION_LIST,
+						},
+					},
+				},
+			},
+			"duplicate accessKey[some_access_key1], already configured in user[some_name]",
+		},
+	}
+	for i, test := range tests {
+		err := CheckDuplicateAccessKey(test.s3cfg)
+		var errString string
+		if err == nil {
+			errString = ""
+		} else {
+			errString = err.Error()
+		}
+		if errString != test.err {
+			t.Errorf("[%d]: got: %s expected: %s", i, errString, test.err)
+		}
+	}
+}

weed/filer/tikv/tikv_store.go

@@ -15,6 +15,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/tikv/client-go/v2/config"
 	"github.com/tikv/client-go/v2/txnkv"
 )
 
@@ -38,21 +39,25 @@ func (store *TikvStore) GetName() string {
 }
 
 func (store *TikvStore) Initialize(config util.Configuration, prefix string) error {
-	pdAddrs := []string{}
-	pdAddrsStr := config.GetString(prefix + "pdaddrs")
-	for _, item := range strings.Split(pdAddrsStr, ",") {
-		pdAddrs = append(pdAddrs, strings.TrimSpace(item))
-	}
+	ca := config.GetString(prefix + "ca_path")
+	cert := config.GetString(prefix + "cert_path")
+	key := config.GetString(prefix + "key_path")
+	verify_cn := strings.Split(config.GetString(prefix+"verify_cn"), ",")
+	pdAddrs := strings.Split(config.GetString(prefix+"pdaddrs"), ",")
+
 	drc := config.GetInt(prefix + "deleterange_concurrency")
 	if drc <= 0 {
 		drc = 1
 	}
 	store.onePC = config.GetBool(prefix + "enable_1pc")
 	store.deleteRangeConcurrency = drc
-	return store.initialize(pdAddrs)
+	return store.initialize(ca, cert, key, verify_cn, pdAddrs)
 }
 
-func (store *TikvStore) initialize(pdAddrs []string) error {
+func (store *TikvStore) initialize(ca, cert, key string, verify_cn, pdAddrs []string) error {
+	config.UpdateGlobal(func(conf *config.Config) {
+		conf.Security = config.NewSecurity(ca, cert, key, verify_cn)
+	})
 	client, err := txnkv.NewClient(pdAddrs)
 	store.client = client
 	return err

weed/iamapi/iamapi_management_handlers.go

@@ -204,6 +204,7 @@ func (iama *IamApiServer) CreatePolicy(s3cfg *iam_pb.S3ApiConfiguration, values
 	return resp, nil
 }
 
+// https://docs.aws.amazon.com/IAM/latest/APIReference/API_PutUserPolicy.html
 func (iama *IamApiServer) PutUserPolicy(s3cfg *iam_pb.S3ApiConfiguration, values url.Values) (resp PutUserPolicyResponse, err error) {
 	userName := values.Get("UserName")
 	policyName := values.Get("PolicyName")
@@ -215,14 +216,15 @@ func (iama *IamApiServer) PutUserPolicy(s3cfg *iam_pb.S3ApiConfiguration, values
 	policyDocuments[policyName] = &policyDocument
 	actions := GetActions(&policyDocument)
 	for _, ident := range s3cfg.Identities {
-		if userName == ident.Name {
-			for _, action := range actions {
-				ident.Actions = append(ident.Actions, action)
-			}
-			break
+		if userName != ident.Name {
+			continue
 		}
+		for _, action := range actions {
+			ident.Actions = append(ident.Actions, action)
+		}
+		return resp, nil
 	}
-	return resp, nil
+	return resp, fmt.Errorf("%s: the user with name %s cannot be found", iam.ErrCodeNoSuchEntityException, userName)
 }
 
 func (iama *IamApiServer) GetUserPolicy(s3cfg *iam_pb.S3ApiConfiguration, values url.Values) (resp GetUserPolicyResponse, err error) {
@@ -412,14 +414,11 @@ func (iama *IamApiServer) DoActions(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	values := r.PostForm
-	var s3cfgLock sync.RWMutex
-	s3cfgLock.RLock()
 	s3cfg := &iam_pb.S3ApiConfiguration{}
 	if err := iama.s3ApiConfig.GetS3ApiConfiguration(s3cfg); err != nil {
 		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		return
 	}
-	s3cfgLock.RUnlock()
 
 	glog.V(4).Infof("DoActions: %+v", values)
 	var response interface{}
@@ -498,9 +497,7 @@ func (iama *IamApiServer) DoActions(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	if changed {
-		s3cfgLock.Lock()
 		err := iama.s3ApiConfig.PutS3ApiConfiguration(s3cfg)
-		s3cfgLock.Unlock()
 		if err != nil {
 			writeIamErrorResponse(w, r, fmt.Errorf(iam.ErrCodeServiceFailureException), "", "", err)
 			return

weed/iamapi/iamapi_server.go

@@ -6,6 +6,8 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"net/http"
+
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
@@ -17,7 +19,6 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/wdclient"
 	"github.com/gorilla/mux"
 	"google.golang.org/grpc"
-	"net/http"
 )
 
 type IamS3ApiConfig interface {
@@ -117,10 +118,10 @@ func (iam IamS3ApiConfigure) GetPolicies(policies *Policies) (err error) {
 		}
 		return nil
 	})
-	if err != nil {
+	if err != nil && err != filer_pb.ErrNotFound {
 		return err
 	}
-	if buf.Len() == 0 {
+	if err == filer_pb.ErrNotFound || buf.Len() == 0 {
 		policies.Policies = make(map[string]PolicyDocument)
 		return nil
 	}

weed/mount/filehandle.go

@@ -98,16 +98,15 @@ func (fh *FileHandle) AddChunks(chunks []*filer_pb.FileChunk) {
 	fh.entryViewCache = nil
 }
 
-func (fh *FileHandle) SetReader(reader *filer.ChunkReadAt) {
+func (fh *FileHandle) CloseReader() {
 	if fh.reader != nil {
 		fh.reader.Close()
 	}
-	fh.reader = reader
 }
 
 func (fh *FileHandle) Release() {
 	fh.dirtyPages.Destroy()
-	fh.SetReader(nil)
+	fh.CloseReader()
 }
 
 func lessThan(a, b *filer_pb.FileChunk) bool {

weed/mount/filehandle_read.go

@@ -62,21 +62,19 @@ func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
 		if chunkResolveErr != nil {
 			return 0, fmt.Errorf("fail to resolve chunk manifest: %v", chunkResolveErr)
 		}
-		fh.SetReader(nil)
+		fh.CloseReader()
 	}
 
-	reader := fh.reader
-	if reader == nil {
+	if fh.reader == nil {
 		chunkViews := filer.ViewFromVisibleIntervals(fh.entryViewCache, 0, fileSize)
 		glog.V(4).Infof("file handle read %s [%d,%d) from %d views", fileFullPath, offset, offset+int64(len(buff)), len(chunkViews))
 		for _, chunkView := range chunkViews {
 			glog.V(4).Infof("  read %s [%d,%d) from chunk %+v", fileFullPath, chunkView.LogicOffset, chunkView.LogicOffset+int64(chunkView.Size), chunkView.FileId)
 		}
-		reader = filer.NewChunkReaderAtFromClient(fh.wfs.LookupFn(), chunkViews, fh.wfs.chunkCache, fileSize)
+		fh.reader = filer.NewChunkReaderAtFromClient(fh.wfs.LookupFn(), chunkViews, fh.wfs.chunkCache, fileSize)
 	}
-	fh.SetReader(reader)
 
-	totalRead, err := reader.ReadAt(buff, offset)
+	totalRead, err := fh.reader.ReadAt(buff, offset)
 
 	if err != nil && err != io.EOF {
 		glog.Errorf("file handle read %s: %v", fileFullPath, err)

weed/mount/inode_to_path.go

@@ -152,7 +152,7 @@ func (i *InodeToPath) RemovePath(path util.FullPath) {
 	}
 }
 
-func (i *InodeToPath) MovePath(sourcePath, targetPath util.FullPath) (replacedInode uint64) {
+func (i *InodeToPath) MovePath(sourcePath, targetPath util.FullPath) (sourceInode, targetInode uint64) {
 	i.Lock()
 	defer i.Unlock()
 	sourceInode, sourceFound := i.path2inode[sourcePath]
@@ -178,7 +178,7 @@ func (i *InodeToPath) MovePath(sourcePath, targetPath util.FullPath) (replacedIn
 	} else {
 		glog.Errorf("MovePath %s to %s: sourceInode %d not found", sourcePath, targetPath, sourceInode)
 	}
-	return targetInode
+	return
 }
 
 func (i *InodeToPath) Forget(inode, nlookup uint64, onForgetDir func(dir util.FullPath)) {

weed/mount/page_writer.go

@@ -29,14 +29,14 @@ func newPageWriter(fh *FileHandle, chunkSize int64) *PageWriter {
 	return pw
 }
 
-func (pw *PageWriter) AddPage(offset int64, data []byte, isSequentail bool) {
+func (pw *PageWriter) AddPage(offset int64, data []byte, isSequential bool) {
 
 	glog.V(4).Infof("%v AddPage [%d, %d)", pw.fh.fh, offset, offset+int64(len(data)))
 
 	chunkIndex := offset / pw.chunkSize
 	for i := chunkIndex; len(data) > 0; i++ {
 		writeSize := min(int64(len(data)), (i+1)*pw.chunkSize-offset)
-		pw.addToOneChunk(i, offset, data[:writeSize], isSequentail)
+		pw.addToOneChunk(i, offset, data[:writeSize], isSequential)
 		offset += writeSize
 		data = data[writeSize:]
 	}

weed/mount/page_writer_pattern.go

@@ -1,9 +1,9 @@
 package mount
 
 type WriterPattern struct {
-	isStreaming     bool
-	lastWriteOffset int64
-	chunkSize       int64
+	isSequentialCounter int64
+	lastWriteStopOffset int64
+	chunkSize           int64
 }
 
 // For streaming write: only cache the first chunk
@@ -12,33 +12,21 @@ type WriterPattern struct {
 
 func NewWriterPattern(chunkSize int64) *WriterPattern {
 	return &WriterPattern{
-		isStreaming:     true,
-		lastWriteOffset: -1,
-		chunkSize:       chunkSize,
+		isSequentialCounter: 0,
+		lastWriteStopOffset: 0,
+		chunkSize:           chunkSize,
 	}
 }
 
 func (rp *WriterPattern) MonitorWriteAt(offset int64, size int) {
-	if rp.lastWriteOffset > offset {
-		rp.isStreaming = false
+	if rp.lastWriteStopOffset == offset {
+		rp.isSequentialCounter++
+	} else {
+		rp.isSequentialCounter--
 	}
-	if rp.lastWriteOffset == -1 {
-		if offset != 0 {
-			rp.isStreaming = false
-		}
-	}
-	rp.lastWriteOffset = offset
+	rp.lastWriteStopOffset = offset + int64(size)
 }
 
-func (rp *WriterPattern) IsStreamingMode() bool {
-	return rp.isStreaming
-}
-
-func (rp *WriterPattern) IsRandomMode() bool {
-	return !rp.isStreaming
-}
-
-func (rp *WriterPattern) Reset() {
-	rp.isStreaming = true
-	rp.lastWriteOffset = -1
+func (rp *WriterPattern) IsSequentialMode() bool {
+	return rp.isSequentialCounter >= 0
 }

weed/mount/weedfs_file_write.go

@@ -58,7 +58,7 @@ func (wfs *WFS) Write(cancel <-chan struct{}, in *fuse.WriteIn, data []byte) (wr
 	entry.Attributes.FileSize = uint64(max(offset+int64(len(data)), int64(entry.Attributes.FileSize)))
 	// glog.V(4).Infof("%v write [%d,%d) %d", fh.f.fullpath(), req.Offset, req.Offset+int64(len(req.Data)), len(req.Data))
 
-	fh.dirtyPages.AddPage(offset, data, fh.dirtyPages.writerPattern.IsStreamingMode())
+	fh.dirtyPages.AddPage(offset, data, fh.dirtyPages.writerPattern.IsSequentialMode())
 
 	written = uint32(len(data))
 

weed/mount/weedfs_rename.go

@@ -233,10 +233,17 @@ func (wfs *WFS) handleRenameResponse(ctx context.Context, resp *filer_pb.StreamR
 		oldPath := oldParent.Child(oldName)
 		newPath := newParent.Child(newName)
 
-		replacedInode := wfs.inodeToPath.MovePath(oldPath, newPath)
-		// invalidate attr and data
-		if replacedInode > 0 {
-			wfs.fuseServer.InodeNotify(replacedInode, 0, -1)
+		sourceInode, targetInode := wfs.inodeToPath.MovePath(oldPath, newPath)
+		if sourceInode != 0 {
+			if fh, foundFh := wfs.fhmap.inode2fh[sourceInode]; foundFh && fh.entry != nil {
+				fh.entry.Name = newName
+			}
+			// invalidate attr and data
+			wfs.fuseServer.InodeNotify(sourceInode, 0, -1)
+		}
+		if targetInode != 0 {
+			// invalidate attr and data
+			wfs.fuseServer.InodeNotify(targetInode, 0, -1)
 		}
 
 	} else if resp.EventNotification.OldEntry != nil {

weed/s3api/auth_credentials.go

@@ -109,6 +109,11 @@ func (iam *IdentityAccessManagement) LoadS3ApiConfigurationFromBytes(content []b
 		glog.Warningf("unmarshal error: %v", err)
 		return fmt.Errorf("unmarshal error: %v", err)
 	}
+
+	if err := filer.CheckDuplicateAccessKey(s3ApiConfiguration); err != nil {
+		return err
+	}
+
 	if err := iam.loadS3ApiConfiguration(s3ApiConfiguration); err != nil {
 		return err
 	}

weed/s3api/s3api_object_handlers.go

@@ -445,9 +445,9 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, uploadUrl string, dataReader
 func setEtag(w http.ResponseWriter, etag string) {
 	if etag != "" {
 		if strings.HasPrefix(etag, "\"") {
-			w.Header().Set("ETag", etag)
+			w.Header()["ETag"] = []string{etag}
 		} else {
-			w.Header().Set("ETag", "\""+etag+"\"")
+			w.Header()["ETag"] = []string{"\"" + etag + "\""}
 		}
 	}
 }

weed/server/filer_grpc_server_sub_meta.go

@@ -90,14 +90,20 @@ func (fs *FilerServer) SubscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq
 
 	peerAddress := findClientAddress(stream.Context(), 0)
 
+	// use negative client id to differentiate from addClient()/deleteClient() used in SubscribeMetadata()
+	req.ClientId = -req.ClientId
+
 	alreadyKnown, clientName := fs.addClient(req.ClientName, peerAddress, req.ClientId)
 	if alreadyKnown {
-		return fmt.Errorf("duplicated local subscription detected for client %s id %d", clientName, req.ClientId)
+		return fmt.Errorf("duplicated local subscription detected for client %s clientId:%d", clientName, req.ClientId)
 	}
-	defer fs.deleteClient(clientName, req.ClientId)
+	defer func() {
+		glog.V(0).Infof(" - %v local subscribe %s clientId:%d", clientName, req.PathPrefix, req.ClientId)
+		fs.deleteClient(clientName, req.ClientId)
+	}()
 
 	lastReadTime := time.Unix(0, req.SinceNs)
-	glog.V(0).Infof(" %v local subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
+	glog.V(0).Infof(" + %v local subscribe %s from %+v clientId:%d", clientName, req.PathPrefix, lastReadTime, req.ClientId)
 
 	eachEventNotificationFn := fs.eachEventNotificationFn(req, stream, clientName)
 
@@ -263,6 +269,9 @@ func (fs *FilerServer) addClient(clientType string, clientAddress string, client
 	if clientId != 0 {
 		fs.knownListenersLock.Lock()
 		_, alreadyKnown = fs.knownListeners[clientId]
+		if !alreadyKnown {
+			fs.knownListeners[clientId] = struct{}{}
+		}
 		fs.knownListenersLock.Unlock()
 	}
 	return

weed/shell/command_s3_configure.go

@@ -4,11 +4,12 @@ import (
 	"bytes"
 	"flag"
 	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/filer"
 	"io"
 	"sort"
 	"strings"
 
+	"github.com/chrislusf/seaweedfs/weed/filer"
+
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/iam_pb"
 )
@@ -164,6 +165,10 @@ func (c *commandS3Configure) Do(args []string, commandEnv *CommandEnv, writer io
 		s3cfg.Identities = append(s3cfg.Identities, &identity)
 	}
 
+	if err = filer.CheckDuplicateAccessKey(s3cfg); err != nil {
+		return err
+	}
+
 	buf.Reset()
 	filer.ProtoToText(&buf, s3cfg)
 

weed/storage/disk_location.go

@@ -4,6 +4,8 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"runtime"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -206,7 +208,21 @@ func (l *DiskLocation) concurrentLoadingVolumes(needleMapKind NeedleMapKind, con
 
 func (l *DiskLocation) loadExistingVolumes(needleMapKind NeedleMapKind) {
 
-	l.concurrentLoadingVolumes(needleMapKind, 10)
+	workerNum := runtime.NumCPU()
+	val, ok := os.LookupEnv("GOMAXPROCS")
+	if ok {
+		num, err := strconv.Atoi(val)
+		if err != nil || num < 1 {
+			num = 10
+			glog.Warningf("failed to set worker number from GOMAXPROCS , set to default:10")
+		}
+		workerNum = num
+	} else {
+		if workerNum <= 10 {
+			workerNum = 10
+		}
+	}
+	l.concurrentLoadingVolumes(needleMapKind, workerNum)
 	glog.V(0).Infof("Store started on dir: %s with %d volumes max %d", l.Directory, len(l.volumes), l.MaxVolumeCount)
 
 	l.loadAllEcShards()

weed/topology/topology_vacuum.go

@@ -185,10 +185,13 @@ func (t *Topology) Vacuum(grpcDialOption grpc.DialOption, garbageThreshold float
 		for _, vl := range c.storageType2VolumeLayout.Items() {
 			if vl != nil {
 				volumeLayout := vl.(*VolumeLayout)
-				if volumeId > 0 && volumeLayout.Lookup(needle.VolumeId(volumeId)) == nil {
-					continue
+				if volumeId > 0 {
+					if volumeLayout.Lookup(needle.VolumeId(volumeId)) != nil {
+						t.vacuumOneVolumeLayout(grpcDialOption, volumeLayout, c, garbageThreshold, preallocate)
+					}
+				} else {
+					t.vacuumOneVolumeLayout(grpcDialOption, volumeLayout, c, garbageThreshold, preallocate)
 				}
-				t.vacuumOneVolumeLayout(grpcDialOption, volumeLayout, c, garbageThreshold, preallocate)
 			}
 		}
 	}

weed/util/constants.go

@@ -5,7 +5,7 @@ import (
 )
 
 var (
-	VERSION_NUMBER = fmt.Sprintf("%.02f", 3.14)
+	VERSION_NUMBER = fmt.Sprintf("%.02f", 3.16)
 	VERSION        = sizeLimit + " " + VERSION_NUMBER
 	COMMIT         = ""
 )