You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

329 lines
12 KiB

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