seaweedfs/weed/server/master_server.go

package weed_server

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httputil"
	"net/url"
	"os"
	"regexp"
	"runtime"
	"strings"
	"sync"
	"time"

	"github.com/seaweedfs/seaweedfs/weed/cluster/maintenance"
	"github.com/seaweedfs/seaweedfs/weed/operation"
	"github.com/seaweedfs/seaweedfs/weed/stats"
	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
	"github.com/seaweedfs/seaweedfs/weed/telemetry"

	"github.com/seaweedfs/seaweedfs/weed/cluster"
	"github.com/seaweedfs/seaweedfs/weed/pb"

	"github.com/gorilla/mux"
	hashicorpRaft "github.com/hashicorp/raft"
	"github.com/seaweedfs/raft"
	"google.golang.org/grpc"

	"github.com/seaweedfs/seaweedfs/weed/glog"
	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
	"github.com/seaweedfs/seaweedfs/weed/security"
	"github.com/seaweedfs/seaweedfs/weed/sequence"
	"github.com/seaweedfs/seaweedfs/weed/shell"
	"github.com/seaweedfs/seaweedfs/weed/topology"
	"github.com/seaweedfs/seaweedfs/weed/util"
	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
	"github.com/seaweedfs/seaweedfs/weed/util/version"
	"github.com/seaweedfs/seaweedfs/weed/wdclient"
)

const (
	SequencerType        = "master.sequencer.type"
	SequencerSnowflakeId = "master.sequencer.sequencer_snowflake_id"
	raftApplyTimeout     = 1 * time.Second
)

type MasterOption struct {
	Master                        pb.ServerAddress
	MetaFolder                    string
	VolumeSizeLimitMB             uint32
	FileSizeLimitMB               int
	VolumePreallocate             bool
	MaxParallelVacuumPerServer    int
	MaxParallelVacuumPerRack      int
	MaxWritableVolumes            int
	VacuumIntervalSeconds         int
	VacuumDeleteEmptyAfterSeconds int
	// PulseSeconds            int
	DefaultReplicaPlacement string
	GarbageThreshold        float64
	WhiteList               []string
	DisableHttp             bool
	MetricsAddress          string
	MetricsIntervalSec      int
	IsFollower              bool
	SingleMaster            bool
	TelemetryUrl            string
	TelemetryEnabled        bool
	VolumeGrowthDisabled    bool
	// ReleaseAllocOnFirstReport lifts the post-mint allocation hold as soon as
	// one full volume-server inventory has been applied, for setups whose
	// embedded volume server is the only possible reporter (weed mini).
	// SingleVolumeReporter pins that reporter's ip:port; any other volume
	// server trying to register is rejected.
	ReleaseAllocOnFirstReport bool
	SingleVolumeReporter      string
	// ExpectedReporterAddr names the one volume server whose first clean full
	// inventory lifts the post-mint allocation hold (a single-master weed
	// server's embedded volume server). Other reporters are still admitted.
	ExpectedReporterAddr string
}

type MasterServer struct {
	master_pb.UnimplementedSeaweedServer
	option *MasterOption
	guard  *security.Guard

	preallocateSize int64

	Topo                    *topology.Topology
	vg                      *topology.VolumeGrowth
	volumeGrowthRequestChan chan *topology.VolumeGrowRequest

	// notifying clients
	clientChansLock sync.RWMutex
	clientChans     map[string]chan *master_pb.KeepConnectedResponse

	grpcDialOption grpc.DialOption

	topologyIdGenLock sync.Mutex

	// masters currently being admitted into the raft quorum, keyed by raft id
	raftPeerAdmissions sync.Map

	MasterClient *wdclient.MasterClient

	adminLocks *AdminLocks

	Cluster *cluster.Cluster
	// orders each stream's cluster list change with its lock ring change
	clusterNodeLock sync.Mutex

	LockRingManager *cluster.LockRingManager

	// telemetry
	telemetryCollector *telemetry.Collector
}

