Merge branch 'master' into messaging

unmaintained/diff_volume_servers/diff_volume_servers.go

@@ -6,6 +6,10 @@ import (
 	"errors"
 	"flag"
 	"fmt"
+	"io"
+	"math"
+	"os"
+
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/operation"
 	"github.com/chrislusf/seaweedfs/weed/pb"
@@ -16,9 +20,6 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/storage/types"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"google.golang.org/grpc"
-	"io"
-	"math"
-	"os"
 )
 
 var (
@@ -155,7 +156,7 @@ func getVolumeFiles(v uint32, addr pb.ServerAddress) (map[types.NeedleId]needleS
 
 	var maxOffset int64
 	files := map[types.NeedleId]needleState{}
-	err = idx.WalkIndexFile(idxFile, func(key types.NeedleId, offset types.Offset, size types.Size) error {
+	err = idx.WalkIndexFile(idxFile, 0, func(key types.NeedleId, offset types.Offset, size types.Size) error {
 		if offset.IsZero() || size.IsDeleted() {
 			files[key] = needleState{
 				state: stateDeleted,

unmaintained/see_idx/see_idx.go

@@ -3,11 +3,12 @@ package main
 import (
 	"flag"
 	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/util"
 	"os"
 	"path"
 	"strconv"
 
+	"github.com/chrislusf/seaweedfs/weed/util"
+
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/storage/idx"
 	"github.com/chrislusf/seaweedfs/weed/storage/types"
@@ -36,7 +37,7 @@ func main() {
 	}
 	defer indexFile.Close()
 
-	idx.WalkIndexFile(indexFile, func(key types.NeedleId, offset types.Offset, size types.Size) error {
+	idx.WalkIndexFile(indexFile, 0, func(key types.NeedleId, offset types.Offset, size types.Size) error {
 		fmt.Printf("key:%v offset:%v size:%v(%v)\n", key, offset, size, util.BytesToHumanReadable(uint64(size)))
 		return nil
 	})

weed/filer/abstract_sql/abstract_sql_store.go

@@ -32,6 +32,8 @@ type AbstractSqlStore struct {
 	dbsLock            sync.Mutex
 }
 
+var _ filer.BucketAware = (*AbstractSqlStore)(nil)
+
 func (store *AbstractSqlStore) CanDropWholeBucket() bool {
 	return store.SupportBucketTable
 }

weed/filer/leveldb3/leveldb3_store.go

@@ -121,23 +121,31 @@ func (store *LevelDB3Store) findDB(fullpath weed_util.FullPath, isForChildren bo
 	}
 
 	store.dbsLock.RUnlock()
-	// upgrade to write lock
+
+	db, err := store.createDB(bucket)
+
+	return db, bucket, shortPath, err
+}
+
+func (store *LevelDB3Store) createDB(bucket string) (*leveldb.DB, error) {
+
 	store.dbsLock.Lock()
 	defer store.dbsLock.Unlock()
 
 	// double check after getting the write lock
 	if db, found := store.dbs[bucket]; found {
-		return db, bucket, shortPath, nil
+		return db, nil
 	}
 
 	// create db
 	db, err := store.loadDB(bucket)
 	if err != nil {
-		return nil, bucket, shortPath, err
+		return nil, err
 	}
+
 	store.dbs[bucket] = db
 
-	return db, bucket, shortPath, nil
+	return db, nil
 }
 
 func (store *LevelDB3Store) closeDB(bucket string) {

weed/filer/leveldb3/leveldb3_store_bucket.go

@@ -0,0 +1,23 @@
+package leveldb
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"os"
+)
+
+var _ filer.BucketAware = (*LevelDB3Store)(nil)
+
+func (store *LevelDB3Store) OnBucketCreation(bucket string) {
+	store.createDB(bucket)
+}
+
+func (store *LevelDB3Store) OnBucketDeletion(bucket string) {
+	store.closeDB(bucket)
+	if bucket != "" { // just to make sure
+		os.RemoveAll(store.dir + "/" + bucket)
+	}
+}
+
+func (store *LevelDB3Store) CanDropWholeBucket() bool {
+	return true
+}

weed/filer/meta_aggregator.go

@@ -7,6 +7,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"io"
+	"strings"
 	"sync"
 	"time"
 
@@ -99,7 +100,11 @@ func (ma *MetaAggregator) loopSubscribeToOneFiler(f *Filer, self pb.ServerAddres
 			return
 		}
 		if err != nil {
-			glog.V(0).Infof("subscribing remote %s meta change: %v", peer, err)
+			errLvl := glog.Level(0)
+			if strings.Contains(err.Error(), "duplicated local subscription detected") {
+				errLvl = glog.Level(1)
+			}
+			glog.V(errLvl).Infof("subscribing remote %s meta change: %v", peer, err)
 		}
 		if lastTsNs < nextLastTsNs {
 			lastTsNs = nextLastTsNs

weed/filer/mysql2/mysql2_store.go

@@ -18,6 +18,8 @@ const (
 	CONNECTION_URL_PATTERN = "%s:%s@tcp(%s:%d)/%s?charset=utf8"
 )
 
+var _ filer.BucketAware = (*MysqlStore2)(nil)
+
 func init() {
 	filer.Stores = append(filer.Stores, &MysqlStore2{})
 }

weed/filer/postgres2/postgres2_store.go

@@ -17,6 +17,8 @@ const (
 	CONNECTION_URL_PATTERN = "host=%s port=%d sslmode=%s connect_timeout=30"
 )
 
+var _ filer.BucketAware = (*PostgresStore2)(nil)
+
 func init() {
 	filer.Stores = append(filer.Stores, &PostgresStore2{})
 }

weed/filer/stream.go

@@ -18,6 +18,12 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/wdclient"
 )
 
+var getLookupFileIdBackoffSchedule = []time.Duration{
+	150 * time.Millisecond,
+	600 * time.Millisecond,
+	1800 * time.Millisecond,
+}
+
 func HasData(entry *filer_pb.Entry) bool {
 
 	if len(entry.Content) > 0 {
@@ -69,14 +75,22 @@ func StreamContent(masterClient wdclient.HasLookupFileIdFunction, writer io.Writ
 	fileId2Url := make(map[string][]string)
 
 	for _, chunkView := range chunkViews {
-
-		urlStrings, err := masterClient.GetLookupFileIdFunction()(chunkView.FileId)
+		var urlStrings []string
+		var err error
+		for _, backoff := range getLookupFileIdBackoffSchedule {
+			urlStrings, err = masterClient.GetLookupFileIdFunction()(chunkView.FileId)
+			if err == nil && len(urlStrings) > 0 {
+				time.Sleep(backoff)
+				break
+			}
+		}
 		if err != nil {
 			glog.V(1).Infof("operation LookupFileId %s failed, err: %v", chunkView.FileId, err)
 			return err
 		} else if len(urlStrings) == 0 {
-			glog.Errorf("operation LookupFileId %s failed, err: urls not found", chunkView.FileId)
-			return fmt.Errorf("operation LookupFileId %s failed, err: urls not found", chunkView.FileId)
+			errUrlNotFound := fmt.Errorf("operation LookupFileId %s failed, err: urls not found", chunkView.FileId)
+			glog.Error(errUrlNotFound)
+			return errUrlNotFound
 		}
 		fileId2Url[chunkView.FileId] = urlStrings
 	}

weed/filer/ydb/ydb_store.go

@@ -320,6 +320,8 @@ func (store *YdbStore) Shutdown() {
 	_ = store.DB.Close(context.Background())
 }
 
+var _ filer.BucketAware = (*YdbStore)(nil)
+
 func (store *YdbStore) CanDropWholeBucket() bool {
 	return store.SupportBucketTable
 }

weed/iamapi/iamapi_management_handlers.go

@@ -219,9 +219,17 @@ func (iama *IamApiServer) PutUserPolicy(s3cfg *iam_pb.S3ApiConfiguration, values
 		if userName != ident.Name {
 			continue
 		}
+
+		existedActions := make(map[string]bool, len(ident.Actions))
+		for _, action := range ident.Actions {
+			existedActions[action] = true
+		}
+
 		for _, action := range actions {
+			if !existedActions[action] {
 				ident.Actions = append(ident.Actions, action)
 			}
+		}
 		return resp, nil
 	}
 	return resp, fmt.Errorf("%s: the user with name %s cannot be found", iam.ErrCodeNoSuchEntityException, userName)
@@ -349,7 +357,8 @@ func (iama *IamApiServer) CreateAccessKey(s3cfg *iam_pb.S3ApiConfiguration, valu
 	}
 	if !changed {
 		s3cfg.Identities = append(s3cfg.Identities,
-			&iam_pb.Identity{Name: userName,
+			&iam_pb.Identity{
+				Name: userName,
 				Credentials: []*iam_pb.Credential{
 					{
 						AccessKey: accessKeyId,

weed/mount/weedfs_rename.go

@@ -235,7 +235,7 @@ func (wfs *WFS) handleRenameResponse(ctx context.Context, resp *filer_pb.StreamR
 
 		sourceInode, targetInode := wfs.inodeToPath.MovePath(oldPath, newPath)
 		if sourceInode != 0 {
-			if fh, foundFh := wfs.fhmap.inode2fh[sourceInode]; foundFh && fh.entry != nil {
+			if fh, foundFh := wfs.fhmap.FindFileHandle(sourceInode); foundFh && fh.entry != nil {
 				fh.entry.Name = newName
 			}
 			// invalidate attr and data

weed/server/master_server.go

@@ -65,8 +65,8 @@ type MasterServer struct {
 
 	boundedLeaderChan chan int
 
-	onPeerUpdatDoneCn      chan string
-	onPeerUpdatDoneCnExist bool
+	onPeerUpdateDoneCn      chan string
+	onPeerUpdateDoneCnExist bool
 
 	// notifying clients
 	clientChansLock sync.RWMutex
@@ -118,7 +118,7 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se
 		Cluster:         cluster.NewCluster(),
 	}
 	ms.boundedLeaderChan = make(chan int, 16)
-	ms.onPeerUpdatDoneCn = make(chan string)
+	ms.onPeerUpdateDoneCn = make(chan string)
 
 	ms.MasterClient.OnPeerUpdate = ms.OnPeerUpdate
 
@@ -366,14 +366,15 @@ func (ms *MasterServer) OnPeerUpdate(update *master_pb.ClusterNodeUpdate, startF
 					hashicorpRaft.ServerAddress(peerAddress.ToGrpcAddress()), 0, 0)
 			}
 		}
-		if ms.onPeerUpdatDoneCnExist {
-			ms.onPeerUpdatDoneCn <- peerName
+		if ms.onPeerUpdateDoneCnExist {
+			ms.onPeerUpdateDoneCn <- peerName
 		}
 	} else if isLeader {
 		go func(peerName string) {
+			raftServerRemovalTimeAfter := time.After(RaftServerRemovalTime)
 			for {
 				select {
-				case <-time.After(RaftServerRemovalTime):
+				case <-raftServerRemovalTimeAfter:
 					err := ms.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
 						_, err := client.RaftRemoveServer(context.Background(), &master_pb.RaftRemoveServerRequest{
 							Id:    peerName,
@@ -384,14 +385,16 @@ func (ms *MasterServer) OnPeerUpdate(update *master_pb.ClusterNodeUpdate, startF
 					if err != nil {
 						glog.Warningf("failed to removing old raft server %s: %v", peerName, err)
 					}
+					glog.V(0).Infof("old raft server %s removed", peerName)
 					return
-				case peerDone := <-ms.onPeerUpdatDoneCn:
+				case peerDone := <-ms.onPeerUpdateDoneCn:
 					if peerName == peerDone {
+						glog.V(0).Infof("raft server %s remove canceled", peerName)
 						return
 					}
 				}
 			}
 		}(peerName)
-		ms.onPeerUpdatDoneCnExist = true
+		ms.onPeerUpdateDoneCnExist = true
 	}
 }

weed/shell/command_volume_configure_replication.go

@@ -68,27 +68,22 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman
 	volumeFilter := getVolumeFilter(replicaPlacement, uint32(vid), *collectionPattern)
 
 	// find all data nodes with volumes that needs replication change
-	var allLocations []location
 	eachDataNode(topologyInfo, func(dc string, rack RackId, dn *master_pb.DataNodeInfo) {
-		loc := newLocation(dc, string(rack), dn)
+		var targetVolumeIds []uint32
 		for _, diskInfo := range dn.DiskInfos {
 			for _, v := range diskInfo.VolumeInfos {
 				if volumeFilter(v) {
-					allLocations = append(allLocations, loc)
-					continue
+					targetVolumeIds = append(targetVolumeIds, v.Id)
 				}
 			}
 		}
-	})
-
-	if len(allLocations) == 0 {
-		return fmt.Errorf("no volume needs change")
+		if len(targetVolumeIds) == 0 {
+			return
 		}
-
-	for _, dst := range allLocations {
-		err := operation.WithVolumeServerClient(false, pb.NewServerAddressFromDataNode(dst.dataNode), commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
+		err = operation.WithVolumeServerClient(false, pb.NewServerAddressFromDataNode(dn), commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
+			for _, targetVolumeId := range targetVolumeIds {
 				resp, configureErr := volumeServerClient.VolumeConfigure(context.Background(), &volume_server_pb.VolumeConfigureRequest{
-				VolumeId:    uint32(vid),
+					VolumeId:    targetVolumeId,
 					Replication: replicaPlacement.String(),
 				})
 				if configureErr != nil {
@@ -97,16 +92,15 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman
 				if resp.Error != "" {
 					return errors.New(resp.Error)
 				}
+			}
 			return nil
 		})
-
 		if err != nil {
-			return err
-		}
-
+			return
 		}
+	})
 
-	return nil
+	return err
 }
 
 func getVolumeFilter(replicaPlacement *super_block.ReplicaPlacement, volumeId uint32, collectionPattern string) func(message *master_pb.VolumeInformationMessage) bool {

weed/shell/command_volume_server_evacuate.go

@@ -18,7 +18,9 @@ func init() {
 }
 
 type commandVolumeServerEvacuate struct {
+	topologyInfo *master_pb.TopologyInfo
 	targetServer string
+	volumeRack   string
 }
 
 func (c *commandVolumeServerEvacuate) Name() string {
@@ -47,7 +49,8 @@ func (c *commandVolumeServerEvacuate) Do(args []string, commandEnv *CommandEnv,
 
 	vsEvacuateCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	volumeServer := vsEvacuateCommand.String("node", "", "<host>:<port> of the volume server")
-	c.targetServer = *vsEvacuateCommand.String("target", "", "<host>:<port> of target volume")
+	volumeRack := vsEvacuateCommand.String("rack", "", "source rack for the volume servers")
+	targetServer := vsEvacuateCommand.String("target", "", "<host>:<port> of target volume")
 	skipNonMoveable := vsEvacuateCommand.Bool("skipNonMoveable", false, "skip volumes that can not be moved")
 	applyChange := vsEvacuateCommand.Bool("force", false, "actually apply the changes")
 	retryCount := vsEvacuateCommand.Int("retry", 0, "how many times to retry")
@@ -56,12 +59,18 @@ func (c *commandVolumeServerEvacuate) Do(args []string, commandEnv *CommandEnv,
 	}
 	infoAboutSimulationMode(writer, *applyChange, "-force")
 
-	if err = commandEnv.confirmIsLocked(args); err != nil {
+	if err = commandEnv.confirmIsLocked(args); err != nil && *applyChange {
 		return
 	}
 
-	if *volumeServer == "" {
-		return fmt.Errorf("need to specify volume server by -node=<host>:<port>")
+	if *volumeServer == "" && *volumeRack == "" {
+		return fmt.Errorf("need to specify volume server by -node=<host>:<port> or source rack")
+	}
+	if *targetServer != "" {
+		c.targetServer = *targetServer
+	}
+	if *volumeRack != "" {
+		c.volumeRack = *volumeRack
 	}
 	for i := 0; i < *retryCount+1; i++ {
 		if err = c.volumeServerEvacuate(commandEnv, *volumeServer, *skipNonMoveable, *applyChange, writer); err == nil {
@@ -80,37 +89,51 @@ func (c *commandVolumeServerEvacuate) volumeServerEvacuate(commandEnv *CommandEn
 
 	// list all the volumes
 	// collect topology information
-	topologyInfo, _, err := collectTopologyInfo(commandEnv, 0)
+	c.topologyInfo, _, err = collectTopologyInfo(commandEnv, 0)
 	if err != nil {
 		return err
 	}
 
-	if err := c.evacuateNormalVolumes(commandEnv, topologyInfo, volumeServer, skipNonMoveable, applyChange, writer); err != nil {
+	if err := c.evacuateNormalVolumes(commandEnv, volumeServer, skipNonMoveable, applyChange, writer); err != nil {
 		return err
 	}
 
-	if err := c.evacuateEcVolumes(commandEnv, topologyInfo, volumeServer, skipNonMoveable, applyChange, writer); err != nil {
+	if err := c.evacuateEcVolumes(commandEnv, volumeServer, skipNonMoveable, applyChange, writer); err != nil {
 		return err
 	}
 
 	return nil
 }
 
-func (c *commandVolumeServerEvacuate) evacuateNormalVolumes(commandEnv *CommandEnv, topologyInfo *master_pb.TopologyInfo, volumeServer string, skipNonMoveable, applyChange bool, writer io.Writer) error {
+func (c *commandVolumeServerEvacuate) evacuateNormalVolumes(commandEnv *CommandEnv, volumeServer string, skipNonMoveable, applyChange bool, writer io.Writer) error {
 	// find this volume server
-	volumeServers := collectVolumeServersByDc(topologyInfo, "")
-	thisNode, otherNodes := nodesOtherThan(volumeServers, volumeServer)
-	if thisNode == nil {
+	volumeServers := collectVolumeServersByDc(c.topologyInfo, "")
+	thisNodes, otherNodes := c.nodesOtherThan(volumeServers, volumeServer)
+	if len(thisNodes) == 0 {
 		return fmt.Errorf("%s is not found in this cluster", volumeServer)
 	}
 
 	// move away normal volumes
-	volumeReplicas, _ := collectVolumeReplicaLocations(topologyInfo)
+	for _, thisNode := range thisNodes {
 		for _, diskInfo := range thisNode.info.DiskInfos {
+			if applyChange {
+				if topologyInfo, _, err := collectTopologyInfo(commandEnv, 0); err != nil {
+					fmt.Fprintf(writer, "update topologyInfo %v", err)
+				} else {
+					_, otherNodesNew := c.nodesOtherThan(
+						collectVolumeServersByDc(topologyInfo, ""), volumeServer)
+					if len(otherNodesNew) > 0 {
+						otherNodes = otherNodesNew
+						c.topologyInfo = topologyInfo
+						fmt.Fprintf(writer, "topologyInfo updated %v\n", len(otherNodes))
+					}
+				}
+			}
+			volumeReplicas, _ := collectVolumeReplicaLocations(c.topologyInfo)
 			for _, vol := range diskInfo.VolumeInfos {
 				hasMoved, err := moveAwayOneNormalVolume(commandEnv, volumeReplicas, vol, thisNode, otherNodes, applyChange)
 				if err != nil {
-				return fmt.Errorf("move away volume %d from %s: %v", vol.Id, volumeServer, err)
+					fmt.Fprintf(writer, "move away volume %d from %s: %v", vol.Id, volumeServer, err)
 				}
 				if !hasMoved {
 					if skipNonMoveable {
@@ -122,23 +145,25 @@ func (c *commandVolumeServerEvacuate) evacuateNormalVolumes(commandEnv *CommandE
 				}
 			}
 		}
+	}
 	return nil
 }
 
-func (c *commandVolumeServerEvacuate) evacuateEcVolumes(commandEnv *CommandEnv, topologyInfo *master_pb.TopologyInfo, volumeServer string, skipNonMoveable, applyChange bool, writer io.Writer) error {
+func (c *commandVolumeServerEvacuate) evacuateEcVolumes(commandEnv *CommandEnv, volumeServer string, skipNonMoveable, applyChange bool, writer io.Writer) error {
 	// find this ec volume server
-	ecNodes, _ := collectEcVolumeServersByDc(topologyInfo, "")
-	thisNode, otherNodes := ecNodesOtherThan(ecNodes, volumeServer)
-	if thisNode == nil {
+	ecNodes, _ := collectEcVolumeServersByDc(c.topologyInfo, "")
+	thisNodes, otherNodes := c.ecNodesOtherThan(ecNodes, volumeServer)
+	if len(thisNodes) == 0 {
 		return fmt.Errorf("%s is not found in this cluster\n", volumeServer)
 	}
 
 	// move away ec volumes
+	for _, thisNode := range thisNodes {
 		for _, diskInfo := range thisNode.info.DiskInfos {
 			for _, ecShardInfo := range diskInfo.EcShardInfos {
 				hasMoved, err := c.moveAwayOneEcVolume(commandEnv, ecShardInfo, thisNode, otherNodes, applyChange)
 				if err != nil {
-				return fmt.Errorf("move away volume %d from %s: %v", ecShardInfo.Id, volumeServer, err)
+					fmt.Fprintf(writer, "move away volume %d from %s: %v", ecShardInfo.Id, volumeServer, err)
 				}
 				if !hasMoved {
 					if skipNonMoveable {
@@ -149,6 +174,7 @@ func (c *commandVolumeServerEvacuate) evacuateEcVolumes(commandEnv *CommandEnv,
 				}
 			}
 		}
+	}
 	return nil
 }
 
@@ -160,9 +186,6 @@ func (c *commandVolumeServerEvacuate) moveAwayOneEcVolume(commandEnv *CommandEnv
 		})
 		for i := 0; i < len(otherNodes); i++ {
 			emptyNode := otherNodes[i]
-			if c.targetServer != "" && c.targetServer != emptyNode.info.Id {
-				continue
-			}
 			collectionPrefix := ""
 			if ecShardInfo.Collection != "" {
 				collectionPrefix = ecShardInfo.Collection + "_"
@@ -207,10 +230,16 @@ func moveAwayOneNormalVolume(commandEnv *CommandEnv, volumeReplicas map[uint32][
 	return
 }
 
-func nodesOtherThan(volumeServers []*Node, thisServer string) (thisNode *Node, otherNodes []*Node) {
+func (c *commandVolumeServerEvacuate) nodesOtherThan(volumeServers []*Node, thisServer string) (thisNodes []*Node, otherNodes []*Node) {
 	for _, node := range volumeServers {
-		if node.info.Id == thisServer {
-			thisNode = node
+		if node.info.Id == thisServer || (c.volumeRack != "" && node.rack == c.volumeRack) {
+			thisNodes = append(thisNodes, node)
+			continue
+		}
+		if c.volumeRack != "" && c.volumeRack == node.rack {
+			continue
+		}
+		if c.targetServer != "" && c.targetServer != node.info.Id {
 			continue
 		}
 		otherNodes = append(otherNodes, node)
@@ -218,10 +247,16 @@ func nodesOtherThan(volumeServers []*Node, thisServer string) (thisNode *Node, o
 	return
 }
 
-func ecNodesOtherThan(volumeServers []*EcNode, thisServer string) (thisNode *EcNode, otherNodes []*EcNode) {
+func (c *commandVolumeServerEvacuate) ecNodesOtherThan(volumeServers []*EcNode, thisServer string) (thisNodes []*EcNode, otherNodes []*EcNode) {
 	for _, node := range volumeServers {
-		if node.info.Id == thisServer {
-			thisNode = node
+		if node.info.Id == thisServer || (c.volumeRack != "" && string(node.rack) == c.volumeRack) {
+			thisNodes = append(thisNodes, node)
+			continue
+		}
+		if c.volumeRack != "" && c.volumeRack == string(node.rack) {
+			continue
+		}
+		if c.targetServer != "" && c.targetServer != node.info.Id {
 			continue
 		}
 		otherNodes = append(otherNodes, node)

weed/shell/command_volume_server_evacuate_test.go

@@ -6,12 +6,11 @@ import (
 )
 
 func TestVolumeServerEvacuate(t *testing.T) {
-	topologyInfo := parseOutput(topoData)
+	c := commandVolumeServerEvacuate{}
+	c.topologyInfo = parseOutput(topoData)
 
 	volumeServer := "192.168.1.4:8080"
-
-	c := commandVolumeServerEvacuate{}
-	if err := c.evacuateNormalVolumes(nil, topologyInfo, volumeServer, true, false, os.Stdout); err != nil {
+	if err := c.evacuateNormalVolumes(nil, volumeServer, true, false, os.Stdout); err != nil {
 		t.Errorf("evacuate: %v", err)
 	}
 

weed/storage/erasure_coding/ec_encoder.go

@@ -294,7 +294,7 @@ func readNeedleMap(baseFileName string) (*needle_map.MemDb, error) {
 	defer indexFile.Close()
 
 	cm := needle_map.NewMemDb()
-	err = idx.WalkIndexFile(indexFile, func(key types.NeedleId, offset types.Offset, size types.Size) error {
+	err = idx.WalkIndexFile(indexFile, 0, func(key types.NeedleId, offset types.Offset, size types.Size) error {
 		if !offset.IsZero() && size != types.TombstoneFileSize {
 			cm.Set(key, offset, size)
 		} else {

weed/storage/idx/walk.go

@@ -9,8 +9,8 @@ import (
 
 // walks through the index file, calls fn function with each key, offset, size
 // stops with the error returned by the fn function
-func WalkIndexFile(r io.ReaderAt, fn func(key types.NeedleId, offset types.Offset, size types.Size) error) error {
-	var readerOffset int64
+func WalkIndexFile(r io.ReaderAt, startFrom uint64, fn func(key types.NeedleId, offset types.Offset, size types.Size) error) error {
+	readerOffset := int64(startFrom * types.NeedleMapEntrySize)
 	bytes := make([]byte, types.NeedleMapEntrySize*RowsToRead)
 	count, e := r.ReadAt(bytes, readerOffset)
 	if count == 0 && e == io.EOF {

weed/storage/needle_map/memdb.go

@@ -111,7 +111,7 @@ func (cm *MemDb) LoadFromIdx(idxName string) (ret error) {
 
 func (cm *MemDb) LoadFromReaderAt(readerAt io.ReaderAt) (ret error) {
 
-	return idx.WalkIndexFile(readerAt, func(key NeedleId, offset Offset, size Size) error {
+	return idx.WalkIndexFile(readerAt, 0, func(key NeedleId, offset Offset, size Size) error {
 		if offset.IsZero() || size.IsDeleted() {
 			return cm.Delete(key)
 		}

weed/storage/needle_map_leveldb.go

@@ -9,6 +9,8 @@ import (
 	"github.com/syndtr/goleveldb/leveldb/opt"
 
 	"github.com/chrislusf/seaweedfs/weed/storage/idx"
+	"github.com/chrislusf/seaweedfs/weed/storage/types"
+	"github.com/chrislusf/seaweedfs/weed/util"
 
 	"github.com/syndtr/goleveldb/leveldb"
 
@@ -17,10 +19,16 @@ import (
 	. "github.com/chrislusf/seaweedfs/weed/storage/types"
 )
 
+//mark it every watermarkBatchSize operations
+const watermarkBatchSize = 10000
+
+var watermarkKey = []byte("idx_entry_watermark")
+
 type LevelDbNeedleMap struct {
 	baseNeedleMapper
 	dbFileName  string
 	db          *leveldb.DB
+	recordCount uint64
 }
 
 func NewLevelDbNeedleMap(dbFileName string, indexFile *os.File, opts *opt.Options) (m *LevelDbNeedleMap, err error) {
@@ -46,7 +54,14 @@ func NewLevelDbNeedleMap(dbFileName string, indexFile *os.File, opts *opt.Option
 			return
 		}
 	}
-	glog.V(1).Infof("Loading %s...", indexFile.Name())
+	glog.V(0).Infof("Loading %s... , watermark: %d", dbFileName, getWatermark(m.db))
+	m.recordCount = uint64(m.indexFileOffset / types.NeedleMapEntrySize)
+	watermark := (m.recordCount / watermarkBatchSize) * watermarkBatchSize
+	err = setWatermark(m.db, watermark)
+	if err != nil {
+		glog.Fatalf("set watermark for %s error: %s\n", dbFileName, err)
+		return
+	}
 	mm, indexLoadError := newNeedleMapMetricFromIndexFile(indexFile)
 	if indexLoadError != nil {
 		return nil, indexLoadError
@@ -78,9 +93,20 @@ func generateLevelDbFile(dbFileName string, indexFile *os.File) error {
 		return err
 	}
 	defer db.Close()
-	return idx.WalkIndexFile(indexFile, func(key NeedleId, offset Offset, size Size) error {
+
+	watermark := getWatermark(db)
+	if stat, err := indexFile.Stat(); err != nil {
+		glog.Fatalf("stat file %s: %v", indexFile.Name(), err)
+		return err
+	} else {
+		if watermark*types.NeedleMapEntrySize > uint64(stat.Size()) {
+			glog.Warningf("wrong watermark %d for filesize %d", watermark, stat.Size())
+		}
+		glog.V(0).Infof("generateLevelDbFile %s, watermark %d, num of entries:%d", dbFileName, watermark, (uint64(stat.Size())-watermark*types.NeedleMapEntrySize)/types.NeedleMapEntrySize)
+	}
+	return idx.WalkIndexFile(indexFile, watermark, func(key NeedleId, offset Offset, size Size) error {
 		if !offset.IsZero() && size.IsValid() {
-			levelDbWrite(db, key, offset, size)
+			levelDbWrite(db, key, offset, size, false, 0)
 		} else {
 			levelDbDelete(db, key)
 		}
@@ -102,6 +128,7 @@ func (m *LevelDbNeedleMap) Get(key NeedleId) (element *needle_map.NeedleValue, o
 
 func (m *LevelDbNeedleMap) Put(key NeedleId, offset Offset, size Size) error {
 	var oldSize Size
+	var watermark uint64
 	if oldNeedle, ok := m.Get(key); ok {
 		oldSize = oldNeedle.Size
 	}
@@ -110,16 +137,54 @@ func (m *LevelDbNeedleMap) Put(key NeedleId, offset Offset, size Size) error {
 	if err := m.appendToIndexFile(key, offset, size); err != nil {
 		return fmt.Errorf("cannot write to indexfile %s: %v", m.indexFile.Name(), err)
 	}
-	return levelDbWrite(m.db, key, offset, size)
+	m.recordCount++
+	if m.recordCount%watermarkBatchSize != 0 {
+		watermark = 0
+	} else {
+		watermark = (m.recordCount / watermarkBatchSize) * watermarkBatchSize
+		glog.V(1).Infof("put cnt:%d for %s,watermark: %d", m.recordCount, m.dbFileName, watermark)
+	}
+	return levelDbWrite(m.db, key, offset, size, watermark == 0, watermark)
 }
 
-func levelDbWrite(db *leveldb.DB, key NeedleId, offset Offset, size Size) error {
+func getWatermark(db *leveldb.DB) uint64 {
+	data, err := db.Get(watermarkKey, nil)
+	if err != nil || len(data) != 8 {
+		glog.Warningf("get watermark from db error: %v, %d", err, len(data))
+		/*
+			if !strings.Contains(strings.ToLower(err.Error()), "not found") {
+				err = setWatermark(db, 0)
+				if err != nil {
+					glog.Errorf("failed to set watermark: %v", err)
+				}
+			}
+		*/
+		return 0
+	}
+	return util.BytesToUint64(data)
+}
+
+func setWatermark(db *leveldb.DB, watermark uint64) error {
+	glog.V(1).Infof("set watermark %d", watermark)
+	var wmBytes = make([]byte, 8)
+	util.Uint64toBytes(wmBytes, watermark)
+	if err := db.Put(watermarkKey, wmBytes, nil); err != nil {
+		return fmt.Errorf("failed to setWatermark: %v", err)
+	}
+	return nil
+}
+
+func levelDbWrite(db *leveldb.DB, key NeedleId, offset Offset, size Size, updateWatermark bool, watermark uint64) error {
 
 	bytes := needle_map.ToBytes(key, offset, size)
 
 	if err := db.Put(bytes[0:NeedleIdSize], bytes[NeedleIdSize:NeedleIdSize+OffsetSize+SizeSize], nil); err != nil {
 		return fmt.Errorf("failed to write leveldb: %v", err)
 	}
+	// set watermark
+	if updateWatermark {
+		return setWatermark(db, watermark)
+	}
 	return nil
 }
 func levelDbDelete(db *leveldb.DB, key NeedleId) error {
@@ -129,6 +194,7 @@ func levelDbDelete(db *leveldb.DB, key NeedleId) error {
 }
 
 func (m *LevelDbNeedleMap) Delete(key NeedleId, offset Offset) error {
+	var watermark uint64
 	oldNeedle, found := m.Get(key)
 	if !found || oldNeedle.Size.IsDeleted() {
 		return nil
@@ -139,8 +205,13 @@ func (m *LevelDbNeedleMap) Delete(key NeedleId, offset Offset) error {
 	if err := m.appendToIndexFile(key, offset, TombstoneFileSize); err != nil {
 		return err
 	}
-
-	return levelDbWrite(m.db, key, oldNeedle.Offset, -oldNeedle.Size)
+	m.recordCount++
+	if m.recordCount%watermarkBatchSize != 0 {
+		watermark = 0
+	} else {
+		watermark = (m.recordCount / watermarkBatchSize) * watermarkBatchSize
+	}
+	return levelDbWrite(m.db, key, oldNeedle.Offset, -oldNeedle.Size, watermark == 0, watermark)
 }
 
 func (m *LevelDbNeedleMap) Close() {

weed/storage/needle_map_memory.go

@@ -33,7 +33,7 @@ func LoadCompactNeedleMap(file *os.File) (*NeedleMap, error) {
 }
 
 func doLoading(file *os.File, nm *NeedleMap) (*NeedleMap, error) {
-	e := idx.WalkIndexFile(file, func(key NeedleId, offset Offset, size Size) error {
+	e := idx.WalkIndexFile(file, 0, func(key NeedleId, offset Offset, size Size) error {
 		nm.MaybeSetMaxFileKey(key)
 		if !offset.IsZero() && size.IsValid() {
 			nm.FileCounter++

weed/util/retry.go

@@ -32,7 +32,7 @@ func Retry(name string, job func() error) (err error) {
 	return err
 }
 
-func RetryForever(name string, job func() error, onErrFn func(err error) bool) {
+func RetryForever(name string, job func() error, onErrFn func(err error) (shouldContinue bool)) {
 	waitTime := time.Second
 	for {
 		err := job()

weed/wdclient/masterclient.go

@@ -2,6 +2,7 @@ package wdclient
 
 import (
 	"context"
+	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/stats"
 	"math/rand"
 	"time"
@@ -46,7 +47,7 @@ func (mc *MasterClient) GetLookupFileIdFunction() LookupFileIdFunctionType {
 
 func (mc *MasterClient) LookupFileIdWithFallback(fileId string) (fullUrls []string, err error) {
 	fullUrls, err = mc.vidMap.LookupFileId(fileId)
-	if err == nil {
+	if err == nil && len(fullUrls) > 0 {
 		return
 	}
 	err = pb.WithMasterClient(false, mc.currentMaster, mc.grpcDialOption, func(client master_pb.SeaweedClient) error {
@@ -54,7 +55,7 @@ func (mc *MasterClient) LookupFileIdWithFallback(fileId string) (fullUrls []stri
 			VolumeOrFileIds: []string{fileId},
 		})
 		if err != nil {
-			return err
+			return fmt.Errorf("LookupVolume failed: %v", err)
 		}
 		for vid, vidLocation := range resp.VolumeIdLocations {
 			for _, vidLoc := range vidLocation.Locations {
@@ -67,7 +68,6 @@ func (mc *MasterClient) LookupFileIdWithFallback(fileId string) (fullUrls []stri
 				fullUrls = append(fullUrls, "http://"+loc.Url+"/"+fileId)
 			}
 		}
-
 		return nil
 	})
 	return