Merge branch 'master' of https://github.com/seaweedfs/seaweedfs

5 files changed, 49 insertions, 11 deletions

diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go
index 7caaf01b2..93569ce0d 100644
--- a/weed/server/master_grpc_server.go
+++ b/weed/server/master_grpc_server.go
@@ -307,6 +307,8 @@ func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServ
 		case <-ticker.C:
 			if !ms.Topo.IsLeader() {
 				stats.MasterRaftIsleader.Set(0)
+				stats.MasterAdminLock.Reset()
+				stats.MasterReplicaPlacementMismatch.Reset()
 				return ms.informNewLeader(stream)
 			} else {
 				stats.MasterRaftIsleader.Set(1)
diff --git a/weed/server/master_grpc_server_admin.go b/weed/server/master_grpc_server_admin.go
index fb2c5bd50..77d11df8a 100644
--- a/weed/server/master_grpc_server_admin.go
+++ b/weed/server/master_grpc_server_admin.go
@@ -3,6 +3,7 @@ package weed_server
 import (
 	"context"
 	"fmt"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"math/rand"
 	"sync"
 	"time"
@@ -112,11 +113,13 @@ func (locks *AdminLocks) generateToken(lockName string, clientName string) (ts t
 		lastClient:     clientName,
 	}
 	locks.locks[lockName] = lock
+	stats.MasterAdminLock.WithLabelValues(clientName).Set(1)
 	return lock.accessLockTime, lock.accessSecret
 }
 
 func (locks *AdminLocks) deleteLock(lockName string) {
 	locks.Lock()
+	stats.MasterAdminLock.WithLabelValues(locks.locks[lockName].lastClient).Set(0)
 	defer locks.Unlock()
 	delete(locks.locks, lockName)
 }
diff --git a/weed/server/master_grpc_server_raft.go b/weed/server/master_grpc_server_raft.go
index 7f8ad70df..e6454704e 100644
--- a/weed/server/master_grpc_server_raft.go
+++ b/weed/server/master_grpc_server_raft.go
@@ -20,6 +20,7 @@ func (ms *MasterServer) RaftListClusterServers(ctx context.Context, req *master_
 	}
 
 	servers := ms.Topo.HashicorpRaft.GetConfiguration().Configuration().Servers
+	_, leaderId := ms.Topo.HashicorpRaft.LeaderWithID()
 	ms.Topo.RaftServerAccessLock.RUnlock()
 
 	for _, server := range servers {
@@ -27,6 +28,7 @@ func (ms *MasterServer) RaftListClusterServers(ctx context.Context, req *master_
 			Id:       string(server.ID),
 			Address:  string(server.Address),
 			Suffrage: server.Suffrage.String(),
+			IsLeader: server.ID == leaderId,
 		})
 	}
 	return resp, nil
diff --git a/weed/server/master_server.go b/weed/server/master_server.go
index 758f212ad..cd20f78fc 100644
--- a/weed/server/master_server.go
+++ b/weed/server/master_server.go
@@ -174,39 +174,45 @@ func (ms *MasterServer) SetRaftServer(raftServer *RaftServer) {
 			glog.V(0).Infof("leader change event: %+v => %+v", e.PrevValue(), e.Value())
 			stats.MasterLeaderChangeCounter.WithLabelValues(fmt.Sprintf("%+v", e.Value())).Inc()
 			if ms.Topo.RaftServer.Leader() != "" {
-				glog.V(0).Infoln("[", ms.Topo.RaftServer.Name(), "]", ms.Topo.RaftServer.Leader(), "becomes leader.")
+				glog.V(0).Infof("[%s] %s becomes leader.", ms.Topo.RaftServer.Name(), ms.Topo.RaftServer.Leader())
 			}
 		})