func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.ServerAddress) *MasterServer {
	v := util.GetViper()
	signingKey := v.GetString("jwt.signing.key")
	v.SetDefault("jwt.signing.expires_after_seconds", 10)
	expiresAfterSec := v.GetInt("jwt.signing.expires_after_seconds")

	readSigningKey := v.GetString("jwt.signing.read.key")
	v.SetDefault("jwt.signing.read.expires_after_seconds", 60)
	readExpiresAfterSec := v.GetInt("jwt.signing.read.expires_after_seconds")

	v.SetDefault("master.replication.treat_replication_as_minimums", false)
	replicationAsMin := v.GetBool("master.replication.treat_replication_as_minimums")

	v.SetDefault("master.volume_growth.copy_1", topology.VolumeGrowStrategy.Copy1Count)
	v.SetDefault("master.volume_growth.copy_2", topology.VolumeGrowStrategy.Copy2Count)
	v.SetDefault("master.volume_growth.copy_3", topology.VolumeGrowStrategy.Copy3Count)
	v.SetDefault("master.volume_growth.copy_other", topology.VolumeGrowStrategy.CopyOtherCount)
	v.SetDefault("master.volume_growth.threshold", topology.VolumeGrowStrategy.Threshold)
	v.SetDefault("master.volume_growth.reservation_timeout", "5m")
	v.SetDefault("master.volume_growth.disable", false)
	option.VolumeGrowthDisabled = v.GetBool("master.volume_growth.disable")
	v.SetDefault("master.volume_growth.freeze_on_topology_mismatch", true)
	v.SetDefault("master.volume_growth.state_loss_recovery_wait", "0s")
	v.SetDefault("master.vacuum.max_parallel_per_rack", 0)
	option.MaxParallelVacuumPerRack = v.GetInt("master.vacuum.max_parallel_per_rack")
	if v.IsSet("master.vacuum.max_parallel_per_server") {
		option.MaxParallelVacuumPerServer = v.GetInt("master.vacuum.max_parallel_per_server")
	}
	v.SetDefault("master.max_writable_volumes_per_layout", option.MaxWritableVolumes)
	option.MaxWritableVolumes = v.GetInt("master.max_writable_volumes_per_layout")

	topology.VolumeGrowStrategy.Copy1Count = v.GetUint32("master.volume_growth.copy_1")
	topology.VolumeGrowStrategy.Copy2Count = v.GetUint32("master.volume_growth.copy_2")
	topology.VolumeGrowStrategy.Copy3Count = v.GetUint32("master.volume_growth.copy_3")
	topology.VolumeGrowStrategy.CopyOtherCount = v.GetUint32("master.volume_growth.copy_other")
	topology.VolumeGrowStrategy.Threshold = v.GetFloat64("master.volume_growth.threshold")
	topology.VolumeGrowStrategy.ReservationTimeout = parseReservationTimeout(v)
	whiteList := util.StringSplit(v.GetString("guard.white_list"), ",")

	var preallocateSize int64
	if option.VolumePreallocate {
		preallocateSize = int64(option.VolumeSizeLimitMB) * (1 << 20)
	}

	grpcDialOption := security.LoadClientTLS(v, "grpc.master")
	ms := &MasterServer{
		option:                  option,
		preallocateSize:         preallocateSize,
		volumeGrowthRequestChan: make(chan *topology.VolumeGrowRequest, 1<<6),
		clientChans:             make(map[string]chan *master_pb.KeepConnectedResponse),
		grpcDialOption:          grpcDialOption,
		MasterClient:            wdclient.NewMasterClient(grpcDialOption, "", cluster.MasterType, option.Master, "", "", *pb.NewServiceDiscoveryFromMap(peers)),
		adminLocks:              NewAdminLocks(),
		Cluster:                 cluster.NewCluster(),
	}

	ms.LockRingManager = cluster.NewLockRingManager(ms.broadcastToClients)

	ms.MasterClient.SetOnPeerUpdateFn(ms.OnPeerUpdate)

	seq := ms.createSequencer(option)
	if nil == seq {
		glog.Fatalf("create sequencer failed.")
	}
	ms.Topo = topology.NewTopology("topo", seq, uint64(ms.option.VolumeSizeLimitMB)*1024*1024, 5, replicationAsMin)
	ms.Topo.SetMaxWritableVolumes(ms.option.MaxWritableVolumes)
	ms.Topo.SetGrowthFreezeConfig(
		v.GetBool("master.volume_growth.freeze_on_topology_mismatch"),
		v.GetDuration("master.volume_growth.state_loss_recovery_wait"))
	if str := strings.TrimSpace(v.GetString("master.volume_growth.allocation_deferral")); str != "" {
		if d, err := time.ParseDuration(str); err != nil {
			glog.Warningf("invalid master.volume_growth.allocation_deferral %q: %v", str, err)
		} else {
			ms.Topo.SetAllocationDeferral(d)
		}
	}
	if option.ReleaseAllocOnFirstReport {
		ms.Topo.ReleaseAllocationOnFirstReport(option.SingleVolumeReporter)
	}
	if option.ExpectedReporterAddr != "" {
		ms.Topo.ReleaseAllocationOnReport(option.ExpectedReporterAddr)
	}
	ms.Topo.QuarantineVolumeFn = func(dn *topology.DataNode, vid needle.VolumeId, readOnly bool) error {
		err := operation.WithVolumeServerClient(false, dn.ServerAddress(), ms.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
			defer cancel()
			if readOnly {
				_, err := client.VolumeMarkReadonly(ctx, &volume_server_pb.VolumeMarkReadonlyRequest{VolumeId: uint32(vid), CanDelete: true})
				return err
			}
			_, err := client.VolumeMarkWritable(ctx, &volume_server_pb.VolumeMarkWritableRequest{VolumeId: uint32(vid)})
			return err
		})
		if err != nil {
			glog.Warningf("quarantine volume %d on %s readOnly=%v: %v", vid, dn.Id(), readOnly, err)
		}
		return err
	}
	ms.vg = topology.NewDefaultVolumeGrowth()
	glog.V(0).Infoln("Volume Size Limit is", ms.option.VolumeSizeLimitMB, "MB")

	// Initialize telemetry after topology is created
	if option.TelemetryEnabled && option.TelemetryUrl != "" {
		telemetryClient := telemetry.NewClient(option.TelemetryUrl, option.TelemetryEnabled)
		ms.telemetryCollector = telemetry.NewCollector(telemetryClient, ms.Topo, ms.Cluster)
		ms.telemetryCollector.SetMasterServer(ms)

		// Set version and OS information
		ms.telemetryCollector.SetVersion(version.VERSION_NUMBER)
		ms.telemetryCollector.SetOS(runtime.GOOS + "/" + runtime.GOARCH)

		// Start periodic telemetry collection (every 24 hours)
		ms.telemetryCollector.StartPeriodicCollection(24 * time.Hour)
	}

	ms.guard = security.NewGuard(append(ms.option.WhiteList, whiteList...), signingKey, expiresAfterSec, readSigningKey, readExpiresAfterSec)

	handleStaticResources2(r)
	r.HandleFunc("/healthz", requestIDMiddleware(ms.healthzHandler)).Methods(http.MethodGet, http.MethodHead)
	r.HandleFunc("/readyz", requestIDMiddleware(ms.readyzHandler)).Methods(http.MethodGet, http.MethodHead)
	r.HandleFunc("/", ms.proxyToLeader(requestIDMiddleware(ms.uiStatusHandler)))
	r.HandleFunc("/ui/index.html", requestIDMiddleware(ms.uiStatusHandler))
	if !ms.option.DisableHttp {
		r.HandleFunc("/dir/assign", ms.proxyToLeader(ms.guard.WhiteList(requestIDMiddleware(ms.dirAssignHandler))))
		r.HandleFunc("/dir/lookup", ms.guard.WhiteList(requestIDMiddleware(ms.dirLookupHandler)))
		r.HandleFunc("/dir/status", ms.proxyToLeader(ms.guard.WhiteList(requestIDMiddleware(ms.dirStatusHandler))))
		r.HandleFunc("/col/delete", ms.proxyToLeader(ms.guard.WhiteList(requestIDMiddleware(ms.collectionDeleteHandler))))
		r.HandleFunc("/vol/grow", ms.proxyToLeader(ms.guard.WhiteList(requestIDMiddleware(ms.volumeGrowHandler))))
		r.HandleFunc("/vol/status", ms.proxyToLeader(ms.guard.WhiteList(requestIDMiddleware(ms.volumeStatusHandler))))
		r.HandleFunc("/vol/vacuum", ms.proxyToLeader(ms.guard.WhiteList(requestIDMiddleware(ms.volumeVacuumHandler))))
		r.HandleFunc("/submit", ms.guard.WhiteList(requestIDMiddleware(ms.submitFromMasterServerHandler)))
		r.HandleFunc("/collection/info", ms.guard.WhiteList(requestIDMiddleware(ms.collectionInfoHandler)))
		/*
			r.HandleFunc("/stats/health", ms.guard.WhiteList(statsHealthHandler))
			r.HandleFunc("/stats/counter", ms.guard.WhiteList(statsCounterHandler))
			r.HandleFunc("/stats/memory", ms.guard.WhiteList(statsMemoryHandler))
		*/
		r.HandleFunc("/{fileId}", requestIDMiddleware(ms.redirectHandler))
	}

	ms.Topo.SetAdminServerConnectedFunc(ms.isAdminServerConnectedFunc)
	ms.Topo.StartRefreshWritableVolumes(
		ms.grpcDialOption,
		ms.option.GarbageThreshold,
		ms.option.MaxParallelVacuumPerServer,
		ms.option.MaxParallelVacuumPerRack,
		topology.VolumeGrowStrategy.Threshold,
		ms.preallocateSize,
		time.Duration(ms.option.VacuumIntervalSeconds)*time.Second,
		time.Duration(ms.option.VacuumDeleteEmptyAfterSeconds)*time.Second,
	)

	ms.ProcessGrowRequest()

	if !option.IsFollower {
		ms.startAdminScripts()
	}

	stats.MasterStartTimeSeconds.Set(float64(time.Now().Unix()))
	return ms
}

