2.49

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

README.md

@@ -6,7 +6,7 @@
 [![Build Status](https://travis-ci.org/chrislusf/seaweedfs.svg?branch=master)](https://travis-ci.org/chrislusf/seaweedfs)
 [![GoDoc](https://godoc.org/github.com/chrislusf/seaweedfs/weed?status.svg)](https://godoc.org/github.com/chrislusf/seaweedfs/weed)
 [![Wiki](https://img.shields.io/badge/docs-wiki-blue.svg)](https://github.com/chrislusf/seaweedfs/wiki)
-[![Docker Pulls](https://img.shields.io/docker/pulls/chrislusf/seaweedfs.svg?maxAge=4800)](https://hub.docker.com/r/chrislusf/seaweedfs/)
+[![Docker Pulls](https://img.shields.io/docker/pulls/chrislusf/seaweedfs?maxAge=4800)](https://hub.docker.com/r/chrislusf/seaweedfs/)
 [![SeaweedFS on Maven Central](https://img.shields.io/maven-central/v/com.github.chrislusf/seaweedfs-client)](https://search.maven.org/search?q=g:com.github.chrislusf)
 
 
@@ -44,7 +44,8 @@ Your support will be really appreciated by me and other supporters!
 - [SeaweedFS Mailing List](https://groups.google.com/d/forum/seaweedfs)
 - [Wiki Documentation](https://github.com/chrislusf/seaweedfs/wiki)
 - [SeaweedFS White Paper](https://github.com/chrislusf/seaweedfs/wiki/SeaweedFS_Architecture.pdf)
-- [SeaweedFS Introduction Slides](https://www.slideshare.net/chrislusf/seaweedfs-introduction)
+- [SeaweedFS Introduction Slides 2021.5](https://docs.google.com/presentation/d/1DcxKWlINc-HNCjhYeERkpGXXm6nTCES8mi2W5G0Z4Ts/edit?usp=sharing)
+- [SeaweedFS Introduction Slides 2019.3](https://www.slideshare.net/chrislusf/seaweedfs-introduction)
 
 Table of Contents
 =================

k8s/seaweedfs/Chart.yaml

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

k8s/seaweedfs/values.yaml

@@ -4,7 +4,7 @@ global:
   registry: ""
   repository: ""
   imageName: chrislusf/seaweedfs
-  # imageTag: "2.48" - started using {.Chart.appVersion}
+  # imageTag: "2.49" - started using {.Chart.appVersion}
   imagePullPolicy: IfNotPresent
   imagePullSecrets: imagepullsecret
   restartPolicy: Always

other/java/client/pom.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.6.4</version>
+    <version>1.6.6</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>

other/java/client/pom.xml.deploy

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.6.4</version>
+    <version>1.6.6</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>

other/java/client/pom_debug.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.6.4</version>
+    <version>1.6.6</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>

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

@@ -39,7 +39,7 @@ public class SeaweedOutputStream extends OutputStream {
     }
 
     public SeaweedOutputStream(FilerClient filerClient, final String fullpath, final String replication) {
-        this(filerClient, fullpath, null, 0, 8 * 1024 * 1024, "000");
+        this(filerClient, fullpath, null, 0, 8 * 1024 * 1024, replication);
     }
 
     public SeaweedOutputStream(FilerClient filerClient, final String path, FilerProto.Entry.Builder entry,

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

@@ -1,5 +1,6 @@
 package seaweedfs.client;
 
+import com.google.common.base.Strings;
 import com.google.protobuf.ByteString;
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpPost;
@@ -45,11 +46,16 @@ public class SeaweedWrite {
         FilerProto.AssignVolumeResponse response = filerClient.getBlockingStub().assignVolume(
                 FilerProto.AssignVolumeRequest.newBuilder()
                         .setCollection(filerClient.getCollection())
-                        .setReplication(replication == null ? filerClient.getReplication() : replication)
+                        .setReplication(Strings.isNullOrEmpty(replication) ? filerClient.getReplication() : replication)
                         .setDataCenter("")
                         .setTtlSec(0)
                         .setPath(path)
                         .build());
+
+        if (!Strings.isNullOrEmpty(response.getError())) {
+            throw new IOException(response.getError());
+        }
+
         String fileId = response.getFileId();
         String auth = response.getAuth();
 

other/java/examples/pom.xml

@@ -11,13 +11,13 @@
         <dependency>
             <groupId>com.github.chrislusf</groupId>
             <artifactId>seaweedfs-client</artifactId>
-            <version>1.6.4</version>
+            <version>1.6.6</version>
             <scope>compile</scope>
         </dependency>
         <dependency>
             <groupId>com.github.chrislusf</groupId>
             <artifactId>seaweedfs-hadoop2-client</artifactId>
-            <version>1.6.4</version>
+            <version>1.6.6</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.4</seaweedfs.client.version>
+    <seaweedfs.client.version>1.6.6</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.4</seaweedfs.client.version>
+        <seaweedfs.client.version>1.6.6</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.4</seaweedfs.client.version>
+    <seaweedfs.client.version>1.6.6</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.4</seaweedfs.client.version>
+        <seaweedfs.client.version>1.6.6</seaweedfs.client.version>
         <hadoop.version>3.1.1</hadoop.version>
     </properties>
 

unmaintained/load_test/load_test_leveldb/load_test_leveldb.go

@@ -1,4 +1,4 @@
-package load_test_leveldb
+package main
 
 import (
 	"crypto/md5"

unmaintained/load_test/load_test_meta_tail/load_test_meta_tail.go

@@ -0,0 +1,108 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"google.golang.org/grpc"
+	"io"
+	"strconv"
+)
+
+var (
+	dir       = flag.String("dir", "/tmp", "directory to create files")
+	n         = flag.Int("n", 100, "the number of metadata")
+	tailFiler = flag.String("filer", "localhost:8888", "the filer address")
+	isWrite = flag.Bool("write", false, "only write")
+)
+
+func main() {
+
+	flag.Parse()
+
+	if *isWrite {
+		startGenerateMetadata()
+		return
+	}
+
+	expected := 0
+	startSubscribeMetadata(func(event *filer_pb.SubscribeMetadataResponse) error {
+		if event.Directory != *dir {
+			return nil
+		}
+		name := event.EventNotification.NewEntry.Name
+		fmt.Printf("=> %s\n", name)
+		id := name[4:]
+		if x, err := strconv.Atoi(id); err == nil {
+			if x != expected {
+				return fmt.Errorf("Expected file%d Actual %s\n", expected, name)
+			}
+			expected++
+		} else {
+			return err
+		}
+		return nil
+	})
+
+}
+
+func startGenerateMetadata() {
+	pb.WithFilerClient(*tailFiler, grpc.WithInsecure(), func(client filer_pb.SeaweedFilerClient) error {
+
+		for i := 0; i < *n; i++ {
+			name := fmt.Sprintf("file%d", i)
+			if err := filer_pb.CreateEntry(client, &filer_pb.CreateEntryRequest{
+				Directory: *dir,
+				Entry: &filer_pb.Entry{
+					Name: name,
+				},
+			}); err != nil {
+				fmt.Printf("create entry %s: %v\n", name, err)
+				return err
+			}
+		}
+
+		return nil
+
+	})
+}
+
+func startSubscribeMetadata(eachEntryFunc func(event *filer_pb.SubscribeMetadataResponse) error) {
+
+	lastTsNs := int64(0)
+
+	tailErr := pb.WithFilerClient(*tailFiler, grpc.WithInsecure(), func(client filer_pb.SeaweedFilerClient) error {
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		stream, err := client.SubscribeMetadata(ctx, &filer_pb.SubscribeMetadataRequest{
+			ClientName: "tail",
+			PathPrefix: *dir,
+			SinceNs:    lastTsNs,
+		})
+		if err != nil {
+			return fmt.Errorf("listen: %v", err)
+		}
+
+		for {
+			resp, listenErr := stream.Recv()
+			if listenErr == io.EOF {
+				return nil
+			}
+			if listenErr != nil {
+				return listenErr
+			}
+			if err = eachEntryFunc(resp); err != nil {
+				return err
+			}
+			lastTsNs = resp.TsNs
+		}
+
+	})
+	if tailErr != nil {
+		fmt.Printf("tail %s: %v\n", *tailFiler, tailErr)
+	}
+}

weed/command/mount.go

@@ -37,7 +37,7 @@ var (
 
 func init() {
 	cmdMount.Run = runMount // break init cycle
-	mountOptions.filer = cmdMount.Flag.String("filer", "localhost:8888", "weed filer location")
+	mountOptions.filer = cmdMount.Flag.String("filer", "localhost:8888", "comma-separated weed filer location")
 	mountOptions.filerMountRootPath = cmdMount.Flag.String("filer.path", "/", "mount this remote path from filer server")
 	mountOptions.dir = cmdMount.Flag.String("dir", ".", "mount weed filer to this directory")
 	mountOptions.dirAutoCreate = cmdMount.Flag.Bool("dirAutoCreate", false, "auto create the directory to mount to")

weed/command/mount_std.go

@@ -51,9 +51,9 @@ func runMount(cmd *Command, args []string) bool {
 
 func RunMount(option *MountOptions, umask os.FileMode) bool {
 
-	filer := *option.filer
+	filers := strings.Split(*option.filer, ",")
 	// parse filer grpc address
-	filerGrpcAddress, err := pb.ParseServerToGrpcAddress(filer)
+	filerGrpcAddresses, err := pb.ParseServersToGrpcAddresses(filers)
 	if err != nil {
 		glog.V(0).Infof("ParseFilerGrpcAddress: %v", err)
 		return true
@@ -64,22 +64,22 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 	grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
 	var cipher bool
 	for i := 0; i < 10; i++ {
-		err = pb.WithGrpcFilerClient(filerGrpcAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+		err = pb.WithOneOfGrpcFilerClients(filerGrpcAddresses, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
 			resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
 			if err != nil {
-				return fmt.Errorf("get filer grpc address %s configuration: %v", filerGrpcAddress, err)
+				return fmt.Errorf("get filer grpc address %v configuration: %v", filerGrpcAddresses, err)
 			}
 			cipher = resp.Cipher
 			return nil
 		})
 		if err != nil {
-			glog.V(0).Infof("failed to talk to filer %s: %v", filerGrpcAddress, err)
+			glog.V(0).Infof("failed to talk to filer %v: %v", filerGrpcAddresses, err)
 			glog.V(0).Infof("wait for %d seconds ...", i+1)
 			time.Sleep(time.Duration(i+1) * time.Second)
 		}
 	}
 	if err != nil {
-		glog.Errorf("failed to talk to filer %s: %v", filerGrpcAddress, err)
+		glog.Errorf("failed to talk to filer %v: %v", filerGrpcAddresses, err)
 		return true
 	}
 
@@ -145,7 +145,7 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 
 	options := []fuse.MountOption{
 		fuse.VolumeName(mountName),
-		fuse.FSName(filer + ":" + filerMountRootPath),
+		fuse.FSName(*option.filer + ":" + filerMountRootPath),
 		fuse.Subtype("seaweedfs"),
 		// fuse.NoAppleDouble(), // include .DS_Store, otherwise can not delete non-empty folders
 		fuse.NoAppleXattr(),
@@ -181,8 +181,8 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 
 	seaweedFileSystem := filesys.NewSeaweedFileSystem(&filesys.Option{
 		MountDirectory:     dir,
-		FilerAddress:       filer,
-		FilerGrpcAddress:   filerGrpcAddress,
+		FilerAddresses:     filers,
+		FilerGrpcAddresses: filerGrpcAddresses,
 		GrpcDialOption:     grpcDialOption,
 		FilerMountRootPath: mountRoot,
 		Collection:         *option.collection,
@@ -218,7 +218,7 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 		c.Close()
 	})
 
-	glog.V(0).Infof("mounted %s%s to %s", filer, mountRoot, dir)
+	glog.V(0).Infof("mounted %s%s to %v", *option.filer, mountRoot, dir)
 	server := fs.New(c, nil)
 	seaweedFileSystem.Server = server
 	err = server.Serve(seaweedFileSystem)

weed/filer/meta_aggregator.go

@@ -71,6 +71,7 @@ func (ma *MetaAggregator) subscribeToOneFiler(f *Filer, self string, peer string
 
 	// when filer store is not shared by multiple filers
 	if peerSignature != f.Signature {
+		lastTsNs = 0
 		if prevTsNs, err := ma.readOffset(f, peer, peerSignature); err == nil {
 			lastTsNs = prevTsNs
 		}

weed/filesys/dirty_pages_temp_file.go

@@ -6,7 +6,6 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"io"
 	"os"
-	"path/filepath"
 	"sync"
 	"time"
 )
@@ -24,14 +23,6 @@ type TempFileDirtyPages struct {
 	replication      string
 }
 
-var (
-	tmpDir = filepath.Join(os.TempDir(), "sw")
-)
-
-func init() {
-	os.Mkdir(tmpDir, 0755)
-}
-
 func newTempFileDirtyPages(file *File, writeOnly bool) *TempFileDirtyPages {
 
 	tempFile := &TempFileDirtyPages{
@@ -49,7 +40,7 @@ func (pages *TempFileDirtyPages) AddPage(offset int64, data []byte) {
 	defer pages.pageAddLock.Unlock()
 
 	if pages.tf == nil {
-		tf, err := os.CreateTemp(tmpDir, "")
+		tf, err := os.CreateTemp(pages.f.wfs.option.getTempFilePageDir(), "")
 		if err != nil {
 			glog.Errorf("create temp file: %v", err)
 			pages.lastErr = err

weed/filesys/wfs.go

@@ -7,8 +7,10 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/storage/types"
 	"github.com/chrislusf/seaweedfs/weed/wdclient"
 	"math"
+	"math/rand"
 	"os"
 	"path"
+	"path/filepath"
 	"sync"
 	"time"
 
@@ -28,8 +30,9 @@ import (
 
 type Option struct {
 	MountDirectory     string
-	FilerAddress       string
-	FilerGrpcAddress   string
+	FilerAddresses     []string
+	filerIndex         int
+	FilerGrpcAddresses []string
 	GrpcDialOption     grpc.DialOption
 	FilerMountRootPath string
 	Collection         string
@@ -52,6 +55,9 @@ type Option struct {
 	VolumeServerAccess string // how to access volume servers
 	Cipher             bool   // whether encrypt data on volume server
 	UidGidMapper       *meta_cache.UidGidMapper
+
+	uniqueCacheDir         string
+	uniqueCacheTempPageDir string
 }
 
 var _ = fs.FS(&WFS{})
@@ -95,14 +101,13 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 		},
 		signature: util.RandomInt32(),
 	}
-	cacheUniqueId := util.Md5String([]byte(option.MountDirectory + option.FilerGrpcAddress + option.FilerMountRootPath + util.Version()))[0:8]
-	cacheDir := path.Join(option.CacheDir, cacheUniqueId)
+	wfs.option.filerIndex = rand.Intn(len(option.FilerAddresses))
+	wfs.option.setupUniqueCacheDirectory()
 	if option.CacheSizeMB > 0 {
-		os.MkdirAll(cacheDir, os.FileMode(0777)&^option.Umask)
-		wfs.chunkCache = chunk_cache.NewTieredChunkCache(256, cacheDir, option.CacheSizeMB, 1024*1024)
+		wfs.chunkCache = chunk_cache.NewTieredChunkCache(256, option.getUniqueCacheDir(), option.CacheSizeMB, 1024*1024)
 	}
 
-	wfs.metaCache = meta_cache.NewMetaCache(path.Join(cacheDir, "meta"), util.FullPath(option.FilerMountRootPath), option.UidGidMapper, func(filePath util.FullPath) {
+	wfs.metaCache = meta_cache.NewMetaCache(path.Join(option.getUniqueCacheDir(), "meta"), util.FullPath(option.FilerMountRootPath), option.UidGidMapper, func(filePath util.FullPath) {
 
 		fsNode := NodeWithId(filePath.AsInode())
 		if err := wfs.Server.InvalidateNodeData(fsNode); err != nil {
@@ -259,11 +264,27 @@ func (wfs *WFS) mapPbIdFromLocalToFiler(entry *filer_pb.Entry) {
 func (wfs *WFS) LookupFn() wdclient.LookupFileIdFunctionType {
 	if wfs.option.VolumeServerAccess == "filerProxy" {
 		return func(fileId string) (targetUrls []string, err error) {
-			return []string{"http://" + wfs.option.FilerAddress + "/?proxyChunkId=" + fileId}, nil
+			return []string{"http://" + wfs.getCurrentFiler() + "/?proxyChunkId=" + fileId}, nil
 		}
 	}
 	return filer.LookupFn(wfs)
+}
+func (wfs *WFS) getCurrentFiler() string {
+	return wfs.option.FilerAddresses[wfs.option.filerIndex]
+}
 
+func (option *Option) setupUniqueCacheDirectory() {
+	cacheUniqueId := util.Md5String([]byte(option.MountDirectory + option.FilerGrpcAddresses[0] + option.FilerMountRootPath + util.Version()))[0:8]
+	option.uniqueCacheDir = path.Join(option.CacheDir, cacheUniqueId)
+	option.uniqueCacheTempPageDir = filepath.Join(option.uniqueCacheDir, "sw")
+	os.MkdirAll(option.uniqueCacheTempPageDir, os.FileMode(0777)&^option.Umask)
+}
+
+func (option *Option) getTempFilePageDir() string {
+	return option.uniqueCacheTempPageDir
+}
+func (option *Option) getUniqueCacheDir() string {
+	return option.uniqueCacheDir
 }
 
 type NodeWithId uint64

weed/filesys/wfs_filer_client.go

@@ -1,6 +1,7 @@
 package filesys
 
 import (
+	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"google.golang.org/grpc"
 
@@ -10,20 +11,36 @@ import (
 
 var _ = filer_pb.FilerClient(&WFS{})
 
-func (wfs *WFS) WithFilerClient(fn func(filer_pb.SeaweedFilerClient) error) error {
+func (wfs *WFS) WithFilerClient(fn func(filer_pb.SeaweedFilerClient) error) (err error) {
 
-	err := util.Retry("filer grpc "+wfs.option.FilerGrpcAddress, func() error {
-		return pb.WithCachedGrpcClient(func(grpcConnection *grpc.ClientConn) error {
-			client := filer_pb.NewSeaweedFilerClient(grpcConnection)
-			return fn(client)
-		}, wfs.option.FilerGrpcAddress, wfs.option.GrpcDialOption)
+	return util.Retry("filer grpc", func() error {
+
+		i := wfs.option.filerIndex
+		n := len(wfs.option.FilerGrpcAddresses)
+		for x := 0; x < n; x++ {
+
+			filerGrpcAddress := wfs.option.FilerGrpcAddresses[i]
+			err = pb.WithCachedGrpcClient(func(grpcConnection *grpc.ClientConn) error {
+				client := filer_pb.NewSeaweedFilerClient(grpcConnection)
+				return fn(client)
+			}, filerGrpcAddress, wfs.option.GrpcDialOption)
+
+			if err != nil {
+				glog.V(0).Infof("WithFilerClient %d %v: %v", x, filerGrpcAddress, err)
+			} else {
+				wfs.option.filerIndex = i
+				return nil
+			}
+
+			i++
+			if i >= n {
+				i = 0
+			}
+
+		}
+		return err
 	})
 
-	if err == nil {
-		return nil
-	}
-	return err
-
 }
 
 func (wfs *WFS) AdjustedUrl(location *filer_pb.Location) string {

weed/filesys/wfs_write.go

@@ -56,7 +56,7 @@ func (wfs *WFS) saveDataAsChunk(fullPath util.FullPath, writeOnly bool) filer.Sa
 
 		fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)
 		if wfs.option.VolumeServerAccess == "filerProxy" {
-			fileUrl = fmt.Sprintf("http://%s/?proxyChunkId=%s", wfs.option.FilerAddress, fileId)
+			fileUrl = fmt.Sprintf("http://%s/?proxyChunkId=%s", wfs.getCurrentFiler(), fileId)
 		}
 		uploadResult, err, data := operation.Upload(fileUrl, filename, wfs.option.Cipher, reader, false, "", nil, auth)
 		if err != nil {

weed/pb/grpc_client_server.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/glog"
+	"math/rand"
 	"net/http"
 	"strconv"
 	"strings"
@@ -24,10 +25,15 @@ const (
 
 var (
 	// cache grpc connections
-	grpcClients     = make(map[string]*grpc.ClientConn)
+	grpcClients     = make(map[string]*versionedGrpcClient)
 	grpcClientsLock sync.Mutex
 )
 
+type versionedGrpcClient struct {
+	*grpc.ClientConn
+	version int
+}
+
 func init() {
 	http.DefaultTransport.(*http.Transport).MaxIdleConnsPerHost = 1024
 	http.DefaultTransport.(*http.Transport).MaxIdleConns = 1024
@@ -79,7 +85,7 @@ func GrpcDial(ctx context.Context, address string, opts ...grpc.DialOption) (*gr
 	return grpc.DialContext(ctx, address, options...)
 }
 
-func getOrCreateConnection(address string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
+func getOrCreateConnection(address string, opts ...grpc.DialOption) (*versionedGrpcClient, error) {
 
 	grpcClientsLock.Lock()
 	defer grpcClientsLock.Unlock()
@@ -94,23 +100,49 @@ func getOrCreateConnection(address string, opts ...grpc.DialOption) (*grpc.Clien
 		return nil, fmt.Errorf("fail to dial %s: %v", address, err)
 	}
 
-	grpcClients[address] = grpcConnection
+	vgc := &versionedGrpcClient{
+		grpcConnection,
+		rand.Int(),
+	}
+	grpcClients[address] = vgc
 
-	return grpcConnection, nil
+	return vgc, nil
 }
 
 func WithCachedGrpcClient(fn func(*grpc.ClientConn) error, address string, opts ...grpc.DialOption) error {
 
-	grpcConnection, err := getOrCreateConnection(address, opts...)
+	vgc, err := getOrCreateConnection(address, opts...)
 	if err != nil {
 		return fmt.Errorf("getOrCreateConnection %s: %v", address, err)
 	}
-	return fn(grpcConnection)
+	executionErr := fn(vgc.ClientConn)
+	if executionErr != nil && strings.Contains(executionErr.Error(), "transport") {
+		grpcClientsLock.Lock()
+		if t, ok := grpcClients[address]; ok {
+			if t.version == vgc.version {
+				vgc.Close()
+				delete(grpcClients, address)
+			}
+		}
+		grpcClientsLock.Unlock()
+	}
+
+	return executionErr
 }
 
 func ParseServerToGrpcAddress(server string) (serverGrpcAddress string, err error) {
 	return ParseServerAddress(server, 10000)
 }
+func ParseServersToGrpcAddresses(servers []string) (serverGrpcAddresses []string, err error) {
+	for _, server := range servers {
+		if serverGrpcAddress, parseErr := ParseServerToGrpcAddress(server); parseErr == nil {
+			serverGrpcAddresses = append(serverGrpcAddresses, serverGrpcAddress)
+		} else {
+			return nil, parseErr
+		}
+	}
+	return
+}
 
 func ParseServerAddress(server string, deltaPort int) (newServerAddress string, err error) {
 
@@ -202,3 +234,18 @@ func WithGrpcFilerClient(filerGrpcAddress string, grpcDialOption grpc.DialOption
 	}, filerGrpcAddress, grpcDialOption)
 
 }
+
+func WithOneOfGrpcFilerClients(filerGrpcAddresses []string, grpcDialOption grpc.DialOption, fn func(client filer_pb.SeaweedFilerClient) error) (err error) {
+
+	for _, filerGrpcAddress := range filerGrpcAddresses {
+		err = WithCachedGrpcClient(func(grpcConnection *grpc.ClientConn) error {
+			client := filer_pb.NewSeaweedFilerClient(grpcConnection)
+			return fn(client)
+		}, filerGrpcAddress, grpcDialOption)
+		if err == nil {
+			return nil
+		}
+	}
+
+	return err
+}

weed/s3api/s3api_object_handlers.go

@@ -62,6 +62,12 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 			writeErrorResponse(w, s3ErrCode, r.URL)
 			return
 		}
+	} else {
+		rAuthType := getRequestAuthType(r)
+		if authTypeAnonymous != rAuthType {
+			writeErrorResponse(w, s3err.ErrAuthNotSetup, r.URL)
+			return
+		}
 	}
 	defer dataReader.Close()
 

weed/s3api/s3err/s3api_errors.go

@@ -91,6 +91,7 @@ const (
 	ErrRequestNotReadyYet
 	ErrMissingDateHeader
 	ErrInvalidRequest
+	ErrAuthNotSetup
 	ErrNotImplemented
 
 	ErrExistingObjectIsDirectory
@@ -341,6 +342,11 @@ var errorCodeResponse = map[ErrorCode]APIError{
 		Description:    "Invalid Request",
 		HTTPStatusCode: http.StatusBadRequest,
 	},
+	ErrAuthNotSetup : {
+		Code:           "InvalidRequest",
+		Description:    "Signed request requires setting up SeaweedFS S3 authentication",
+		HTTPStatusCode: http.StatusBadRequest,
+	},
 	ErrNotImplemented: {
 		Code:           "NotImplemented",
 		Description:    "A header you provided implies functionality that is not implemented",

weed/server/filer_server.go

@@ -128,7 +128,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 	fs.filer.DirBucketsPath = v.GetString("filer.options.buckets_folder")
 	// TODO deprecated, will be be removed after 2020-12-31
 	// replaced by https://github.com/chrislusf/seaweedfs/wiki/Path-Specific-Configuration
-	fs.filer.FsyncBuckets = v.GetStringSlice("filer.options.buckets_fsync")
+	// fs.filer.FsyncBuckets = v.GetStringSlice("filer.options.buckets_fsync")
 	fs.filer.LoadConfiguration(v)
 
 	notification.LoadConfiguration(v, "notification.")

weed/shell/command_fs_meta_save.go

@@ -38,7 +38,7 @@ func (c *commandFsMetaSave) Help() string {
 	fs.meta.save                 # save from current directory
 
 	The meta data will be saved into a local <filer_host>-<port>-<time>.meta file.
-	These meta data can be later loaded by fs.meta.load command, 
+	These meta data can be later loaded by fs.meta.load command
 
 `
 }

weed/storage/volume_vacuum.go

@@ -158,10 +158,10 @@ func (v *Volume) cleanupCompact() error {
 
 	e1 := os.Remove(v.FileName(".cpd"))
 	e2 := os.Remove(v.FileName(".cpx"))
-	if e1 != nil {
+	if e1 != nil && !os.IsNotExist(e1) {
 		return e1
 	}
-	if e2 != nil {
+	if e2 != nil && !os.IsNotExist(e2) {
 		return e2
 	}
 	return nil

weed/topology/data_node.go

@@ -177,7 +177,13 @@ func (dn *DataNode) GetVolumesById(id needle.VolumeId) (vInfo storage.VolumeInfo
 
 func (dn *DataNode) GetDataCenter() *DataCenter {
 	rack := dn.Parent()
+	if rack == nil {
+		return nil
+	}
 	dcNode := rack.Parent()
+	if dcNode == nil {
+		return nil
+	}
 	dcValue := dcNode.GetValue()
 	return dcValue.(*DataCenter)
 }

weed/util/constants.go

@@ -5,7 +5,7 @@ import (
 )
 
 var (
-	VERSION = fmt.Sprintf("%s %d.%02d", sizeLimit, 2, 48)
+	VERSION = fmt.Sprintf("%s %d.%02d", sizeLimit, 2, 49)
 	COMMIT  = ""
 )
 

weed/wdclient/masterclient.go

@@ -36,6 +36,7 @@ func NewMasterClient(grpcDialOption grpc.DialOption, clientType string, clientHo
 }
 
 func (mc *MasterClient) GetMaster() string {
+	mc.WaitUntilConnected()
 	return mc.currentMaster
 }