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.

211 lines
7.5 KiB

5 years ago
6 years ago
5 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
6 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
5 years ago
5 years ago
4 years ago
5 years ago
5 years ago
5 years ago
6 years ago
6 years ago
5 years ago
5 years ago
6 years ago
6 years ago
  1. package command
  2. import (
  3. "github.com/chrislusf/raft/protobuf"
  4. "github.com/gorilla/mux"
  5. "google.golang.org/grpc/reflection"
  6. "net/http"
  7. "os"
  8. "sort"
  9. "strconv"
  10. "strings"
  11. "time"
  12. "github.com/chrislusf/seaweedfs/weed/util/grace"
  13. "github.com/chrislusf/seaweedfs/weed/glog"
  14. "github.com/chrislusf/seaweedfs/weed/pb"
  15. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  16. "github.com/chrislusf/seaweedfs/weed/security"
  17. "github.com/chrislusf/seaweedfs/weed/server"
  18. "github.com/chrislusf/seaweedfs/weed/storage/backend"
  19. "github.com/chrislusf/seaweedfs/weed/util"
  20. )
  21. var (
  22. m MasterOptions
  23. )
  24. type MasterOptions struct {
  25. port *int
  26. ip *string
  27. ipBind *string
  28. metaFolder *string
  29. peers *string
  30. volumeSizeLimitMB *uint
  31. volumePreallocate *bool
  32. // pulseSeconds *int
  33. defaultReplication *string
  34. garbageThreshold *float64
  35. whiteList *string
  36. disableHttp *bool
  37. metricsAddress *string
  38. metricsIntervalSec *int
  39. raftResumeState *bool
  40. }
  41. func init() {
  42. cmdMaster.Run = runMaster // break init cycle
  43. m.port = cmdMaster.Flag.Int("port", 9333, "http listen port")
  44. m.ip = cmdMaster.Flag.String("ip", util.DetectedHostAddress(), "master <ip>|<server> address, also used as identifier")
  45. m.ipBind = cmdMaster.Flag.String("ip.bind", "", "ip address to bind to")
  46. m.metaFolder = cmdMaster.Flag.String("mdir", os.TempDir(), "data directory to store meta data")
  47. 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")
  48. m.volumeSizeLimitMB = cmdMaster.Flag.Uint("volumeSizeLimitMB", 30*1000, "Master stops directing writes to oversized volumes.")
  49. m.volumePreallocate = cmdMaster.Flag.Bool("volumePreallocate", false, "Preallocate disk space for volumes.")
  50. // m.pulseSeconds = cmdMaster.Flag.Int("pulseSeconds", 5, "number of seconds between heartbeats")
  51. m.defaultReplication = cmdMaster.Flag.String("defaultReplication", "", "Default replication type if not specified.")
  52. m.garbageThreshold = cmdMaster.Flag.Float64("garbageThreshold", 0.3, "threshold to vacuum and reclaim spaces")
  53. m.whiteList = cmdMaster.Flag.String("whiteList", "", "comma separated Ip addresses having write permission. No limit if empty.")
  54. m.disableHttp = cmdMaster.Flag.Bool("disableHttp", false, "disable http requests, only gRPC operations are allowed.")
  55. m.metricsAddress = cmdMaster.Flag.String("metrics.address", "", "Prometheus gateway address <host>:<port>")
  56. m.metricsIntervalSec = cmdMaster.Flag.Int("metrics.intervalSeconds", 15, "Prometheus push interval in seconds")
  57. m.raftResumeState = cmdMaster.Flag.Bool("resumeState", false, "resume previous state on start master server")
  58. }
  59. var cmdMaster = &Command{
  60. UsageLine: "master -port=9333",
  61. Short: "start a master server",
  62. Long: `start a master server to provide volume=>location mapping service and sequence number of file ids
  63. The configuration file "security.toml" is read from ".", "$HOME/.seaweedfs/", "/usr/local/etc/seaweedfs/", or "/etc/seaweedfs/", in that order.
  64. The example security.toml configuration file can be generated by "weed scaffold -config=security"
  65. `,
  66. }
  67. var (
  68. masterCpuProfile = cmdMaster.Flag.String("cpuprofile", "", "cpu profile output file")
  69. masterMemProfile = cmdMaster.Flag.String("memprofile", "", "memory profile output file")
  70. )
  71. func runMaster(cmd *Command, args []string) bool {
  72. util.LoadConfiguration("security", false)
  73. util.LoadConfiguration("master", false)
  74. grace.SetupProfiling(*masterCpuProfile, *masterMemProfile)
  75. parent, _ := util.FullPath(*m.metaFolder).DirAndName()
  76. if util.FileExists(string(parent)) && !util.FileExists(*m.metaFolder) {
  77. os.MkdirAll(*m.metaFolder, 0755)
  78. }
  79. if err := util.TestFolderWritable(util.ResolvePath(*m.metaFolder)); err != nil {
  80. glog.Fatalf("Check Meta Folder (-mdir) Writable %s : %s", *m.metaFolder, err)
  81. }
  82. var masterWhiteList []string
  83. if *m.whiteList != "" {
  84. masterWhiteList = strings.Split(*m.whiteList, ",")
  85. }
  86. if *m.volumeSizeLimitMB > util.VolumeSizeLimitGB*1000 {
  87. glog.Fatalf("volumeSizeLimitMB should be smaller than 30000")
  88. }
  89. startMaster(m, masterWhiteList)
  90. return true
  91. }
  92. func startMaster(masterOption MasterOptions, masterWhiteList []string) {
  93. backend.LoadConfiguration(util.GetViper())
  94. myMasterAddress, peers := checkPeers(*masterOption.ip, *masterOption.port, *masterOption.peers)
  95. r := mux.NewRouter()
  96. ms := weed_server.NewMasterServer(r, masterOption.toMasterOption(masterWhiteList), peers)
  97. listeningAddress := *masterOption.ipBind + ":" + strconv.Itoa(*masterOption.port)
  98. glog.V(0).Infof("Start Seaweed Master %s at %s", util.Version(), listeningAddress)
  99. masterListener, e := util.NewListener(listeningAddress, 0)
  100. if e != nil {
  101. glog.Fatalf("Master startup error: %v", e)
  102. }
  103. // start raftServer
  104. raftServer, err := weed_server.NewRaftServer(security.LoadClientTLS(util.GetViper(), "grpc.master"),
  105. peers, myMasterAddress, util.ResolvePath(*masterOption.metaFolder), ms.Topo, *masterOption.raftResumeState)
  106. if raftServer == nil {
  107. glog.Fatalf("please verify %s is writable, see https://github.com/chrislusf/seaweedfs/issues/717: %s", *masterOption.metaFolder, err)
  108. }
  109. ms.SetRaftServer(raftServer)
  110. r.HandleFunc("/cluster/status", raftServer.StatusHandler).Methods("GET")
  111. // starting grpc server
  112. grpcPort := *masterOption.port + 10000
  113. grpcL, err := util.NewListener(*masterOption.ipBind+":"+strconv.Itoa(grpcPort), 0)
  114. if err != nil {
  115. glog.Fatalf("master failed to listen on grpc port %d: %v", grpcPort, err)
  116. }
  117. grpcS := pb.NewGrpcServer(security.LoadServerTLS(util.GetViper(), "grpc.master"))
  118. master_pb.RegisterSeaweedServer(grpcS, ms)
  119. protobuf.RegisterRaftServer(grpcS, raftServer)
  120. reflection.Register(grpcS)
  121. glog.V(0).Infof("Start Seaweed Master %s grpc server at %s:%d", util.Version(), *masterOption.ipBind, grpcPort)
  122. go grpcS.Serve(grpcL)
  123. go func() {
  124. time.Sleep(1500 * time.Millisecond)
  125. if ms.Topo.RaftServer.Leader() == "" && ms.Topo.RaftServer.IsLogEmpty() && isTheFirstOne(myMasterAddress, peers) {
  126. if ms.MasterClient.FindLeaderFromOtherPeers(myMasterAddress) == "" {
  127. raftServer.DoJoinCommand()
  128. }
  129. }
  130. }()
  131. go ms.MasterClient.KeepConnectedToMaster()
  132. // start http server
  133. httpS := &http.Server{Handler: r}
  134. go httpS.Serve(masterListener)
  135. select {}
  136. }
  137. func checkPeers(masterIp string, masterPort int, peers string) (masterAddress string, cleanedPeers []string) {
  138. glog.V(0).Infof("current: %s:%d peers:%s", masterIp, masterPort, peers)
  139. masterAddress = masterIp + ":" + strconv.Itoa(masterPort)
  140. if peers != "" {
  141. cleanedPeers = strings.Split(peers, ",")
  142. }
  143. hasSelf := false
  144. for _, peer := range cleanedPeers {
  145. if peer == masterAddress {
  146. hasSelf = true
  147. break
  148. }
  149. }
  150. if !hasSelf {
  151. cleanedPeers = append(cleanedPeers, masterAddress)
  152. }
  153. if len(cleanedPeers)%2 == 0 {
  154. glog.Fatalf("Only odd number of masters are supported!")
  155. }
  156. return
  157. }
  158. func isTheFirstOne(self string, peers []string) bool {
  159. sort.Strings(peers)
  160. if len(peers) <= 0 {
  161. return true
  162. }
  163. return self == peers[0]
  164. }
  165. func (m *MasterOptions) toMasterOption(whiteList []string) *weed_server.MasterOption {
  166. return &weed_server.MasterOption{
  167. Host: *m.ip,
  168. Port: *m.port,
  169. MetaFolder: *m.metaFolder,
  170. VolumeSizeLimitMB: uint32(*m.volumeSizeLimitMB),
  171. VolumePreallocate: *m.volumePreallocate,
  172. // PulseSeconds: *m.pulseSeconds,
  173. DefaultReplicaPlacement: *m.defaultReplication,
  174. GarbageThreshold: *m.garbageThreshold,
  175. WhiteList: whiteList,
  176. DisableHttp: *m.disableHttp,
  177. MetricsAddress: *m.metricsAddress,
  178. MetricsIntervalSec: *m.metricsIntervalSec,
  179. }
  180. }