func (ms *MasterServer) healthzHandler(w http.ResponseWriter, r *http.Request) {
	// Liveness: process is alive. Keep this fast and simple.
	w.WriteHeader(http.StatusOK)
}

func (ms *MasterServer) readyzHandler(w http.ResponseWriter, r *http.Request) {
	// Readiness: check we can serve traffic. Answer from what raft knows now
	// rather than waiting out an election, so the probe's own timeout decides.
	leader, err := ms.Topo.MaybeLeader()
	if err != nil || leader == "" {
		w.WriteHeader(http.StatusServiceUnavailable)
		return
	}
	if ms.option.Master.Equals(leader) {
		isLocked, err := ms.Topo.IsChildLocked()
		if err != nil {
			glog.Errorf("readyzHandler: %+v", err)
		}
		if isLocked {
			w.WriteHeader(http.StatusLocked)
			return
		}
	}
	w.WriteHeader(http.StatusOK)
}

func (ms *MasterServer) SetRaftServer(raftServer *RaftServer) {
	var raftServerName string

	ms.Topo.RaftServerAccessLock.Lock()
	if raftServer.raftServer != nil {
		ms.Topo.RaftServer = raftServer.raftServer
		ms.Topo.RaftServer.AddEventListener(raft.LeaderChangeEventType, func(e raft.Event) {
			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).Infof("[%s] %s becomes leader.", ms.Topo.RaftServer.Name(), ms.Topo.RaftServer.Leader())
				ms.Topo.SetLastLeaderChangeTime(time.Now())
				if pb.ServerAddress(ms.Topo.RaftServer.Leader()).Equals(pb.ServerAddress(ms.Topo.RaftServer.Name())) {
					go ms.ensureTopologyId()
				}
			}
		})
		raftServerName = fmt.Sprintf("[%s]", ms.Topo.RaftServer.Name())
	} else if raftServer.RaftHashicorp != nil {
		ms.Topo.HashicorpRaft = raftServer.RaftHashicorp
		raftServerName = ms.Topo.HashicorpRaft.String()
	}
	ms.Topo.RaftServerAccessLock.Unlock()

	if ms.Topo.IsLeader() {
		// Seed the warmup timestamp so IsWarmingUp() is active even if the
		// leader change event hasn't fired yet (e.g. node is already leader
		// on startup). Followers don't need warmup state.
		ms.Topo.SetLastLeaderChangeTime(time.Now())
		glog.V(0).Infof("%s I am the leader!", raftServerName)
		go ms.ensureTopologyId()
	} else {
		var raftServerLeader string
		ms.Topo.RaftServerAccessLock.RLock()
		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)
	}
}

