diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go index 9893b75457..7d7b97818c 100644 --- a/weed/server/master_grpc_server.go +++ b/weed/server/master_grpc_server.go @@ -518,14 +518,14 @@ func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServ // buffer by 1 so we don't end up getting stuck writing to stopChan forever stopChan := make(chan bool, 1) + ms.clusterNodeLock.Lock() clientName, messageChan := ms.addClient(req.FilerGroup, req.ClientType, peerAddress) for _, update := range ms.Cluster.AddClusterNode(req.FilerGroup, req.ClientType, cluster.DataCenter(req.DataCenter), cluster.Rack(req.Rack), peerAddress, req.Version) { glog.V(1).Infof("Cluster: %s node %s added to group '%s'", req.ClientType, peerAddress, req.FilerGroup) ms.broadcastToClients(update) } - if req.ClientType == cluster.FilerType && !req.LeaveLockRing { - ms.LockRingManager.AddServer(cluster.FilerGroupName(req.FilerGroup), peerAddress) - } + ms.updateLockRing(req, peerAddress) + ms.clusterNodeLock.Unlock() if req.ClientType == cluster.MasterType { // Only the leader gets this far, and a master that starts with no raft // state cannot campaign its way in, so this registration is where it @@ -535,12 +535,7 @@ func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServ } defer func() { - for _, update := range ms.Cluster.RemoveClusterNode(req.FilerGroup, req.ClientType, peerAddress) { - ms.broadcastToClients(update) - } - if req.ClientType == cluster.FilerType { - ms.LockRingManager.RemoveServer(cluster.FilerGroupName(req.FilerGroup), peerAddress) - } + ms.removeClusterNode(req, peerAddress) ms.deleteClient(clientName, messageChan) }() @@ -634,6 +629,33 @@ func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServ } +// A filer that reconnects while leaving may overlap a stream that has not ended +// yet, so its registration takes the address out of the ring itself. +func (ms *MasterServer) updateLockRing(req *master_pb.KeepConnectedRequest, peerAddress pb.ServerAddress) { + if req.ClientType != cluster.FilerType { + return + } + if req.LeaveLockRing { + ms.LockRingManager.RemoveServer(cluster.FilerGroupName(req.FilerGroup), peerAddress) + return + } + ms.LockRingManager.AddServer(cluster.FilerGroupName(req.FilerGroup), peerAddress) +} + +func (ms *MasterServer) removeClusterNode(req *master_pb.KeepConnectedRequest, peerAddress pb.ServerAddress) { + ms.clusterNodeLock.Lock() + defer ms.clusterNodeLock.Unlock() + removed := ms.Cluster.RemoveClusterNode(req.FilerGroup, req.ClientType, peerAddress) + for _, update := range removed { + ms.broadcastToClients(update) + } + // A replacement that registered before this stream ended still holds the + // address, so only its last registration leaves the ring. + if req.ClientType == cluster.FilerType && len(removed) > 0 { + ms.LockRingManager.RemoveServer(cluster.FilerGroupName(req.FilerGroup), peerAddress) + } +} + func (ms *MasterServer) onKeepConnectedMessage(registered *master_pb.KeepConnectedRequest, peerAddress pb.ServerAddress, message *master_pb.KeepConnectedRequest) { if registered.ClientType == cluster.FilerType && message.LeaveLockRing { glog.V(0).Infof("LockRing: filer %s leaving group '%s'", peerAddress, registered.FilerGroup) diff --git a/weed/server/master_grpc_server_test.go b/weed/server/master_grpc_server_test.go index b692c14b04..787d1f1913 100644 --- a/weed/server/master_grpc_server_test.go +++ b/weed/server/master_grpc_server_test.go @@ -5,6 +5,7 @@ import ( "time" "github.com/seaweedfs/seaweedfs/weed/cluster" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -50,6 +51,92 @@ func TestLeaveLockRingDropsOnlyTheLeavingFiler(t *testing.T) { assert.ElementsMatch(t, []string{"filer2:8888"}, ms.LockRingManager.GetServers("group-a")) } +// A filer's replacement can register under the same address before the master +// ends the old stream; that stream's cleanup must leave the live filer in the ring. +func TestOverlappedFilerRegistrationStaysInLockRing(t *testing.T) { + ms := &MasterServer{ + Cluster: cluster.NewCluster(), + LockRingManager: cluster.NewLockRingManager(nil), + } + registered := &master_pb.KeepConnectedRequest{ClientType: cluster.FilerType, FilerGroup: "group-a"} + register := func(address pb.ServerAddress) { + ms.Cluster.AddClusterNode(registered.FilerGroup, registered.ClientType, "", "", address, "") + ms.updateLockRing(registered, address) + } + ringServers := func() []string { + ms.LockRingManager.FlushPending("group-a") + return ms.LockRingManager.GetServers("group-a") + } + + register("filer0:8888") + register("filer1:8888") + register("filer0:8888") + + ms.removeClusterNode(registered, "filer0:8888") + assert.True(t, ms.Cluster.IsKnownNode(cluster.FilerType, "filer0:8888")) + assert.ElementsMatch(t, []string{"filer0:8888", "filer1:8888"}, ringServers()) + + ms.removeClusterNode(registered, "filer0:8888") + assert.False(t, ms.Cluster.IsKnownNode(cluster.FilerType, "filer0:8888")) + assert.ElementsMatch(t, []string{"filer1:8888"}, ringServers()) +} + +// A leaving filer's reconnect can register before the master ends its broken +// stream; the filer must leave the ring without waiting for either stream. +func TestLeavingFilerReconnectLeavesLockRing(t *testing.T) { + ms := &MasterServer{ + Cluster: cluster.NewCluster(), + LockRingManager: cluster.NewLockRingManager(nil), + } + registered := &master_pb.KeepConnectedRequest{ClientType: cluster.FilerType, FilerGroup: "group-a"} + leaving := &master_pb.KeepConnectedRequest{ClientType: cluster.FilerType, FilerGroup: "group-a", LeaveLockRing: true} + register := func(req *master_pb.KeepConnectedRequest, address pb.ServerAddress) { + ms.Cluster.AddClusterNode(req.FilerGroup, req.ClientType, "", "", address, "") + ms.updateLockRing(req, address) + } + ringServers := func() []string { + ms.LockRingManager.FlushPending("group-a") + return ms.LockRingManager.GetServers("group-a") + } + + register(registered, "filer0:8888") + register(registered, "filer1:8888") + register(leaving, "filer0:8888") + assert.ElementsMatch(t, []string{"filer1:8888"}, ringServers()) + + ms.removeClusterNode(registered, "filer0:8888") + assert.True(t, ms.Cluster.IsKnownNode(cluster.FilerType, "filer0:8888")) + assert.ElementsMatch(t, []string{"filer1:8888"}, ringServers()) + + ms.removeClusterNode(leaving, "filer0:8888") + assert.False(t, ms.Cluster.IsKnownNode(cluster.FilerType, "filer0:8888")) + assert.ElementsMatch(t, []string{"filer1:8888"}, ringServers()) +} + +// A filer that left the ring and is replaced before its stream ends must not +// take the replacement out of the ring with it. +func TestLeftFilerStreamEndKeepsReplacementInLockRing(t *testing.T) { + ms := &MasterServer{ + Cluster: cluster.NewCluster(), + LockRingManager: cluster.NewLockRingManager(nil), + } + registered := &master_pb.KeepConnectedRequest{ClientType: cluster.FilerType, FilerGroup: "group-a"} + for _, address := range []pb.ServerAddress{"filer0:8888", "filer1:8888"} { + ms.Cluster.AddClusterNode(registered.FilerGroup, registered.ClientType, "", "", address, "") + ms.LockRingManager.AddServer(cluster.FilerGroupName(registered.FilerGroup), address) + } + + ms.onKeepConnectedMessage(registered, "filer0:8888", &master_pb.KeepConnectedRequest{LeaveLockRing: true}) + ms.LockRingManager.FlushPending("group-a") + assert.ElementsMatch(t, []string{"filer1:8888"}, ms.LockRingManager.GetServers("group-a")) + + ms.Cluster.AddClusterNode(registered.FilerGroup, registered.ClientType, "", "", "filer0:8888", "") + ms.LockRingManager.AddServer(cluster.FilerGroupName(registered.FilerGroup), "filer0:8888") + ms.removeClusterNode(registered, "filer0:8888") + ms.LockRingManager.FlushPending("group-a") + assert.ElementsMatch(t, []string{"filer0:8888", "filer1:8888"}, ms.LockRingManager.GetServers("group-a")) +} + func TestInitialLockRingUpdateSkipsNonFilers(t *testing.T) { ms := &MasterServer{ LockRingManager: cluster.NewLockRingManager(nil), diff --git a/weed/server/master_server.go b/weed/server/master_server.go index cefed50232..465c925cdf 100644 --- a/weed/server/master_server.go +++ b/weed/server/master_server.go @@ -110,6 +110,8 @@ type MasterServer struct { adminLocks *AdminLocks Cluster *cluster.Cluster + // orders each stream's cluster list change with its lock ring change + clusterNodeLock sync.Mutex LockRingManager *cluster.LockRingManager