2.80

minor

.github/workflows/binaries_dev.yml

@@ -37,14 +37,14 @@ jobs:
         run: echo BUILD_TIME=$(date -u +%Y%m%d-%H%M) >> ${GITHUB_ENV}
 
       - name: Go Release Binaries Large Disk
-        uses: wangyoucao577/go-release-action@bugfix/upload-fail
+        uses: wangyoucao577/go-release-action@v1.22
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           goos: ${{ matrix.goos }}
           goarch: ${{ matrix.goarch }}
           release_tag: dev
           overwrite: true
-          pre_command: export CGO_ENABLED=0
+          pre_command: export CGO_ENABLED=0 && export GODEBUG=http2client=0
           build_flags: -tags 5BytesOffset # optional, default is
           ldflags: -extldflags -static -X github.com/chrislusf/seaweedfs/weed/util.COMMIT=${{github.sha}}
           # Where to run `go build .`
@@ -53,14 +53,14 @@ jobs:
           asset_name: "weed-large-disk-${{ env.BUILD_TIME }}-${{ matrix.goos }}-${{ matrix.goarch }}"
 
       - name: Go Release Binaries Normal Volume Size
-        uses: wangyoucao577/go-release-action@bugfix/upload-fail
+        uses: wangyoucao577/go-release-action@v1.22
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           goos: ${{ matrix.goos }}
           goarch: ${{ matrix.goarch }}
           release_tag: dev
           overwrite: true
-          pre_command: export CGO_ENABLED=0
+          pre_command: export CGO_ENABLED=0 && export GODEBUG=http2client=0
           ldflags: -extldflags -static -X github.com/chrislusf/seaweedfs/weed/util.COMMIT=${{github.sha}}
           # Where to run `go build .`
           project_path: weed
@@ -84,14 +84,14 @@ jobs:
         run: echo BUILD_TIME=$(date -u +%Y%m%d-%H%M) >> ${GITHUB_ENV}
 
       - name: Go Release Binaries Large Disk
-        uses: wangyoucao577/go-release-action@bugfix/upload-fail
+        uses: wangyoucao577/go-release-action@v1.22
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           goos: ${{ matrix.goos }}
           goarch: ${{ matrix.goarch }}
           release_tag: dev
           overwrite: true
-          pre_command: export CGO_ENABLED=0
+          pre_command: export CGO_ENABLED=0 && export GODEBUG=http2client=0
           build_flags: -tags 5BytesOffset # optional, default is
           ldflags: -extldflags -static -X github.com/chrislusf/seaweedfs/weed/util.COMMIT=${{github.sha}}
           # Where to run `go build .`
@@ -100,14 +100,14 @@ jobs:
           asset_name: "weed-large-disk-${{ env.BUILD_TIME }}-${{ matrix.goos }}-${{ matrix.goarch }}"
 
       - name: Go Release Binaries Normal Volume Size
-        uses: wangyoucao577/go-release-action@bugfix/upload-fail
+        uses: wangyoucao577/go-release-action@v1.22
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           goos: ${{ matrix.goos }}
           goarch: ${{ matrix.goarch }}
           release_tag: dev
           overwrite: true
-          pre_command: export CGO_ENABLED=0
+          pre_command: export CGO_ENABLED=0 && export GODEBUG=http2client=0
           ldflags: -extldflags -static -X github.com/chrislusf/seaweedfs/weed/util.COMMIT=${{github.sha}}
           # Where to run `go build .`
           project_path: weed

.github/workflows/binaries_release0.yml

@@ -25,13 +25,13 @@ jobs:
       # Checks-out your repository under $GITHUB_WORKSPACE, so your job can access it
       - uses: actions/checkout@v2
       - name: Go Release Binaries Normal Volume Size
-        uses: wangyoucao577/go-release-action@v1.20
+        uses: wangyoucao577/go-release-action@v1.22
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           goos: ${{ matrix.goos }}
           goarch: ${{ matrix.goarch }}
           overwrite: true
-          pre_command: export CGO_ENABLED=0
+          pre_command: export CGO_ENABLED=0 && export GODEBUG=http2client=0
           # build_flags: -tags 5BytesOffset # optional, default is
           ldflags: -extldflags -static -X github.com/chrislusf/seaweedfs/weed/util.COMMIT=${{github.sha}}
           # Where to run `go build .`
@@ -39,13 +39,13 @@ jobs:
           binary_name: weed
           asset_name: "${{ matrix.goos }}_${{ matrix.goarch }}"
       - name: Go Release Large Disk Binaries
-        uses: wangyoucao577/go-release-action@v1.20
+        uses: wangyoucao577/go-release-action@v1.22
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           goos: ${{ matrix.goos }}
           goarch: ${{ matrix.goarch }}
           overwrite: true
-          pre_command: export CGO_ENABLED=0
+          pre_command: export CGO_ENABLED=0 && export GODEBUG=http2client=0
           build_flags: -tags 5BytesOffset # optional, default is
           ldflags: -extldflags -static -X github.com/chrislusf/seaweedfs/weed/util.COMMIT=${{github.sha}}
           # Where to run `go build .`

.github/workflows/binaries_release1.yml

@@ -25,13 +25,13 @@ jobs:
       # Checks-out your repository under $GITHUB_WORKSPACE, so your job can access it
       - uses: actions/checkout@v2
       - name: Go Release Binaries Normal Volume Size
-        uses: wangyoucao577/go-release-action@v1.20
+        uses: wangyoucao577/go-release-action@v1.22
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           goos: ${{ matrix.goos }}
           goarch: ${{ matrix.goarch }}
           overwrite: true
-          pre_command: export CGO_ENABLED=0
+          pre_command: export CGO_ENABLED=0 && export GODEBUG=http2client=0
           # build_flags: -tags 5BytesOffset # optional, default is
           ldflags: -extldflags -static -X github.com/chrislusf/seaweedfs/weed/util.COMMIT=${{github.sha}}
           # Where to run `go build .`
@@ -39,13 +39,13 @@ jobs:
           binary_name: weed
           asset_name: "${{ matrix.goos }}_${{ matrix.goarch }}"
       - name: Go Release Large Disk Binaries
-        uses: wangyoucao577/go-release-action@v1.20
+        uses: wangyoucao577/go-release-action@v1.22
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           goos: ${{ matrix.goos }}
           goarch: ${{ matrix.goarch }}
           overwrite: true
-          pre_command: export CGO_ENABLED=0
+          pre_command: export CGO_ENABLED=0 && export GODEBUG=http2client=0
           build_flags: -tags 5BytesOffset # optional, default is
           ldflags: -extldflags -static -X github.com/chrislusf/seaweedfs/weed/util.COMMIT=${{github.sha}}
           # Where to run `go build .`

.github/workflows/binaries_release2.yml

@@ -25,13 +25,13 @@ jobs:
       # Checks-out your repository under $GITHUB_WORKSPACE, so your job can access it
       - uses: actions/checkout@v2
       - name: Go Release Binaries Normal Volume Size
-        uses: wangyoucao577/go-release-action@v1.20
+        uses: wangyoucao577/go-release-action@v1.22
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           goos: ${{ matrix.goos }}
           goarch: ${{ matrix.goarch }}
           overwrite: true
-          pre_command: export CGO_ENABLED=0
+          pre_command: export CGO_ENABLED=0 && export GODEBUG=http2client=0
           # build_flags: -tags 5BytesOffset # optional, default is
           ldflags: -extldflags -static -X github.com/chrislusf/seaweedfs/weed/util.COMMIT=${{github.sha}}
           # Where to run `go build .`
@@ -39,13 +39,13 @@ jobs:
           binary_name: weed
           asset_name: "${{ matrix.goos }}_${{ matrix.goarch }}"
       - name: Go Release Large Disk Binaries
-        uses: wangyoucao577/go-release-action@v1.20
+        uses: wangyoucao577/go-release-action@v1.22
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           goos: ${{ matrix.goos }}
           goarch: ${{ matrix.goarch }}
           overwrite: true