func (ms *MasterServer) syncRaftForTopologyId(topologyId string) error {
	ms.Topo.RaftServerAccessLock.RLock()
	defer ms.Topo.RaftServerAccessLock.RUnlock()

	if ms.Topo.RaftServer != nil {
		_, err := ms.Topo.RaftServer.Do(topology.NewMaxVolumeIdCommand(ms.Topo.GetMaxVolumeId(), topologyId))
		return err
	} else if ms.Topo.HashicorpRaft != nil {
		b, err := json.Marshal(topology.NewMaxVolumeIdCommand(ms.Topo.GetMaxVolumeId(), topologyId))
		if err != nil {
			return fmt.Errorf("failed marshal NewMaxVolumeIdCommand: %v", err)
		}
		if future := ms.Topo.HashicorpRaft.Apply(b, raftApplyTimeout); future.Error() != nil {
			return future.Error()
		}
		return nil
	}
	return fmt.Errorf("no raft server configured")
}

func (ms *MasterServer) ensureTopologyId() {
	ms.topologyIdGenLock.Lock()
	defer ms.topologyIdGenLock.Unlock()

	// Send a no-op command to ensure all previous logs are applied (barrier)
	// This handles the case where log replay is still in progress
	glog.V(1).Infof("ensureTopologyId: sending barrier command")
	for {
		if !ms.Topo.IsLeader() {
			glog.V(1).Infof("lost leadership while sending barrier command for topologyId")
			return
		}
		if err := ms.syncRaftForTopologyId(ms.Topo.GetTopologyId()); err != nil {
			glog.Errorf("failed to sync raft for topologyId: %v, retrying in 1s", err)
			time.Sleep(time.Second)
			continue
		}
		break
	}
	glog.V(1).Infof("ensureTopologyId: barrier command completed")

	if !ms.Topo.IsLeader() {
		return
	}

	currentId := ms.Topo.GetTopologyId()
	glog.V(1).Infof("ensureTopologyId: current TopologyId after barrier: %s", currentId)

	prevId := ms.Topo.GetTopologyId()

	EnsureTopologyId(ms.Topo, func() bool {
		return ms.Topo.IsLeader()
	}, func(topologyId string) error {
		return ms.syncRaftForTopologyId(topologyId)
	})

	// If a new TopologyId was generated, take a snapshot so it survives
	// raft state cleanup on future non-resume restarts.
	if prevId == "" && ms.Topo.GetTopologyId() != "" {
		ms.Topo.RaftServerAccessLock.RLock()
		if ms.Topo.RaftServer != nil {
			if err := ms.Topo.RaftServer.TakeSnapshot(); err != nil {
				glog.Warningf("snapshot after TopologyId generation: %v", err)
			} else {
				glog.V(0).Infof("snapshot taken to persist TopologyId %s", ms.Topo.GetTopologyId())
			}
		}
		// Hashicorp raft snapshots are handled automatically.
		ms.Topo.RaftServerAccessLock.RUnlock()
	}
}

