Merge pull request #3059 from guol-fnst/avoid_dup_vol

avoid loading duplicated volume directory

weed/command/volume.go

@@ -267,7 +267,7 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
 
 	stopChan := make(chan bool)
 	grace.OnInterrupt(func() {
-		fmt.Println("volume server has be killed")
+		fmt.Println("volume server has been killed")
 
 		// Stop heartbeats
 		if !volumeServer.StopHeartbeat() {

weed/pb/master.proto

@@ -70,7 +70,7 @@ message Heartbeat {
 
   map<string, uint32> max_volume_counts = 4;
   uint32 grpc_port = 20;
-
+  repeated string location_uuids = 21;
 }
 
 message HeartbeatResponse {
@@ -79,6 +79,7 @@ message HeartbeatResponse {
   string metrics_address = 3;
   uint32 metrics_interval_seconds = 4;
   repeated StorageBackend storage_backends = 5;
+  repeated string duplicated_uuids = 6;
 }
 
 message VolumeInformationMessage {

weed/server/master_grpc_server.go

@@ -2,12 +2,16 @@ package weed_server
 
 import (
 	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sort"
+	"time"
+
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/stats"
 	"github.com/chrislusf/seaweedfs/weed/storage/backend"
 	"github.com/chrislusf/seaweedfs/weed/util"
-	"net"
-	"time"
 
 	"github.com/chrislusf/raft"
 	"google.golang.org/grpc/peer"
@@ -18,6 +22,41 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/topology"
 )
 
+func (ms *MasterServer) RegisterUuids(heartbeat *master_pb.Heartbeat) (duplicated_uuids []string, err error) {
+	ms.Topo.UuidAccessLock.Lock()
+	defer ms.Topo.UuidAccessLock.Unlock()
+	key := fmt.Sprintf("%s:%d", heartbeat.Ip, heartbeat.Port)
+	if ms.Topo.UuidMap == nil {
+		ms.Topo.UuidMap = make(map[string][]string)
+	}
+	// find whether new uuid exists
+	for k, v := range ms.Topo.UuidMap {
+		sort.Strings(v)
+		for _, id := range heartbeat.LocationUuids {
+			index := sort.SearchStrings(v, id)
+			if index < len(v) && v[index] == id {
+				duplicated_uuids = append(duplicated_uuids, id)
+				glog.Errorf("directory of %s on %s has been loaded", id, k)
+			}
+		}
+	}
+	if len(duplicated_uuids) > 0 {
+		return duplicated_uuids, errors.New("volume: Duplicated volume directories were loaded")
+	}
+
+	ms.Topo.UuidMap[key] = heartbeat.LocationUuids
+	glog.V(0).Infof("found new uuid:%v %v , %v", key, heartbeat.LocationUuids, ms.Topo.UuidMap)
+	return nil, nil
+}
+
+func (ms *MasterServer) UnRegisterUuids(ip string, port int) {
+	ms.Topo.UuidAccessLock.Lock()
+	defer ms.Topo.UuidAccessLock.Unlock()
+	key := fmt.Sprintf("%s:%d", ip, port)
+	delete(ms.Topo.UuidMap, key)
+	glog.V(0).Infof("remove volume server %v, online volume server: %v", key, ms.Topo.UuidMap)
+}
+
 func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServer) error {
 	var dn *topology.DataNode
 
@@ -32,6 +71,7 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 			//  the unregister and register can race with each other
 			ms.Topo.UnRegisterDataNode(dn)
 			glog.V(0).Infof("unregister disconnected volume server %s:%d", dn.Ip, dn.Port)
+			ms.UnRegisterUuids(dn.Ip, dn.Port)
 
 			message := &master_pb.VolumeLocation{
 				Url:       dn.Url(),
@@ -69,7 +109,18 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 			dc := ms.Topo.GetOrCreateDataCenter(dcName)
 			rack := dc.GetOrCreateRack(rackName)
 			dn = rack.GetOrCreateDataNode(heartbeat.Ip, int(heartbeat.Port), int(heartbeat.GrpcPort), heartbeat.PublicUrl, heartbeat.MaxVolumeCounts)
-			glog.V(0).Infof("added volume server %d: %v:%d", dn.Counter, heartbeat.GetIp(), heartbeat.GetPort())
+			glog.V(0).Infof("added volume server %d: %v:%d %v", dn.Counter, heartbeat.GetIp(), heartbeat.GetPort(), heartbeat.LocationUuids)
+			uuidlist, err := ms.RegisterUuids(heartbeat)
+			if err != nil {
+				if stream_err := stream.Send(&master_pb.HeartbeatResponse{
+					DuplicatedUuids:    uuidlist,
+				}); stream_err != nil {
+					glog.Warningf("SendHeartbeat.Send DuplicatedDirectory response to %s:%d %v", dn.Ip, dn.Port, stream_err)
+					return stream_err
+				}
+				return err
+			}
+
 			if err := stream.Send(&master_pb.HeartbeatResponse{
 				VolumeSizeLimit: uint64(ms.option.VolumeSizeLimitMB) * 1024 * 1024,
 			}); err != nil {

weed/server/volume_grpc_client_to_master.go

@@ -2,9 +2,11 @@ package weed_server
 
 import (
 	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/operation"
+	"os"
 	"time"
 
+	"github.com/chrislusf/seaweedfs/weed/operation"
+
 	"google.golang.org/grpc"
 
 	"github.com/chrislusf/seaweedfs/weed/pb"
@@ -116,6 +118,18 @@ func (vs *VolumeServer) doHeartbeat(masterAddress pb.ServerAddress, grpcDialOpti
 				doneChan <- err
 				return
 			}
+			if len(in.DuplicatedUuids) > 0 {
+				var duplictedDir []string
+				for _, loc := range vs.store.Locations {
+					for _, uuid := range in.DuplicatedUuids {
+						if uuid == loc.DirectoryUuid {
+							duplictedDir = append(duplictedDir, loc.Directory)
+						}
+					}
+				}
+				glog.Errorf("Shut down Volume Server due to duplicated volume directories: %v", duplictedDir)
+				os.Exit(1)
+			}
 			if in.GetVolumeSizeLimit() != 0 && vs.store.GetVolumeSizeLimit() != in.GetVolumeSizeLimit() {
 				vs.store.SetVolumeSizeLimit(in.GetVolumeSizeLimit())
 				if vs.store.MaybeAdjustVolumeMax() {

weed/server/volume_server.go

@@ -1,11 +1,12 @@
 package weed_server
 
 import (
+	"net/http"
+	"sync"
+
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/chrislusf/seaweedfs/weed/storage/types"
-	"net/http"
-	"sync"
 
 	"google.golang.org/grpc"
 

weed/server/volume_server_tcp_handlers_write.go

@@ -3,12 +3,13 @@ package weed_server
 import (
 	"bufio"
 	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/glog"
-	"github.com/chrislusf/seaweedfs/weed/storage/needle"
-	"github.com/chrislusf/seaweedfs/weed/util"
 	"io"
 	"net"
 	"strings"
+
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/storage/needle"
+	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
 func (vs *VolumeServer) HandleTcpConnection(c net.Conn) {

weed/storage/disk_location.go

@@ -14,10 +14,12 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 	"github.com/chrislusf/seaweedfs/weed/storage/types"
 	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/google/uuid"
 )
 
 type DiskLocation struct {
 	Directory              string
+	DirectoryUuid          string
 	IdxDirectory           string
 	DiskType               types.DiskType
 	MaxVolumeCount         int
@@ -33,6 +35,29 @@ type DiskLocation struct {
 	isDiskSpaceLow bool
 }
 
+func GenerateDirUuid(dir string) (dirUuidString string, err error) {
+	glog.V(1).Infof("Getting uuid of volume directory:%s", dir)
+	dirUuidString = ""
+	fileName := dir + "/vol_dir.uuid"
+	if !util.FileExists(fileName) {
+		dirUuid, _ := uuid.NewRandom()
+		dirUuidString = dirUuid.String()
+		writeErr := util.WriteFile(fileName, []byte(dirUuidString), 0644)
+		if writeErr != nil {
+			glog.Warningf("failed to write uuid to %s : %v", fileName, writeErr)
+			return "", fmt.Errorf("failed to write uuid to %s : %v", fileName, writeErr)
+		}
+	} else {
+		uuidData, readErr := os.ReadFile(fileName)
+		if readErr != nil {
+			glog.Warningf("failed to read uuid from %s : %v", fileName, readErr)
+			return "", fmt.Errorf("failed to read uuid from %s : %v", fileName, readErr)
+		}
+		dirUuidString = string(uuidData)
+	}
+	return dirUuidString, nil
+}
+
 func NewDiskLocation(dir string, maxVolumeCount int, minFreeSpace util.MinFreeSpace, idxDir string, diskType types.DiskType) *DiskLocation {
 	dir = util.ResolvePath(dir)
 	if idxDir == "" {
@@ -40,8 +65,10 @@ func NewDiskLocation(dir string, maxVolumeCount int, minFreeSpace util.MinFreeSp
 	} else {
 		idxDir = util.ResolvePath(idxDir)
 	}
+	dirUuid, _ := GenerateDirUuid(dir)
 	location := &DiskLocation{
 		Directory:              dir,
+		DirectoryUuid:          dirUuid,
 		IdxDirectory:           idxDir,
 		DiskType:               diskType,
 		MaxVolumeCount:         maxVolumeCount,

weed/storage/store.go

@@ -2,13 +2,14 @@ package storage
 
 import (
 	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/pb"
-	"github.com/chrislusf/seaweedfs/weed/storage/volume_info"
-	"github.com/chrislusf/seaweedfs/weed/util"
 	"path/filepath"
 	"strings"
 	"sync/atomic"
 
+	"github.com/chrislusf/seaweedfs/weed/pb"
+	"github.com/chrislusf/seaweedfs/weed/storage/volume_info"
+	"github.com/chrislusf/seaweedfs/weed/util"
+
 	"google.golang.org/grpc"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
@@ -300,6 +301,11 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 		}
 	}
 
+	var uuidList []string
+	for _, loc := range s.Locations {
+		uuidList = append(uuidList, loc.DirectoryUuid)
+	}
+
 	for col, size := range collectionVolumeSize {
 		stats.VolumeServerDiskSizeGauge.WithLabelValues(col, "normal").Set(float64(size))
 	}
@@ -321,6 +327,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 		Rack:            s.rack,
 		Volumes:         volumeMessages,
 		HasNoVolumes:    len(volumeMessages) == 0,
+		LocationUuids:   uuidList,
 	}
 
 }

weed/storage/volume_stream_write.go

@@ -3,10 +3,11 @@ package storage
 import (
 	"bufio"
 	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/util"
 	"io"
 	"time"
 
+	"github.com/chrislusf/seaweedfs/weed/util"
+
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/storage/backend"
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"

weed/topology/topology.go

@@ -4,12 +4,13 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/pb"
-	"github.com/chrislusf/seaweedfs/weed/storage/types"
 	"math/rand"
 	"sync"
 	"time"
 
+	"github.com/chrislusf/seaweedfs/weed/pb"
+	"github.com/chrislusf/seaweedfs/weed/storage/types"
+
 	"github.com/chrislusf/raft"
 	hashicorpRaft "github.com/hashicorp/raft"
 
@@ -44,6 +45,8 @@ type Topology struct {
 
 	RaftServer     raft.Server
 	HashicorpRaft  *hashicorpRaft.Raft
+	UuidAccessLock sync.RWMutex
+	UuidMap        map[string][]string
 }
 
 func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, pulse int, replicationAsMin bool) *Topology {