2.77

following https://github.com/wangyoucao577/go-release-action/issues/63#issuecomment-962417798

.github/workflows/binaries_dev.yml

@@ -37,7 +37,7 @@ 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@v1.20
+        uses: wangyoucao577/go-release-action@bugfix/upload-fail
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           goos: ${{ matrix.goos }}
@@ -53,7 +53,7 @@ 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@v1.20
+        uses: wangyoucao577/go-release-action@bugfix/upload-fail
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           goos: ${{ matrix.goos }}
@@ -84,7 +84,7 @@ 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@v1.20
+        uses: wangyoucao577/go-release-action@bugfix/upload-fail
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           goos: ${{ matrix.goos }}
@@ -100,7 +100,7 @@ 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@v1.20
+        uses: wangyoucao577/go-release-action@bugfix/upload-fail
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           goos: ${{ matrix.goos }}

docker/Dockerfile.go_build

@@ -1,4 +1,4 @@
-FROM amd64/golang:1.17-alpine as builder
+FROM golang:1.17-alpine as builder
 RUN apk add git g++ fuse
 RUN mkdir -p /go/src/github.com/chrislusf/
 RUN git clone https://github.com/chrislusf/seaweedfs /go/src/github.com/chrislusf/seaweedfs

docker/Dockerfile.go_build_large

@@ -1,4 +1,4 @@
-FROM amd64/golang:1.17-alpine as builder
+FROM golang:1.17-alpine as builder
 RUN apk add git g++ fuse
 RUN mkdir -p /go/src/github.com/chrislusf/
 RUN git clone https://github.com/chrislusf/seaweedfs /go/src/github.com/chrislusf/seaweedfs

k8s/helm_charts2/Chart.yaml

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

k8s/helm_charts2/templates/filer-statefulset.yaml

@@ -133,11 +133,6 @@ spec:
               -encryptVolumeData \
               {{- end }}
               -ip=${POD_IP} \