func (ms *MasterServer) proxyToLeader(f http.HandlerFunc) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		if ms.Topo.IsLeader() {
			f(w, r)
			return
		}

		// get the current raft leader
		leaderAddr, _ := ms.Topo.MaybeLeader()
		raftServerLeader := leaderAddr.ToHttpAddress()
		if raftServerLeader == "" {
			f(w, r)
			return
		}

		// determine the scheme based on HTTPS client configuration
		scheme := util_http.GetGlobalHttpClient().GetHttpScheme()

		targetUrl, err := url.Parse(scheme + "://" + raftServerLeader)
		if err != nil {
			writeJsonError(w, r, http.StatusInternalServerError,
				fmt.Errorf("Leader URL %s://%s Parse Error: %v", scheme, raftServerLeader, err))
			return
		}

		// proxy to leader
		glog.V(4).Infoln("proxying to leader", raftServerLeader, "using", scheme)
		proxy := httputil.NewSingleHostReverseProxy(targetUrl)
		proxy.Transport = util_http.GetGlobalHttpClient().GetClientTransport()
		proxy.ServeHTTP(w, r)
	}
}

func (ms *MasterServer) isAdminServerConnectedFunc() bool {
	if ms == nil || ms.adminLocks == nil {
		return false
	}
	_, _, isLocked := ms.adminLocks.isLocked(cluster.AdminServerPresenceLockName)
	return isLocked
}

