Merge pull request #15 from chrislusf/master

sync

12 files changed, 423 insertions, 57 deletions

diff --git a/weed/command/command.go b/weed/command/command.go
index 9a41a8a7c..0df22b575 100644
--- a/weed/command/command.go
+++ b/weed/command/command.go
@@ -16,6 +16,7 @@ var Commands = []*Command{
 	cmdExport,
 	cmdFiler,
 	cmdFilerReplicate,
+	cmdFilerSynchronize,
 	cmdFix,
 	cmdMaster,
 	cmdMount,
diff --git a/weed/command/download.go b/weed/command/download.go
index 7d4dad2d4..f7588fbf0 100644
--- a/weed/command/download.go
+++ b/weed/command/download.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"net/http"
 	"os"
 	"path"
 	"strings"
@@ -59,7 +60,7 @@ func downloadToFile(server, fileId, saveDir string) error {
 	if err != nil {
 		return err
 	}
-	defer rc.Close()
+	defer util.CloseResponse(rc)
 	if filename == "" {
 		filename = fileId
 	}
@@ -71,12 +72,11 @@ func downloadToFile(server, fileId, saveDir string) error {
 	}
 	f, err := os.OpenFile(path.Join(saveDir, filename), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, os.ModePerm)
 	if err != nil {
-		io.Copy(ioutil.Discard, rc)
 		return err
 	}
 	defer f.Close()
 	if isFileList {
-		content, err := ioutil.ReadAll(rc)
+		content, err := ioutil.ReadAll(rc.Body)
 		if err != nil {
 			return err
 		}
@@ -95,7 +95,7 @@ func downloadToFile(server, fileId, saveDir string) error {
 			}
 		}
 	} else {
-		if _, err = io.Copy(f, rc); err != nil {
+		if _, err = io.Copy(f, rc.Body); err != nil {
 			return err
 		}
 
@@ -108,12 +108,12 @@ func fetchContent(server string, fileId string) (filename string, content []byte
 	if lookupError != nil {
 		return "", nil, lookupError
 	}
-	var rc io.ReadCloser
+	var rc *http.Response
 	if filename, _, rc, e = util.DownloadFile(fileUrl); e != nil {
 		return "", nil, e
 	}
-	content, e = ioutil.ReadAll(rc)
-	rc.Close()
+	defer util.CloseResponse(rc)
+	content, e = ioutil.ReadAll(rc.Body)
 	return
 }
 
diff --git a/weed/command/export.go b/weed/command/export.go
index 3ea4b00d3..78d75ef52 100644
--- a/weed/command/export.go
+++ b/weed/command/export.go
@@ -23,7 +23,7 @@ import (
 )
 
 const (
-	defaultFnFormat = `{{.Mime}}/{{.Id}}:{{.Name}}`
+	defaultFnFormat = `{{.Id}}_{{.Name}}{{.Ext}}`
 	timeFormat      = "2006-01-02T15:04:05"
 )
 
@@ -56,7 +56,7 @@ func init() {
 
 var (
 	output      = cmdExport.Flag.String("o", "", "output tar file name, must ends with .tar, or just a \"-\" for stdout")
-	format      = cmdExport.Flag.String("fileNameFormat", defaultFnFormat, "filename formatted with {{.Mime}} {{.Id}} {{.Name}} {{.Ext}}")
+	format      = cmdExport.Flag.String("fileNameFormat", defaultFnFormat, "filename formatted with {{.Id}} {{.Name}} {{.Ext}}")
 	newer       = cmdExport.Flag.String("newer", "", "export only files newer than this time, default is all files. Must be specified in RFC3339 without timezone, e.g. 2006-01-02T15:04:05")
 	showDeleted = cmdExport.Flag.Bool("deleted", false, "export deleted files. only applies if -o is not specified")
 	limit       = cmdExport.Flag.Int("limit", 0, "only show first n entries if specified")
@@ -70,13 +70,13 @@ var (
 	localLocation, _             = time.LoadLocation("Local")
 )
 
-func printNeedle(vid needle.VolumeId, n *needle.Needle, version needle.Version, deleted bool) {
+func printNeedle(vid needle.VolumeId, n *needle.Needle, version needle.Version, deleted bool, offset int64, onDiskSize int64) {
 	key := needle.NewFileIdFromNeedle(vid, n).String()
 	size := int32(n.DataSize)
 	if version == needle.Version1 {
 		size = int32(n.Size)
 	}
-	fmt.Printf("%s\t%s\t%d\t%t\t%s\t%s\t%s\t%t\n",
+	fmt.Printf("%s\t%s\t%d\t%t\t%s\t%s\t%s\t%t\t%d\t%d\n",
 		key,
 		n.Name,
 		size,
@@ -85,6 +85,8 @@ func printNeedle(vid needle.VolumeId, n *needle.Needle, version needle.Version,
 		n.LastModifiedString(),
 		n.Ttl.String(),
 		deleted,
+		offset,
+		offset+onDiskSize,
 	)
 }
 
@@ -111,7 +113,7 @@ func (scanner *VolumeFileScanner4Export) VisitNeedle(n *needle.Needle, offset in
 	nv, ok := needleMap.Get(n.Id)
 	glog.V(3).Infof("key %d offset %d size %d disk_size %d compressed %v ok %v nv %+v",
 		n.Id, offset, n.Size, n.DiskSize(scanner.version), n.IsCompressed(), ok, nv)
-	if ok && nv.Size.IsValid() && nv.Offset.ToAcutalOffset() == offset {
+	if *showDeleted && n.Size > 0 || ok && nv.Size.IsValid() && nv.Offset.ToAcutalOffset() == offset {
 		if newerThanUnix >= 0 && n.HasLastModifiedDate() && n.LastModified < uint64(newerThanUnix) {
 			glog.V(3).Infof("Skipping this file, as it's old enough: LastModified %d vs %d",
 				n.LastModified, newerThanUnix)
@@ -124,17 +126,17 @@ func (scanner *VolumeFileScanner4Export) VisitNeedle(n *needle.Needle, offset in
 		if tarOutputFile != nil {
 			return writeFile(vid, n)
 		} else {
-			printNeedle(vid, n, scanner.version, false)
+			printNeedle(vid, n, scanner.version, false, offset, n.DiskSize(scanner.version))
 			return nil
 		}
 	}
 	if !ok {
 		if *showDeleted && tarOutputFile == nil {
 			if n.DataSize > 0 {
-				printNeedle(vid, n, scanner.version, true)
+				printNeedle(vid, n, scanner.version, true, offset, n.DiskSize(scanner.version))
 			} else {
 				n.Name = []byte("*tombstone")
-				printNeedle(vid, n, scanner.version, true)
+				printNeedle(vid, n, scanner.version, true, offset, n.DiskSize(scanner.version))
 			}
 		}
 		glog.V(2).Infof("This seems deleted %d size %d", n.Id, n.Size)
@@ -208,7 +210,7 @@ func runExport(cmd *Command, args []string) bool {
 	}
 
 	if tarOutputFile == nil {
-		fmt.Printf("key\tname\tsize\tgzip\tmime\tmodified\tttl\tdeleted\n")
+		fmt.Printf("key\tname\tsize\tgzip\tmime\tmodified\tttl\tdeleted\tstart\tstop\n")
 	}
 
 	err = storage.ScanVolumeFile(util.ResolvePath(*export.dir), *export.collection, vid, storage.NeedleMapInMemory, volumeFileScanner)
diff --git a/weed/command/filer.go b/weed/command/filer.go
index c36c43e93..a6670b063 100644
--- a/weed/command/filer.go
+++ b/weed/command/filer.go
@@ -152,7 +152,7 @@ func (fo *FilerOptions) startFiler() {
 
 	// starting grpc server
 	grpcPort := *fo.port + 10000
-	grpcL, err := util.NewListener(":"+strconv.Itoa(grpcPort), 0)
+	grpcL, err := util.NewListener(*fo.bindIp+":"+strconv.Itoa(grpcPort), 0)
 	if err != nil {
 		glog.Fatalf("failed to listen on grpc port %d: %v", grpcPort, err)
 	}
diff --git a/weed/command/filer_copy.go b/weed/command/filer_copy.go
index 2d6ba94d6..88148acc5 100644
--- a/weed/command/filer_copy.go
+++ b/weed/command/filer_copy.go
@@ -72,7 +72,7 @@ var cmdCopy = &Command{
 
   If "maxMB" is set to a positive number, files larger than it would be split into chunks.
 
-  `,
+`,
 }
 
 func runCopy(cmd *Command, args []string) bool {
diff --git a/weed/command/filer_sync.go b/weed/command/filer_sync.go
new file mode 100644
index 000000000..af0a624b1
--- /dev/null
+++ b/weed/command/filer_sync.go
@@ -0,0 +1,337 @@
+package command
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/replication"
+	"github.com/chrislusf/seaweedfs/weed/replication/sink/filersink"
+	"github.com/chrislusf/seaweedfs/weed/replication/source"
+	"github.com/chrislusf/seaweedfs/weed/security"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/chrislusf/seaweedfs/weed/util/grace"
+	"google.golang.org/grpc"
+	"io"
+	"strings"
+	"time"
+)
+
+type SyncOptions struct {
+	isActivePassive *bool
+	filerA          *string
+	filerB          *string
+	aPath           *string
+	bPath           *string
+	aReplication    *string
+	bReplication    *string
+	aCollection     *string
+	bCollection     *string
+	aTtlSec         *int
+	bTtlSec         *int
+	aDebug          *bool
+	bDebug          *bool
+}
+
+var (
+	syncOptions    SyncOptions
+	syncCpuProfile *string
+	syncMemProfile *string
+)
+
+func init() {
+	cmdFilerSynchronize.Run = runFilerSynchronize // break init cycle
+	syncOptions.isActivePassive = cmdFilerSynchronize.Flag.Bool("isActivePassive", false, "one directional follow if true")
+	syncOptions.filerA = cmdFilerSynchronize.Flag.String("a", "", "filer A in one SeaweedFS cluster")
+	syncOptions.filerB = cmdFilerSynchronize.Flag.String("b", "", "filer B in the other SeaweedFS cluster")
+	syncOptions.aPath = cmdFilerSynchronize.Flag.String("a.path", "/", "directory to sync on filer A")
+	syncOptions.bPath = cmdFilerSynchronize.Flag.String("b.path", "/", "directory to sync on filer B")
+	syncOptions.aReplication = cmdFilerSynchronize.Flag.String("a.replication", "", "replication on filer A")
+	syncOptions.bReplication = cmdFilerSynchronize.Flag.String("b.replication", "", "replication on filer B")
+	syncOptions.aCollection = cmdFilerSynchronize.Flag.String("a.collection", "", "collection on filer A")
+	syncOptions.bCollection = cmdFilerSynchronize.Flag.String("b.collection", "", "collection on filer B")
+	syncOptions.aTtlSec = cmdFilerSynchronize.Flag.Int("a.ttlSec", 0, "ttl in seconds on filer A")
+	syncOptions.bTtlSec = cmdFilerSynchronize.Flag.Int("b.ttlSec", 0, "ttl in seconds on filer B")
+	syncOptions.aDebug = cmdFilerSynchronize.Flag.Bool("a.debug", false, "debug mode to print out filer A received files")
+	syncOptions.bDebug = cmdFilerSynchronize.Flag.Bool("b.debug", false, "debug mode to print out filer B received files")
+	syncCpuProfile = cmdFilerSynchronize.Flag.String("cpuprofile", "", "cpu profile output file")
+	syncMemProfile = cmdFilerSynchronize.Flag.String("memprofile", "", "memory profile output file")
+}
+
+var cmdFilerSynchronize = &Command{
+	UsageLine: "filer.sync -a=<oneFilerHost>:<oneFilerPort> -b=<otherFilerHost>:<otherFilerPort>",
+	Short:     "continuously synchronize between two active-active or active-passive SeaweedFS clusters",
+	Long: `continuously synchronize file changes between two active-active or active-passive filers
+
+	filer.sync listens on filer notifications. If any file is updated, it will fetch the updated content,
+	and write to the other destination. Different from filer.replicate:
+
+	* filer.sync only works between two filers.
+	* filer.sync does not need any special message queue setup.
+	* filer.sync supports both active-active and active-passive modes.
+	
+	If restarted, the synchronization will resume from the previous checkpoints, persisted every minute.
+	A fresh sync will start from the earliest metadata logs.
+
+`,
+}
+
+func runFilerSynchronize(cmd *Command, args []string) bool {
+
+	grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
+
+	grace.SetupProfiling(*syncCpuProfile, *syncMemProfile)
+
+	go func() {
+		for {
+			err := doSubscribeFilerMetaChanges(grpcDialOption, *syncOptions.filerA, *syncOptions.aPath, *syncOptions.filerB,
+				*syncOptions.bPath, *syncOptions.bReplication, *syncOptions.bCollection, *syncOptions.bTtlSec, *syncOptions.bDebug)
+			if err != nil {
+				glog.Errorf("sync from %s to %s: %v", *syncOptions.filerA, *syncOptions.filerB, err)
+				time.Sleep(1747 * time.Millisecond)
+			}
+		}
+	}()
+
+	if !*syncOptions.isActivePassive {
+		go func() {
+			for {
+				err := doSubscribeFilerMetaChanges(grpcDialOption, *syncOptions.filerB, *syncOptions.bPath, *syncOptions.filerA,
+					*syncOptions.aPath, *syncOptions.aReplication, *syncOptions.aCollection, *syncOptions.aTtlSec, *syncOptions.aDebug)
+				if err != nil {
+					glog.Errorf("sync from %s to %s: %v", *syncOptions.filerB, *syncOptions.filerA, err)
+					time.Sleep(2147 * time.Millisecond)
+				}
+			}
+		}()
+	}
+
+	select {}
+
+	return true
+}
+
+func doSubscribeFilerMetaChanges(grpcDialOption grpc.DialOption, sourceFiler, sourcePath, targetFiler, targetPath string,
+	replicationStr, collection string, ttlSec int, debug bool) error {
+
+	// read source filer signature
+	sourceFilerSignature, sourceErr := replication.ReadFilerSignature(grpcDialOption, sourceFiler)
+	if sourceErr != nil {
+		return sourceErr
+	}
+	// read target filer signature
+	targetFilerSignature, targetErr := replication.ReadFilerSignature(grpcDialOption, targetFiler)
+	if targetErr != nil {
+		return targetErr
+	}
+
+	// if first time, start from now
+	// if has previously synced, resume from that point of time
+	sourceFilerOffsetTsNs, err := readSyncOffset(grpcDialOption, targetFiler, sourceFilerSignature)
+	if err != nil {
+		return err
+	}
+
+	glog.V(0).Infof("start sync %s(%d) => %s(%d) from %v(%d)", sourceFiler, sourceFilerSignature, targetFiler, targetFilerSignature, time.Unix(0, sourceFilerOffsetTsNs), sourceFilerOffsetTsNs)
+
+	// create filer sink
+	filerSource := &source.FilerSource{}
+	filerSource.DoInitialize(pb.ServerToGrpcAddress(sourceFiler), sourcePath)
+	filerSink := &filersink.FilerSink{}
+	filerSink.DoInitialize(pb.ServerToGrpcAddress(targetFiler), targetPath, replicationStr, collection, ttlSec, grpcDialOption)
+	filerSink.SetSourceFiler(filerSource)
+
+	processEventFn := func(resp *filer_pb.SubscribeMetadataResponse) error {
+		message := resp.EventNotification
+
+		var sourceOldKey, sourceNewKey util.FullPath
+		if message.OldEntry != nil {
+			sourceOldKey = util.FullPath(resp.Directory).Child(message.OldEntry.Name)
+		}
+		if message.NewEntry != nil {
+			sourceNewKey = util.FullPath(message.NewParentPath).Child(message.NewEntry.Name)
+		}
+
+		for _, sig := range message.Signatures {
+			if sig == targetFilerSignature && targetFilerSignature != 0 {
+				fmt.Printf("%s skipping %s change to %v\n", targetFiler, sourceFiler, message)
+				return nil
+			}
+		}
+		if debug {
+			fmt.Printf("%s check %s change %s,%s sig %v, target sig: %v\n", targetFiler, sourceFiler, sourceOldKey, sourceNewKey, message.Signatures, targetFilerSignature)
+		}
+
+		if !strings.HasPrefix(resp.Directory, sourcePath) {
+			return nil
+		}
+
+		// handle deletions
+		if message.OldEntry != nil && message.NewEntry == nil {
+			if !strings.HasPrefix(string(sourceOldKey), sourcePath) {
+				return nil
+			}
+			key := util.Join(targetPath, string(sourceOldKey)[len(sourcePath):])
+			return filerSink.DeleteEntry(key, message.OldEntry.IsDirectory, message.DeleteChunks, message.Signatures)
+		}
+
+		// handle new entries
+		if message.OldEntry == nil && message.NewEntry != nil {
+			if !strings.HasPrefix(string(sourceNewKey), sourcePath) {
+				return nil
+			}
+			key := util.Join(targetPath, string(sourceNewKey)[len(sourcePath):])
+			return filerSink.CreateEntry(key, message.NewEntry, message.Signatures)
+		}
+
+		// this is something special?
+		if message.OldEntry == nil && message.NewEntry == nil {
+			return nil
+		}
+
+		// handle updates
+		if strings.HasPrefix(string(sourceOldKey), sourcePath) {
+			// old key is in the watched directory
+			if strings.HasPrefix(string(sourceNewKey), sourcePath) {
+				// new key is also in the watched directory
+				oldKey := util.Join(targetPath, string(sourceOldKey)[len(sourcePath):])
+				message.NewParentPath = util.Join(targetPath, message.NewParentPath[len(sourcePath):])
+				foundExisting, err := filerSink.UpdateEntry(string(oldKey), message.OldEntry, message.NewParentPath, message.NewEntry, message.DeleteChunks, message.Signatures)
+				if foundExisting {
+					return err
+				}
+
+				// not able to find old entry
+				if err = filerSink.DeleteEntry(string(oldKey), message.OldEntry.IsDirectory, false, message.Signatures); err != nil {
+					return fmt.Errorf("delete old entry %v: %v", oldKey, err)
+				}
+
+				// create the new entry
+				newKey := util.Join(targetPath, string(sourceNewKey)[len(sourcePath):])
+				return filerSink.CreateEntry(newKey, message.NewEntry, message.Signatures)
+
+			} else {
+				// new key is outside of the watched directory
+				key := util.Join(targetPath, string(sourceOldKey)[len(sourcePath):])
+				return filerSink.DeleteEntry(key, message.OldEntry.IsDirectory, message.DeleteChunks, message.Signatures)
+			}
+		} else {
+			// old key is outside of the watched directory
+			if strings.HasPrefix(string(sourceNewKey), sourcePath) {
+				// new key is in the watched directory
+				key := util.Join(targetPath, string(sourceNewKey)[len(sourcePath):])
+				return filerSink.CreateEntry(key, message.NewEntry, message.Signatures)
+			} else {
+				// new key is also outside of the watched directory
+				// skip
+			}
+		}
+
+		return nil
+	}
+
+	return pb.WithFilerClient(sourceFiler, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		stream, err := client.SubscribeMetadata(ctx, &filer_pb.SubscribeMetadataRequest{
+			ClientName: "syncTo_" + targetFiler,
+			PathPrefix: sourcePath,
+			SinceNs:    sourceFilerOffsetTsNs,
+			Signature:  targetFilerSignature,
+		})
+		if err != nil {
+			return fmt.Errorf("listen: %v", err)
+		}
+
+		var counter int64
+		var lastWriteTime time.Time
+		for {
+			resp, listenErr := stream.Recv()
+			if listenErr == io.EOF {
+				return nil
+			}
+			if listenErr != nil {
+				return listenErr
+			}
+
+			if err := processEventFn(resp); err != nil {
+				return err
+			}
+
+			counter++
+			if lastWriteTime.Add(3 * time.Second).Before(time.Now()) {
+				glog.V(0).Infof("sync %s => %s progressed to %v %0.2f/sec", sourceFiler, targetFiler, time.Unix(0, resp.TsNs), float64(counter)/float64(3))
+				counter = 0
+				lastWriteTime = time.Now()
+				if err := writeSyncOffset(grpcDialOption, targetFiler, sourceFilerSignature, resp.TsNs); err != nil {
+					return err
+				}
+			}
+
+		}
+
+	})
+
+}
+
+const (
+	SyncKeyPrefix = "sync."
+)
+
+func readSyncOffset(grpcDialOption grpc.DialOption, filer string, filerSignature int32) (lastOffsetTsNs int64, readErr error) {
+
+	readErr = pb.WithFilerClient(filer, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+		syncKey := []byte(SyncKeyPrefix + "____")
+		util.Uint32toBytes(syncKey[len(SyncKeyPrefix):len(SyncKeyPrefix)+4], uint32(filerSignature))
+
+		resp, err := client.KvGet(context.Background(), &filer_pb.KvGetRequest{Key: syncKey})
+		if err != nil {
+			return err
+		}
+
+		if len(resp.Error) != 0 {
+			return errors.New(resp.Error)
+		}
+		if len(resp.Value) < 8 {
+			return nil
+		}
+
+		lastOffsetTsNs = int64(util.BytesToUint64(resp.Value))
+
+		return nil
+	})
+
+	return
+
+}
+
+func writeSyncOffset(grpcDialOption grpc.DialOption, filer string, filerSignature int32, offsetTsNs int64) error {
+	return pb.WithFilerClient(filer, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+
+		syncKey := []byte(SyncKeyPrefix + "____")
+		util.Uint32toBytes(syncKey[len(SyncKeyPrefix):len(SyncKeyPrefix)+4], uint32(filerSignature))
+
+		valueBuf := make([]byte, 8)
+		util.Uint64toBytes(valueBuf, uint64(offsetTsNs))
+
+		resp, err := client.KvPut(context.Background(), &filer_pb.KvPutRequest{
+			Key:   syncKey,
+			Value: valueBuf,
+		})
+		if err != nil {
+			return err
+		}
+
+		if len(resp.Error) != 0 {
+			return errors.New(resp.Error)
+		}
+
+		return nil
+
+	})
+
+}
diff --git a/weed/command/mount.go b/weed/command/mount.go
index a0e573423..7bf59cdc7 100644
--- a/weed/command/mount.go
+++ b/weed/command/mount.go
@@ -20,6 +20,8 @@ type MountOptions struct {
 	umaskString                 *string
 	nonempty                    *bool
 	outsideContainerClusterMode *bool
+	uidMap                      *string
+	gidMap                      *string
 }
 
 var (
@@ -47,6 +49,8 @@ func init() {
 	mountCpuProfile = cmdMount.Flag.String("cpuprofile", "", "cpu profile output file")
 	mountMemProfile = cmdMount.Flag.String("memprofile", "", "memory profile output file")
 	mountOptions.outsideContainerClusterMode = cmdMount.Flag.Bool("outsideContainerClusterMode", false, "allows other users to access the file system")
+	mountOptions.uidMap = cmdMount.Flag.String("map.uid", "", "map local uid to uid on filer, comma-separated <local_uid>:<filer_uid>")
+	mountOptions.gidMap = cmdMount.Flag.String("map.gid", "", "map local gid to gid on filer, comma-separated <local_gid>:<filer_gid>")
 }
 
 var cmdMount = &Command{
diff --git a/weed/command/mount_std.go b/weed/command/mount_std.go
index 3975575e9..44e945f23 100644
--- a/weed/command/mount_std.go
+++ b/weed/command/mount_std.go
@@ -5,6 +5,7 @@ package command
 import (
 	"context"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/filesys/meta_cache"
 	"os"
 	"os/user"
 	"path"
@@ -115,6 +116,13 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 		}
 	}
 
+	// mapping uid, gid
+	uidGidMapper, err := meta_cache.NewUidGidMapper(*option.uidMap, *option.gidMap)
+	if err != nil {
+		fmt.Printf("failed to parse %s %s: %v\n", *option.uidMap, *option.gidMap, err)
+		return false
+	}
+
 	// Ensure target mount point availability
 	if isValid := checkMountPointAvailable(dir); !isValid {
 		glog.Fatalf("Expected mount to still be active, target mount point: %s, please check!", dir)
@@ -174,6 +182,7 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 		Umask:                       umask,
 		OutsideContainerClusterMode: *mountOptions.outsideContainerClusterMode,
 		Cipher:                      cipher,
+		UidGidMapper:                uidGidMapper,
 	})
 
 	// mount
diff --git a/weed/command/scaffold.go b/weed/command/scaffold.go
index b199f2d2d..dd12f12a2 100644
--- a/weed/command/scaffold.go
+++ b/weed/command/scaffold.go
@@ -173,6 +173,20 @@ enabled = false
 uri = "mongodb://localhost:27017"
 option_pool_size = 0
 database = "seaweedfs"
+
+[elastic7]
+enabled = false
+servers = [
+    "http://localhost1:9200",
+    "http://localhost2:9200",
+    "http://localhost3:9200",
+]
+username = ""
+password = ""
+sniff_enabled = false
+healthcheck_enabled = false
+# increase the value is recommend, be sure the value in Elastic is greater or equal here
+index.max_result_window = 10000
 `
 
 	NOTIFICATION_TOML_EXAMPLE = `
@@ -377,7 +391,7 @@ default = "localhost:8888"    # used by maintenance scripts if the scripts needs
 
 
 [master.sequencer]
-type = "memory"     # Choose [memory|etcd] type for storing the file id sequence
+type = "raft"     # Choose [raft|etcd] type for storing the file id sequence
 # when sequencer.type = etcd, set listen client urls of etcd cluster that store file id sequence
 # example : http://127.0.0.1:2379,http://127.0.0.1:2389
 sequencer_etcd_urls = "http://127.0.0.1:2379"
diff --git a/weed/command/server.go b/weed/command/server.go
index 4aeecbff0..6f40263bb 100644
--- a/weed/command/server.go
+++ b/weed/command/server.go
@@ -98,7 +98,7 @@ func init() {
 	serverOptions.v.compactionMBPerSecond = cmdServer.Flag.Int("volume.compactionMBps", 0, "limit compaction speed in mega bytes per second")
 	serverOptions.v.fileSizeLimitMB = cmdServer.Flag.Int("volume.fileSizeLimitMB", 1024, "limit file size to avoid out of memory")
 	serverOptions.v.publicUrl = cmdServer.Flag.String("volume.publicUrl", "", "publicly accessible address")
-	serverOptions.v.preStopSeconds = cmdServer.Flag.Int("volume.preStopSeconds", 30, "number of seconds between stop send heartbeats and stop volume server")
+	serverOptions.v.preStopSeconds = cmdServer.Flag.Int("volume.preStopSeconds", 10, "number of seconds between stop send heartbeats and stop volume server")
 	serverOptions.v.pprof = &False
 
 	s3Options.port = cmdServer.Flag.Int("s3.port", 8333, "s3 server http listen port")
diff --git a/weed/command/volume.go b/weed/command/volume.go
index c8f24802c..92f83f945 100644
--- a/weed/command/volume.go
+++ b/weed/command/volume.go
@@ -67,7 +67,7 @@ func init() {
 	v.publicUrl = cmdVolume.Flag.String("publicUrl", "", "Publicly accessible address")
 	v.bindIp = cmdVolume.Flag.String("ip.bind", "0.0.0.0", "ip address to bind to")
 	v.masters = cmdVolume.Flag.String("mserver", "localhost:9333", "comma-separated master servers")
-	v.preStopSeconds = cmdVolume.Flag.Int("preStopSeconds", 30, "number of seconds between stop send heartbeats and stop volume server")
+	v.preStopSeconds = cmdVolume.Flag.Int("preStopSeconds", 10, "number of seconds between stop send heartbeats and stop volume server")
 	// v.pulseSeconds = cmdVolume.Flag.Int("pulseSeconds", 5, "number of seconds between heartbeats, must be smaller than or equal to the master's setting")
 	v.idleConnectionTimeout = cmdVolume.Flag.Int("idleTimeout", 30, "connection idle seconds")
 	v.dataCenter = cmdVolume.Flag.String("dataCenter", "", "current volume server's data center name")
@@ -223,52 +223,48 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
 	// starting the cluster http server
 	clusterHttpServer := v.startClusterHttpService(volumeMux)
 
-	stopChain := make(chan struct{})
+	stopChan := make(chan bool)
 	grace.OnInterrupt(func() {
 		fmt.Println("volume server has be killed")
-		var startTime time.Time
 
 		// Stop heartbeats
-		glog.V(0).Infof("stop send heartbeat and wait %d seconds until shutdown ...", *v.preStopSeconds)
-		volumeServer.SendHeartbeat = false
-		time.Sleep(time.Duration(*v.preStopSeconds) * time.Second)
-		glog.V(0).Infof("end sleep %d sec", *v.preStopSeconds)
-		// firstly, stop the public http service to prevent from receiving new user request
-		if nil != publicHttpDown {
-			startTime = time.Now()
-			if err := publicHttpDown.Stop(); err != nil {
-				glog.Warningf("stop the public http server failed, %v", err)
-			}
-			delta := time.Now().Sub(startTime).Nanoseconds() / 1e6
-			glog.V(0).Infof("stop public http server, elapsed %dms", delta)
+		if !volumeServer.StopHeartbeat() {
+			glog.V(0).Infof("stop send heartbeat and wait %d seconds until shutdown ...", *v.preStopSeconds)
+			time.Sleep(time.Duration(*v.preStopSeconds) * time.Second)
 		}
 
-		startTime = time.Now()
-		if err := clusterHttpServer.Stop(); err != nil {
-			glog.Warningf("stop the cluster http server failed, %v", err)
-		}
-		delta := time.Now().Sub(startTime).Nanoseconds() / 1e6
-		glog.V(0).Infof("graceful stop cluster http server, elapsed [%d]", delta)
+		shutdown(publicHttpDown, clusterHttpServer, grpcS, volumeServer)
+		stopChan <- true
+	})
 
-		startTime = time.Now()
-		grpcS.GracefulStop()
-		delta = time.Now().Sub(startTime).Nanoseconds() / 1e6
-		glog.V(0).Infof("graceful stop gRPC, elapsed [%d]", delta)
+	select {
+	case <-stopChan:
+	}
 
-		startTime = time.Now()
-		volumeServer.Shutdown()
-		delta = time.Now().Sub(startTime).Nanoseconds() / 1e6
-		glog.V(0).Infof("stop volume server, elapsed [%d]", delta)
+}
 
-		pprof.StopCPUProfile()
+func shutdown(publicHttpDown httpdown.Server, clusterHttpServer httpdown.Server, grpcS *grpc.Server, volumeServer *weed_server.VolumeServer) {
 
-		close(stopChain) // notify exit
-	})
+	// firstly, stop the public http service to prevent from receiving new user request
+	if nil != publicHttpDown {
+		glog.V(0).Infof("stop public http server ... ")
+		if err := publicHttpDown.Stop(); err != nil {
+			glog.Warningf("stop the public http server failed, %v", err)
+		}
+	}
 
-	select {
-	case <-stopChain:
+	glog.V(0).Infof("graceful stop cluster http server ... ")
+	if err := clusterHttpServer.Stop(); err != nil {
+		glog.Warningf("stop the cluster http server failed, %v", err)
 	}
-	glog.Warningf("the volume server exit.")
+
+	glog.V(0).Infof("graceful stop gRPC ...")
+	grpcS.GracefulStop()
+
+	volumeServer.Shutdown()
+
+	pprof.StopCPUProfile()
+
 }
 
 // check whether configure the public port
diff --git a/weed/command/watch.go b/weed/command/watch.go
index 9340db141..fd7dd6fb2 100644
--- a/weed/command/watch.go
+++ b/weed/command/watch.go
@@ -78,7 +78,10 @@ func runWatch(cmd *Command, args []string) bool {
 
 	watchErr := pb.WithFilerClient(*watchFiler, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
 
-		stream, err := client.SubscribeMetadata(context.Background(), &filer_pb.SubscribeMetadataRequest{
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		stream, err := client.SubscribeMetadata(ctx, &filer_pb.SubscribeMetadataRequest{
 			ClientName: "watch",
 			PathPrefix: *watchTarget,
 			SinceNs:    time.Now().Add(-*watchStart).UnixNano(),
@@ -98,7 +101,7 @@ func runWatch(cmd *Command, args []string) bool {
 			if !shouldPrint(resp) {
 				continue
 			}
-			fmt.Printf("%+v\n", resp.EventNotification)
+			fmt.Printf("dir:%s %+v\n", resp.Directory, resp.EventNotification)
 		}
 
 	})