-              {{- if .Values.filer.enable_peers }}
-              {{- if gt (.Values.filer.replicas | int) 1 }}
-              -peers=$(echo -n "{{ range $index := until (.Values.filer.replicas | int) }}${SEAWEEDFS_FULLNAME}-filer-{{ $index }}.${SEAWEEDFS_FULLNAME}-filer:{{ $.Values.filer.port }}{{ if lt $index (sub ($.Values.filer.replicas | int) 1) }},{{ end }}{{ end }}" | sed "s/$HOSTNAME.${SEAWEEDFS_FULLNAME}-filer:{{ $.Values.filer.port }}//" | sed 's/,$//; 's/^,//'; s/,,/,/;' )  \
-              {{- end }}
-              {{- end }}
               {{- if .Values.filer.s3.enabled }}
               -s3 \
               -s3.port={{ .Values.filer.s3.port }} \

k8s/helm_charts2/values.yaml

@@ -246,8 +246,6 @@ filer:
   maxMB: null
   # encrypt data on volume servers
   encryptVolumeData: false
-  # enable peers sync metadata, for leveldb (localdb for filer but with sync across)
-  enable_peers: false
 
   # Whether proxy or redirect to volume server during file GET request
   redirectOnRead: false

other/java/client/pom.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.6.9</version>
+    <version>1.7.0</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>
@@ -60,7 +60,7 @@
         <dependency>
             <groupId>org.apache.httpcomponents</groupId>
             <artifactId>httpmime</artifactId>
-            <version>4.5.6</version>
+            <version>4.5.13</version>
         </dependency>
         <dependency>
             <groupId>junit</groupId>

other/java/client/pom.xml.deploy

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.6.9</version>
+    <version>1.7.0</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>
@@ -60,7 +60,7 @@
         <dependency>
             <groupId>org.apache.httpcomponents</groupId>
             <artifactId>httpmime</artifactId>
-            <version>4.5.6</version>
+            <version>4.5.13</version>
         </dependency>
         <dependency>
             <groupId>junit</groupId>

other/java/client/pom_debug.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.6.9</version>
+    <version>1.7.0</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>
@@ -60,7 +60,7 @@
         <dependency>
             <groupId>org.apache.httpcomponents</groupId>
             <artifactId>httpmime</artifactId>
-            <version>4.5.6</version>
+            <version>4.5.13</version>
         </dependency>
         <dependency>
             <groupId>junit</groupId>

other/java/client/src/main/java/seaweedfs/client/RemoteUtil.java

@@ -14,10 +14,10 @@ public class RemoteUtil {
         String dir = SeaweedOutputStream.getParentDirectory(fullpath);
         String name = SeaweedOutputStream.getFileName(fullpath);
 
-        final FilerProto.CacheRemoteObjectToLocalClusterResponse downloadToLocalResponse = filerClient.getBlockingStub()
-                .downloadToLocal(FilerProto.CacheRemoteObjectToLocalClusterRequest.newBuilder()
+        final FilerProto.CacheRemoteObjectToLocalClusterResponse response = filerClient.getBlockingStub()
+                .cacheRemoteObjectToLocalCluster(FilerProto.CacheRemoteObjectToLocalClusterRequest.newBuilder()
                 .setDirectory(dir).setName(name).build());
 
-        return downloadToLocalResponse.getEntry();
+        return response.getEntry();
     }
 }

other/java/examples/pom.xml

@@ -11,13 +11,13 @@
         <dependency>
             <groupId>com.github.chrislusf</groupId>
             <artifactId>seaweedfs-client</artifactId>
-            <version>1.6.9</version>
+            <version>1.7.0</version>
             <scope>compile</scope>
         </dependency>
         <dependency>
             <groupId>com.github.chrislusf</groupId>
             <artifactId>seaweedfs-hadoop2-client</artifactId>
-            <version>1.6.9</version>
+            <version>1.7.0</version>
             <scope>compile</scope>
         </dependency>
         <dependency>

other/java/hdfs2/dependency-reduced-pom.xml

@@ -301,7 +301,7 @@
     </snapshotRepository>
   </distributionManagement>
   <properties>
-    <seaweedfs.client.version>1.6.9</seaweedfs.client.version>
+    <seaweedfs.client.version>1.7.0</seaweedfs.client.version>
     <hadoop.version>2.9.2</hadoop.version>
   </properties>
 </project>

other/java/hdfs2/pom.xml

@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <properties>
-        <seaweedfs.client.version>1.6.9</seaweedfs.client.version>
+        <seaweedfs.client.version>1.7.0</seaweedfs.client.version>
         <hadoop.version>2.9.2</hadoop.version>
     </properties>
 

other/java/hdfs3/dependency-reduced-pom.xml

@@ -309,7 +309,7 @@
     </snapshotRepository>
   </distributionManagement>
   <properties>
-    <seaweedfs.client.version>1.6.9</seaweedfs.client.version>
+    <seaweedfs.client.version>1.7.0</seaweedfs.client.version>
     <hadoop.version>3.1.1</hadoop.version>
   </properties>
 </project>

other/java/hdfs3/pom.xml

@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <properties>
-        <seaweedfs.client.version>1.6.9</seaweedfs.client.version>
+        <seaweedfs.client.version>1.7.0</seaweedfs.client.version>
         <hadoop.version>3.1.1</hadoop.version>
     </properties>
 

weed/command/filer.go

@@ -4,7 +4,6 @@ import (
 	"fmt"
 	"net/http"
 	"os"
-	"strings"
 	"time"
 
 	"google.golang.org/grpc/reflection"
@@ -46,7 +45,6 @@ type FilerOptions struct {
 	enableNotification      *bool
 	disableHttp             *bool
 	cipher                  *bool
-	peers                   *string
 	metricsHttpPort         *int
 	saveToFilerLimit        *int
 	defaultLevelDbDirectory *string
@@ -72,7 +70,6 @@ func init() {
 	f.rack = cmdFiler.Flag.String("rack", "", "prefer to write to volumes in this rack")
 	f.disableHttp = cmdFiler.Flag.Bool("disableHttp", false, "disable http request, only gRpc operations are allowed")
 	f.cipher = cmdFiler.Flag.Bool("encryptVolumeData", false, "encrypt data on volume servers")
-	f.peers = cmdFiler.Flag.String("peers", "", "all filers sharing the same filer store in comma separated ip:port list")
 	f.metricsHttpPort = cmdFiler.Flag.Int("metricsPort", 0, "Prometheus metrics listen port")
 	f.saveToFilerLimit = cmdFiler.Flag.Int("saveToFilerLimit", 0, "files smaller than this limit will be saved in filer store")
 	f.defaultLevelDbDirectory = cmdFiler.Flag.String("defaultStoreDir", ".", "if filer.toml is empty, use an embedded filer store in the directory")
@@ -186,11 +183,6 @@ func (fo *FilerOptions) startFiler() {
 
 	defaultLevelDbDirectory := util.ResolvePath(*fo.defaultLevelDbDirectory + "/filerldb2")
 
-	var peers []string
-	if *fo.peers != "" {
-		peers = strings.Split(*fo.peers, ",")
-	}
-
 	filerAddress := pb.NewServerAddress(*fo.ip, *fo.port, *fo.portGrpc)
 
 	fs, nfs_err := weed_server.NewFilerServer(defaultMux, publicVolumeMux, &weed_server.FilerOption{
@@ -207,7 +199,6 @@ func (fo *FilerOptions) startFiler() {
 		Host:                  filerAddress,
 		Cipher:                *fo.cipher,
 		SaveToFilerLimit:      int64(*fo.saveToFilerLimit),
-		Filers:                pb.FromAddressStrings(peers),
 		ConcurrentUploadLimit: int64(*fo.concurrentUploadLimitMB) * 1024 * 1024,
 	})
 	if nfs_err != nil {

weed/command/server.go

@@ -106,7 +106,6 @@ func init() {
 	filerOptions.maxMB = cmdServer.Flag.Int("filer.maxMB", 4, "split files larger than the limit")
 	filerOptions.dirListingLimit = cmdServer.Flag.Int("filer.dirListLimit", 1000, "limit sub dir listing size")
 	filerOptions.cipher = cmdServer.Flag.Bool("filer.encryptVolumeData", false, "encrypt data on volume servers")
-	filerOptions.peers = cmdServer.Flag.String("filer.peers", "", "all filers sharing the same filer store in comma separated ip:port list")
 	filerOptions.saveToFilerLimit = cmdServer.Flag.Int("filer.saveToFilerLimit", 0, "Small files smaller than this limit can be cached in filer store.")
 	filerOptions.concurrentUploadLimitMB = cmdServer.Flag.Int("filer.concurrentUploadLimitMB", 64, "limit total concurrent upload size")
 

weed/command/shell.go

@@ -37,7 +37,7 @@ func runShell(command *Command, args []string) bool {
 	util.LoadConfiguration("security", false)
 	shellOptions.GrpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client")
 
-	if *shellOptions.Masters == "" && *shellInitialFiler == "" {
+	if *shellOptions.Masters == "" {
 		util.LoadConfiguration("shell", false)
 		v := util.GetViper()
 		cluster := v.GetString("cluster.default")
@@ -45,15 +45,14 @@ func runShell(command *Command, args []string) bool {
 			cluster = *shellCluster
 		}
 		if cluster == "" {
-			*shellOptions.Masters, *shellInitialFiler = "localhost:9333", "localhost:8888"
+			*shellOptions.Masters = "localhost:9333"
 		} else {
 			*shellOptions.Masters = v.GetString("cluster." + cluster + ".master")
 			*shellInitialFiler = v.GetString("cluster." + cluster + ".filer")
+			fmt.Printf("master: %s filer: %s\n", *shellOptions.Masters, *shellInitialFiler)
 		}
 	}
 
-	fmt.Printf("master: %s filer: %s\n", *shellOptions.Masters, *shellInitialFiler)
-
 	shellOptions.FilerAddress = pb.ServerAddress(*shellInitialFiler)
 	shellOptions.Directory = "/"
 

weed/election/cluster.go

@@ -0,0 +1,207 @@
+package election
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
+	"math"
+	"sync"
+	"time"
+)
+
+type ClusterNode struct {
+	Address   pb.ServerAddress
+	Version   string
+	counter   int
+	createdTs time.Time
+}
+
+type Leaders struct {
+	leaders [3]pb.ServerAddress
+}
+
+type Cluster struct {
+	nodes     map[pb.ServerAddress]*ClusterNode
+	nodesLock sync.RWMutex
+	leaders   *Leaders
+}
+
+func NewCluster() *Cluster {
+	return &Cluster{
+		nodes:   make(map[pb.ServerAddress]*ClusterNode),
+		leaders: &Leaders{},
+	}
+}
+
+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 {
+			existingNode.counter++
+			return nil
+		}
+		cluster.nodes[address] = &ClusterNode{
+			Address:   address,
+			Version:   version,
+			counter:   1,
+			createdTs: time.Now(),
+		}
+		return cluster.ensureLeader(true, nodeType, address)
+	case "master":
+	}
+	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 {
+			return nil
+		} else {
+			existingNode.counter--
+			if existingNode.counter <= 0 {
+				delete(cluster.nodes, address)
+				return cluster.ensureLeader(false, nodeType, address)
+			}
+		}
+	case "master":
+	}
+	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 {
+			nodes = append(nodes, node)
+		}
+	case "master":
+	}
+	return
+}
+
+func (cluster *Cluster) IsOneLeader(address pb.ServerAddress) bool {
+	return cluster.leaders.isOneLeader(address)
+}
+
+func (cluster *Cluster) ensureLeader(isAdd bool, nodeType string, address pb.ServerAddress) (result []*master_pb.KeepConnectedResponse) {
+	if isAdd {
+		if cluster.leaders.addLeaderIfVacant(address) {
+			// has added the address as one leader
+			result = append(result, &master_pb.KeepConnectedResponse{
+				ClusterNodeUpdate: &master_pb.ClusterNodeUpdate{
+					NodeType: nodeType,
+					Address:  string(address),
+					IsLeader: true,
+					IsAdd:    true,
+				},
+			})
+		} else {
+			result = append(result, &master_pb.KeepConnectedResponse{
+				ClusterNodeUpdate: &master_pb.ClusterNodeUpdate{
+					NodeType: nodeType,
+					Address:  string(address),
+					IsLeader: false,
+					IsAdd:    true,
+				},
+			})
+		}
+	} else {
+		if cluster.leaders.removeLeaderIfExists(address) {
+
+			result = append(result, &master_pb.KeepConnectedResponse{
+				ClusterNodeUpdate: &master_pb.ClusterNodeUpdate{
+					NodeType: nodeType,
+					Address:  string(address),
+					IsLeader: true,
+					IsAdd:    false,
+				},
+			})
+
+			// pick the freshest one, since it is less likely to go away
+			var shortestDuration int64 = math.MaxInt64
+			now := time.Now()
+			var candidateAddress pb.ServerAddress
+			for _, node := range cluster.nodes {
+				if cluster.leaders.isOneLeader(node.Address) {
+					continue
+				}
+				duration := now.Sub(node.createdTs).Nanoseconds()
+				if duration < shortestDuration {
+					shortestDuration = duration
+					candidateAddress = node.Address
+				}
+			}
+			if candidateAddress != "" {
+				cluster.leaders.addLeaderIfVacant(candidateAddress)
+				// added a new leader
+				result = append(result, &master_pb.KeepConnectedResponse{
+					ClusterNodeUpdate: &master_pb.ClusterNodeUpdate{
+						NodeType: nodeType,
+						Address:  string(candidateAddress),
+						IsLeader: true,
+						IsAdd:    true,
+					},
+				})
+			}
+		} else {
+			result = append(result, &master_pb.KeepConnectedResponse{
+				ClusterNodeUpdate: &master_pb.ClusterNodeUpdate{
+					NodeType: nodeType,
+					Address:  string(address),
+					IsLeader: false,
+					IsAdd:    false,
+				},
+			})
+		}
+	}
+	return
+}
+
+func (leaders *Leaders) addLeaderIfVacant(address pb.ServerAddress) (hasChanged bool) {
+	if leaders.isOneLeader(address) {
+		return
+	}
+	for i := 0; i < len(leaders.leaders); i++ {
+		if leaders.leaders[i] == "" {
+			leaders.leaders[i] = address
+			hasChanged = true
+			return
+		}
+	}
+	return
+}
+func (leaders *Leaders) removeLeaderIfExists(address pb.ServerAddress) (hasChanged bool) {
+	if !leaders.isOneLeader(address) {
+		return
+	}
+	for i := 0; i < len(leaders.leaders); i++ {
+		if leaders.leaders[i] == address {
+			leaders.leaders[i] = ""
+			hasChanged = true
+			return
+		}
+	}
+	return
+}
+func (leaders *Leaders) isOneLeader(address pb.ServerAddress) bool {
+	for i := 0; i < len(leaders.leaders); i++ {
+		if leaders.leaders[i] == address {
+			return true
+		}
+	}
+	return false
+}
+func (leaders *Leaders) GetLeaders() (addresses []pb.ServerAddress) {
+	for i := 0; i < len(leaders.leaders); i++ {
+		if leaders.leaders[i] != "" {
+			addresses = append(addresses, leaders.leaders[i])
+		}
+	}
+	return
+}

weed/election/cluster_test.go

@@ -0,0 +1,47 @@
+package election
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/pb"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestClusterAddRemoveNodes(t *testing.T) {
+	c := NewCluster()
+
+	c.AddClusterNode("filer", pb.ServerAddress("111:1"), "23.45")
+	c.AddClusterNode("filer", pb.ServerAddress("111:2"), "23.45")
+	assert.Equal(t, []pb.ServerAddress{
+		pb.ServerAddress("111:1"),
+		pb.ServerAddress("111:2"),
+	}, c.leaders.GetLeaders())
+
+	c.AddClusterNode("filer", pb.ServerAddress("111:3"), "23.45")
+	c.AddClusterNode("filer", pb.ServerAddress("111:4"), "23.45")
+	assert.Equal(t, []pb.ServerAddress{
+		pb.ServerAddress("111:1"),
+		pb.ServerAddress("111:2"),
+		pb.ServerAddress("111:3"),
+	}, c.leaders.GetLeaders())
+
+	c.AddClusterNode("filer", pb.ServerAddress("111:5"), "23.45")
+	c.AddClusterNode("filer", pb.ServerAddress("111:6"), "23.45")
+	c.RemoveClusterNode("filer", pb.ServerAddress("111:4"))
+	assert.Equal(t, []pb.ServerAddress{
+		pb.ServerAddress("111:1"),
+		pb.ServerAddress("111:2"),
+		pb.ServerAddress("111:3"),
+	}, c.leaders.GetLeaders())
+
+	// remove oldest
+	c.RemoveClusterNode("filer", pb.ServerAddress("111:1"))
+	assert.Equal(t, []pb.ServerAddress{
+		pb.ServerAddress("111:6"),
+		pb.ServerAddress("111:2"),
+		pb.ServerAddress("111:3"),
+	}, c.leaders.GetLeaders())
+
+	// remove oldest
+	c.RemoveClusterNode("filer", pb.ServerAddress("111:1"))
+
+}

weed/filer/filechunk_manifest.go

@@ -6,6 +6,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/wdclient"
 	"io"
 	"math"
+	"math/rand"
 	"net/url"
 	"strings"
 	"time"
@@ -142,6 +143,9 @@ 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/filechunks_read.go

@@ -40,7 +40,7 @@ func readResolvedChunks(chunks []*filer_pb.FileChunk) (visibles []VisibleInterva
 	for _, point := range points {
 		if point.isStart {
 			if len(queue) > 0 {
-				lastIndex := len(queue) -1
+				lastIndex := len(queue) - 1
 				lastPoint := queue[lastIndex]
 				if point.x != prevX && lastPoint.ts < point.ts {
 					visibles = addToVisibles(visibles, prevX, lastPoint, point)

weed/filer/filechunks_read_test.go

@@ -52,7 +52,7 @@ func TestReadResolvedChunks(t *testing.T) {
 
 func TestRandomizedReadResolvedChunks(t *testing.T) {
 
-	var limit int64 = 1024*1024
+	var limit int64 = 1024 * 1024
 	array := make([]int64, limit)
 	var chunks []*filer_pb.FileChunk
 	for ts := int64(0); ts < 1024; ts++ {
@@ -75,7 +75,7 @@ func TestRandomizedReadResolvedChunks(t *testing.T) {
 	visibles := readResolvedChunks(chunks)
 
 	for _, visible := range visibles {
-		for i := visible.start; i<visible.stop;i++{
+		for i := visible.start; i < visible.stop; i++ {
 			if array[i] != visible.modifiedTime {
 				t.Errorf("position %d expected ts %d actual ts %d", i, array[i], visible.modifiedTime)
 			}
@@ -101,12 +101,12 @@ func randomWrite(array []int64, start int64, size int64, ts int64) *filer_pb.Fil
 
 func TestSequentialReadResolvedChunks(t *testing.T) {
 
-	var chunkSize int64 = 1024*1024*2
+	var chunkSize int64 = 1024 * 1024 * 2
 	var chunks []*filer_pb.FileChunk
 	for ts := int64(0); ts < 13; ts++ {
 		chunks = append(chunks, &filer_pb.FileChunk{
 			FileId: "",
-			Offset: chunkSize*ts,
+			Offset: chunkSize * ts,
 			Size:   uint64(chunkSize),
 			Mtime:  1,
 		})

weed/filer/filer.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 	"os"
 	"strings"
 	"time"
@@ -66,22 +67,38 @@ func NewFiler(masters []pb.ServerAddress, grpcDialOption grpc.DialOption,
 	return f
 }
 
-func (f *Filer) AggregateFromPeers(self pb.ServerAddress, filers []pb.ServerAddress) {
+func (f *Filer) AggregateFromPeers(self pb.ServerAddress) {
 
-	// set peers
-	found := false
-	for _, peer := range filers {
-		if peer == self {
-			found = true
+	f.MetaAggregator = NewMetaAggregator(f, self, f.GrpcDialOption)
+	f.MasterClient.OnPeerUpdate = f.MetaAggregator.OnPeerUpdate
+
+	for _, peerUpdate := range f.ListExistingPeerUpdates() {
+		f.MetaAggregator.OnPeerUpdate(peerUpdate)
+	}
+
+}
+
+func (f *Filer) ListExistingPeerUpdates() (existingNodes []*master_pb.ClusterNodeUpdate){
+
+	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",
+		})
+
+		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",
+				Address:  node.Address,
+				IsLeader: node.IsLeader,
+				IsAdd:    true,
+			})
 		}
+		return err
+	}); grpcErr != nil {
+		glog.V(0).Infof("connect to %s: %v", f.MasterClient.GetMaster(), grpcErr)
 	}
-	if !found {
-		filers = append(filers, self)
-	}
-
-	f.MetaAggregator = NewMetaAggregator(filers, f.GrpcDialOption)
-	f.MetaAggregator.StartLoopSubscribe(f, self)
-
+	return
 }
 
 func (f *Filer) SetStore(store FilerStore) {
@@ -117,7 +134,7 @@ func (fs *Filer) GetMaster() pb.ServerAddress {
 	return fs.MasterClient.GetMaster()
 }
 
-func (fs *Filer) KeepConnectedToMaster() {
+func (fs *Filer) KeepMasterClientConnected() {
 	fs.MasterClient.KeepConnectedToMaster()
 }
 

weed/filer/filer_delete_entry.go

@@ -15,6 +15,9 @@ const (
 	MsgFailDelNonEmptyFolder = "fail to delete non-empty folder"
 )
 
+type OnChunksFunc func([]*filer_pb.FileChunk) error
+type OnHardLinkIdsFunc func([]HardLinkId) error
+
 func (f *Filer) DeleteEntryMetaAndData(ctx context.Context, p util.FullPath, isRecursive, ignoreRecursiveError, shouldDeleteChunks, isFromOtherCluster bool, signatures []int32) (err error) {
 	if p == "/" {
 		return nil
@@ -27,20 +30,29 @@ func (f *Filer) DeleteEntryMetaAndData(ctx context.Context, p util.FullPath, isR
 
 	isDeleteCollection := f.isBucket(entry)
 
-	var chunks []*filer_pb.FileChunk
-	var hardLinkIds []HardLinkId
-	chunks = append(chunks, entry.Chunks...)
 	if entry.IsDirectory() {
 		// delete the folder children, not including the folder itself
-		var dirChunks []*filer_pb.FileChunk
-		var dirHardLinkIds []HardLinkId
-		dirChunks, dirHardLinkIds, err = f.doBatchDeleteFolderMetaAndData(ctx, entry, isRecursive, ignoreRecursiveError, shouldDeleteChunks && !isDeleteCollection, isDeleteCollection, isFromOtherCluster, signatures)
+		err = f.doBatchDeleteFolderMetaAndData(ctx, entry, isRecursive, ignoreRecursiveError, shouldDeleteChunks && !isDeleteCollection, isDeleteCollection, isFromOtherCluster, signatures, func(chunks []*filer_pb.FileChunk) error {
+			if shouldDeleteChunks && !isDeleteCollection {
+				f.DirectDeleteChunks(chunks)
+			}
+			return nil
+		}, func(hardLinkIds []HardLinkId) error {
+			// A case not handled:
+			// what if the chunk is in a different collection?
+			if shouldDeleteChunks {
+				f.maybeDeleteHardLinks(hardLinkIds)
+			}
+			return nil
+		})
 		if err != nil {
 			glog.V(0).Infof("delete directory %s: %v", p, err)
 			return fmt.Errorf("delete directory %s: %v", p, err)
 		}
-		chunks = append(chunks, dirChunks...)
-		hardLinkIds = append(hardLinkIds, dirHardLinkIds...)
+	}
+
+	if shouldDeleteChunks && !isDeleteCollection {
+		f.DirectDeleteChunks(entry.Chunks)
 	}
 
 	// delete the file or folder
@@ -49,15 +61,6 @@ func (f *Filer) DeleteEntryMetaAndData(ctx context.Context, p util.FullPath, isR
 		return fmt.Errorf("delete file %s: %v", p, err)
 	}
 
-	if shouldDeleteChunks && !isDeleteCollection {
-		f.DirectDeleteChunks(chunks)
-	}
-	// A case not handled:
-	// what if the chunk is in a different collection?
-	if shouldDeleteChunks {
-		f.maybeDeleteHardLinks(hardLinkIds)
-	}
-
 	if isDeleteCollection {
 		collectionName := entry.Name()
 		f.doDeleteCollection(collectionName)
@@ -67,7 +70,7 @@ func (f *Filer) DeleteEntryMetaAndData(ctx context.Context, p util.FullPath, isR
 	return nil
 }
 
-func (f *Filer) doBatchDeleteFolderMetaAndData(ctx context.Context, entry *Entry, isRecursive, ignoreRecursiveError, shouldDeleteChunks, isDeletingBucket, isFromOtherCluster bool, signatures []int32) (chunks []*filer_pb.FileChunk, hardlinkIds []HardLinkId, err error) {
+func (f *Filer) doBatchDeleteFolderMetaAndData(ctx context.Context, entry *Entry, isRecursive, ignoreRecursiveError, shouldDeleteChunks, isDeletingBucket, isFromOtherCluster bool, signatures []int32, onChunksFn OnChunksFunc, onHardLinkIdsFn OnHardLinkIdsFunc) (err error) {
 
 	lastFileName := ""
 	includeLastFile := false
@@ -76,34 +79,30 @@ func (f *Filer) doBatchDeleteFolderMetaAndData(ctx context.Context, entry *Entry
 			entries, _, err := f.ListDirectoryEntries(ctx, entry.FullPath, lastFileName, includeLastFile, PaginationSize, "", "", "")
 			if err != nil {
 				glog.Errorf("list folder %s: %v", entry.FullPath, err)
-				return nil, nil, fmt.Errorf("list folder %s: %v", entry.FullPath, err)
+				return fmt.Errorf("list folder %s: %v", entry.FullPath, err)
 			}
 			if lastFileName == "" && !isRecursive && len(entries) > 0 {
 				// only for first iteration in the loop
 				glog.V(0).Infof("deleting a folder %s has children: %+v ...", entry.FullPath, entries[0].Name())
-				return nil, nil, fmt.Errorf("%s: %s", MsgFailDelNonEmptyFolder, entry.FullPath)
+				return fmt.Errorf("%s: %s", MsgFailDelNonEmptyFolder, entry.FullPath)
 			}
 
 			for _, sub := range entries {
 				lastFileName = sub.Name()
-				var dirChunks []*filer_pb.FileChunk
-				var dirHardLinkIds []HardLinkId
 				if sub.IsDirectory() {
 					subIsDeletingBucket := f.isBucket(sub)
-					dirChunks, dirHardLinkIds, err = f.doBatchDeleteFolderMetaAndData(ctx, sub, isRecursive, ignoreRecursiveError, shouldDeleteChunks, subIsDeletingBucket, false, nil)
-					chunks = append(chunks, dirChunks...)
-					hardlinkIds = append(hardlinkIds, dirHardLinkIds...)
+					err = f.doBatchDeleteFolderMetaAndData(ctx, sub, isRecursive, ignoreRecursiveError, shouldDeleteChunks, subIsDeletingBucket, false, nil, onChunksFn, onHardLinkIdsFn)
 				} else {
 					f.NotifyUpdateEvent(ctx, sub, nil, shouldDeleteChunks, isFromOtherCluster, nil)
 					if len(sub.HardLinkId) != 0 {
 						// hard link chunk data are deleted separately
-						hardlinkIds = append(hardlinkIds, sub.HardLinkId)
+						err = onHardLinkIdsFn([]HardLinkId{sub.HardLinkId})
 					} else {
-						chunks = append(chunks, sub.Chunks...)
+						err = onChunksFn(sub.Chunks)
 					}
 				}
 				if err != nil && !ignoreRecursiveError {
-					return nil, nil, err
+					return err
 				}
 			}
 
@@ -113,15 +112,15 @@ func (f *Filer) doBatchDeleteFolderMetaAndData(ctx context.Context, entry *Entry
 		}
 	}
 
-	glog.V(3).Infof("deleting directory %v delete %d chunks: %v", entry.FullPath, len(chunks), shouldDeleteChunks)
+	glog.V(3).Infof("deleting directory %v delete chunks: %v", entry.FullPath, shouldDeleteChunks)
 
 	if storeDeletionErr := f.Store.DeleteFolderChildren(ctx, entry.FullPath); storeDeletionErr != nil {
-		return nil, nil, fmt.Errorf("filer store delete: %v", storeDeletionErr)
+		return fmt.Errorf("filer store delete: %v", storeDeletionErr)
 	}
 
 	f.NotifyUpdateEvent(ctx, entry, nil, shouldDeleteChunks, isFromOtherCluster, signatures)
 
-	return chunks, hardlinkIds, nil
+	return nil
 }
 
 func (f *Filer) doDeleteEntryMetaAndData(ctx context.Context, entry *Entry, shouldDeleteChunks bool, isFromOtherCluster bool, signatures []int32) (err error) {

weed/filer/meta_aggregator.go

@@ -3,6 +3,7 @@ package filer
 import (
 	"context"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"io"
 	"sync"
@@ -18,9 +19,13 @@ import (
 )
 
 type MetaAggregator struct {
-	filers         []pb.ServerAddress
-	grpcDialOption grpc.DialOption
-	MetaLogBuffer  *log_buffer.LogBuffer
+	filer           *Filer
+	self            pb.ServerAddress
+	isLeader        bool
+	grpcDialOption  grpc.DialOption
+	MetaLogBuffer   *log_buffer.LogBuffer
+	peerStatues     map[pb.ServerAddress]struct{}
+	peerStatuesLock sync.Mutex
 	// notifying clients
 	ListenersLock sync.Mutex
 	ListenersCond *sync.Cond
@@ -28,10 +33,12 @@ type MetaAggregator struct {
 
 // MetaAggregator only aggregates data "on the fly". The logs are not re-persisted to disk.
 // The old data comes from what each LocalMetadata persisted on disk.
-func NewMetaAggregator(filers []pb.ServerAddress, grpcDialOption grpc.DialOption) *MetaAggregator {
+func NewMetaAggregator(filer *Filer, self pb.ServerAddress, grpcDialOption grpc.DialOption) *MetaAggregator {
 	t := &MetaAggregator{
-		filers:         filers,
+		filer:          filer,
+		self:           self,
 		grpcDialOption: grpcDialOption,
+		peerStatues:    make(map[pb.ServerAddress]struct{}),
 	}
 	t.ListenersCond = sync.NewCond(&t.ListenersLock)
 	t.MetaLogBuffer = log_buffer.NewLogBuffer("aggr", LogFlushInterval, nil, func() {
@@ -40,10 +47,35 @@ func NewMetaAggregator(filers []pb.ServerAddress, grpcDialOption grpc.DialOption
 	return t
 }
 
-func (ma *MetaAggregator) StartLoopSubscribe(f *Filer, self pb.ServerAddress) {
-	for _, filer := range ma.filers {
-		go ma.subscribeToOneFiler(f, self, filer)
+func (ma *MetaAggregator) OnPeerUpdate(update *master_pb.ClusterNodeUpdate) {
+	if update.NodeType != "filer" {
+		return
 	}
+
+	address := pb.ServerAddress(update.Address)
+	if update.IsAdd {
+		// every filer should subscribe to a new filer
+		ma.setActive(address, true)
+		go ma.subscribeToOneFiler(ma.filer, ma.self, address)
+	} else {
+		ma.setActive(address, false)
+	}
+}
+
+func (ma *MetaAggregator) setActive(address pb.ServerAddress, isActive bool) {
+	ma.peerStatuesLock.Lock()
+	defer ma.peerStatuesLock.Unlock()
+	if isActive {
+		ma.peerStatues[address] = struct{}{}
+	} else {
+		delete(ma.peerStatues, address)
+	}
+}
+func (ma *MetaAggregator) isActive(address pb.ServerAddress)(isActive bool) {
+	ma.peerStatuesLock.Lock()
+	defer ma.peerStatuesLock.Unlock()
+	_, isActive = ma.peerStatues[address]
+	return
 }
 
 func (ma *MetaAggregator) subscribeToOneFiler(f *Filer, self pb.ServerAddress, peer pb.ServerAddress) {
@@ -149,6 +181,10 @@ func (ma *MetaAggregator) subscribeToOneFiler(f *Filer, self pb.ServerAddress, p
 
 			}
 		})
+		if !ma.isActive(peer) {
+			glog.V(0).Infof("stop subscribing remote %s meta change", peer)
+			return
+		}
 		if err != nil {
 			glog.V(0).Infof("subscribing remote %s meta change: %v", peer, err)
 			time.Sleep(1733 * time.Millisecond)

weed/filer/redis2/universal_redis_store.go

@@ -134,8 +134,8 @@ func (store *UniversalRedis2Store) DeleteFolderChildren(ctx context.Context, ful
 	}
 
 	members, err := store.Client.ZRangeByLex(ctx, genDirectoryListKey(string(fullpath)), &redis.ZRangeBy{
-		Min:    "-",
-		Max:    "+",
+		Min: "-",
+		Max: "+",
 	}).Result()
 	if err != nil {
 		return fmt.Errorf("DeleteFolderChildren %s : %v", fullpath, err)

weed/filesys/dir_rename.go

@@ -67,7 +67,6 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
 	return nil
 }
 
-
 func (dir *Dir) handleRenameResponse(ctx context.Context, resp *filer_pb.StreamRenameEntryResponse) error {
 	// comes from filer StreamRenameEntry, can only be create or delete entry
 
@@ -119,7 +118,7 @@ func (dir *Dir) handleRenameResponse(ctx context.Context, resp *filer_pb.StreamR
 		}
 		dir.wfs.handlesLock.Unlock()
 
-	}else if resp.EventNotification.OldEntry != nil {
+	} else if resp.EventNotification.OldEntry != nil {
 		// without new entry, only old entry name exists. This is the second step to delete old entry
 		if err := dir.wfs.metaCache.AtomicUpdateEntryFromFiler(ctx, util.NewFullPath(resp.Directory, resp.EventNotification.OldEntry.Name), nil); err != nil {
 			return err

weed/messaging/broker/broker_grpc_server_discovery.go

@@ -93,14 +93,16 @@ func (broker *MessageBroker) checkFilers() {
 	for !found {
 		for _, master := range masters {
 			err := broker.withMasterClient(master, func(client master_pb.SeaweedClient) error {
-				resp, err := client.ListMasterClients(context.Background(), &master_pb.ListMasterClientsRequest{
+				resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
 					ClientType: "filer",
 				})
 				if err != nil {
 					return err
 				}
 
-				filers = append(filers, pb.FromAddressStrings(resp.GrpcAddresses)...)
+				for _, clusterNode := range resp.ClusterNodes {
+					filers = append(filers, pb.ServerAddress(clusterNode.Address))
+				}
 
 				return nil
 			})

weed/pb/master.proto

@@ -7,209 +7,222 @@ option go_package = "github.com/chrislusf/seaweedfs/weed/pb/master_pb";
 //////////////////////////////////////////////////
 
 service Seaweed {
-    rpc SendHeartbeat (stream Heartbeat) returns (stream HeartbeatResponse) {
-    }
-    rpc KeepConnected (stream KeepConnectedRequest) returns (stream VolumeLocation) {
-    }
-    rpc LookupVolume (LookupVolumeRequest) returns (LookupVolumeResponse) {
-    }
-    rpc Assign (AssignRequest) returns (AssignResponse) {
-    }
-    rpc Statistics (StatisticsRequest) returns (StatisticsResponse) {
-    }
-    rpc CollectionList (CollectionListRequest) returns (CollectionListResponse) {
-    }
-    rpc CollectionDelete (CollectionDeleteRequest) returns (CollectionDeleteResponse) {
-    }
-    rpc VolumeList (VolumeListRequest) returns (VolumeListResponse) {
-    }
-    rpc LookupEcVolume (LookupEcVolumeRequest) returns (LookupEcVolumeResponse) {
-    }
-    rpc VacuumVolume (VacuumVolumeRequest) returns (VacuumVolumeResponse) {
-    }
-    rpc GetMasterConfiguration (GetMasterConfigurationRequest) returns (GetMasterConfigurationResponse) {
-    }
-    rpc ListMasterClients (ListMasterClientsRequest) returns (ListMasterClientsResponse) {
-    }
-    rpc LeaseAdminToken (LeaseAdminTokenRequest) returns (LeaseAdminTokenResponse) {
-    }
-    rpc ReleaseAdminToken (ReleaseAdminTokenRequest) returns (ReleaseAdminTokenResponse) {
-    }
+  rpc SendHeartbeat (stream Heartbeat) returns (stream HeartbeatResponse) {
+  }
+  rpc KeepConnected (stream KeepConnectedRequest) returns (stream KeepConnectedResponse) {
+  }
+  rpc LookupVolume (LookupVolumeRequest) returns (LookupVolumeResponse) {
+  }
+  rpc Assign (AssignRequest) returns (AssignResponse) {
+  }
+  rpc Statistics (StatisticsRequest) returns (StatisticsResponse) {
+  }
+  rpc CollectionList (CollectionListRequest) returns (CollectionListResponse) {
+  }
+  rpc CollectionDelete (CollectionDeleteRequest) returns (CollectionDeleteResponse) {
+  }
+  rpc VolumeList (VolumeListRequest) returns (VolumeListResponse) {
+  }
+  rpc LookupEcVolume (LookupEcVolumeRequest) returns (LookupEcVolumeResponse) {
+  }
+  rpc VacuumVolume (VacuumVolumeRequest) returns (VacuumVolumeResponse) {
+  }
+  rpc GetMasterConfiguration (GetMasterConfigurationRequest) returns (GetMasterConfigurationResponse) {
+  }
+  rpc ListClusterNodes (ListClusterNodesRequest) returns (ListClusterNodesResponse) {
+  }
+  rpc LeaseAdminToken (LeaseAdminTokenRequest) returns (LeaseAdminTokenResponse) {
+  }
+  rpc ReleaseAdminToken (ReleaseAdminTokenRequest) returns (ReleaseAdminTokenResponse) {
+  }
 
 }
 
 //////////////////////////////////////////////////
 
 message Heartbeat {
-    string ip = 1;
-    uint32 port = 2;
-    string public_url = 3;
-    uint64 max_file_key = 5;
-    string data_center = 6;
-    string rack = 7;
-    uint32 admin_port = 8;
-    repeated VolumeInformationMessage volumes = 9;
-    // delta volumes
-    repeated VolumeShortInformationMessage new_volumes = 10;
-    repeated VolumeShortInformationMessage deleted_volumes = 11;
-    bool has_no_volumes = 12;
+  string ip = 1;
+  uint32 port = 2;
+  string public_url = 3;
+  uint64 max_file_key = 5;
+  string data_center = 6;
+  string rack = 7;
+  uint32 admin_port = 8;
+  repeated VolumeInformationMessage volumes = 9;
+  // delta volumes
+  repeated VolumeShortInformationMessage new_volumes = 10;
+  repeated VolumeShortInformationMessage deleted_volumes = 11;
+  bool has_no_volumes = 12;
 
-    // erasure coding
-    repeated VolumeEcShardInformationMessage ec_shards = 16;
-    // delta erasure coding shards
-    repeated VolumeEcShardInformationMessage new_ec_shards = 17;
-    repeated VolumeEcShardInformationMessage deleted_ec_shards = 18;
-    bool has_no_ec_shards = 19;
+  // erasure coding
+  repeated VolumeEcShardInformationMessage ec_shards = 16;
+  // delta erasure coding shards
+  repeated VolumeEcShardInformationMessage new_ec_shards = 17;
+  repeated VolumeEcShardInformationMessage deleted_ec_shards = 18;
+  bool has_no_ec_shards = 19;
 
-    map<string, uint32> max_volume_counts = 4;
-    uint32 grpc_port = 20;
+  map<string, uint32> max_volume_counts = 4;
+  uint32 grpc_port = 20;
 
 }
 
 message HeartbeatResponse {
-    uint64 volume_size_limit = 1;
-    string leader = 2;
-    string metrics_address = 3;
-    uint32 metrics_interval_seconds = 4;
-    repeated StorageBackend storage_backends = 5;
+  uint64 volume_size_limit = 1;
+  string leader = 2;
+  string metrics_address = 3;
+  uint32 metrics_interval_seconds = 4;
+  repeated StorageBackend storage_backends = 5;
 }
 
 message VolumeInformationMessage {
-    uint32 id = 1;
-    uint64 size = 2;
-    string collection = 3;
-    uint64 file_count = 4;
-    uint64 delete_count = 5;
-    uint64 deleted_byte_count = 6;
-    bool read_only = 7;
-    uint32 replica_placement = 8;
-    uint32 version = 9;
-    uint32 ttl = 10;
-    uint32 compact_revision = 11;
-    int64 modified_at_second = 12;
-    string remote_storage_name = 13;
-    string remote_storage_key = 14;
-    string disk_type = 15;
+  uint32 id = 1;
+  uint64 size = 2;
+  string collection = 3;
+  uint64 file_count = 4;
+  uint64 delete_count = 5;
+  uint64 deleted_byte_count = 6;
+  bool read_only = 7;
+  uint32 replica_placement = 8;
+  uint32 version = 9;
+  uint32 ttl = 10;
+  uint32 compact_revision = 11;
+  int64 modified_at_second = 12;
+  string remote_storage_name = 13;
+  string remote_storage_key = 14;
+  string disk_type = 15;
 }
 
 message VolumeShortInformationMessage {
-    uint32 id = 1;
-    string collection = 3;
-    uint32 replica_placement = 8;
-    uint32 version = 9;
-    uint32 ttl = 10;
-    string disk_type = 15;
+  uint32 id = 1;
+  string collection = 3;
+  uint32 replica_placement = 8;
+  uint32 version = 9;
+  uint32 ttl = 10;
+  string disk_type = 15;
 }
 
 message VolumeEcShardInformationMessage {
-    uint32 id = 1;
-    string collection = 2;
-    uint32 ec_index_bits = 3;
-    string disk_type = 4;
+  uint32 id = 1;
+  string collection = 2;
+  uint32 ec_index_bits = 3;
+  string disk_type = 4;
 }
 
 message StorageBackend {
-    string type = 1;
-    string id = 2;
-    map<string, string> properties = 3;
+  string type = 1;
+  string id = 2;
+  map<string, string> properties = 3;
 }
 
 message Empty {
 }
 
 message SuperBlockExtra {
-    message ErasureCoding {
-        uint32 data = 1;
-        uint32 parity = 2;
-        repeated uint32 volume_ids = 3;
-    }
-    ErasureCoding erasure_coding = 1;
+  message ErasureCoding {
+    uint32 data = 1;
+    uint32 parity = 2;
+    repeated uint32 volume_ids = 3;
+  }
+  ErasureCoding erasure_coding = 1;
 }
 
 message KeepConnectedRequest {
-    string name = 1;
-    string client_address = 3;
+  string client_type = 1;
+  string client_address = 3;
+  string version = 4;
 }
 
 message VolumeLocation {
-    string url = 1;
-    string public_url = 2;
-    repeated uint32 new_vids = 3;
-    repeated uint32 deleted_vids = 4;
-    string leader = 5; // optional when leader is not itself
-    string data_center = 6; // optional when DataCenter is in use
-    uint32 grpc_port = 7;
+  string url = 1;
+  string public_url = 2;
+  repeated uint32 new_vids = 3;
+  repeated uint32 deleted_vids = 4;
+  string leader = 5; // optional when leader is not itself
+  string data_center = 6; // optional when DataCenter is in use
+  uint32 grpc_port = 7;
+}
+
+message ClusterNodeUpdate {
+  string node_type = 1;
+  string address = 2;
+  bool is_leader = 3;
+  bool is_add = 4;
+}
+
+message KeepConnectedResponse {
+  VolumeLocation volume_location = 1;
+  ClusterNodeUpdate cluster_node_update = 2;
 }
 
 message LookupVolumeRequest {
-    repeated string volume_or_file_ids = 1;
-    string collection = 2; // optional, a bit faster if provided.
+  repeated string volume_or_file_ids = 1;
+  string collection = 2; // optional, a bit faster if provided.
 }
 message LookupVolumeResponse {
-    message VolumeIdLocation {
-        string volume_or_file_id = 1;
-        repeated Location locations = 2;
-        string error = 3;
-        string auth = 4;
-    }
-    repeated VolumeIdLocation volume_id_locations = 1;
+  message VolumeIdLocation {
+    string volume_or_file_id = 1;
+    repeated Location locations = 2;
+    string error = 3;
+    string auth = 4;
+  }
+  repeated VolumeIdLocation volume_id_locations = 1;
 }
 
 message Location {
-    string url = 1;
-    string public_url = 2;
-    uint32 grpc_port = 3;
+  string url = 1;
+  string public_url = 2;
+  uint32 grpc_port = 3;
 }
 
 message AssignRequest {
-    uint64 count = 1;
-    string replication = 2;
-    string collection = 3;
-    string ttl = 4;
-    string data_center = 5;
-    string rack = 6;
-    string data_node = 7;
-    uint32 memory_map_max_size_mb = 8;
-    uint32 Writable_volume_count = 9;
-    string disk_type = 10;
+  uint64 count = 1;
+  string replication = 2;
+  string collection = 3;
+  string ttl = 4;
+  string data_center = 5;
+  string rack = 6;
+  string data_node = 7;
+  uint32 memory_map_max_size_mb = 8;
+  uint32 Writable_volume_count = 9;
+  string disk_type = 10;
 }
 message AssignResponse {
-    string fid = 1;
-    uint64 count = 4;
-    string error = 5;
-    string auth = 6;
-    repeated Location replicas = 7;
-    Location location = 8;
+  string fid = 1;
+  uint64 count = 4;
+  string error = 5;
+  string auth = 6;
+  repeated Location replicas = 7;
+  Location location = 8;
 }
 
 message StatisticsRequest {
-    string replication = 1;
-    string collection = 2;
-    string ttl = 3;
-    string disk_type = 4;
+  string replication = 1;
+  string collection = 2;
+  string ttl = 3;
+  string disk_type = 4;
 }
 message StatisticsResponse {
-    uint64 total_size = 4;
-    uint64 used_size = 5;
-    uint64 file_count = 6;
+  uint64 total_size = 4;
+  uint64 used_size = 5;
+  uint64 file_count = 6;
 }
 
 //
 // collection related
 //
 message Collection {
-    string name = 1;
+  string name = 1;
 }
 message CollectionListRequest {
-    bool include_normal_volumes = 1;
-    bool include_ec_volumes = 2;
+  bool include_normal_volumes = 1;
+  bool include_ec_volumes = 2;
 }
 message CollectionListResponse {
-    repeated Collection collections = 1;
+  repeated Collection collections = 1;
 }
 
 message CollectionDeleteRequest {
-    string name = 1;
+  string name = 1;
 }
 message CollectionDeleteResponse {
 }
@@ -218,56 +231,56 @@ message CollectionDeleteResponse {
 // volume related
 //
 message DiskInfo {
-    string type = 1;
-    int64 volume_count = 2;
-    int64 max_volume_count = 3;
-    int64 free_volume_count = 4;
-    int64 active_volume_count = 5;
-    repeated VolumeInformationMessage volume_infos = 6;
-    repeated VolumeEcShardInformationMessage ec_shard_infos = 7;
-    int64 remote_volume_count = 8;
+  string type = 1;
+  int64 volume_count = 2;
+  int64 max_volume_count = 3;
+  int64 free_volume_count = 4;
+  int64 active_volume_count = 5;
+  repeated VolumeInformationMessage volume_infos = 6;
+  repeated VolumeEcShardInformationMessage ec_shard_infos = 7;
+  int64 remote_volume_count = 8;
 }
 message DataNodeInfo {
-    string id = 1;
-    map<string, DiskInfo> diskInfos = 2;
-    uint32 grpc_port = 3;
+  string id = 1;
+  map<string, DiskInfo> diskInfos = 2;
+  uint32 grpc_port = 3;
 }
 message RackInfo {
-    string id = 1;
-    repeated DataNodeInfo data_node_infos = 2;
-    map<string, DiskInfo> diskInfos = 3;
+  string id = 1;
+  repeated DataNodeInfo data_node_infos = 2;
+  map<string, DiskInfo> diskInfos = 3;
 }
 message DataCenterInfo {
-    string id = 1;
-    repeated RackInfo rack_infos = 2;
-    map<string, DiskInfo> diskInfos = 3;
+  string id = 1;
+  repeated RackInfo rack_infos = 2;
+  map<string, DiskInfo> diskInfos = 3;
 }
 message TopologyInfo {
-    string id = 1;
-    repeated DataCenterInfo data_center_infos = 2;
-    map<string, DiskInfo> diskInfos = 3;
+  string id = 1;
+  repeated DataCenterInfo data_center_infos = 2;
+  map<string, DiskInfo> diskInfos = 3;
 }
 message VolumeListRequest {
 }
 message VolumeListResponse {
-    TopologyInfo topology_info = 1;
-    uint64 volume_size_limit_mb = 2;
+  TopologyInfo topology_info = 1;
+  uint64 volume_size_limit_mb = 2;
 }
 
 message LookupEcVolumeRequest {
-    uint32 volume_id = 1;
+  uint32 volume_id = 1;
 }
 message LookupEcVolumeResponse {
-    uint32 volume_id = 1;
-    message EcShardIdLocation {
-        uint32 shard_id = 1;
-        repeated Location locations = 2;
-    }
-    repeated EcShardIdLocation shard_id_locations = 2;
+  uint32 volume_id = 1;
+  message EcShardIdLocation {
+    uint32 shard_id = 1;
+    repeated Location locations = 2;
+  }
+  repeated EcShardIdLocation shard_id_locations = 2;
 }
 
 message VacuumVolumeRequest {
-    float garbage_threshold = 1;
+  float garbage_threshold = 1;
 }
 message VacuumVolumeResponse {
 }
@@ -275,37 +288,42 @@ message VacuumVolumeResponse {
 message GetMasterConfigurationRequest {
 }
 message GetMasterConfigurationResponse {
-    string metrics_address = 1;
-    uint32 metrics_interval_seconds = 2;
-    repeated StorageBackend storage_backends = 3;
-    string default_replication = 4;
-    string leader = 5;
-    uint32 volume_size_limit_m_b = 6;
-    bool volume_preallocate = 7;
+  string metrics_address = 1;
+  uint32 metrics_interval_seconds = 2;
+  repeated StorageBackend storage_backends = 3;
+  string default_replication = 4;
+  string leader = 5;
+  uint32 volume_size_limit_m_b = 6;
+  bool volume_preallocate = 7;
 }
 
-message ListMasterClientsRequest {
-    string client_type = 1;
+message ListClusterNodesRequest {
+  string client_type = 1;
 }
-message ListMasterClientsResponse {
-    repeated string grpc_addresses = 1;
+message ListClusterNodesResponse {
+  message ClusterNode {
+    string address = 1;
+    string version = 2;
+    bool is_leader = 3;
+  }
+  repeated ClusterNode cluster_nodes = 1;
 }
 
 message LeaseAdminTokenRequest {
-    int64 previous_token = 1;
-    int64 previous_lock_time = 2;
-    string lock_name = 3;
-    string client_name = 4;
+  int64 previous_token = 1;
+  int64 previous_lock_time = 2;
+  string lock_name = 3;
+  string client_name = 4;
 }
 message LeaseAdminTokenResponse {
-    int64 token = 1;
-    int64 lock_ts_ns = 2;
+  int64 token = 1;
+  int64 lock_ts_ns = 2;
 }
 
 message ReleaseAdminTokenRequest {
-    int64 previous_token = 1;
-    int64 previous_lock_time = 2;
-    string lock_name = 3;
+  int64 previous_token = 1;
+  int64 previous_lock_time = 2;
+  string lock_name = 3;
 }
 message ReleaseAdminTokenResponse {
 }

weed/remote_storage/s3/s3_storage_client.go

@@ -83,6 +83,7 @@ func (s *s3RemoteStorageClient) Traverse(remote *remote_pb.RemoteStorageLocation
 	}
 	isLastPage := false
 	for !isLastPage && err == nil {
+		var localErr error
 		listErr := s.conn.ListObjectsV2Pages(listInput, func(page *s3.ListObjectsV2Output, lastPage bool) bool {
 			for _, content := range page.Contents {
 				key := *content.Key
@@ -94,6 +95,7 @@ func (s *s3RemoteStorageClient) Traverse(remote *remote_pb.RemoteStorageLocation
 					RemoteETag:  *content.ETag,
 					StorageName: s.conf.Name,
 				}); err != nil {
+					localErr = err
 					return false
 				}
 			}
@@ -104,6 +106,9 @@ func (s *s3RemoteStorageClient) Traverse(remote *remote_pb.RemoteStorageLocation
 		if listErr != nil {
 			err = fmt.Errorf("list %v: %v", remote, listErr)
 		}
+		if localErr != nil {
+			err = fmt.Errorf("process %v: %v", remote, localErr)
+		}
 	}
 	return
 }

weed/s3api/auth_credentials.go

@@ -146,11 +146,13 @@ func (iam *IdentityAccessManagement) lookupByAccessKey(accessKey string) (identi
 
 	for _, ident := range iam.identities {
 		for _, cred := range ident.Credentials {
+			// println("checking", ident.Name, cred.AccessKey)
 			if cred.AccessKey == accessKey {
 				return ident, cred, true
 			}
 		}
 	}
+	glog.V(1).Infof("could not find accessKey %s", accessKey)
 	return nil, nil, false
 }
 

weed/s3api/s3api_bucket_handlers.go

@@ -111,6 +111,13 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request)
 		return
 	}
 
+	if s3a.iam.isEnabled() {
+		if _, errCode = s3a.iam.authRequest(r, s3_constants.ACTION_ADMIN); errCode != s3err.ErrNone {
+			s3err.WriteErrorResponse(w, r, errCode)
+			return
+		}
+	}
+
 	fn := func(entry *filer_pb.Entry) {
 		if identityId := r.Header.Get(xhttp.AmzIdentityId); identityId != "" {
 			if entry.Extended == nil {

weed/s3api/s3api_objects_list_handlers.go

@@ -220,12 +220,16 @@ func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, d
 			err = subErr
 			return
 		}
+		counter += subCounter
 		isTruncated = isTruncated || subIsTruncated
 		maxKeys -= subCounter
 		nextMarker = subDir + "/" + subNextMarker
 		// finished processing this sub directory
 		marker = subDir
 	}
+	if maxKeys <= 0 {
+		return
+	}
 
 	// now marker is also a direct child of dir
 	request := &filer_pb.ListEntriesRequest{

weed/s3api/s3api_server.go

@@ -66,7 +66,7 @@ func (s3a *S3ApiServer) registerRouter(router *mux.Router) {
 		// HeadObject
 		bucket.Methods("HEAD").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.HeadObjectHandler, ACTION_READ), "GET"))
 		// HeadBucket
-		bucket.Methods("HEAD").HandlerFunc(track(s3a.iam.Auth(s3a.HeadBucketHandler, ACTION_ADMIN), "GET"))
+		bucket.Methods("HEAD").HandlerFunc(track(s3a.iam.Auth(s3a.HeadBucketHandler, ACTION_READ), "GET"))
 
 		// CopyObjectPart
 		bucket.Methods("PUT").Path("/{object:.+}").HeadersRegexp("X-Amz-Copy-Source", `.*?(\/|%2F).*?`).HandlerFunc(track(s3a.iam.Auth(s3a.CopyObjectPartHandler, ACTION_WRITE), "PUT")).Queries("partNumber", "{partNumber:[0-9]+}", "uploadId", "{uploadId:.*}")
@@ -137,7 +137,7 @@ func (s3a *S3ApiServer) registerRouter(router *mux.Router) {
 		bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.ListObjectsV1Handler, ACTION_LIST), "LIST"))
 
 		// PutBucket
-		bucket.Methods("PUT").HandlerFunc(track(s3a.iam.Auth(s3a.PutBucketHandler, ACTION_ADMIN), "PUT"))
+		bucket.Methods("PUT").HandlerFunc(track(s3a.PutBucketHandler, "PUT"))
 
 		// DeleteBucket
 		bucket.Methods("DELETE").HandlerFunc(track(s3a.iam.Auth(s3a.DeleteBucketHandler, ACTION_WRITE), "DELETE"))

weed/server/filer_grpc_server_rename.go

@@ -161,7 +161,7 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee
 		if err := stream.Send(&filer_pb.StreamRenameEntryResponse{
 			Directory: string(newParent),
 			EventNotification: &filer_pb.EventNotification{
-				OldEntry:           &filer_pb.Entry{
+				OldEntry: &filer_pb.Entry{
 					Name: entry.Name(),
 				},
 				NewEntry:           newEntry.ToProtoEntry(),

weed/server/filer_server.go

@@ -61,7 +61,6 @@ type FilerOption struct {
 	recursiveDelete       bool
 	Cipher                bool
 	SaveToFilerLimit      int64
-	Filers                []pb.ServerAddress
 	ConcurrentUploadLimit int64
 }
 
@@ -108,7 +107,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 	fs.checkWithMaster()
 
 	go stats.LoopPushingMetric("filer", string(fs.option.Host), fs.metricsAddress, fs.metricsIntervalSec)
-	go fs.filer.KeepConnectedToMaster()
+	go fs.filer.KeepMasterClientConnected()
 
 	v := util.GetViper()
 	if !util.LoadConfiguration("filer", false) {
@@ -143,7 +142,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 		readonlyMux.HandleFunc("/", fs.readonlyFilerHandler)
 	}
 
-	fs.filer.AggregateFromPeers(option.Host, option.Filers)
+	fs.filer.AggregateFromPeers(option.Host)
 
 	fs.filer.LoadBuckets()
 

weed/server/master_grpc_server.go

@@ -6,7 +6,6 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/storage/backend"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"net"
-	"strings"
 	"time"
 
 	"github.com/chrislusf/raft"
@@ -45,11 +44,7 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 			}
 
 			if len(message.DeletedVids) > 0 {
-				ms.clientChansLock.RLock()
-				for _, ch := range ms.clientChans {
-					ch <- message
-				}
-				ms.clientChansLock.RUnlock()
+				ms.broadcastToClients(&master_pb.KeepConnectedResponse{VolumeLocation: message})
 			}
 		}
 	}()
@@ -154,12 +149,7 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 
 		}
 		if len(message.NewVids) > 0 || len(message.DeletedVids) > 0 {
-			ms.clientChansLock.RLock()
-			for host, ch := range ms.clientChans {
-				glog.V(0).Infof("master send to %s: %s", host, message.String())
-				ch <- message
-			}
-			ms.clientChansLock.RUnlock()
+			ms.broadcastToClients(&master_pb.KeepConnectedResponse{VolumeLocation: message})
 		}
 
 		// tell the volume servers about the leader
@@ -195,12 +185,20 @@ func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServ
 	// buffer by 1 so we don't end up getting stuck writing to stopChan forever
 	stopChan := make(chan bool, 1)
 
-	clientName, messageChan := ms.addClient(req.Name, peerAddress)
+	clientName, messageChan := ms.addClient(req.ClientType, peerAddress)
+	for _, update := range ms.Cluster.AddClusterNode(req.ClientType, peerAddress, req.Version) {
+		ms.broadcastToClients(update)
+	}
 
-	defer ms.deleteClient(clientName)
+	defer func() {
+		for _, update := range ms.Cluster.RemoveClusterNode(req.ClientType, peerAddress) {
+			ms.broadcastToClients(update)
+		}
+		ms.deleteClient(clientName)
+	}()
 
 	for _, message := range ms.Topo.ToVolumeLocations() {
-		if sendErr := stream.Send(message); sendErr != nil {
+		if sendErr := stream.Send(&master_pb.KeepConnectedResponse{VolumeLocation: message}); sendErr != nil {
 			return sendErr
 		}
 	}
@@ -235,21 +233,31 @@ func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServ
 
 }
 
+func (ms *MasterServer) broadcastToClients(message *master_pb.KeepConnectedResponse) {
+	ms.clientChansLock.RLock()
+	for _, ch := range ms.clientChans {
+		ch <- message
+	}
+	ms.clientChansLock.RUnlock()
+}
+
 func (ms *MasterServer) informNewLeader(stream master_pb.Seaweed_KeepConnectedServer) error {
 	leader, err := ms.Topo.Leader()
 	if err != nil {
 		glog.Errorf("topo leader: %v", err)
 		return raft.NotLeaderError
 	}
-	if err := stream.Send(&master_pb.VolumeLocation{
-		Leader: string(leader),
+	if err := stream.Send(&master_pb.KeepConnectedResponse{
+		VolumeLocation: &master_pb.VolumeLocation{
+			Leader: string(leader),
+		},
 	}); err != nil {
 		return err
 	}
 	return nil
 }
 
-func (ms *MasterServer) addClient(clientType string, clientAddress pb.ServerAddress) (clientName string, messageChan chan *master_pb.VolumeLocation) {
+func (ms *MasterServer) addClient(clientType string, clientAddress pb.ServerAddress) (clientName string, messageChan chan *master_pb.KeepConnectedResponse) {
 	clientName = clientType + "@" + string(clientAddress)
 	glog.V(0).Infof("+ client %v", clientName)
 
@@ -258,7 +266,7 @@ func (ms *MasterServer) addClient(clientType string, clientAddress pb.ServerAddr
 	// trying to send to it in SendHeartbeat and so we can't lock the
 	// clientChansLock to remove the channel and we're stuck writing to it
 	// 100 is probably overkill
-	messageChan = make(chan *master_pb.VolumeLocation, 100)
+	messageChan = make(chan *master_pb.KeepConnectedResponse, 100)
 
 	ms.clientChansLock.Lock()
 	ms.clientChans[clientName] = messageChan
@@ -295,19 +303,6 @@ func findClientAddress(ctx context.Context, grpcPort uint32) string {
 
 }
 
-func (ms *MasterServer) ListMasterClients(ctx context.Context, req *master_pb.ListMasterClientsRequest) (*master_pb.ListMasterClientsResponse, error) {
-	resp := &master_pb.ListMasterClientsResponse{}
-	ms.clientChansLock.RLock()
-	defer ms.clientChansLock.RUnlock()
-
-	for k := range ms.clientChans {
-		if strings.HasPrefix(k, req.ClientType+"@") {
-			resp.GrpcAddresses = append(resp.GrpcAddresses, k[len(req.ClientType)+1:])
-		}
-	}
-	return resp, nil
-}
-
 func (ms *MasterServer) GetMasterConfiguration(ctx context.Context, req *master_pb.GetMasterConfigurationRequest) (*master_pb.GetMasterConfigurationResponse, error) {
 
 	// tell the volume servers about the leader

weed/server/master_grpc_server_cluster.go

@@ -0,0 +1,21 @@
+package weed_server
+
+import (
+	"context"
+	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
+)
+
+func (ms *MasterServer) ListClusterNodes(ctx context.Context, req *master_pb.ListClusterNodesRequest) (*master_pb.ListClusterNodesResponse, error) {
+	resp := &master_pb.ListClusterNodesResponse{}
+
+	clusterNodes := ms.Cluster.ListClusterNode(req.ClientType)
+
+	for _, node := range clusterNodes {
+		resp.ClusterNodes = append(resp.ClusterNodes, &master_pb.ListClusterNodesResponse_ClusterNode{
+			Address: string(node.Address),
+			Version: node.Version,
+			IsLeader: ms.Cluster.IsOneLeader(node.Address),
+		})
+	}
+	return resp, nil
+}

weed/server/master_server.go

@@ -2,6 +2,7 @@ package weed_server
 
 import (
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/election"
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"net/http"
 	"net/http/httputil"
@@ -60,13 +61,15 @@ type MasterServer struct {
 
 	// notifying clients
 	clientChansLock sync.RWMutex
-	clientChans     map[string]chan *master_pb.VolumeLocation
+	clientChans     map[string]chan *master_pb.KeepConnectedResponse
 
 	grpcDialOption grpc.DialOption
 
 	MasterClient *wdclient.MasterClient
 
 	adminLocks *AdminLocks
+
+	Cluster *election.Cluster
 }
 
 func NewMasterServer(r *mux.Router, option *MasterOption, peers []pb.ServerAddress) *MasterServer {
@@ -99,10 +102,11 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers []pb.ServerAddre
 		option:          option,
 		preallocateSize: preallocateSize,
 		vgCh:            make(chan *topology.VolumeGrowRequest, 1<<6),
-		clientChans:     make(map[string]chan *master_pb.VolumeLocation),
+		clientChans:     make(map[string]chan *master_pb.KeepConnectedResponse),
 		grpcDialOption:  grpcDialOption,
 		MasterClient:    wdclient.NewMasterClient(grpcDialOption, "master", option.Master, "", peers),
 		adminLocks:      NewAdminLocks(),
+		Cluster:         election.NewCluster(),
 	}
 	ms.boundedLeaderChan = make(chan int, 16)
 

weed/server/volume_grpc_copy.go

@@ -80,7 +80,7 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre
 
 		// println("source:", volFileInfoResp.String())
 		copyResponse := &volume_server_pb.VolumeCopyResponse{}
-		reportInterval := int64(1024*1024*128)
+		reportInterval := int64(1024 * 1024 * 128)
 		nextReportTarget := reportInterval
 		var modifiedTsNs int64
 		var sendErr error

weed/server/volume_grpc_vacuum.go

@@ -27,7 +27,7 @@ func (vs *VolumeServer) VacuumVolumeCheck(ctx context.Context, req *volume_serve
 func (vs *VolumeServer) VacuumVolumeCompact(req *volume_server_pb.VacuumVolumeCompactRequest, stream volume_server_pb.VolumeServer_VacuumVolumeCompactServer) error {
 
 	resp := &volume_server_pb.VacuumVolumeCompactResponse{}
-	reportInterval := int64(1024*1024*128)
+	reportInterval := int64(1024 * 1024 * 128)
 	nextReportTarget := reportInterval
 
 	var sendErr error

weed/shell/command_cluster_ps.go

@@ -0,0 +1,54 @@
+package shell
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+
+	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
+)
+
+func init() {
+	Commands = append(Commands, &commandClusterPs{})
+}
+
+type commandClusterPs struct {
+}
+
+func (c *commandClusterPs) Name() string {
+	return "cluster.ps"
+}
+
+func (c *commandClusterPs) Help() string {
+	return `check current cluster process status
+
+	cluster.ps
+
+`
+}
+
+func (c *commandClusterPs) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
+
+	clusterPsCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	if err = clusterPsCommand.Parse(args); err != nil {
+		return nil
+	}
+
+	err = commandEnv.MasterClient.WithClient(func(client master_pb.SeaweedClient) error {
+		resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
+			ClientType: "filer",
+		})
+
+		fmt.Fprintf(writer, "the cluster has %d filers\n", len(resp.ClusterNodes))
+		for _, node := range resp.ClusterNodes {
+			fmt.Fprintf(writer, "  * %s (%v)\n", node.Address, node.Version)
+		}
+		return err
+	})
+	if err != nil {
+		return
+	}
+
+	return nil
+}

weed/shell/command_ec_encode.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"io"
+	"math/rand"
 	"sync"
 	"time"
 
@@ -33,7 +34,7 @@ func (c *commandEcEncode) Name() string {
 func (c *commandEcEncode) Help() string {
 	return `apply erasure coding to a volume
 
-	ec.encode [-collection=""] [-fullPercent=95] [-quietFor=1h]
+	ec.encode [-collection=""] [-fullPercent=95 -quietFor=1h]
 	ec.encode [-collection=""] [-volumeId=<volume_id>]
 
 	This command will:
@@ -248,7 +249,7 @@ func parallelCopyEcShardsFromSource(grpcDialOption grpc.DialOption, targetServer
 func balancedEcDistribution(servers []*EcNode) (allocated [][]uint32) {
 	allocated = make([][]uint32, len(servers))
 	allocatedShardIdIndex := uint32(0)
-	serverIndex := 0
+	serverIndex := rand.Intn(len(servers))
 	for allocatedShardIdIndex < erasure_coding.TotalShardsCount {
 		if servers[serverIndex].freeEcSlot > 0 {
 			allocated[serverIndex] = append(allocated[serverIndex], allocatedShardIdIndex)

weed/shell/command_remote_cache.go

@@ -26,7 +26,7 @@ func (c *commandRemoteCache) Help() string {
 	return `cache the file content for mounted directories or files
 
 	# assume a remote storage is configured to name "cloud1"
-	remote.configure -name=cloud1 -type=s3 -access_key=xxx -secret_key=yyy
+	remote.configure -name=cloud1 -type=s3 -s3.access_key=xxx -s3.secret_key=yyy
 	# mount and pull one bucket
 	remote.mount -dir=/xxx -remote=cloud1/bucket
 

weed/shell/command_remote_meta_sync.go

@@ -27,7 +27,7 @@ func (c *commandRemoteMetaSync) Help() string {
 	return `synchronize the local file meta data with the remote file metadata
 
 	# assume a remote storage is configured to name "cloud1"
-	remote.configure -name=cloud1 -type=s3 -access_key=xxx -secret_key=yyy
+	remote.configure -name=cloud1 -type=s3 -s3.access_key=xxx -s3.secret_key=yyy
 	# mount and pull one bucket
 	remote.mount -dir=/xxx -remote=cloud1/bucket
 

weed/shell/command_remote_mount.go

@@ -32,7 +32,7 @@ func (c *commandRemoteMount) Help() string {
 	return `mount remote storage and pull its metadata
 
 	# assume a remote storage is configured to name "cloud1"
-	remote.configure -name=cloud1 -type=s3 -access_key=xxx -secret_key=yyy
+	remote.configure -name=cloud1 -type=s3 -s3.access_key=xxx -s3.secret_key=yyy
 
 	# mount and pull one bucket
 	remote.mount -dir=/xxx -remote=cloud1/bucket

weed/shell/command_remote_mount_buckets.go

@@ -27,7 +27,7 @@ func (c *commandRemoteMountBuckets) Help() string {
 	return `mount all buckets in remote storage and pull its metadata
 
 	# assume a remote storage is configured to name "cloud1"
-	remote.configure -name=cloud1 -type=s3 -access_key=xxx -secret_key=yyy
+	remote.configure -name=cloud1 -type=s3 -s3.access_key=xxx -s3.secret_key=yyy
 
 	# mount all buckets
 	remote.mount.buckets -remote=cloud1

weed/shell/command_remote_unmount.go

@@ -27,7 +27,7 @@ func (c *commandRemoteUnmount) Help() string {
 	return `unmount remote storage
 
 	# assume a remote storage is configured to name "s3_1"
-	remote.configure -name=s3_1 -type=s3 -access_key=xxx -secret_key=yyy
+	remote.configure -name=s3_1 -type=s3 -s3.access_key=xxx -s3.secret_key=yyy
 	# mount and pull one bucket
 	remote.mount -dir=/xxx -remote=s3_1/bucket
 

weed/shell/command_volume_fix_replication_test.go

@@ -261,6 +261,29 @@ func TestSatisfyReplicaPlacement00x(t *testing.T) {
 
 }
 
+func TestSatisfyReplicaPlacement100(t *testing.T) {
+
+	var tests = []testcase{
+		{
+			name:        "test 100",
+			replication: "100",
+			replicas: []*VolumeReplica{
+				{
+					location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn1"}},
+				},
+				{
+					location: &location{"dc1", "r2", &master_pb.DataNodeInfo{Id: "dn2"}},
+				},
+			},
+			possibleLocation: location{"dc2", "r3", &master_pb.DataNodeInfo{Id: "dn3"}},
+			expected:         true,
+		},
+	}
+
+	runTests(tests, t)
+
+}
+
 func runTests(tests []testcase, t *testing.T) {
 	for _, tt := range tests {
 		replicaPlacement, _ := super_block.NewReplicaPlacementFromString(tt.replication)

weed/shell/shell_liner.go

@@ -3,9 +3,12 @@ package shell
 import (
 	"context"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 	"github.com/chrislusf/seaweedfs/weed/util/grace"
 	"io"
+	"math/rand"
 	"os"
 	"path"
 	"regexp"
@@ -47,6 +50,29 @@ func RunShell(options ShellOptions) {
 	go commandEnv.MasterClient.KeepConnectedToMaster()
 	commandEnv.MasterClient.WaitUntilConnected()
 
+	if commandEnv.option.FilerAddress == "" {
+		var filers []pb.ServerAddress
+		commandEnv.MasterClient.WithClient(func(client master_pb.SeaweedClient) error {
+			resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
+				ClientType: "filer",
+			})
+			if err != nil {
+				return err
+			}
+
+			for _, clusterNode := range resp.ClusterNodes {
+				filers = append(filers, pb.ServerAddress(clusterNode.Address))
+			}
+			return nil
+		})
+		fmt.Printf("master: %s ", *options.Masters)
+		if len(filers) > 0 {
+			fmt.Printf("filers: %v", filers)
+			commandEnv.option.FilerAddress = filers[rand.Intn(len(filers))]
+		}
+		fmt.Println()
+	}
+
 	if commandEnv.option.FilerAddress != "" {
 		commandEnv.WithFilerClient(func(filerClient filer_pb.SeaweedFilerClient) error {
 			resp, err := filerClient.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})

weed/storage/backend/s3_backend/s3_upload.go

@@ -49,10 +49,10 @@ func uploadToS3(sess s3iface.S3API, filename string, destBucket string, destKey
 	// Upload the file to S3.
 	var result *s3manager.UploadOutput
 	result, err = uploader.Upload(&s3manager.UploadInput{
-		Bucket:               aws.String(destBucket),
-		Key:                  aws.String(destKey),
-		Body:                 fileReader,
-		StorageClass:         aws.String("STANDARD_IA"),
+		Bucket:       aws.String(destBucket),
+		Key:          aws.String(destKey),
+		Body:         fileReader,
+		StorageClass: aws.String("STANDARD_IA"),
 	})
 
 	//in case it fails to upload

weed/util/constants.go

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

weed/wdclient/masterclient.go

@@ -21,6 +21,8 @@ type MasterClient struct {
 	grpcDialOption grpc.DialOption
 
 	vidMap
+
+	OnPeerUpdate func(update *master_pb.ClusterNodeUpdate)
 }
 
 func NewMasterClient(grpcDialOption grpc.DialOption, clientType string, clientHost pb.ServerAddress, clientDataCenter string, masters []pb.ServerAddress) *MasterClient {
@@ -93,7 +95,7 @@ func (mc *MasterClient) tryAllMasters() {
 }
 
 func (mc *MasterClient) tryConnectToMaster(master pb.ServerAddress) (nextHintedLeader pb.ServerAddress) {
-	glog.V(0).Infof("%s masterClient Connecting to master %v", mc.clientType, master)
+	glog.V(1).Infof("%s masterClient Connecting to master %v", mc.clientType, master)
 	gprcErr := pb.WithMasterClient(master, mc.grpcDialOption, func(client master_pb.SeaweedClient) error {
 
 		ctx, cancel := context.WithCancel(context.Background())
@@ -105,7 +107,11 @@ func (mc *MasterClient) tryConnectToMaster(master pb.ServerAddress) (nextHintedL
 			return err
 		}
 
-		if err = stream.Send(&master_pb.KeepConnectedRequest{Name: mc.clientType, ClientAddress: string(mc.clientHost)}); err != nil {
+		if err = stream.Send(&master_pb.KeepConnectedRequest{
+			ClientType:    mc.clientType,
+			ClientAddress: string(mc.clientHost),
+			Version:       util.Version(),
+		}); err != nil {
 			glog.V(0).Infof("%s masterClient failed to send to %s: %v", mc.clientType, master, err)
 			return err
 		}
@@ -114,34 +120,49 @@ func (mc *MasterClient) tryConnectToMaster(master pb.ServerAddress) (nextHintedL
 		mc.currentMaster = master
 
 		for {
-			volumeLocation, err := stream.Recv()
+			resp, err := stream.Recv()
 			if err != nil {
 				glog.V(0).Infof("%s masterClient failed to receive from %s: %v", mc.clientType, master, err)
 				return err
 			}
 
-			// maybe the leader is changed
-			if volumeLocation.Leader != "" {
-				glog.V(0).Infof("redirected to leader %v", volumeLocation.Leader)
-				nextHintedLeader = pb.ServerAddress(volumeLocation.Leader)
-				return nil
+			if resp.VolumeLocation != nil {
+				// maybe the leader is changed
+				if resp.VolumeLocation.Leader != "" {
+					glog.V(0).Infof("redirected to leader %v", resp.VolumeLocation.Leader)
+					nextHintedLeader = pb.ServerAddress(resp.VolumeLocation.Leader)
+					return nil
+				}
+
+				// process new volume location
+				loc := Location{
+					Url:        resp.VolumeLocation.Url,
+					PublicUrl:  resp.VolumeLocation.PublicUrl,
+					DataCenter: resp.VolumeLocation.DataCenter,
+					GrpcPort:   int(resp.VolumeLocation.GrpcPort),
+				}
+				for _, newVid := range resp.VolumeLocation.NewVids {
+					glog.V(1).Infof("%s: %s masterClient adds volume %d", mc.clientType, loc.Url, newVid)
+					mc.addLocation(newVid, loc)
+				}
+				for _, deletedVid := range resp.VolumeLocation.DeletedVids {
+					glog.V(1).Infof("%s: %s masterClient removes volume %d", mc.clientType, loc.Url, deletedVid)
+					mc.deleteLocation(deletedVid, loc)
+				}
 			}
 
-			// process new volume location
-			loc := Location{
-				Url:        volumeLocation.Url,
-				PublicUrl:  volumeLocation.PublicUrl,
-				DataCenter: volumeLocation.DataCenter,
-				GrpcPort:   int(volumeLocation.GrpcPort),
-			}
-			for _, newVid := range volumeLocation.NewVids {
-				glog.V(1).Infof("%s: %s masterClient adds volume %d", mc.clientType, loc.Url, newVid)
-				mc.addLocation(newVid, loc)
-			}
-			for _, deletedVid := range volumeLocation.DeletedVids {
-				glog.V(1).Infof("%s: %s masterClient removes volume %d", mc.clientType, loc.Url, deletedVid)
-				mc.deleteLocation(deletedVid, loc)
+			if resp.ClusterNodeUpdate != nil {
+				update := resp.ClusterNodeUpdate
+				if mc.OnPeerUpdate != nil {
+					if update.IsAdd {
+						glog.V(0).Infof("+ %s %s leader:%v\n", update.NodeType, update.Address, update.IsLeader)
+					} else {
+						glog.V(0).Infof("- %s %s leader:%v\n", update.NodeType, update.Address, update.IsLeader)
+					}
+					mc.OnPeerUpdate(update)
+				}
 			}
+
 		}
 
 	})