func (ms *MasterServer) startAdminScripts() {
	v := util.GetViper()
	adminScripts := v.GetString("master.maintenance.scripts")
	if adminScripts == "" {
		return
	}
	glog.V(0).Infof("adminScripts: %v", adminScripts)

	sleepMinutes := v.GetFloat64("master.maintenance.sleep_minutes")
	if sleepMinutes <= 0 {
		sleepMinutes = float64(maintenance.DefaultMaintenanceSleepMinutes)
	}

	scriptLines := strings.Split(adminScripts, "\n")
	if !strings.Contains(adminScripts, "lock") {
		scriptLines = append(append([]string{}, "lock"), scriptLines...)
		scriptLines = append(scriptLines, "unlock")
	}

	masterAddress := string(ms.option.Master)

	var shellOptions shell.ShellOptions
	shellOptions.GrpcDialOption = security.LoadClientTLS(v, "grpc.master")
	shellOptions.Masters = &masterAddress

	shellOptions.Directory = "/"
	emptyFilerGroup := ""
	shellOptions.FilerGroup = &emptyFilerGroup

	commandEnv := shell.NewCommandEnv(&shellOptions)

	reg, _ := regexp.Compile(`'.*?'|".*?"|\S+`)

	go commandEnv.MasterClient.KeepConnectedToMaster(context.Background())

	go func() {
		for {
			time.Sleep(time.Duration(sleepMinutes) * time.Minute)
			if ms.Topo.IsLeader() && ms.MasterClient.GetMaster(context.Background()) != "" {
				if ms.isAdminServerConnectedFunc() {
					glog.V(1).Infof("Skipping master maintenance scripts because admin server is connected")
					continue
				}
				shellOptions.FilerAddress = ms.GetOneFiler(cluster.FilerGroupName(*shellOptions.FilerGroup))
				if shellOptions.FilerAddress == "" {
					continue
				}
				for _, line := range scriptLines {
					for _, c := range strings.Split(line, ";") {
						processEachCmd(reg, c, commandEnv)
					}
				}
			}
		}
	}()
}

func processEachCmd(reg *regexp.Regexp, line string, commandEnv *shell.CommandEnv) {
	cmds := reg.FindAllString(line, -1)
	if len(cmds) == 0 {
		return
	}
	args := make([]string, len(cmds[1:]))
	for i := range args {
		args[i] = strings.Trim(string(cmds[1+i]), "\"'")
	}
	cmd := cmds[0]

	for _, c := range shell.Commands {
		if c.Name() == cmd {
			if c.HasTag(shell.ResourceHeavy) {
				glog.Warningf("%s is resource heavy and should not run on master", cmd)
				continue
			}
			glog.V(0).Infof("executing: %s %v", cmd, args)
			// noLock belongs to the invocation that set it, not to the
			// CommandEnv this loop reuses for every line of the script. Without
			// this, a dry run earlier in the script leaves the mutations after
			// it unlocked.
			commandEnv.SetNoLock(false)
			if err := c.Do(args, commandEnv, os.Stdout); err != nil {
				glog.V(0).Infof("error: %v", err)
			}
		}
	}
}