-          pre_command: export CGO_ENABLED=0
+          pre_command: export CGO_ENABLED=0 && export GODEBUG=http2client=0
           build_flags: -tags 5BytesOffset # optional, default is
           ldflags: -extldflags -static -X github.com/chrislusf/seaweedfs/weed/util.COMMIT=${{github.sha}}
           # Where to run `go build .`

.github/workflows/binaries_release3.yml

@@ -25,13 +25,13 @@ jobs:
       # Checks-out your repository under $GITHUB_WORKSPACE, so your job can access it
       - uses: actions/checkout@v2
       - name: Go Release Binaries Normal Volume Size
-        uses: wangyoucao577/go-release-action@v1.20
+        uses: wangyoucao577/go-release-action@v1.22
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           goos: ${{ matrix.goos }}
           goarch: ${{ matrix.goarch }}
           overwrite: true
-          pre_command: export CGO_ENABLED=0
+          pre_command: export CGO_ENABLED=0 && export GODEBUG=http2client=0
           # build_flags: -tags 5BytesOffset # optional, default is
           ldflags: -extldflags -static -X github.com/chrislusf/seaweedfs/weed/util.COMMIT=${{github.sha}}
           # Where to run `go build .`
@@ -39,13 +39,13 @@ jobs:
           binary_name: weed
           asset_name: "${{ matrix.goos }}_${{ matrix.goarch }}"
       - name: Go Release Large Disk Binaries
-        uses: wangyoucao577/go-release-action@v1.20
+        uses: wangyoucao577/go-release-action@v1.22
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           goos: ${{ matrix.goos }}
           goarch: ${{ matrix.goarch }}
           overwrite: true
-          pre_command: export CGO_ENABLED=0
+          pre_command: export CGO_ENABLED=0 && export GODEBUG=http2client=0
           build_flags: -tags 5BytesOffset # optional, default is
           ldflags: -extldflags -static -X github.com/chrislusf/seaweedfs/weed/util.COMMIT=${{github.sha}}
           # Where to run `go build .`

backers.md

@@ -5,12 +5,16 @@
 
 <h2 align="center">Generous Backers ($50+)</h2>
 
