mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-27 23:40:41 +01:00
* master, volume, filer: start on a read-only mounted directory and serve existing data read-only A read-only mount was treated as writable (permission bits only), so every service failed on its first write: the master panicked writing raft conf.tmp, the filer exited opening the leveldb LOCK, and the volume server skipped every volume while still offering the directory as free capacity. - util.IsReadOnlyMount: access(W_OK) == EROFS. - volume: on a read-only -dir take the existing O_RDONLY/noWriteOrDelete load, index .idx in memory when no .sdx exists, offer no free slots (same branches as isDiskSpaceLow), skip the vid stamp there. - master: on a read-only -mdir run raft from a private copy and refuse new volume ids. - filer: leveldb2/leveldb3 set their existing ReadOnly option. * volume: skip read-only mounts when picking EC shard copy targets A shard copy landing on a read-only mount fails at write time; offer only writable locations like the other placement paths do. * volume: mirror the read-only mount handling in the Rust volume server The .dat open only fell back to read-only on EACCES, so a real read-only mount (EROFS) still failed the load. The index side already degrades once no_write_or_delete is set. Add is_read_only_mount via access(W_OK) == EROFS, mark DiskLocation.read_only_mount, and mirror the Go checks: no free slots, heartbeat max = used, no vid-stamp writes, no EC shard copy targets, and a read-only index dir forces the volume read-only. * volume: write vid stamps while only the index dir is read-only * filer: skip the write-bit check when a leveldb store opens read-only * filer: serve reads of buckets with no leveldb3 dir on a read-only store * volume: derive a stable dir uuid when a read-only mount cannot store one * master: keep the read-only raft copy across restarts so votes survive * volume: index a missing .idx in memory on a read-only mount * volume: release the .idx handle after an in-memory read-only load * volume: mount EC shards read-only when their index dir is on a read-only mount * volume: mark the index dir read-only before the tail repair opens .idx * volume: write vid stamps while only the index dir is read-only (Rust) * volume: derive a stable dir uuid and report no free slots on a read-only mount (Rust) * volume: index a missing .idx in memory on a read-only mount (Rust) * volume: refuse an EC copy whose shard owner is on a read-only disk * master: key the read-only raft copy by the absolute -mdir Two masters sharing the same relative -mdir text from different working directories would have collided on one temp copy and shared raft state. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume: mix device and host into the read-only dir uuid Deriving the uuid from the path alone let independent servers at the same mount point report one uuid, which RegisterUuids rejects. Hash in the device id and hostname so restarts stay stable but disks differ. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: vic6996-web <320442181+vic6996-web@users.noreply.github.com> Co-authored-by: user.email <chrislusf@users.noreply.github.com> Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
600 lines
23 KiB
Go
600 lines
23 KiB
Go
package command
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha1"
|
|
"crypto/tls"
|
|
"fmt"
|
|
"io/fs"
|
|
"math/rand/v2"
|
|
"net"
|
|
"net/http"
|
|
"os"
|
|
"path"
|
|
"path/filepath"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/util/version"
|
|
|
|
hashicorpRaft "github.com/hashicorp/raft"
|
|
|
|
"slices"
|
|
|
|
"github.com/gorilla/mux"
|
|
"github.com/seaweedfs/raft/protobuf"
|
|
"github.com/spf13/viper"
|
|
"google.golang.org/grpc/reflection"
|
|
|
|
stats_collect "github.com/seaweedfs/seaweedfs/weed/stats"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/util/grace"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/glog"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/security"
|
|
weed_server "github.com/seaweedfs/seaweedfs/weed/server"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/backend"
|
|
"github.com/seaweedfs/seaweedfs/weed/util"
|
|
)
|
|
|
|
var (
|
|
m MasterOptions
|
|
)
|
|
|
|
const (
|
|
raftJoinCheckDelay = 1500 * time.Millisecond // delay before checking if we should join a raft cluster
|
|
)
|
|
|
|
type MasterOptions struct {
|
|
port *int
|
|
portGrpc *int
|
|
ip *string
|
|
ipBind *string
|
|
listenOnLocalhost *bool
|
|
metaFolder *string
|
|
peers *string
|
|
mastersDeprecated *string // deprecated, for backward compatibility in master.follower
|
|
volumeSizeLimitMB *uint
|
|
fileSizeLimitMB *int
|
|
volumePreallocate *bool
|
|
maxParallelVacuumPerServer *int
|
|
vacuumIntervalSeconds *int
|
|
vacuumDeleteEmptyAfterSeconds *int
|
|
// pulseSeconds *int
|
|
defaultReplication *string
|
|
garbageThreshold *float64
|
|
whiteList *string
|
|
disableHttp *bool
|
|
metricsAddress *string
|
|
metricsIntervalSec *int
|
|
raftResumeState *bool
|
|
metricsHttpPort *int
|
|
metricsHttpIp *string
|
|
heartbeatInterval *time.Duration
|
|
electionTimeout *time.Duration
|
|
raftHashicorp *bool
|
|
raftBootstrap *bool
|
|
telemetryUrl *string
|
|
telemetryEnabled *bool
|
|
debug *bool
|
|
debugPort *int
|
|
// shutdownCtx, when non-nil, tells startMaster to shut down once the ctx
|
|
// is cancelled. Used by integration tests and by weed mini; nil for
|
|
// standalone weed master.
|
|
shutdownCtx context.Context
|
|
// releaseAllocOnFirstReport lifts the post-mint allocation hold as soon as
|
|
// one full volume-server inventory has been applied. weed mini sets it in
|
|
// single-master mode, with singleReporterAddr pinning the embedded volume
|
|
// server as the only admitted reporter: any other volume server is refused,
|
|
// so the warmup window after a fresh topology id is dead time.
|
|
// Standalone weed master / weed server can face late-arriving veterans and
|
|
// must keep the window.
|
|
releaseAllocOnFirstReport bool
|
|
singleReporterAddr string
|
|
}
|
|
|
|
func init() {
|
|
cmdMaster.Run = runMaster // break init cycle
|
|
m.port = cmdMaster.Flag.Int("port", 9333, "http listen port")
|
|
m.portGrpc = cmdMaster.Flag.Int("port.grpc", 0, "grpc listen port")
|
|
m.ip = cmdMaster.Flag.String("ip", util.DetectedHostAddress(), "master <ip>|<server> address, also used as identifier")
|
|
m.ipBind = cmdMaster.Flag.String("ip.bind", "", "ip address to bind to. If empty, default to same as -ip option.")
|
|
m.metaFolder = cmdMaster.Flag.String("mdir", os.TempDir(), "data directory to store meta data")
|
|
m.peers = cmdMaster.Flag.String("peers", "", "all master nodes in comma separated ip:port list, example: 127.0.0.1:9093,127.0.0.1:9094,127.0.0.1:9095; use 'none' for single-master mode")
|
|
m.volumeSizeLimitMB = cmdMaster.Flag.Uint("volumeSizeLimitMB", util.DefaultVolumeSizeLimitMB, "Master stops directing writes to oversized volumes.")
|
|
m.fileSizeLimitMB = cmdMaster.Flag.Int("fileSizeLimitMB", 256, "limit the file size accepted by /submit, should match the volume servers' -fileSizeLimitMB (-volume.fileSizeLimitMB under weed server or weed mini, which set this for you)")
|
|
m.volumePreallocate = cmdMaster.Flag.Bool("volumePreallocate", false, "Preallocate disk space for volumes.")
|
|
m.maxParallelVacuumPerServer = cmdMaster.Flag.Int("maxParallelVacuumPerServer", 1, "maximum number of volumes to vacuum in parallel per volume server; 0 for no limit (deprecated, use master.vacuum.max_parallel_per_server in master.toml)")
|
|
m.vacuumIntervalSeconds = cmdMaster.Flag.Int("vacuumIntervalSeconds", 840, "seconds between automatic vacuum sweeps")
|
|
m.vacuumDeleteEmptyAfterSeconds = cmdMaster.Flag.Int("vacuumDeleteEmptyAfterSeconds", 0, "automatic sweep deletes volume copies that stay empty this many seconds; 0 disables")
|
|
// m.pulseSeconds = cmdMaster.Flag.Int("pulseSeconds", 5, "number of seconds between heartbeats")
|
|
m.defaultReplication = cmdMaster.Flag.String("defaultReplication", "", "Default replication type if not specified.")
|
|
m.garbageThreshold = cmdMaster.Flag.Float64("garbageThreshold", 0.3, "threshold to vacuum and reclaim spaces")
|
|
m.whiteList = cmdMaster.Flag.String("whiteList", "", "comma separated Ip addresses having write permission. No limit if empty.")
|
|
m.disableHttp = cmdMaster.Flag.Bool("disableHttp", false, "disable http requests, only gRPC operations are allowed.")
|
|
m.metricsAddress = cmdMaster.Flag.String("metrics.address", "", "Prometheus gateway address <host>:<port>")
|
|
m.metricsIntervalSec = cmdMaster.Flag.Int("metrics.intervalSeconds", 15, "Prometheus push interval in seconds")
|
|
m.metricsHttpPort = cmdMaster.Flag.Int("metricsPort", 0, "Prometheus metrics listen port")
|
|
m.metricsHttpIp = cmdMaster.Flag.String("metricsIp", "", "metrics listen ip. If empty, default to same as -ip.bind option.")
|
|
m.raftResumeState = cmdMaster.Flag.Bool("resumeState", true, "resume previous state on start master server")
|
|
m.heartbeatInterval = cmdMaster.Flag.Duration("heartbeatInterval", 300*time.Millisecond, "heartbeat interval of master servers, and will be randomly multiplied by [1, 1.25)")
|
|
m.electionTimeout = cmdMaster.Flag.Duration("electionTimeout", 10*time.Second, "election timeout of master servers")
|
|
m.raftHashicorp = cmdMaster.Flag.Bool("raftHashicorp", false, "use hashicorp raft")
|
|
m.raftBootstrap = cmdMaster.Flag.Bool("raftBootstrap", false, "deprecated and ignored: the first master in -peers mints the Raft cluster on its own once it sees no leader anywhere")
|
|
m.telemetryUrl = cmdMaster.Flag.String("telemetry.url", "https://telemetry.seaweedfs.com/api/collect", "telemetry server URL to send usage statistics")
|
|
m.telemetryEnabled = cmdMaster.Flag.Bool("telemetry", true, "report anonymous cluster statistics to telemetry.url, use -telemetry=false to opt out")
|
|
m.debug = cmdMaster.Flag.Bool("debug", false, "serves runtime profiling data via pprof on the port specified by -debug.port")
|
|
m.debugPort = cmdMaster.Flag.Int("debug.port", 6060, "http port for debugging")
|
|
}
|
|
|
|
var cmdMaster = &Command{
|
|
UsageLine: "master -port=9333",
|
|
Short: "start a master server",
|
|
Long: `start a master server to provide volume=>location mapping service and sequence number of file ids
|
|
|
|
The configuration file "security.toml" is read from ".", "$HOME/.seaweedfs/", "/usr/local/etc/seaweedfs/", or "/etc/seaweedfs/", in that order.
|
|
|
|
The example security.toml configuration file can be generated by "weed scaffold -config=security"
|
|
|
|
For single-master setups, use -peers=none to skip Raft quorum wait and enable instant startup.
|
|
This is ideal for development or standalone deployments.
|
|
|
|
`,
|
|
}
|
|
|
|
var (
|
|
masterCpuProfile = cmdMaster.Flag.String("cpuprofile", "", "cpu profile output file")
|
|
masterMemProfile = cmdMaster.Flag.String("memprofile", "", "memory profile output file")
|
|
)
|
|
|
|
func runMaster(cmd *Command, args []string) bool {
|
|
if *m.debug {
|
|
grace.StartDebugServer(*m.debugPort)
|
|
}
|
|
|
|
util.LoadSecurityConfiguration()
|
|
util.LoadConfiguration("master", false)
|
|
|
|
// bind viper configuration to command line flags
|
|
if v := util.GetViper().GetString("master.mdir"); v != "" {
|
|
*m.metaFolder = v
|
|
}
|
|
|
|
*m.metaFolder = util.ResolvePath(*m.metaFolder)
|
|
*masterCpuProfile = util.ResolvePath(*masterCpuProfile)
|
|
*masterMemProfile = util.ResolvePath(*masterMemProfile)
|
|
grace.SetupProfiling(*masterCpuProfile, *masterMemProfile)
|
|
|
|
parent, _ := util.FullPath(*m.metaFolder).DirAndName()
|
|
if util.FileExists(string(parent)) && !util.FileExists(*m.metaFolder) {
|
|
if err := os.MkdirAll(*m.metaFolder, 0755); err != nil {
|
|
glog.Fatalf("Could not create Meta Folder %s: %v", *m.metaFolder, err)
|
|
}
|
|
}
|
|
if err := util.TestFolderWritable(*m.metaFolder); err != nil {
|
|
glog.Fatalf("Check Meta Folder (-mdir) Writable %s : %s", *m.metaFolder, err)
|
|
}
|
|
|
|
masterWhiteList := util.StringSplit(*m.whiteList, ",")
|
|
if *m.volumeSizeLimitMB > util.MaxVolumeSizeLimitMB {
|
|
glog.Fatalf("volumeSizeLimitMB should not exceed %d", util.MaxVolumeSizeLimitMB)
|
|
}
|
|
|
|
switch {
|
|
case *m.metricsHttpIp != "":
|
|
// noting to do, use m.metricsHttpIp
|
|
case *m.ipBind != "":
|
|
*m.metricsHttpIp = *m.ipBind
|
|
case *m.ip != "":
|
|
*m.metricsHttpIp = *m.ip
|
|
}
|
|
go stats_collect.StartMetricsServer(*m.metricsHttpIp, *m.metricsHttpPort)
|
|
go stats_collect.LoopPushingMetric("masterServer", util.JoinHostPort(*m.ip, *m.port), *m.metricsAddress, *m.metricsIntervalSec)
|
|
startMaster(m, masterWhiteList)
|
|
return true
|
|
}
|
|
|
|
func startMaster(masterOption MasterOptions, masterWhiteList []string) {
|
|
|
|
backend.LoadConfiguration(util.GetViper())
|
|
|
|
if *masterOption.portGrpc == 0 {
|
|
*masterOption.portGrpc = 10000 + *masterOption.port
|
|
}
|
|
// The loopback companion listener is only kept while -ip.bind is unset
|
|
// and the bind address falls back to -ip. An explicit -ip.bind pins each
|
|
// port to exactly that address.
|
|
listenOnLocalhost := *masterOption.ipBind == ""
|
|
if masterOption.listenOnLocalhost != nil {
|
|
listenOnLocalhost = *masterOption.listenOnLocalhost
|
|
}
|
|
if *masterOption.ipBind == "" {
|
|
*masterOption.ipBind = *masterOption.ip
|
|
}
|
|
util.SetOutboundLocalIP(*masterOption.ipBind)
|
|
|
|
myMasterAddress, peers := checkPeers(*masterOption.ip, *masterOption.port, *masterOption.portGrpc, *masterOption.peers)
|
|
|
|
masterPeers := make(map[string]pb.ServerAddress)
|
|
for _, peer := range peers {
|
|
masterPeers[string(peer)] = peer
|
|
}
|
|
|
|
r := mux.NewRouter()
|
|
ms := weed_server.NewMasterServer(r, masterOption.toMasterOption(masterWhiteList), masterPeers)
|
|
listeningAddress := util.JoinHostPort(*masterOption.ipBind, *masterOption.port)
|
|
glog.V(0).Infof("Start Seaweed Master %s at %s", version.Version(), listeningAddress)
|
|
masterListener, masterLocalListener, e := util.NewIpAndLocalListeners(*masterOption.ipBind, *masterOption.port, 0, listenOnLocalhost)
|
|
if e != nil {
|
|
glog.Fatalf("Master startup error: %v", e)
|
|
}
|
|
|
|
// start raftServer
|
|
metaFolder := util.ResolvePath(*masterOption.metaFolder)
|
|
metaDir := path.Join(metaFolder, fmt.Sprintf("m%d", *masterOption.port))
|
|
if util.IsReadOnlyMount(metaFolder) {
|
|
// raft rewrites conf/state even to start. Run it from a private copy and
|
|
// allocate no volume ids, so nothing is lost when the copy is dropped.
|
|
roCopy, err := copyDirForReadOnly(metaDir)
|
|
if err != nil {
|
|
glog.Fatalf("-mdir %s is read-only and could not be copied: %v", *masterOption.metaFolder, err)
|
|
}
|
|
glog.V(0).Infof("-mdir %s is on a read-only filesystem: raft runs from %s, volume growth disabled", *masterOption.metaFolder, roCopy)
|
|
metaDir = roCopy
|
|
ms.Topo.SetMetaReadOnly(true)
|
|
}
|
|
|
|
isSingleMaster := isSingleMasterMode(*masterOption.peers)
|
|
|
|
if *masterOption.raftBootstrap {
|
|
glog.V(0).Infof("-raftBootstrap is ignored: masters mint a cluster on their own when no peer has a leader, and never over existing raft state")
|
|
}
|
|
|
|
raftServerOption := &weed_server.RaftServerOption{
|
|
GrpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.master"),
|
|
Peers: masterPeers,
|
|
ServerAddr: myMasterAddress,
|
|
DataDir: metaDir,
|
|
Topo: ms.Topo,
|
|
RaftResumeState: *masterOption.raftResumeState,
|
|
SingleMaster: isSingleMaster,
|
|
HeartbeatInterval: *masterOption.heartbeatInterval,
|
|
ElectionTimeout: *masterOption.electionTimeout,
|
|
}
|
|
var raftServer *weed_server.RaftServer
|
|
var err error
|
|
if *masterOption.raftHashicorp {
|
|
if raftServer, err = weed_server.NewHashicorpRaftServer(raftServerOption); err != nil {
|
|
glog.Fatalf("NewHashicorpRaftServer: %s", err)
|
|
}
|
|
} else {
|
|
raftServer, err = weed_server.NewRaftServer(raftServerOption)
|
|
if raftServer == nil {
|
|
glog.Fatalf("please verify %s is writable, see https://github.com/seaweedfs/seaweedfs/issues/717: %s", *masterOption.metaFolder, err)
|
|
}
|
|
}
|
|
// For single-master mode with a fresh log, initialize cluster immediately.
|
|
// When resuming with existing state, the server is already a member and
|
|
// will self-elect via fastResume — sending another JoinCommand would block
|
|
// because goraft's setCommitIndex returns early on JoinCommand entries,
|
|
// preventing the new entry's event from being notified when old uncommitted
|
|
// JoinCommands exist in the log.
|
|
if isSingleMaster && !raftServer.HasExistingState() {
|
|
glog.V(0).Infof("Single-master mode: initializing cluster immediately")
|
|
if err := raftServer.Bootstrap(); err != nil {
|
|
glog.Errorf("fail to bootstrap the cluster: %v", err)
|
|
}
|
|
}
|
|
ms.SetRaftServer(raftServer)
|
|
r.HandleFunc("/cluster/status", raftServer.StatusHandler).Methods(http.MethodGet, http.MethodHead)
|
|
r.HandleFunc("/cluster/healthz", raftServer.HealthzHandler).Methods(http.MethodGet, http.MethodHead)
|
|
if *masterOption.raftHashicorp {
|
|
r.HandleFunc("/raft/stats", raftServer.StatsRaftHandler).Methods(http.MethodGet)
|
|
}
|
|
// starting grpc server
|
|
grpcPort := *masterOption.portGrpc
|
|
grpcL, grpcLocalL, err := util.NewIpAndLocalListeners(*masterOption.ipBind, grpcPort, 0, listenOnLocalhost)
|
|
if err != nil {
|
|
glog.Fatalf("master failed to listen on grpc port %d: %v", grpcPort, err)
|
|
}
|
|
grpcS := pb.NewGrpcServer(security.LoadServerTLS(util.GetViper(), "grpc.master"))
|
|
master_pb.RegisterSeaweedServer(grpcS, ms)
|
|
if *masterOption.raftHashicorp {
|
|
raftServer.TransportManager.Register(grpcS)
|
|
} else {
|
|
protobuf.RegisterRaftServer(grpcS, raftServer)
|
|
}
|
|
reflection.Register(grpcS)
|
|
glog.V(0).Infof("Start Seaweed Master %s grpc server at %s:%d", version.Version(), *masterOption.ipBind, grpcPort)
|
|
if grpcLocalL != nil {
|
|
go grpcS.Serve(grpcLocalL)
|
|
}
|
|
go grpcS.Serve(grpcL)
|
|
pb.ServeGrpcOnLocalSocket(grpcS, grpcPort)
|
|
|
|
// A master that starts with no raft state cannot elect on its own — neither
|
|
// raft implementation lets a server outside the configuration campaign — so
|
|
// it has to be pulled in by a leader. Keep asking the peers who the leader is
|
|
// until we are in: the leader admits us once our master client registers, and
|
|
// only when nobody has one does the first peer mint a new cluster. Keeping
|
|
// that one peer the sole authority is what stops a partition from minting
|
|
// two clusters. Restarting a master alone, or scaling the peer list up,
|
|
// both land here.
|
|
if !isSingleMaster {
|
|
go func() {
|
|
// Stagger bootstrap by peer index so masters don't all check
|
|
// simultaneously. Peer 0 waits ~1.5s, peer 1 ~3s, etc.
|
|
idx := peerIndex(myMasterAddress, peers)
|
|
delay := time.Duration(float64(raftJoinCheckDelay) * (rand.Float64()*0.25 + 1) * float64(idx+1))
|
|
glog.V(0).Infof("bootstrap check in %v (peer index %d of %d)", delay, idx, len(peers))
|
|
time.Sleep(delay)
|
|
|
|
for {
|
|
if raftServer.HasExistingState() {
|
|
return
|
|
}
|
|
if leader, err := ms.Topo.MaybeLeader(); err == nil && leader != "" {
|
|
return
|
|
}
|
|
if existingLeader := ms.MasterClient.FindLeaderFromOtherPeers(myMasterAddress); existingLeader != "" {
|
|
glog.V(0).Infof("waiting to be admitted by existing leader %s", existingLeader)
|
|
} else if idx == 0 {
|
|
if err := raftServer.Bootstrap(); err != nil {
|
|
glog.Errorf("fail to bootstrap the cluster: %v", err)
|
|
}
|
|
} else {
|
|
glog.V(0).Infof("skip bootstrap: %v is not the first master in peers (index %d)", myMasterAddress, idx)
|
|
}
|
|
time.Sleep(raftJoinCheckDelay)
|
|
}
|
|
}()
|
|
}
|
|
|
|
go ms.MasterClient.KeepConnectedToMaster(context.Background())
|
|
|
|
// start http server
|
|
var (
|
|
clientCertFile,
|
|
certFile,
|
|
keyFile string
|
|
)
|
|
useTLS := false
|
|
useMTLS := false
|
|
|
|
if viper.GetString("https.master.key") != "" {
|
|
useTLS = true
|
|
certFile = viper.GetString("https.master.cert")
|
|
keyFile = viper.GetString("https.master.key")
|
|
}
|
|
|
|
if viper.GetString("https.master.ca") != "" {
|
|
useMTLS = true
|
|
clientCertFile = viper.GetString("https.master.ca")
|
|
}
|
|
|
|
if masterLocalListener != nil {
|
|
go newHttpServer(r, nil).Serve(masterLocalListener)
|
|
}
|
|
|
|
var tlsConfig *tls.Config
|
|
if useMTLS {
|
|
tlsConfig = security.LoadClientTLSHTTP(clientCertFile)
|
|
if err := security.FixTlsConfig(util.GetViper(), tlsConfig); err != nil {
|
|
glog.Fatalf("failed to fix TLS config: %v", err)
|
|
}
|
|
}
|
|
|
|
if useTLS {
|
|
getCert, certProvider, err := security.NewReloadingServerCertificate(certFile, keyFile)
|
|
if err != nil {
|
|
glog.Fatalf("failed to load master HTTPS certificate: %v", err)
|
|
}
|
|
// Master runs ServeTLS in a goroutine and this function then blocks
|
|
// on shutdownCtx / select{}; tie the pem refresh goroutine to the
|
|
// existing interrupt hook instead of a local defer that would fire
|
|
// while the server is still running.
|
|
grace.OnInterrupt(certProvider.Close)
|
|
if tlsConfig == nil {
|
|
tlsConfig = &tls.Config{}
|
|
}
|
|
tlsConfig.GetCertificate = getCert
|
|
go newHttpServer(r, tlsConfig).ServeTLS(masterListener, "", "")
|
|
} else {
|
|
go newHttpServer(r, nil).Serve(masterListener)
|
|
}
|
|
|
|
grace.OnInterrupt(ms.Shutdown)
|
|
grace.OnInterrupt(grpcS.Stop)
|
|
grace.OnReload(ms.Reload)
|
|
grace.OnReload(func() {
|
|
if ms.Topo.HashicorpRaft != nil && ms.Topo.HashicorpRaft.State() == hashicorpRaft.Leader {
|
|
ms.Topo.HashicorpRaft.LeadershipTransfer()
|
|
}
|
|
})
|
|
if masterOption.shutdownCtx != nil {
|
|
<-masterOption.shutdownCtx.Done()
|
|
ms.Shutdown()
|
|
grpcS.Stop()
|
|
} else {
|
|
select {}
|
|
}
|
|
}
|
|
|
|
// copyDirForReadOnly copies the master's raft directory (a few KB) into a
|
|
// writable directory under os.TempDir keyed by src, and returns its path.
|
|
// A previous copy is reused so a restart keeps the votes and committed
|
|
// entries the frozen -mdir could not record; it is rebuilt only when the
|
|
// on-disk state is newer, i.e. the mount was writable in between. A missing
|
|
// src yields an empty copy.
|
|
func copyDirForReadOnly(src string) (string, error) {
|
|
// Key by the absolute path: a relative -mdir from different working
|
|
// directories must not collide on one copy.
|
|
abs, err := filepath.Abs(src)
|
|
if err != nil {
|
|
abs = src
|
|
}
|
|
sum := sha1.Sum([]byte(abs))
|
|
dst := filepath.Join(os.TempDir(), fmt.Sprintf("weed-master-ro-%x", sum[:8]))
|
|
if fi, err := os.Stat(dst); err == nil && fi.IsDir() &&
|
|
fi.Mode().Perm() == 0700 && util.DirOwnedBySelf(fi) {
|
|
if !newestMtime(dst).Before(newestMtime(src)) {
|
|
return dst, nil
|
|
}
|
|
os.RemoveAll(dst)
|
|
}
|
|
staging, err := os.MkdirTemp("", "weed-master-ro-")
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
err = filepath.WalkDir(src, func(p string, d fs.DirEntry, walkErr error) error {
|
|
if walkErr != nil {
|
|
if p == src && os.IsNotExist(walkErr) {
|
|
return nil
|
|
}
|
|
return walkErr
|
|
}
|
|
rel, err := filepath.Rel(src, p)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
target := filepath.Join(staging, rel)
|
|
if d.IsDir() {
|
|
return os.MkdirAll(target, 0700)
|
|
}
|
|
data, err := os.ReadFile(p)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return os.WriteFile(target, data, 0600)
|
|
})
|
|
if err != nil {
|
|
os.RemoveAll(staging)
|
|
return "", err
|
|
}
|
|
if err = os.Rename(staging, dst); err != nil {
|
|
return staging, nil
|
|
}
|
|
return dst, nil
|
|
}
|
|
|
|
// newestMtime returns the latest modification time under dir, or the zero
|
|
// time for a missing or empty one.
|
|
func newestMtime(dir string) (latest time.Time) {
|
|
filepath.WalkDir(dir, func(_ string, d fs.DirEntry, err error) error {
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
if info, statErr := d.Info(); statErr == nil && info.ModTime().After(latest) {
|
|
latest = info.ModTime()
|
|
}
|
|
return nil
|
|
})
|
|
return latest
|
|
}
|
|
|
|
func isSingleMasterMode(peers string) bool {
|
|
p := strings.ToLower(strings.TrimSpace(peers))
|
|
return p == "none"
|
|
}
|
|
|
|
func checkPeers(masterIp string, masterPort int, masterGrpcPort int, peers string) (masterAddress pb.ServerAddress, cleanedPeers []pb.ServerAddress) {
|
|
glog.V(0).Infof("current: %s:%d peers:%s", masterIp, masterPort, peers)
|
|
masterAddress = pb.NewServerAddress(masterIp, masterPort, masterGrpcPort)
|
|
|
|
// Handle special case: -peers=none for single-master setup
|
|
if isSingleMasterMode(peers) {
|
|
glog.V(0).Infof("Running in single-master mode (peers=none), no quorum required")
|
|
cleanedPeers = []pb.ServerAddress{masterAddress}
|
|
return
|
|
}
|
|
|
|
peers = strings.TrimSpace(peers)
|
|
seenPeers := make(map[string]struct{})
|
|
for _, peer := range pb.ServerAddresses(peers).ToAddresses() {
|
|
normalizedPeer := normalizeMasterPeerAddress(peer, masterAddress)
|
|
key := string(normalizedPeer)
|
|
if _, found := seenPeers[key]; found {
|
|
continue
|
|
}
|
|
seenPeers[key] = struct{}{}
|
|
cleanedPeers = append(cleanedPeers, normalizedPeer)
|
|
}
|
|
|
|
hasSelf := false
|
|
for _, peer := range cleanedPeers {
|
|
if peer.ToHttpAddress() == masterAddress.ToHttpAddress() {
|
|
hasSelf = true
|
|
break
|
|
}
|
|
}
|
|
|
|
if !hasSelf {
|
|
cleanedPeers = append(cleanedPeers, masterAddress)
|
|
}
|
|
if len(cleanedPeers)%2 == 0 {
|
|
glog.Fatalf("Only odd number of masters are supported: %+v", cleanedPeers)
|
|
}
|
|
return
|
|
}
|
|
|
|
func normalizeMasterPeerAddress(peer pb.ServerAddress, self pb.ServerAddress) pb.ServerAddress {
|
|
if peer.ToHttpAddress() == self.ToHttpAddress() {
|
|
return self
|
|
}
|
|
|
|
_, grpcPort, err := net.SplitHostPort(peer.ToGrpcAddress())
|
|
if err != nil {
|
|
return peer
|
|
}
|
|
grpcPortValue, err := strconv.Atoi(grpcPort)
|
|
if err != nil {
|
|
return peer
|
|
}
|
|
|
|
return pb.NewServerAddressWithGrpcPort(peer.ToHttpAddress(), grpcPortValue)
|
|
}
|
|
|
|
// peerIndex returns the 0-based position of self in the sorted peer list.
|
|
// Peer 0 is the designated bootstrap node. Returns -1 if self is not found.
|
|
func peerIndex(self pb.ServerAddress, peers []pb.ServerAddress) int {
|
|
slices.SortFunc(peers, func(a, b pb.ServerAddress) int {
|
|
return strings.Compare(a.ToHttpAddress(), b.ToHttpAddress())
|
|
})
|
|
for i, peer := range peers {
|
|
if peer.ToHttpAddress() == self.ToHttpAddress() {
|
|
return i
|
|
}
|
|
}
|
|
glog.Warningf("peerIndex: self %s not found in peers %v", self, peers)
|
|
return -1
|
|
}
|
|
|
|
func (m *MasterOptions) toMasterOption(whiteList []string) *weed_server.MasterOption {
|
|
masterAddress := pb.NewServerAddress(*m.ip, *m.port, *m.portGrpc)
|
|
return &weed_server.MasterOption{
|
|
Master: masterAddress,
|
|
MetaFolder: *m.metaFolder,
|
|
VolumeSizeLimitMB: uint32(*m.volumeSizeLimitMB),
|
|
FileSizeLimitMB: *m.fileSizeLimitMB,
|
|
VolumePreallocate: *m.volumePreallocate,
|
|
MaxParallelVacuumPerServer: *m.maxParallelVacuumPerServer,
|
|
VacuumIntervalSeconds: *m.vacuumIntervalSeconds,
|
|
VacuumDeleteEmptyAfterSeconds: *m.vacuumDeleteEmptyAfterSeconds,
|
|
// PulseSeconds: *m.pulseSeconds,
|
|
DefaultReplicaPlacement: *m.defaultReplication,
|
|
GarbageThreshold: *m.garbageThreshold,
|
|
WhiteList: whiteList,
|
|
DisableHttp: *m.disableHttp,
|
|
MetricsAddress: *m.metricsAddress,
|
|
MetricsIntervalSec: *m.metricsIntervalSec,
|
|
TelemetryUrl: *m.telemetryUrl,
|
|
TelemetryEnabled: *m.telemetryEnabled,
|
|
SingleMaster: isSingleMasterMode(*m.peers),
|
|
ReleaseAllocOnFirstReport: m.releaseAllocOnFirstReport,
|
|
SingleVolumeReporter: m.singleReporterAddr,
|
|
}
|
|
}
|