func (ms *MasterServer) createSequencer(option *MasterOption) sequence.Sequencer {
	var seq sequence.Sequencer
	v := util.GetViper()
	seqType := strings.ToLower(v.GetString(SequencerType))
	glog.V(1).Infof("[%s] : [%s]", SequencerType, seqType)
	switch strings.ToLower(seqType) {
	case "snowflake":
		var err error
		snowflakeId := v.GetInt(SequencerSnowflakeId)
		seq, err = sequence.NewSnowflakeSequencer(string(option.Master), snowflakeId)
		if err != nil {
			glog.Error(err)
			seq = nil
		}
	case "raft":
		fallthrough
	default:
		seq = sequence.NewMemorySequencer()
	}
	return seq
}

func (ms *MasterServer) OnPeerUpdate(update *master_pb.ClusterNodeUpdate, startFrom time.Time) {
	if update.NodeType != cluster.MasterType {
		return
	}
	glog.V(4).Infof("OnPeerUpdate: %+v", update)

	peerAddress := pb.ServerAddress(update.Address)
	if update.IsAdd {
		ms.AdmitRaftPeer(peerAddress)
		return
	}

	ms.Topo.RaftServerAccessLock.RLock()
	defer ms.Topo.RaftServerAccessLock.RUnlock()

	// goraft rebuilds its peer set from -peers on every start, so a departed
	// master is already out of the list there and would come back on the next
	// restart anyway; only hashicorp raft carries membership across restarts.
	if ms.Topo.HashicorpRaft == nil || ms.Topo.HashicorpRaft.State() != hashicorpRaft.Leader {
		return
	}
	// A master that is merely down is still a member: -peers is what declares
	// membership, and updatePeers reconciles the configuration against it on
	// every leadership change. Evicting one here would shrink the quorum behind
	// the operator's back, and a restart then races the eviction — the master
	// gets re-admitted, the removal lands after it, and it is left out of the
	// configuration with nobody left to vote it back in.
	for _, peer := range ms.MasterClient.GetMasters(context.Background()) {
		if peer.ToHttpAddress() == peerAddress.ToHttpAddress() {
			return
		}
	}

	peerName := raftServerID(peerAddress)
	pb.WithMasterClient(context.Background(), false, peerAddress, ms.grpcDialOption, true, func(client master_pb.SeaweedClient) error {
		ctx, cancel := context.WithTimeout(context.TODO(), 15*time.Second)
		defer cancel()
		if _, err := client.Ping(ctx, &master_pb.PingRequest{Target: string(peerAddress), TargetType: cluster.MasterType}); err != nil {
			glog.V(0).Infof("master %s didn't respond to pings. remove raft server", peerName)
			// We are the leader here, so drop the dead peer through the local
			// raft handle instead of dialing our own RaftRemoveServer RPC.
			if err := ms.Topo.HashicorpRaft.RemoveServer(hashicorpRaft.ServerID(peerName), 0, 0).Error(); err != nil {
				glog.Warningf("failed removing old raft server: %v", err)
				return err
			}
		} else {
			glog.V(0).Infof("master %s successfully responded to ping", peerName)
		}
		return nil
	})
}