-- [4Sight Imaging](https://www.4sightimaging.com/)
 - [Evercam Camera Management Software](https://evercam.io/)
-- [Admiral](https://getadmiral.com)
 
 <h2 align="center">Backers</h2>
 
 - [ColorfulClouds Tech Co. Ltd.](https://caiyunai.com/)
 - [Haravan - Ecommerce Platform](https://www.haravan.com)
 - PeterCxy - Creator of Shelter App
+- [Hive Games](https://playhive.com/)
+- Flowm
+- Yoni Nakache
+- Catalin Constantin
+- MingLi Yuan
+- Leroy van Logchem

docker/Dockerfile

@@ -48,6 +48,7 @@ EXPOSE 7333
 RUN mkdir -p /data/filerldb2
 
 VOLUME /data
+WORKDIR /data
 
 COPY filer.toml /etc/seaweedfs/filer.toml
 COPY entrypoint.sh /entrypoint.sh

docker/Dockerfile.gccgo_build

@@ -37,6 +37,7 @@ EXPOSE 7333
 RUN mkdir -p /data/filerldb2
 
 VOLUME /data
+WORKDIR /data
 
 RUN chmod +x /entrypoint.sh
 

docker/Dockerfile.go_build

@@ -36,6 +36,7 @@ EXPOSE 7333
 RUN mkdir -p /data/filerldb2
 
 VOLUME /data
+WORKDIR /data
 
 RUN chmod +x /entrypoint.sh
 

docker/Dockerfile.go_build_large

@@ -36,6 +36,7 @@ EXPOSE 7333
 RUN mkdir -p /data/filerldb2
 
 VOLUME /data
+WORKDIR /data
 
 RUN chmod +x /entrypoint.sh
 

docker/Dockerfile.local

@@ -26,6 +26,7 @@ EXPOSE 7333
 RUN mkdir -p /data/filerldb2
 
 VOLUME /data
+WORKDIR /data
 
 RUN chmod +x /entrypoint.sh
 

docker/Dockerfile.rocksdb_large

@@ -54,6 +54,8 @@ RUN mkdir -p /data/filerldb2
 
 VOLUME /data
 
+WORKDIR /data
+
 RUN chmod +x /entrypoint.sh
 
 ENTRYPOINT ["/entrypoint.sh"]

k8s/helm_charts2/Chart.yaml

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

k8s/helm_charts2/templates/ingress.yaml

@@ -1,61 +1,67 @@
+{{- if .Values.filer.ingress.enabled }}
+{{- if semverCompare ">=1.19-0" .Capabilities.KubeVersion.GitVersion }}
+apiVersion: networking.k8s.io/v1
+{{- else if semverCompare ">=1.14-0" .Capabilities.KubeVersion.GitVersion }}
+apiVersion: networking.k8s.io/v1beta1
+{{- else }}
 apiVersion: extensions/v1beta1
+{{- end }}
 kind: Ingress
 metadata:
- name: ingress-{{ template "seaweedfs.name" . }}-filer
- namespace: {{ .Release.Namespace }}
- annotations:
-   kubernetes.io/ingress.class: "nginx"
-   nginx.ingress.kubernetes.io/auth-type: "basic"
-   nginx.ingress.kubernetes.io/auth-secret: "default/ingress-basic-auth-secret"
-   nginx.ingress.kubernetes.io/auth-realm: 'Authentication Required - SW-Filer'
-   nginx.ingress.kubernetes.io/service-upstream: "true"
-   nginx.ingress.kubernetes.io/rewrite-target: /$1
-   nginx.ingress.kubernetes.io/use-regex: "true"
-   nginx.ingress.kubernetes.io/enable-rewrite-log: "true"
-   nginx.ingress.kubernetes.io/ssl-redirect: "false"
-   nginx.ingress.kubernetes.io/force-ssl-redirect: "false"
-   nginx.ingress.kubernetes.io/configuration-snippet: |
-     sub_filter '<head>' '<head> <base href="/sw-filer/">'; #add base url
-     sub_filter '="/' '="./';                               #make absolute paths to relative
-     sub_filter '=/' '=./';
-     sub_filter '/seaweedfsstatic' './seaweedfsstatic';
-     sub_filter_once off;
+  name: ingress-{{ template "seaweedfs.name" . }}-filer
+  namespace: {{ .Release.Namespace }}
+  annotations:
+    {{ omit .Values.filer.ingress.annotations "kubernetes.io/ingress.class" | toYaml | nindent 4 }}
 spec:
- rules:
- - http:
-     paths:
-     - path: /sw-filer/?(.*)
-       backend:
-         serviceName: {{ template "seaweedfs.name" . }}-filer
-         servicePort: {{ .Values.filer.port }}
+  ingressClassName: {{ .Values.filer.ingress.className | quote }}
+  rules:
+  - http:
+      paths:
+      - path: /sw-filer/?(.*)
+        pathType: ImplementationSpecific
+        backend:
+{{- if semverCompare ">=1.19-0" .Capabilities.KubeVersion.GitVersion }}
+          service:
+            name: {{ template "seaweedfs.name" . }}-filer
+            port:
+              number: {{ .Values.filer.port }}
+              #name:
+{{- else }}
+          serviceName: {{ template "seaweedfs.name" . }}-filer
+          servicePort: {{ .Values.filer.port }}
+{{- end }}
+{{- end }}
 ---
+{{- if .Values.master.ingress.enabled }}
+{{- if semverCompare ">=1.19-0" .Capabilities.KubeVersion.GitVersion }}
+apiVersion: networking.k8s.io/v1
+{{- else if semverCompare ">=1.14-0" .Capabilities.KubeVersion.GitVersion }}
+apiVersion: networking.k8s.io/v1beta1
+{{- else }}
 apiVersion: extensions/v1beta1
+{{- end }}
 kind: Ingress
 metadata:
   name: ingress-{{ template "seaweedfs.name" . }}-master
   namespace: {{ .Release.Namespace }}
   annotations:
-    kubernetes.io/ingress.class: "nginx"
-    nginx.ingress.kubernetes.io/auth-type: "basic"
-    nginx.ingress.kubernetes.io/auth-secret: "default/ingress-basic-auth-secret"
-    nginx.ingress.kubernetes.io/auth-realm: 'Authentication Required - SW-Master'
-    nginx.ingress.kubernetes.io/service-upstream: "true"
-    nginx.ingress.kubernetes.io/rewrite-target: /$1
-    nginx.ingress.kubernetes.io/use-regex: "true"
-    nginx.ingress.kubernetes.io/enable-rewrite-log: "true"
-    nginx.ingress.kubernetes.io/ssl-redirect: "false"
-    nginx.ingress.kubernetes.io/force-ssl-redirect: "false"
-    nginx.ingress.kubernetes.io/configuration-snippet: |
-      sub_filter '<head>' '<head> <base href="/sw-master/">'; #add base url
-      sub_filter '="/' '="./';                                #make absolute paths to relative
-      sub_filter '=/' '=./';
-      sub_filter '/seaweedfsstatic' './seaweedfsstatic';
-      sub_filter_once off;
+    {{ omit .Values.master.ingress.annotations "kubernetes.io/ingress.class" | toYaml | nindent 4 }}
 spec:
+  ingressClassName: {{ .Values.master.ingress.className | quote }}
   rules:
   - http:
       paths:
       - path: /sw-master/?(.*)
+        pathType: ImplementationSpecific
         backend:
+{{- if semverCompare ">=1.19-0" .Capabilities.KubeVersion.GitVersion }}
+          service:
+            name: {{ template "seaweedfs.name" . }}-master
+            port:
+              number: {{ .Values.master.port }}
+              #name:
+{{- else }}
           serviceName: {{ template "seaweedfs.name" . }}-master
           servicePort: {{ .Values.master.port }}
+{{- end }}
+{{- end }}

k8s/helm_charts2/templates/service-account.yaml

@@ -1,7 +1,7 @@
 #hack for delete pod master after migration
 ---
 kind: ClusterRole
-apiVersion: rbac.authorization.k8s.io/v1beta1
+apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: seaweefds-rw-cr
 rules:
@@ -16,7 +16,7 @@ metadata:
   namespace: {{ .Release.Namespace }}
 ---
 kind: ClusterRoleBinding
-apiVersion: rbac.authorization.k8s.io/v1beta1
+apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: system:serviceaccount:seaweefds-rw-sa:default
 subjects:
@@ -26,4 +26,4 @@ subjects:
 roleRef:
   apiGroup: rbac.authorization.k8s.io
   kind: ClusterRole
-  name: seaweefds-rw-cr
+  name: seaweefds-rw-cr

k8s/helm_charts2/values.yaml

@@ -109,6 +109,26 @@ master:
   # ref: https://kubernetes.io/docs/concepts/configuration/pod-priority-preemption/
   priorityClassName: ""
 
+  ingress:
+    enabled: false
+    className: "nginx"
+    annotations:
+      nginx.ingress.kubernetes.io/auth-type: "basic"
+      nginx.ingress.kubernetes.io/auth-secret: "default/ingress-basic-auth-secret"
+      nginx.ingress.kubernetes.io/auth-realm: 'Authentication Required - SW-Master'
+      nginx.ingress.kubernetes.io/service-upstream: "true"
+      nginx.ingress.kubernetes.io/rewrite-target: /$1
+      nginx.ingress.kubernetes.io/use-regex: "true"
+      nginx.ingress.kubernetes.io/enable-rewrite-log: "true"
+      nginx.ingress.kubernetes.io/ssl-redirect: "false"
+      nginx.ingress.kubernetes.io/force-ssl-redirect: "false"
+      nginx.ingress.kubernetes.io/configuration-snippet: |
+        sub_filter '<head>' '<head> <base href="/sw-master/">'; #add base url
+        sub_filter '="/' '="./';                                #make absolute paths to relative
+        sub_filter '=/' '=./';
+        sub_filter '/seaweedfsstatic' './seaweedfsstatic';
+        sub_filter_once off;
+
   extraEnvironmentVars:
     WEED_MASTER_VOLUME_GROWTH_COPY_1: 7
     WEED_MASTER_VOLUME_GROWTH_COPY_2: 6
@@ -309,6 +329,26 @@ filer:
   # ref: https://kubernetes.io/docs/concepts/configuration/pod-priority-preemption/
   priorityClassName: ""
 
+  ingress:
+    enabled: false
+    className: "nginx"
+    annotations:
+      nginx.ingress.kubernetes.io/auth-type: "basic"
+      nginx.ingress.kubernetes.io/auth-secret: "default/ingress-basic-auth-secret"
+      nginx.ingress.kubernetes.io/auth-realm: 'Authentication Required - SW-Filer'
+      nginx.ingress.kubernetes.io/service-upstream: "true"
+      nginx.ingress.kubernetes.io/rewrite-target: /$1
+      nginx.ingress.kubernetes.io/use-regex: "true"
+      nginx.ingress.kubernetes.io/enable-rewrite-log: "true"
+      nginx.ingress.kubernetes.io/ssl-redirect: "false"
+      nginx.ingress.kubernetes.io/force-ssl-redirect: "false"
+      nginx.ingress.kubernetes.io/configuration-snippet: |
+        sub_filter '<head>' '<head> <base href="/sw-filer/">'; #add base url
+        sub_filter '="/' '="./';                               #make absolute paths to relative
+        sub_filter '=/' '=./';
+        sub_filter '/seaweedfsstatic' './seaweedfsstatic';
+        sub_filter_once off;
+
   # extraEnvVars is a list of extra enviroment variables to set with the stateful set.
   extraEnvironmentVars:
     WEED_MYSQL_ENABLED: "true"

weed/Makefile

@@ -2,10 +2,13 @@ BINARY = weed
 
 SOURCE_DIR = .
 
-all: debug_mount
+all: install
 
 .PHONY : clean debug_mount
 
+install:
+	go install
+
 clean:
 	go clean $(SOURCE_DIR)
 	rm -f $(BINARY)

weed/cluster/cluster.go

@@ -1,4 +1,4 @@
-package election
+package cluster
 
 import (
 	"github.com/chrislusf/seaweedfs/weed/pb"
@@ -8,6 +8,12 @@ import (
 	"time"
 )
 
+const (
+	MasterType = "master"
+	FilerType  = "filer"
+	BrokerType = "broker"
+)
+
 type ClusterNode struct {
 	Address   pb.ServerAddress
 	Version   string
@@ -20,78 +26,129 @@ type Leaders struct {
 }
 
 type Cluster struct {
-	nodes     map[pb.ServerAddress]*ClusterNode
-	nodesLock sync.RWMutex
-	leaders   *Leaders
+	filers       map[pb.ServerAddress]*ClusterNode
+	filersLock   sync.RWMutex
+	filerLeaders *Leaders
+	brokers      map[pb.ServerAddress]*ClusterNode
+	brokersLock  sync.RWMutex
 }
 
 func NewCluster() *Cluster {
 	return &Cluster{
-		nodes:   make(map[pb.ServerAddress]*ClusterNode),
-		leaders: &Leaders{},
+		filers:       make(map[pb.ServerAddress]*ClusterNode),
+		filerLeaders: &Leaders{},
+		brokers:      make(map[pb.ServerAddress]*ClusterNode),
 	}
 }
 
 func (cluster *Cluster) AddClusterNode(nodeType string, address pb.ServerAddress, version string) []*master_pb.KeepConnectedResponse {
 	switch nodeType {
-	case "filer":
-		cluster.nodesLock.Lock()
-		defer cluster.nodesLock.Unlock()
-		if existingNode, found := cluster.nodes[address]; found {
+	case FilerType:
+		cluster.filersLock.Lock()
+		defer cluster.filersLock.Unlock()
+		if existingNode, found := cluster.filers[address]; found {
 			existingNode.counter++
 			return nil
 		}
-		cluster.nodes[address] = &ClusterNode{
+		cluster.filers[address] = &ClusterNode{
 			Address:   address,
 			Version:   version,
 			counter:   1,
 			createdTs: time.Now(),
 		}
-		return cluster.ensureLeader(true, nodeType, address)
-	case "master":
+		return cluster.ensureFilerLeaders(true, nodeType, address)
+	case BrokerType:
+		cluster.brokersLock.Lock()
+		defer cluster.brokersLock.Unlock()
+		if existingNode, found := cluster.brokers[address]; found {
+			existingNode.counter++
+			return nil
+		}
+		cluster.brokers[address] = &ClusterNode{
+			Address:   address,
+			Version:   version,
+			counter:   1,
+			createdTs: time.Now(),
+		}
+		return []*master_pb.KeepConnectedResponse{
+			{
+				ClusterNodeUpdate: &master_pb.ClusterNodeUpdate{
+					NodeType: nodeType,
+					Address:  string(address),
+					IsAdd:    true,
+				},
+			},
+		}
+	case MasterType:
 	}
 	return nil
 }
 
 func (cluster *Cluster) RemoveClusterNode(nodeType string, address pb.ServerAddress) []*master_pb.KeepConnectedResponse {
 	switch nodeType {
-	case "filer":
-		cluster.nodesLock.Lock()
-		defer cluster.nodesLock.Unlock()
-		if existingNode, found := cluster.nodes[address]; !found {
+	case FilerType:
+		cluster.filersLock.Lock()
+		defer cluster.filersLock.Unlock()
+		if existingNode, found := cluster.filers[address]; !found {
 			return nil
 		} else {
 			existingNode.counter--
 			if existingNode.counter <= 0 {
-				delete(cluster.nodes, address)
-				return cluster.ensureLeader(false, nodeType, address)
+				delete(cluster.filers, address)
+				return cluster.ensureFilerLeaders(false, nodeType, address)
 			}
 		}
-	case "master":
+	case BrokerType:
+		cluster.brokersLock.Lock()
+		defer cluster.brokersLock.Unlock()
+		if existingNode, found := cluster.brokers[address]; !found {
+			return nil
+		} else {
+			existingNode.counter--
+			if existingNode.counter <= 0 {
+				delete(cluster.brokers, address)
+				return []*master_pb.KeepConnectedResponse{
+					{
+						ClusterNodeUpdate: &master_pb.ClusterNodeUpdate{
+							NodeType: nodeType,
+							Address:  string(address),
+							IsAdd:    false,
+						},
+					},
+				}
+			}
+		}
+	case MasterType:
 	}
 	return nil
 }
 
 func (cluster *Cluster) ListClusterNode(nodeType string) (nodes []*ClusterNode) {
 	switch nodeType {
-	case "filer":
-		cluster.nodesLock.RLock()
-		defer cluster.nodesLock.RUnlock()
-		for _, node := range cluster.nodes {
+	case FilerType:
+		cluster.filersLock.RLock()
+		defer cluster.filersLock.RUnlock()
+		for _, node := range cluster.filers {
 			nodes = append(nodes, node)
 		}
-	case "master":
+	case BrokerType:
+		cluster.brokersLock.RLock()
+		defer cluster.brokersLock.RUnlock()
+		for _, node := range cluster.brokers {
+			nodes = append(nodes, node)
+		}
+	case MasterType:
 	}
 	return
 }
 
 func (cluster *Cluster) IsOneLeader(address pb.ServerAddress) bool {
-	return cluster.leaders.isOneLeader(address)
+	return cluster.filerLeaders.isOneLeader(address)
 }
 
-func (cluster *Cluster) ensureLeader(isAdd bool, nodeType string, address pb.ServerAddress) (result []*master_pb.KeepConnectedResponse) {
+func (cluster *Cluster) ensureFilerLeaders(isAdd bool, nodeType string, address pb.ServerAddress) (result []*master_pb.KeepConnectedResponse) {
 	if isAdd {
-		if cluster.leaders.addLeaderIfVacant(address) {
+		if cluster.filerLeaders.addLeaderIfVacant(address) {
 			// has added the address as one leader
 			result = append(result, &master_pb.KeepConnectedResponse{
 				ClusterNodeUpdate: &master_pb.ClusterNodeUpdate{
@@ -112,7 +169,7 @@ func (cluster *Cluster) ensureLeader(isAdd bool, nodeType string, address pb.Ser
 			})
 		}
 	} else {
-		if cluster.leaders.removeLeaderIfExists(address) {
+		if cluster.filerLeaders.removeLeaderIfExists(address) {
 
 			result = append(result, &master_pb.KeepConnectedResponse{
 				ClusterNodeUpdate: &master_pb.ClusterNodeUpdate{
@@ -127,8 +184,8 @@ func (cluster *Cluster) ensureLeader(isAdd bool, nodeType string, address pb.Ser
 			var shortestDuration int64 = math.MaxInt64
 			now := time.Now()
 			var candidateAddress pb.ServerAddress
-			for _, node := range cluster.nodes {
-				if cluster.leaders.isOneLeader(node.Address) {
+			for _, node := range cluster.filers {
+				if cluster.filerLeaders.isOneLeader(node.Address) {
 					continue
 				}
 				duration := now.Sub(node.createdTs).Nanoseconds()
@@ -138,7 +195,7 @@ func (cluster *Cluster) ensureLeader(isAdd bool, nodeType string, address pb.Ser
 				}
 			}
 			if candidateAddress != "" {
-				cluster.leaders.addLeaderIfVacant(candidateAddress)
+				cluster.filerLeaders.addLeaderIfVacant(candidateAddress)
 				// added a new leader
 				result = append(result, &master_pb.KeepConnectedResponse{
 					ClusterNodeUpdate: &master_pb.ClusterNodeUpdate{

weed/cluster/cluster_test.go

@@ -1,4 +1,4 @@
-package election
+package cluster
 
 import (
 	"github.com/chrislusf/seaweedfs/weed/pb"
@@ -14,7 +14,7 @@ func TestClusterAddRemoveNodes(t *testing.T) {
 	assert.Equal(t, []pb.ServerAddress{
 		pb.ServerAddress("111:1"),
 		pb.ServerAddress("111:2"),
-	}, c.leaders.GetLeaders())
+	}, c.filerLeaders.GetLeaders())
 
 	c.AddClusterNode("filer", pb.ServerAddress("111:3"), "23.45")
 	c.AddClusterNode("filer", pb.ServerAddress("111:4"), "23.45")
@@ -22,7 +22,7 @@ func TestClusterAddRemoveNodes(t *testing.T) {
 		pb.ServerAddress("111:1"),
 		pb.ServerAddress("111:2"),
 		pb.ServerAddress("111:3"),
-	}, c.leaders.GetLeaders())
+	}, c.filerLeaders.GetLeaders())
 
 	c.AddClusterNode("filer", pb.ServerAddress("111:5"), "23.45")
 	c.AddClusterNode("filer", pb.ServerAddress("111:6"), "23.45")
@@ -31,7 +31,7 @@ func TestClusterAddRemoveNodes(t *testing.T) {
 		pb.ServerAddress("111:1"),
 		pb.ServerAddress("111:2"),
 		pb.ServerAddress("111:3"),
-	}, c.leaders.GetLeaders())
+	}, c.filerLeaders.GetLeaders())
 
 	// remove oldest
 	c.RemoveClusterNode("filer", pb.ServerAddress("111:1"))
@@ -39,7 +39,7 @@ func TestClusterAddRemoveNodes(t *testing.T) {
 		pb.ServerAddress("111:6"),
 		pb.ServerAddress("111:2"),
 		pb.ServerAddress("111:3"),
-	}, c.leaders.GetLeaders())
+	}, c.filerLeaders.GetLeaders())
 
 	// remove oldest
 	c.RemoveClusterNode("filer", pb.ServerAddress("111:1"))

weed/command/filer_remote_gateway_buckets.go

@@ -199,8 +199,7 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour
 				return client.WriteDirectory(dest, message.NewEntry)
 			}
 			glog.V(0).Infof("create %s", remote_storage.FormatLocation(dest))
-			reader := filer.NewFileReader(filerSource, message.NewEntry)
-			remoteEntry, writeErr := client.WriteFile(dest, message.NewEntry, reader)
+			remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewEntry, dest)
 			if writeErr != nil {
 				return writeErr
 			}
@@ -264,9 +263,7 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour
 						return client.UpdateFileMetadata(oldDest, message.OldEntry, message.NewEntry)
 					} else {
 						newDest := toRemoteStorageLocation(newBucket, util.NewFullPath(message.NewParentPath, message.NewEntry.Name), newRemoteStorageMountLocation)
-						reader := filer.NewFileReader(filerSource, message.NewEntry)
-						glog.V(0).Infof("create %s", remote_storage.FormatLocation(newDest))
-						remoteEntry, writeErr := client.WriteFile(newDest, message.NewEntry, reader)
+						remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewEntry, newDest)
 						if writeErr != nil {
 							return writeErr
 						}
@@ -303,9 +300,7 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour
 				if message.NewEntry.IsDirectory {
 					return client.WriteDirectory(newDest, message.NewEntry)
 				}
-				reader := filer.NewFileReader(filerSource, message.NewEntry)
-				glog.V(0).Infof("create %s", remote_storage.FormatLocation(newDest))
-				remoteEntry, writeErr := client.WriteFile(newDest, message.NewEntry, reader)
+				remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewEntry, newDest)
 				if writeErr != nil {
 					return writeErr
 				}

weed/command/filer_remote_sync_dir.go

@@ -3,6 +3,10 @@ package command
 import (
 	"context"
 	"fmt"
+	"os"
+	"strings"
+	"time"
+
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb"
@@ -13,9 +17,6 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"github.com/golang/protobuf/proto"
 	"google.golang.org/grpc"
-	"os"
-	"strings"
-	"time"
 )
 
 func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *source.FilerSource, mountedDir string) error {
@@ -108,8 +109,7 @@ func makeEventProcessor(remoteStorage *remote_pb.RemoteConf, mountedDir string,
 				return client.WriteDirectory(dest, message.NewEntry)
 			}
 			glog.V(0).Infof("create %s", remote_storage.FormatLocation(dest))
-			reader := filer.NewFileReader(filerSource, message.NewEntry)
-			remoteEntry, writeErr := client.WriteFile(dest, message.NewEntry, reader)
+			remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewEntry, dest)
 			if writeErr != nil {
 				return writeErr
 			}
@@ -146,9 +146,7 @@ func makeEventProcessor(remoteStorage *remote_pb.RemoteConf, mountedDir string,
 			if err := client.DeleteFile(oldDest); err != nil {
 				return err
 			}
-			reader := filer.NewFileReader(filerSource, message.NewEntry)
-			glog.V(0).Infof("create %s", remote_storage.FormatLocation(dest))
-			remoteEntry, writeErr := client.WriteFile(dest, message.NewEntry, reader)
+			remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewEntry, dest)
 			if writeErr != nil {
 				return writeErr
 			}
@@ -160,6 +158,23 @@ func makeEventProcessor(remoteStorage *remote_pb.RemoteConf, mountedDir string,
 	return eachEntryFunc, nil
 }
 
+func retriedWriteFile(client remote_storage.RemoteStorageClient, filerSource *source.FilerSource, newEntry *filer_pb.Entry, dest *remote_pb.RemoteStorageLocation) (remoteEntry *filer_pb.RemoteEntry, err error) {
+	var writeErr error
+	err = util.Retry("writeFile", func() error {
+		reader := filer.NewFileReader(filerSource, newEntry)
+		glog.V(0).Infof("create %s", remote_storage.FormatLocation(dest))
+		remoteEntry, writeErr = client.WriteFile(dest, newEntry, reader)
+		if writeErr != nil {
+			return writeErr
+		}
+		return nil
+	})
+	if err != nil {
+		glog.Errorf("write to %s: %v", dest, err)
+	}
+	return
+}
+
 func collectLastSyncOffset(filerClient filer_pb.FilerClient, grpcDialOption grpc.DialOption, filerAddress pb.ServerAddress, mountedDir string, timeAgo time.Duration) time.Time {
 	// 1. specified by timeAgo
 	// 2. last offset timestamp for this directory

weed/command/scaffold/filer.toml

@@ -61,12 +61,12 @@ connection_max_lifetime_seconds = 0
 interpolateParams = false
 # if insert/upsert failing, you can disable upsert or update query syntax to match your RDBMS syntax:
 enableUpsert = true
-upsertQuery = """INSERT INTO ` + "`%s`" + ` (dirhash,name,directory,meta) VALUES(?,?,?,?) ON DUPLICATE KEY UPDATE meta = VALUES(meta)"""
+upsertQuery = """INSERT INTO `%s` (dirhash,name,directory,meta) VALUES(?,?,?,?) ON DUPLICATE KEY UPDATE meta = VALUES(meta)"""
 
 [mysql2]  # or memsql, tidb
 enabled = false
 createTable = """
-  CREATE TABLE IF NOT EXISTS ` + "`%s`" + ` (
+  CREATE TABLE IF NOT EXISTS `%s` (
     dirhash BIGINT,
     name VARCHAR(1000) BINARY,
     directory TEXT BINARY,
@@ -85,7 +85,7 @@ connection_max_lifetime_seconds = 0
 interpolateParams = false
 # if insert/upsert failing, you can disable upsert or update query syntax to match your RDBMS syntax:
 enableUpsert = true
-upsertQuery = """INSERT INTO ` + "`%s`" + ` (dirhash,name,directory,meta) VALUES(?,?,?,?) ON DUPLICATE KEY UPDATE meta = VALUES(meta)"""
+upsertQuery = """INSERT INTO `%s` (dirhash,name,directory,meta) VALUES(?,?,?,?) ON DUPLICATE KEY UPDATE meta = VALUES(meta)"""
 
 [postgres] # or cockroachdb, YugabyteDB
 # CREATE TABLE IF NOT EXISTS filemeta (
@@ -167,6 +167,14 @@ database = 0
 # This changes the data layout. Only add new directories. Removing/Updating will cause data loss.
 superLargeDirectories = []
 
+[redis2_sentinel]
+enabled = false
+addresses = ["172.22.12.7:26379","172.22.12.8:26379","172.22.12.9:26379"]
+masterName = "master"
+username = ""
+password = ""
+database = 0
+
 [redis_cluster2]
 enabled = false
 addresses = [

weed/command/scaffold/master.toml

@@ -14,6 +14,7 @@ scripts = """
   volume.deleteEmpty -quietFor=24h -force
   volume.balance -force
   volume.fix.replication
+  s3.clean.uploads -timeAgo=24h
   unlock
 """
 sleep_minutes = 17          # sleep minutes between each script execution

weed/filer/filechunk_manifest.go

@@ -6,7 +6,6 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/wdclient"
 	"io"
 	"math"
-	"math/rand"
 	"net/url"
 	"strings"
 	"time"
@@ -143,9 +142,6 @@ func retriedStreamFetchChunkData(writer io.Writer, urlStrings []string, cipherKe
 	var shouldRetry bool
 	var totalWritten int
 
-	rand.Shuffle(len(urlStrings), func(i, j int) {
-		urlStrings[i], urlStrings[j] = urlStrings[j], urlStrings[i]
-	})
 	for waitTime := time.Second; waitTime < util.RetryWaitTime; waitTime += waitTime / 2 {
 		for _, urlString := range urlStrings {
 			var localProcesed int

weed/filer/filer.go

@@ -3,6 +3,7 @@ package filer
 import (
 	"context"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/cluster"
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 	"os"
@@ -51,7 +52,7 @@ type Filer struct {
 func NewFiler(masters []pb.ServerAddress, grpcDialOption grpc.DialOption,
 	filerHost pb.ServerAddress, collection string, replication string, dataCenter string, notifyFn func()) *Filer {
 	f := &Filer{
-		MasterClient:        wdclient.NewMasterClient(grpcDialOption, "filer", filerHost, dataCenter, masters),
+		MasterClient:        wdclient.NewMasterClient(grpcDialOption, cluster.FilerType, filerHost, dataCenter, masters),
 		fileIdDeletionQueue: util.NewUnboundedQueue(),
 		GrpcDialOption:      grpcDialOption,
 		FilerConf:           NewFilerConf(),
@@ -82,13 +83,13 @@ func (f *Filer) ListExistingPeerUpdates() (existingNodes []*master_pb.ClusterNod
 
 	if grpcErr := pb.WithMasterClient(f.MasterClient.GetMaster(), f.GrpcDialOption, func(client master_pb.SeaweedClient) error {
 		resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
-			ClientType: "filer",
+			ClientType: cluster.FilerType,
 		})
 
 		glog.V(0).Infof("the cluster has %d filers\n", len(resp.ClusterNodes))
 		for _, node := range resp.ClusterNodes {
 			existingNodes = append(existingNodes, &master_pb.ClusterNodeUpdate{
-				NodeType: "filer",
+				NodeType: cluster.FilerType,
 				Address:  node.Address,
 				IsLeader: node.IsLeader,
 				IsAdd:    true,

weed/filer/filer_deletion.go

@@ -136,6 +136,7 @@ func (f *Filer) deleteChunksIfNotNew(oldEntry, newEntry *Entry) {
 	}
 	if newEntry == nil {
 		f.DeleteChunks(oldEntry.Chunks)
+		return
 	}
 
 	var toDelete []*filer_pb.FileChunk

weed/filer/meta_aggregator.go

@@ -3,6 +3,7 @@ package filer
 import (
 	"context"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/cluster"
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"io"
@@ -48,7 +49,7 @@ func NewMetaAggregator(filer *Filer, self pb.ServerAddress, grpcDialOption grpc.
 }
 
 func (ma *MetaAggregator) OnPeerUpdate(update *master_pb.ClusterNodeUpdate) {
-	if update.NodeType != "filer" {
+	if update.NodeType != cluster.FilerType {
 		return
 	}
 

weed/filer/redis2/redis_sentinel_store.go

@@ -0,0 +1,45 @@
+package redis2
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/go-redis/redis/v8"
+	"time"
+)
+
+func init() {
+	filer.Stores = append(filer.Stores, &Redis2SentinelStore{})
+}
+
+type Redis2SentinelStore struct {
+	UniversalRedis2Store
+}
+
+func (store *Redis2SentinelStore) GetName() string {
+	return "redis2_sentinel"
+}
+
+func (store *Redis2SentinelStore) Initialize(configuration util.Configuration, prefix string) (err error) {
+	return store.initialize(
+		configuration.GetStringSlice(prefix+"addresses"),
+		configuration.GetString(prefix+"masterName"),
+		configuration.GetString(prefix+"username"),
+		configuration.GetString(prefix+"password"),
+		configuration.GetInt(prefix+"database"),
+	)
+}
+
+func (store *Redis2SentinelStore) initialize(addresses []string, masterName string, username string, password string, database int) (err error) {
+	store.Client = redis.NewFailoverClient(&redis.FailoverOptions{
+		MasterName:      masterName,
+		SentinelAddrs:   addresses,
+		Username:        username,
+		Password:        password,
+		DB:              database,
+		MinRetryBackoff: time.Millisecond * 100,
+		MaxRetryBackoff: time.Minute * 1,
+		ReadTimeout:     time.Second * 30,
+		WriteTimeout:    time.Second * 5,
+	})
+	return
+}

weed/filer/stream.go

@@ -297,7 +297,7 @@ func (c *ChunkStreamReader) fetchChunkToBuffer(chunkView *ChunkView) error {
 	var buffer bytes.Buffer
 	var shouldRetry bool
 	for _, urlString := range urlStrings {
-		shouldRetry, err = util.ReadUrlAsStream(urlString, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size), func(data []byte) {
+		shouldRetry, err = util.ReadUrlAsStream(urlString+"?readDeleted=true", chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size), func(data []byte) {
 			buffer.Write(data)
 		})
 		if !shouldRetry {

weed/iamapi/iamapi_server.go

@@ -49,7 +49,7 @@ var s3ApiConfigure IamS3ApiConfig
 func NewIamApiServer(router *mux.Router, option *IamServerOption) (iamApiServer *IamApiServer, err error) {
 	s3ApiConfigure = IamS3ApiConfigure{
 		option:       option,
-		masterClient: wdclient.NewMasterClient(option.GrpcDialOption, pb.AdminShellClient, "", "", option.Masters),
+		masterClient: wdclient.NewMasterClient(option.GrpcDialOption, "iam", "", "", option.Masters),
 	}
 	s3Option := s3api.S3ApiServerOption{Filer: option.Filer}
 	iamApiServer = &IamApiServer{

weed/messaging/broker/broker_grpc_server_discovery.go

@@ -3,6 +3,7 @@ package broker
 import (
 	"context"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/cluster"
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"time"
 
@@ -94,7 +95,7 @@ func (broker *MessageBroker) checkFilers() {
 		for _, master := range masters {
 			err := broker.withMasterClient(master, func(client master_pb.SeaweedClient) error {
 				resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
-					ClientType: "filer",
+					ClientType: cluster.FilerType,
 				})
 				if err != nil {
 					return err

weed/s3api/auth_credentials.go

@@ -5,6 +5,7 @@ import (
 	"net/http"
 	"os"
 	"strings"
+	"sync"
 
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/glog"
@@ -23,6 +24,8 @@ type Iam interface {
 }
 
 type IdentityAccessManagement struct {
+	m sync.RWMutex
+
 	identities []*Identity
 	domain     string
 }
@@ -131,19 +134,23 @@ func (iam *IdentityAccessManagement) loadS3ApiConfiguration(config *iam_pb.S3Api
 		}
 		identities = append(identities, t)
 	}
-
+	iam.m.Lock()
 	// atomically switch
 	iam.identities = identities
+	iam.m.Unlock()
 	return nil
 }
 
 func (iam *IdentityAccessManagement) isEnabled() bool {
-
+	iam.m.RLock()
+	defer iam.m.RUnlock()
 	return len(iam.identities) > 0
 }
 
 func (iam *IdentityAccessManagement) lookupByAccessKey(accessKey string) (identity *Identity, cred *Credential, found bool) {
 
+	iam.m.RLock()
+	defer iam.m.RUnlock()
 	for _, ident := range iam.identities {
 		for _, cred := range ident.Credentials {
 			// println("checking", ident.Name, cred.AccessKey)
@@ -157,7 +164,8 @@ func (iam *IdentityAccessManagement) lookupByAccessKey(accessKey string) (identi
 }
 
 func (iam *IdentityAccessManagement) lookupAnonymous() (identity *Identity, found bool) {
-
+	iam.m.RLock()
+	defer iam.m.RUnlock()
 	for _, ident := range iam.identities {
 		if ident.Name == "anonymous" {
 			return ident, true
@@ -179,6 +187,8 @@ func (iam *IdentityAccessManagement) Auth(f http.HandlerFunc, action Action) htt
 				r.Header.Set(xhttp.AmzIdentityId, identity.Name)
 				if identity.isAdmin() {
 					r.Header.Set(xhttp.AmzIsAdmin, "true")
+				} else if _, ok := r.Header[xhttp.AmzIsAdmin]; ok {
+					r.Header.Del(xhttp.AmzIsAdmin)
 				}
 			}
 			f(w, r)

weed/s3api/s3api_bucket_handlers.go

@@ -177,8 +177,8 @@ func (s3a *S3ApiServer) HeadBucketHandler(w http.ResponseWriter, r *http.Request
 	bucket, _ := getBucketAndObject(r)
 	glog.V(3).Infof("HeadBucketHandler %s", bucket)
 
-	if err := s3a.checkBucket(r, bucket); err != s3err.ErrNone {
-		s3err.WriteErrorResponse(w, r, err)
+	if entry, err := s3a.getEntry(s3a.option.BucketsPath, bucket); entry == nil || err == filer_pb.ErrNotFound {
+		s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchBucket)
 		return
 	}
 

weed/s3api/stats.go

@@ -2,7 +2,6 @@ package s3api
 
 import (
 	stats_collect "github.com/chrislusf/seaweedfs/weed/stats"
-	"github.com/chrislusf/seaweedfs/weed/util"
 	"net/http"
 	"strconv"
 	"time"
@@ -28,7 +27,7 @@ func (r *StatusRecorder) Flush() {
 
 func track(f http.HandlerFunc, action string) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
-		w.Header().Set("Server", "SeaweedFS S3 "+util.VERSION)
+		w.Header().Set("Server", "SeaweedFS S3")
 		recorder := NewStatusResponseWriter(w)
 		start := time.Now()
 		f(recorder, r)

weed/server/filer_grpc_server_remote.go

@@ -3,6 +3,10 @@ package weed_server
 import (
 	"context"
 	"fmt"
+	"strings"
+	"sync"
+	"time"
+
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/operation"
 	"github.com/chrislusf/seaweedfs/weed/pb"
@@ -12,9 +16,6 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"github.com/golang/protobuf/proto"
-	"strings"
-	"sync"
-	"time"
 )
 
 func (fs *FilerServer) CacheRemoteObjectToLocalCluster(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) {
@@ -178,6 +179,7 @@ func (fs *FilerServer) CacheRemoteObjectToLocalCluster(ctx context.Context, req
 	// this skips meta data log events
 
 	if err := fs.filer.Store.UpdateEntry(context.Background(), newEntry); err != nil {
+		fs.filer.DeleteChunks(chunks)
 		return nil, err
 	}
 	fs.filer.DeleteChunks(garbage)

weed/server/filer_server_handlers_write_autochunk.go

@@ -97,6 +97,9 @@ func (fs *FilerServer) doPostAutoChunk(ctx context.Context, w http.ResponseWrite
 
 	md5bytes = md5Hash.Sum(nil)
 	filerResult, replyerr = fs.saveMetaData(ctx, r, fileName, contentType, so, md5bytes, fileChunks, chunkOffset, smallContent)
+	if replyerr != nil {
+		fs.filer.DeleteChunks(fileChunks)
+	}
 
 	return
 }
@@ -116,6 +119,9 @@ func (fs *FilerServer) doPutAutoChunk(ctx context.Context, w http.ResponseWriter
 
 	md5bytes = md5Hash.Sum(nil)
 	filerResult, replyerr = fs.saveMetaData(ctx, r, fileName, contentType, so, md5bytes, fileChunks, chunkOffset, smallContent)
+	if replyerr != nil {
+		fs.filer.DeleteChunks(fileChunks)
+	}
 
 	return
 }
@@ -218,13 +224,12 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
 	entry.Extended = SaveAmzMetaData(r, entry.Extended, false)
 
 	for k, v := range r.Header {
-		if len(v) > 0 && (strings.HasPrefix(k, needle.PairNamePrefix) || k == "Cache-Control" || k == "Expires" || k == "Content-Disposition") {
+		if len(v) > 0 && len(v[0]) > 0 && (strings.HasPrefix(k, needle.PairNamePrefix) || k == "Cache-Control" || k == "Expires" || k == "Content-Disposition") {
 			entry.Extended[k] = []byte(v[0])
 		}
 	}
 
 	if dbErr := fs.filer.CreateEntry(ctx, entry, false, false, nil); dbErr != nil {
-		fs.filer.DeleteChunks(fileChunks)
 		replyerr = dbErr
 		filerResult.Error = dbErr.Error()
 		glog.V(0).Infof("failing to write %s to filer server : %v", path, dbErr)

weed/server/master_server.go

@@ -2,8 +2,6 @@ package weed_server
 
 import (
 	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/election"
-	"github.com/chrislusf/seaweedfs/weed/pb"
 	"net/http"
 	"net/http/httputil"
 	"net/url"
@@ -13,6 +11,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/chrislusf/seaweedfs/weed/cluster"
+	"github.com/chrislusf/seaweedfs/weed/pb"
+
 	"github.com/chrislusf/raft"
 	"github.com/gorilla/mux"
 	"google.golang.org/grpc"
@@ -69,7 +70,7 @@ type MasterServer struct {
 
 	adminLocks *AdminLocks
 
-	Cluster *election.Cluster
+	Cluster *cluster.Cluster
 }
 
 func NewMasterServer(r *mux.Router, option *MasterOption, peers []pb.ServerAddress) *MasterServer {
@@ -104,9 +105,9 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers []pb.ServerAddre
 		vgCh:            make(chan *topology.VolumeGrowRequest, 1<<6),
 		clientChans:     make(map[string]chan *master_pb.KeepConnectedResponse),
 		grpcDialOption:  grpcDialOption,
-		MasterClient:    wdclient.NewMasterClient(grpcDialOption, "master", option.Master, "", peers),
+		MasterClient:    wdclient.NewMasterClient(grpcDialOption, cluster.MasterType, option.Master, "", peers),
 		adminLocks:      NewAdminLocks(),
-		Cluster:         election.NewCluster(),
+		Cluster:         cluster.NewCluster(),
 	}
 	ms.boundedLeaderChan = make(chan int, 16)
 
@@ -210,10 +211,20 @@ func (ms *MasterServer) startAdminScripts() {
 
 	v := util.GetViper()
 	adminScripts := v.GetString("master.maintenance.scripts")
-	glog.V(0).Infof("adminScripts:\n%v", adminScripts)
 	if adminScripts == "" {
-		return
+		adminScripts = `
+		lock
+		ec.encode -fullPercent=95 -quietFor=1h
+		ec.rebuild -force
+		ec.balance -force
+		volume.deleteEmpty -quietFor=24h -force
+		volume.balance -force
+		volume.fix.replication
+		s3.clean.uploads -timeAgo=24h
+		unlock
+	`
 	}
+	glog.V(0).Infof("adminScripts: %v", adminScripts)
 
 	v.SetDefault("master.maintenance.sleep_minutes", 17)
 	sleepMinutes := v.GetInt("master.maintenance.sleep_minutes")

weed/server/volume_grpc_copy.go

@@ -226,7 +226,7 @@ func writeToFile(client volume_server_pb.VolumeServer_CopyFileClient, fileName s
 		if receiveErr == io.EOF {
 			break
 		}
-		if resp.ModifiedTsNs != 0 {
+		if resp!=nil && resp.ModifiedTsNs != 0 {
 			modifiedTsNs = resp.ModifiedTsNs
 		}
 		if receiveErr != nil {

weed/shell/command_cluster_ps.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/cluster"
 	"io"
 
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
@@ -37,7 +38,7 @@ func (c *commandClusterPs) Do(args []string, commandEnv *CommandEnv, writer io.W
 
 	err = commandEnv.MasterClient.WithClient(func(client master_pb.SeaweedClient) error {
 		resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
-			ClientType: "filer",
+			ClientType: cluster.FilerType,
 		})
 
 		fmt.Fprintf(writer, "the cluster has %d filers\n", len(resp.ClusterNodes))

weed/shell/command_remote_uncache.go

@@ -4,12 +4,13 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/filer"
-	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
-	"github.com/chrislusf/seaweedfs/weed/util"
 	"io"
 	"path/filepath"
 	"strings"
+
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
 func init() {
@@ -41,12 +42,12 @@ func (c *commandRemoteUncache) Help() string {
 
 func (c *commandRemoteUncache) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	remoteUnmountCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	remoteUncacheCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 
-	dir := remoteUnmountCommand.String("dir", "", "a directory in filer")
-	fileFiler := newFileFilter(remoteUnmountCommand)
+	dir := remoteUncacheCommand.String("dir", "", "a directory in filer")
+	fileFiler := newFileFilter(remoteUncacheCommand)
 
-	if err = remoteUnmountCommand.Parse(args); err != nil {
+	if err = remoteUncacheCommand.Parse(args); err != nil {
 		return nil
 	}
 

weed/shell/command_volume_configure_replication.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"io"
+	"path/filepath"
 
 	"github.com/chrislusf/seaweedfs/weed/operation"
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
@@ -34,11 +35,12 @@ func (c *commandVolumeConfigureReplication) Help() string {
 `
 }
 
-func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
+func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *CommandEnv, _ io.Writer) (err error) {
 
 	configureReplicationCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	volumeIdInt := configureReplicationCommand.Int("volumeId", 0, "the volume id")
 	replicationString := configureReplicationCommand.String("replication", "", "the intended replication value")
+	collectionPattern := configureReplicationCommand.String("collectionPattern", "", "match with wildcard characters '*' and '?'")
 	if err = configureReplicationCommand.Parse(args); err != nil {
 		return nil
 	}
@@ -55,7 +57,6 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman
 	if err != nil {
 		return fmt.Errorf("replication format: %v", err)
 	}
-	replicaPlacementInt32 := uint32(replicaPlacement.Byte())
 
 	// collect topology information
 	topologyInfo, _, err := collectTopologyInfo(commandEnv)
@@ -64,6 +65,7 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman
 	}
 
 	vid := needle.VolumeId(*volumeIdInt)
+	volumeFilter := getVolumeFilter(replicaPlacement, uint32(vid), *collectionPattern)
 
 	// find all data nodes with volumes that needs replication change
 	var allLocations []location
@@ -71,7 +73,7 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman
 		loc := newLocation(dc, string(rack), dn)
 		for _, diskInfo := range dn.DiskInfos {
 			for _, v := range diskInfo.VolumeInfos {
-				if v.Id == uint32(vid) && v.ReplicaPlacement != replicaPlacementInt32 {
+				if volumeFilter(v) {
 					allLocations = append(allLocations, loc)
 					continue
 				}
@@ -106,3 +108,19 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman
 
 	return nil
 }
+
+func getVolumeFilter(replicaPlacement *super_block.ReplicaPlacement, volumeId uint32, collectionPattern string) func(message *master_pb.VolumeInformationMessage) bool {
+	replicaPlacementInt32 := uint32(replicaPlacement.Byte())
+	if volumeId > 0 {
+		return func(v *master_pb.VolumeInformationMessage) bool {
+			return v.Id == volumeId && v.ReplicaPlacement != replicaPlacementInt32
+		}
+	}
+	return func(v *master_pb.VolumeInformationMessage) bool {
+		matched, err := filepath.Match(collectionPattern, v.Collection)
+		if err != nil {
+			return false
+		}
+		return matched
+	}
+}

weed/shell/command_volume_fsck.go

@@ -180,7 +180,6 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(volumeIdToServer map[uint
 		return nil
 	})
 
-	return nil
 }
 
 func (c *commandVolumeFsck) findFilerChunksMissingInVolumeServers(volumeIdToVInfo map[uint32]VInfo, tempFolder string, writer io.Writer, verbose bool, applyPurging *bool) error {
@@ -217,11 +216,11 @@ func (c *commandVolumeFsck) findExtraChunksInVolumeServers(volumeIdToVInfo map[u
 			}
 			if inUseCount == 0 {
 				if err := deleteVolume(c.env.option.GrpcDialOption, needle.VolumeId(volumeId), vinfo.server); err != nil {
-					return fmt.Errorf("delete volume %d: %v\n", volumeId, err)
+					return fmt.Errorf("delete volume %d: %v", volumeId, err)
 				}
 			} else {
 				if err := c.purgeFileIdsForOneVolume(volumeId, orphanFileIds, writer); err != nil {
-					return fmt.Errorf("purge for volume %d: %v\n", volumeId, err)
+					return fmt.Errorf("purge for volume %d: %v", volumeId, err)
 				}
 			}
 		}

weed/shell/shell_liner.go

@@ -3,6 +3,7 @@ package shell
 import (
 	"context"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/cluster"
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
@@ -54,7 +55,7 @@ func RunShell(options ShellOptions) {
 		var filers []pb.ServerAddress
 		commandEnv.MasterClient.WithClient(func(client master_pb.SeaweedClient) error {
 			resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
-				ClientType: "filer",
+				ClientType: cluster.FilerType,
 			})
 			if err != nil {
 				return err

weed/util/constants.go

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

weed/util/mem/slot_pool.go

@@ -0,0 +1,44 @@
+package mem
+
+import "sync"
+
+var pools []*sync.Pool
+
+const (
+	min_size = 1024
+)
+
+func bitCount(size int) (count int) {
+	for ; size > min_size; count++ {
+		size = size >> 1
+	}
+	return
+}
+
+func init() {
+	// 1KB ~ 256MB
+	pools = make([]*sync.Pool, bitCount(1024*1024*256))
+	for i := 0; i < len(pools); i++ {
+		slotSize := 1024 << i
+		pools[i] = &sync.Pool{
+			New: func() interface{} {
+				buffer := make([]byte, slotSize)
+				return &buffer
+			},
+		}
+	}
+}
+
+func getSlotPool(size int) *sync.Pool {
+	index := bitCount(size)
+	return pools[index]
+}
+
+func Allocate(size int) []byte {
+	slab := *getSlotPool(size).Get().(*[]byte)
+	return slab[:size]
+}
+
+func Free(buf []byte) {
+	getSlotPool(cap(buf)).Put(&buf)
+}

weed/util/mem/slot_pool_test.go

@@ -0,0 +1,37 @@
+package mem
+
+import (
+	"testing"
+)
+
+func TestAllocateFree(t *testing.T) {
+	buf := Allocate(12)
+	Free(buf)
+	if cap(buf) != min_size {
+		t.Errorf("min size error allocated capacity=%d", cap(buf))
+	}
+	if len(buf) != 12 {
+		t.Errorf("size error")
+	}
+
+	buf = Allocate(4883)
+	Free(buf)
+	if cap(buf) != 1024<<bitCount(4883) {
+		t.Errorf("min size error allocated capacity=%d", cap(buf))
+	}
+	if len(buf) != 4883 {
+		t.Errorf("size error")
+	}
+
+}
+
+func TestBitCount(t *testing.T) {
+	count := bitCount(12)
+	if count != 0 {
+		t.Errorf("bitCount error count=%d", count)
+	}
+	if count != bitCount(min_size) {
+		t.Errorf("bitCount error count=%d", count)
+	}
+
+}

weed/wdclient/vid_map.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/pb"
+	"math/rand"
 	"strconv"
 	"strings"
 	"sync"
@@ -69,13 +70,21 @@ func (vc *vidMap) LookupVolumeServerUrl(vid string) (serverUrls []string, err er
 	if !found {
 		return nil, fmt.Errorf("volume %d not found", id)
 	}
+	var sameDcServers, otherDcServers []string
 	for _, loc := range locations {
 		if vc.DataCenter == "" || loc.DataCenter == "" || vc.DataCenter != loc.DataCenter {
-			serverUrls = append(serverUrls, loc.Url)
+			otherDcServers = append(otherDcServers, loc.Url)
 		} else {
-			serverUrls = append([]string{loc.Url}, serverUrls...)
+			sameDcServers = append(sameDcServers, loc.Url)
 		}
 	}
+	rand.Shuffle(len(sameDcServers), func(i, j int) {
+		sameDcServers[i], sameDcServers[j] = sameDcServers[j], sameDcServers[i]
+	})
+	rand.Shuffle(len(otherDcServers), func(i, j int) {
+		otherDcServers[i], otherDcServers[j] = otherDcServers[j], otherDcServers[i]
+	})
+	serverUrls = append(sameDcServers, otherDcServers...)
 	return
 }