master.go 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315
  1. package command
  2. import (
  3. "fmt"
  4. "net/http"
  5. "os"
  6. "path"
  7. "strings"
  8. "time"
  9. hashicorpRaft "github.com/hashicorp/raft"
  10. "golang.org/x/exp/slices"
  11. "github.com/gorilla/mux"
  12. "github.com/seaweedfs/raft/protobuf"
  13. "github.com/spf13/viper"
  14. "google.golang.org/grpc/reflection"
  15. stats_collect "github.com/seaweedfs/seaweedfs/weed/stats"
  16. "github.com/seaweedfs/seaweedfs/weed/util/grace"
  17. "github.com/seaweedfs/seaweedfs/weed/glog"
  18. "github.com/seaweedfs/seaweedfs/weed/pb"
  19. "github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
  20. "github.com/seaweedfs/seaweedfs/weed/security"
  21. weed_server "github.com/seaweedfs/seaweedfs/weed/server"
  22. "github.com/seaweedfs/seaweedfs/weed/storage/backend"
  23. "github.com/seaweedfs/seaweedfs/weed/util"
  24. )
  25. var (
  26. m MasterOptions
  27. )
  28. type MasterOptions struct {
  29. port *int
  30. portGrpc *int
  31. ip *string
  32. ipBind *string
  33. metaFolder *string
  34. peers *string
  35. volumeSizeLimitMB *uint
  36. volumePreallocate *bool
  37. // pulseSeconds *int
  38. defaultReplication *string
  39. garbageThreshold *float64
  40. whiteList *string
  41. disableHttp *bool
  42. metricsAddress *string
  43. metricsIntervalSec *int
  44. raftResumeState *bool
  45. metricsHttpPort *int
  46. heartbeatInterval *time.Duration
  47. electionTimeout *time.Duration
  48. raftHashicorp *bool
  49. raftBootstrap *bool
  50. }
  51. func init() {
  52. cmdMaster.Run = runMaster // break init cycle
  53. m.port = cmdMaster.Flag.Int("port", 9333, "http listen port")
  54. m.portGrpc = cmdMaster.Flag.Int("port.grpc", 0, "grpc listen port")
  55. m.ip = cmdMaster.Flag.String("ip", util.DetectedHostAddress(), "master <ip>|<server> address, also used as identifier")
  56. m.ipBind = cmdMaster.Flag.String("ip.bind", "", "ip address to bind to. If empty, default to same as -ip option.")
  57. m.metaFolder = cmdMaster.Flag.String("mdir", os.TempDir(), "data directory to store meta data")
  58. 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")
  59. m.volumeSizeLimitMB = cmdMaster.Flag.Uint("volumeSizeLimitMB", 30*1000, "Master stops directing writes to oversized volumes.")
  60. m.volumePreallocate = cmdMaster.Flag.Bool("volumePreallocate", false, "Preallocate disk space for volumes.")
  61. // m.pulseSeconds = cmdMaster.Flag.Int("pulseSeconds", 5, "number of seconds between heartbeats")
  62. m.defaultReplication = cmdMaster.Flag.String("defaultReplication", "", "Default replication type if not specified.")
  63. m.garbageThreshold = cmdMaster.Flag.Float64("garbageThreshold", 0.3, "threshold to vacuum and reclaim spaces")
  64. m.whiteList = cmdMaster.Flag.String("whiteList", "", "comma separated Ip addresses having write permission. No limit if empty.")
  65. m.disableHttp = cmdMaster.Flag.Bool("disableHttp", false, "disable http requests, only gRPC operations are allowed.")
  66. m.metricsAddress = cmdMaster.Flag.String("metrics.address", "", "Prometheus gateway address <host>:<port>")
  67. m.metricsIntervalSec = cmdMaster.Flag.Int("metrics.intervalSeconds", 15, "Prometheus push interval in seconds")
  68. m.metricsHttpPort = cmdMaster.Flag.Int("metricsPort", 0, "Prometheus metrics listen port")
  69. m.raftResumeState = cmdMaster.Flag.Bool("resumeState", false, "resume previous state on start master server")
  70. m.heartbeatInterval = cmdMaster.Flag.Duration("heartbeatInterval", 300*time.Millisecond, "heartbeat interval of master servers, and will be randomly multiplied by [1, 1.25)")
  71. m.electionTimeout = cmdMaster.Flag.Duration("electionTimeout", 10*time.Second, "election timeout of master servers")
  72. m.raftHashicorp = cmdMaster.Flag.Bool("raftHashicorp", false, "use hashicorp raft")
  73. m.raftBootstrap = cmdMaster.Flag.Bool("raftBootstrap", false, "Whether to bootstrap the Raft cluster")
  74. }
  75. var cmdMaster = &Command{
  76. UsageLine: "master -port=9333",
  77. Short: "start a master server",
  78. Long: `start a master server to provide volume=>location mapping service and sequence number of file ids
  79. The configuration file "security.toml" is read from ".", "$HOME/.seaweedfs/", "/usr/local/etc/seaweedfs/", or "/etc/seaweedfs/", in that order.
  80. The example security.toml configuration file can be generated by "weed scaffold -config=security"
  81. `,
  82. }
  83. var (
  84. masterCpuProfile = cmdMaster.Flag.String("cpuprofile", "", "cpu profile output file")
  85. masterMemProfile = cmdMaster.Flag.String("memprofile", "", "memory profile output file")
  86. )
  87. func runMaster(cmd *Command, args []string) bool {
  88. util.LoadConfiguration("security", false)
  89. util.LoadConfiguration("master", false)
  90. grace.SetupProfiling(*masterCpuProfile, *masterMemProfile)
  91. parent, _ := util.FullPath(*m.metaFolder).DirAndName()
  92. if util.FileExists(string(parent)) && !util.FileExists(*m.metaFolder) {
  93. os.MkdirAll(*m.metaFolder, 0755)
  94. }
  95. if err := util.TestFolderWritable(util.ResolvePath(*m.metaFolder)); err != nil {
  96. glog.Fatalf("Check Meta Folder (-mdir) Writable %s : %s", *m.metaFolder, err)
  97. }
  98. masterWhiteList := util.StringSplit(*m.whiteList, ",")
  99. if *m.volumeSizeLimitMB > util.VolumeSizeLimitGB*1000 {
  100. glog.Fatalf("volumeSizeLimitMB should be smaller than 30000")
  101. }
  102. go stats_collect.StartMetricsServer(*m.ipBind, *m.metricsHttpPort)
  103. startMaster(m, masterWhiteList)
  104. return true
  105. }
  106. func startMaster(masterOption MasterOptions, masterWhiteList []string) {
  107. backend.LoadConfiguration(util.GetViper())
  108. if *masterOption.portGrpc == 0 {
  109. *masterOption.portGrpc = 10000 + *masterOption.port
  110. }
  111. if *masterOption.ipBind == "" {
  112. *masterOption.ipBind = *masterOption.ip
  113. }
  114. myMasterAddress, peers := checkPeers(*masterOption.ip, *masterOption.port, *masterOption.portGrpc, *masterOption.peers)
  115. masterPeers := make(map[string]pb.ServerAddress)
  116. for _, peer := range peers {
  117. masterPeers[string(peer)] = peer
  118. }
  119. r := mux.NewRouter()
  120. ms := weed_server.NewMasterServer(r, masterOption.toMasterOption(masterWhiteList), masterPeers)
  121. listeningAddress := util.JoinHostPort(*masterOption.ipBind, *masterOption.port)
  122. glog.V(0).Infof("Start Seaweed Master %s at %s", util.Version(), listeningAddress)
  123. masterListener, masterLocalListener, e := util.NewIpAndLocalListeners(*masterOption.ipBind, *masterOption.port, 0)
  124. if e != nil {
  125. glog.Fatalf("Master startup error: %v", e)
  126. }
  127. // start raftServer
  128. metaDir := path.Join(*masterOption.metaFolder, fmt.Sprintf("m%d", *masterOption.port))
  129. raftServerOption := &weed_server.RaftServerOption{
  130. GrpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.master"),
  131. Peers: masterPeers,
  132. ServerAddr: myMasterAddress,
  133. DataDir: util.ResolvePath(metaDir),
  134. Topo: ms.Topo,
  135. RaftResumeState: *masterOption.raftResumeState,
  136. HeartbeatInterval: *masterOption.heartbeatInterval,
  137. ElectionTimeout: *masterOption.electionTimeout,
  138. RaftBootstrap: *masterOption.raftBootstrap,
  139. }
  140. var raftServer *weed_server.RaftServer
  141. var err error
  142. if *masterOption.raftHashicorp {
  143. if raftServer, err = weed_server.NewHashicorpRaftServer(raftServerOption); err != nil {
  144. glog.Fatalf("NewHashicorpRaftServer: %s", err)
  145. }
  146. } else {
  147. raftServer, err = weed_server.NewRaftServer(raftServerOption)
  148. if raftServer == nil {
  149. glog.Fatalf("please verify %s is writable, see https://github.com/seaweedfs/seaweedfs/issues/717: %s", *masterOption.metaFolder, err)
  150. }
  151. }
  152. ms.SetRaftServer(raftServer)
  153. r.HandleFunc("/cluster/status", raftServer.StatusHandler).Methods("GET")
  154. r.HandleFunc("/cluster/healthz", raftServer.HealthzHandler).Methods("GET", "HEAD")
  155. if *masterOption.raftHashicorp {
  156. r.HandleFunc("/raft/stats", raftServer.StatsRaftHandler).Methods("GET")
  157. }
  158. // starting grpc server
  159. grpcPort := *masterOption.portGrpc
  160. grpcL, grpcLocalL, err := util.NewIpAndLocalListeners(*masterOption.ipBind, grpcPort, 0)
  161. if err != nil {
  162. glog.Fatalf("master failed to listen on grpc port %d: %v", grpcPort, err)
  163. }
  164. grpcS := pb.NewGrpcServer(security.LoadServerTLS(util.GetViper(), "grpc.master"))
  165. master_pb.RegisterSeaweedServer(grpcS, ms)
  166. if *masterOption.raftHashicorp {
  167. raftServer.TransportManager.Register(grpcS)
  168. } else {
  169. protobuf.RegisterRaftServer(grpcS, raftServer)
  170. }
  171. reflection.Register(grpcS)
  172. glog.V(0).Infof("Start Seaweed Master %s grpc server at %s:%d", util.Version(), *masterOption.ipBind, grpcPort)
  173. if grpcLocalL != nil {
  174. go grpcS.Serve(grpcLocalL)
  175. }
  176. go grpcS.Serve(grpcL)
  177. timeSleep := 1500 * time.Millisecond
  178. if !*masterOption.raftHashicorp {
  179. go func() {
  180. time.Sleep(timeSleep)
  181. ms.Topo.RaftServerAccessLock.RLock()
  182. isEmptyMaster := ms.Topo.RaftServer.Leader() == "" && ms.Topo.RaftServer.IsLogEmpty()
  183. if isEmptyMaster && isTheFirstOne(myMasterAddress, peers) && ms.MasterClient.FindLeaderFromOtherPeers(myMasterAddress) == "" {
  184. raftServer.DoJoinCommand()
  185. }
  186. ms.Topo.RaftServerAccessLock.RUnlock()
  187. }()
  188. }
  189. go ms.MasterClient.KeepConnectedToMaster()
  190. // start http server
  191. var (
  192. clientCertFile,
  193. certFile,
  194. keyFile string
  195. )
  196. useTLS := false
  197. useMTLS := false
  198. if viper.GetString("https.master.key") != "" {
  199. useTLS = true
  200. certFile = viper.GetString("https.master.cert")
  201. keyFile = viper.GetString("https.master.key")
  202. }
  203. if viper.GetString("https.master.ca") != "" {
  204. useMTLS = true
  205. clientCertFile = viper.GetString("https.master.ca")
  206. }
  207. httpS := &http.Server{Handler: r}
  208. if masterLocalListener != nil {
  209. go httpS.Serve(masterLocalListener)
  210. }
  211. if useMTLS {
  212. httpS.TLSConfig = security.LoadClientTLSHTTP(clientCertFile)
  213. }
  214. if useTLS {
  215. go httpS.ServeTLS(masterListener, certFile, keyFile)
  216. } else {
  217. go httpS.Serve(masterListener)
  218. }
  219. grace.OnInterrupt(ms.Shutdown)
  220. grace.OnInterrupt(grpcS.Stop)
  221. grace.OnReload(func() {
  222. if ms.Topo.HashicorpRaft != nil && ms.Topo.HashicorpRaft.State() == hashicorpRaft.Leader {
  223. ms.Topo.HashicorpRaft.LeadershipTransfer()
  224. }
  225. })
  226. select {}
  227. }
  228. func checkPeers(masterIp string, masterPort int, masterGrpcPort int, peers string) (masterAddress pb.ServerAddress, cleanedPeers []pb.ServerAddress) {
  229. glog.V(0).Infof("current: %s:%d peers:%s", masterIp, masterPort, peers)
  230. masterAddress = pb.NewServerAddress(masterIp, masterPort, masterGrpcPort)
  231. cleanedPeers = pb.ServerAddresses(peers).ToAddresses()
  232. hasSelf := false
  233. for _, peer := range cleanedPeers {
  234. if peer.ToHttpAddress() == masterAddress.ToHttpAddress() {
  235. hasSelf = true
  236. break
  237. }
  238. }
  239. if !hasSelf {
  240. cleanedPeers = append(cleanedPeers, masterAddress)
  241. }
  242. if len(cleanedPeers)%2 == 0 {
  243. glog.Fatalf("Only odd number of masters are supported: %+v", cleanedPeers)
  244. }
  245. return
  246. }
  247. func isTheFirstOne(self pb.ServerAddress, peers []pb.ServerAddress) bool {
  248. slices.SortFunc(peers, func(a, b pb.ServerAddress) bool {
  249. return strings.Compare(string(a), string(b)) < 0
  250. })
  251. if len(peers) <= 0 {
  252. return true
  253. }
  254. return self == peers[0]
  255. }
  256. func (m *MasterOptions) toMasterOption(whiteList []string) *weed_server.MasterOption {
  257. masterAddress := pb.NewServerAddress(*m.ip, *m.port, *m.portGrpc)
  258. return &weed_server.MasterOption{
  259. Master: masterAddress,
  260. MetaFolder: *m.metaFolder,
  261. VolumeSizeLimitMB: uint32(*m.volumeSizeLimitMB),
  262. VolumePreallocate: *m.volumePreallocate,
  263. // PulseSeconds: *m.pulseSeconds,
  264. DefaultReplicaPlacement: *m.defaultReplication,
  265. GarbageThreshold: *m.garbageThreshold,
  266. WhiteList: whiteList,
  267. DisableHttp: *m.disableHttp,
  268. MetricsAddress: *m.metricsAddress,
  269. MetricsIntervalSec: *m.metricsIntervalSec,
  270. }
  271. }