// AdmitRaftPeer pulls a master into the raft quorum, if we are the leader and
// do not have it yet. A master that starts with no raft state cannot campaign
// under either implementation, so this is its only way in. The returned
// channel resolves with the admission result, or nil when no admission ran.
func (ms *MasterServer) AdmitRaftPeer(peerAddress pb.ServerAddress) <-chan error {
	if ms.option.SingleMaster || peerAddress.ToHttpAddress() == ms.option.Master.ToHttpAddress() {
		return nil
	}

	peerName, ok := ms.missingRaftPeerName(peerAddress)
	if !ok {
		return nil
	}
	admission := make(chan error, 1)
	if inFlight, alreadyAdmitting := ms.raftPeerAdmissions.LoadOrStore(peerName, admission); alreadyAdmitting {
		return inFlight.(chan error)
	}
	glog.V(0).Infof("adding new raft server: %s", peerName)
	// The join commits through raft, which waits on the other peers, so keep it
	// off the caller: this runs on the peer update stream and on the grpc
	// handler that a joining master is still blocked in.
	go func() {
		defer ms.raftPeerAdmissions.Delete(peerName)
		defer close(admission)
		if err := ms.raftAddServer(peerName, peerAddress.ToGrpcAddress(), true); err != nil {
			glog.Warningf("failed adding raft server %s: %v", peerName, err)
			admission <- err
			return
		}
		admission <- nil
	}()
	return admission
}

// missingRaftPeerName returns the raft id to admit peerAddress under, and
// whether this master is the leader and is missing that peer.
func (ms *MasterServer) missingRaftPeerName(peerAddress pb.ServerAddress) (string, bool) {
	ms.Topo.RaftServerAccessLock.RLock()
	defer ms.Topo.RaftServerAccessLock.RUnlock()

	if ms.Topo.RaftServer != nil {
		if ms.Topo.RaftServer.State() != raft.Leader {
			return "", false
		}
		// Peers are keyed by the name a master calls itself, which carries the
		// grpc port; match on the http address so a peer already known under
		// another spelling is not added twice.
		for name := range ms.Topo.RaftServer.Peers() {
			if pb.ServerAddress(name).ToHttpAddress() == peerAddress.ToHttpAddress() {
				return "", false
			}
		}
		return string(peerAddress), true
	}

	if ms.Topo.HashicorpRaft == nil || ms.Topo.HashicorpRaft.State() != hashicorpRaft.Leader {
		return "", false
	}
	peerName := raftServerID(peerAddress)
	for _, server := range ms.Topo.HashicorpRaft.GetConfiguration().Configuration().Servers {
		if string(server.ID) == peerName {
			return "", false
		}
	}
	return peerName, true
}

func (ms *MasterServer) Shutdown() {
	if ms.Topo == nil {
		return
	}
	ms.Topo.RaftServerAccessLock.RLock()
	raftServer := ms.Topo.RaftServer
	hashRaft := ms.Topo.HashicorpRaft
	ms.Topo.RaftServerAccessLock.RUnlock()

	if hashRaft != nil {
		if hashRaft.State() == hashicorpRaft.Leader {
			hashRaft.LeadershipTransfer()
		}
		hashRaft.Shutdown()
	}
	// Stop the goraft server too. Without this, the raft event loop
	// goroutine (leaderLoop/followerLoop) keeps running after the master
	// shuts down, leaking goroutines across test runs and occasionally
	// processing stale events that trigger safety assertions.
	if raftServer != nil {
		raftServer.Stop()
	}
}

func (ms *MasterServer) Reload() {
	glog.V(0).Infoln("Reload master server...")

	util.LoadConfiguration("security", false)
	v := util.GetViper()
	ms.guard.UpdateWhiteList(append(ms.option.WhiteList,
		util.StringSplit(v.GetString("guard.white_list"), ",")...),
	)
	ms.guard.UpdateSigningKeys(
		v.GetString("jwt.signing.key"),
		v.GetInt("jwt.signing.expires_after_seconds"),
		v.GetString("jwt.signing.read.key"),
		v.GetInt("jwt.signing.read.expires_after_seconds"),
	)
}

func parseReservationTimeout(v *util.ViperProxy) time.Duration {
	if str := strings.TrimSpace(v.GetString("master.volume_growth.reservation_timeout")); str != "" {
		if d, err := time.ParseDuration(str); err == nil && d > 0 {
			return d
		}
	}
	// A bare number in the config means seconds; GetDuration would read it as nanoseconds.
	if sec := v.GetInt("master.volume_growth.reservation_timeout"); sec > 0 {
		return time.Duration(sec) * time.Second
	}
	return topology.DefaultReservationTimeout
}