-		raftServerName = ms.Topo.RaftServer.Name()
+		raftServerName = fmt.Sprintf("[%s]", ms.Topo.RaftServer.Name())
 	} else if raftServer.RaftHashicorp != nil {
 		ms.Topo.HashicorpRaft = raftServer.RaftHashicorp
 		leaderCh := raftServer.RaftHashicorp.LeaderCh()
-		prevLeader := ms.Topo.HashicorpRaft.Leader()
+		prevLeader, _ := ms.Topo.HashicorpRaft.LeaderWithID()
+		raftServerName = ms.Topo.HashicorpRaft.String()
 		go func() {
 			for {
 				select {
 				case isLeader := <-leaderCh:
-					leader := ms.Topo.HashicorpRaft.Leader()
+					ms.Topo.RaftServerAccessLock.RLock()
+					leader, _ := ms.Topo.HashicorpRaft.LeaderWithID()
+					ms.Topo.RaftServerAccessLock.RUnlock()
 					glog.V(0).Infof("is leader %+v change event: %+v => %+v", isLeader, prevLeader, leader)
 					stats.MasterLeaderChangeCounter.WithLabelValues(fmt.Sprintf("%+v", leader)).Inc()
 					prevLeader = leader
 				}
 			}
 		}()
-		raftServerName = ms.Topo.HashicorpRaft.String()
 	}
 	ms.Topo.RaftServerAccessLock.Unlock()
 
 	if ms.Topo.IsLeader() {
-		glog.V(0).Infoln("[", raftServerName, "]", "I am the leader!")
+		glog.V(0).Infof("%s I am the leader!", raftServerName)
 	} else {
+		var raftServerLeader string
 		ms.Topo.RaftServerAccessLock.RLock()
-		if ms.Topo.RaftServer != nil && ms.Topo.RaftServer.Leader() != "" {
-			glog.V(0).Infoln("[", ms.Topo.RaftServer.Name(), "]", ms.Topo.RaftServer.Leader(), "is the leader.")
-		} else if ms.Topo.HashicorpRaft != nil && ms.Topo.HashicorpRaft.Leader() != "" {
-			glog.V(0).Infoln("[", ms.Topo.HashicorpRaft.String(), "]", ms.Topo.HashicorpRaft.Leader(), "is the leader.")
+		if ms.Topo.RaftServer != nil {
+			raftServerLeader = ms.Topo.RaftServer.Leader()
+		} else if ms.Topo.HashicorpRaft != nil {
+			raftServerName = ms.Topo.HashicorpRaft.String()
+			raftServerLeaderAddr, _ := ms.Topo.HashicorpRaft.LeaderWithID()
+			raftServerLeader = string(raftServerLeaderAddr)
 		}
 		ms.Topo.RaftServerAccessLock.RUnlock()
+		glog.V(0).Infof("%s %s - is the leader.", raftServerName, raftServerLeader)
 	}
 }
 
@@ -388,8 +394,17 @@ func (ms *MasterServer) OnPeerUpdate(update *master_pb.ClusterNodeUpdate, startF
 			} else {
 				glog.V(0).Infof("master %s successfully responded to ping", peerName)
 			}
-
 			return nil
 		})
 	}
 }
+
+func (ms *MasterServer) Shutdown() {
+	if ms.Topo == nil || ms.Topo.HashicorpRaft == nil {
+		return
+	}
+	if ms.Topo.HashicorpRaft.State() == hashicorpRaft.Leader {
+		ms.Topo.HashicorpRaft.LeadershipTransfer()
+	}
+	ms.Topo.HashicorpRaft.Shutdown()
+}
diff --git a/weed/server/raft_hashicorp.go b/weed/server/raft_hashicorp.go
index 8f81924d1..76468b5e1 100644
--- a/weed/server/raft_hashicorp.go
+++ b/weed/server/raft_hashicorp.go
@@ -6,10 +6,13 @@ package weed_server
 import (
 	"fmt"
 	transport "github.com/Jille/raft-grpc-transport"
+	"github.com/armon/go-metrics"
+	"github.com/armon/go-metrics/prometheus"
 	"github.com/hashicorp/raft"
 	boltdb "github.com/hashicorp/raft-boltdb/v2"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"google.golang.org/grpc"
 	"math/rand"
 	"os"
@@ -182,5 +185,18 @@ func NewHashicorpRaftServer(option *RaftServerOption) (*RaftServer, error) {
 		}()
 	}
 
+	// Configure a prometheus sink as the raft metrics sink
+	if sink, err := prometheus.NewPrometheusSinkFrom(prometheus.PrometheusOpts{
+		Registerer: stats.Gather,
+	}); err != nil {
+		return nil, fmt.Errorf("NewPrometheusSink: %v", err)
+	} else {
+		metricsConf := metrics.DefaultConfig(stats.Namespace)
+		metricsConf.EnableRuntimeMetrics = false
+		if _, err = metrics.NewGlobal(metricsConf, sink); err != nil {
+			return nil, fmt.Errorf("metrics.NewGlobal: %v", err)
+		}
+	}
+
 	return s, nil
 }