master_server.go 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417
  1. package weed_server
  2. import (
  3. "context"
  4. "fmt"
  5. "net/http"
  6. "net/http/httputil"
  7. "net/url"
  8. "os"
  9. "regexp"
  10. "strings"
  11. "sync"
  12. "time"
  13. "github.com/seaweedfs/seaweedfs/weed/stats"
  14. "github.com/seaweedfs/seaweedfs/weed/cluster"
  15. "github.com/seaweedfs/seaweedfs/weed/pb"
  16. "github.com/gorilla/mux"
  17. hashicorpRaft "github.com/hashicorp/raft"
  18. "github.com/seaweedfs/raft"
  19. "google.golang.org/grpc"
  20. "github.com/seaweedfs/seaweedfs/weed/glog"
  21. "github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
  22. "github.com/seaweedfs/seaweedfs/weed/security"
  23. "github.com/seaweedfs/seaweedfs/weed/sequence"
  24. "github.com/seaweedfs/seaweedfs/weed/shell"
  25. "github.com/seaweedfs/seaweedfs/weed/topology"
  26. "github.com/seaweedfs/seaweedfs/weed/util"
  27. "github.com/seaweedfs/seaweedfs/weed/wdclient"
  28. )
  29. const (
  30. SequencerType = "master.sequencer.type"
  31. SequencerSnowflakeId = "master.sequencer.sequencer_snowflake_id"
  32. )
  33. type MasterOption struct {
  34. Master pb.ServerAddress
  35. MetaFolder string
  36. VolumeSizeLimitMB uint32
  37. VolumePreallocate bool
  38. // PulseSeconds int
  39. DefaultReplicaPlacement string
  40. GarbageThreshold float64
  41. WhiteList []string
  42. DisableHttp bool
  43. MetricsAddress string
  44. MetricsIntervalSec int
  45. IsFollower bool
  46. }
  47. type MasterServer struct {
  48. master_pb.UnimplementedSeaweedServer
  49. option *MasterOption
  50. guard *security.Guard
  51. preallocateSize int64
  52. Topo *topology.Topology
  53. vg *topology.VolumeGrowth
  54. volumeGrowthRequestChan chan *topology.VolumeGrowRequest
  55. boundedLeaderChan chan int
  56. // notifying clients
  57. clientChansLock sync.RWMutex
  58. clientChans map[string]chan *master_pb.KeepConnectedResponse
  59. grpcDialOption grpc.DialOption
  60. MasterClient *wdclient.MasterClient
  61. adminLocks *AdminLocks
  62. Cluster *cluster.Cluster
  63. }
  64. func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.ServerAddress) *MasterServer {
  65. v := util.GetViper()
  66. signingKey := v.GetString("jwt.signing.key")
  67. v.SetDefault("jwt.signing.expires_after_seconds", 10)
  68. expiresAfterSec := v.GetInt("jwt.signing.expires_after_seconds")
  69. readSigningKey := v.GetString("jwt.signing.read.key")
  70. v.SetDefault("jwt.signing.read.expires_after_seconds", 60)
  71. readExpiresAfterSec := v.GetInt("jwt.signing.read.expires_after_seconds")
  72. v.SetDefault("master.replication.treat_replication_as_minimums", false)
  73. replicationAsMin := v.GetBool("master.replication.treat_replication_as_minimums")
  74. v.SetDefault("master.volume_growth.copy_1", 7)
  75. v.SetDefault("master.volume_growth.copy_2", 6)
  76. v.SetDefault("master.volume_growth.copy_3", 3)
  77. v.SetDefault("master.volume_growth.copy_other", 1)
  78. v.SetDefault("master.volume_growth.threshold", 0.9)
  79. topology.VolumeGrowStrategy.Copy1Count = v.GetInt("master.volume_growth.copy_1")
  80. topology.VolumeGrowStrategy.Copy2Count = v.GetInt("master.volume_growth.copy_2")
  81. topology.VolumeGrowStrategy.Copy3Count = v.GetInt("master.volume_growth.copy_3")
  82. topology.VolumeGrowStrategy.CopyOtherCount = v.GetInt("master.volume_growth.copy_other")
  83. topology.VolumeGrowStrategy.Threshold = v.GetFloat64("master.volume_growth.threshold")
  84. var preallocateSize int64
  85. if option.VolumePreallocate {
  86. preallocateSize = int64(option.VolumeSizeLimitMB) * (1 << 20)
  87. }
  88. grpcDialOption := security.LoadClientTLS(v, "grpc.master")
  89. ms := &MasterServer{
  90. option: option,
  91. preallocateSize: preallocateSize,
  92. volumeGrowthRequestChan: make(chan *topology.VolumeGrowRequest, 1<<6),
  93. clientChans: make(map[string]chan *master_pb.KeepConnectedResponse),
  94. grpcDialOption: grpcDialOption,
  95. MasterClient: wdclient.NewMasterClient(grpcDialOption, "", cluster.MasterType, option.Master, "", "", *pb.NewServiceDiscoveryFromMap(peers)),
  96. adminLocks: NewAdminLocks(),
  97. Cluster: cluster.NewCluster(),
  98. }
  99. ms.boundedLeaderChan = make(chan int, 16)
  100. ms.MasterClient.SetOnPeerUpdateFn(ms.OnPeerUpdate)
  101. seq := ms.createSequencer(option)
  102. if nil == seq {
  103. glog.Fatalf("create sequencer failed.")
  104. }
  105. ms.Topo = topology.NewTopology("topo", seq, uint64(ms.option.VolumeSizeLimitMB)*1024*1024, 5, replicationAsMin)
  106. ms.vg = topology.NewDefaultVolumeGrowth()
  107. glog.V(0).Infoln("Volume Size Limit is", ms.option.VolumeSizeLimitMB, "MB")
  108. ms.guard = security.NewGuard(ms.option.WhiteList, signingKey, expiresAfterSec, readSigningKey, readExpiresAfterSec)
  109. handleStaticResources2(r)
  110. r.HandleFunc("/", ms.proxyToLeader(ms.uiStatusHandler))
  111. r.HandleFunc("/ui/index.html", ms.uiStatusHandler)
  112. if !ms.option.DisableHttp {
  113. r.HandleFunc("/dir/assign", ms.proxyToLeader(ms.guard.WhiteList(ms.dirAssignHandler)))
  114. r.HandleFunc("/dir/lookup", ms.guard.WhiteList(ms.dirLookupHandler))
  115. r.HandleFunc("/dir/status", ms.proxyToLeader(ms.guard.WhiteList(ms.dirStatusHandler)))
  116. r.HandleFunc("/col/delete", ms.proxyToLeader(ms.guard.WhiteList(ms.collectionDeleteHandler)))
  117. r.HandleFunc("/vol/grow", ms.proxyToLeader(ms.guard.WhiteList(ms.volumeGrowHandler)))
  118. r.HandleFunc("/vol/status", ms.proxyToLeader(ms.guard.WhiteList(ms.volumeStatusHandler)))
  119. r.HandleFunc("/vol/vacuum", ms.proxyToLeader(ms.guard.WhiteList(ms.volumeVacuumHandler)))
  120. r.HandleFunc("/submit", ms.guard.WhiteList(ms.submitFromMasterServerHandler))
  121. /*
  122. r.HandleFunc("/stats/health", ms.guard.WhiteList(statsHealthHandler))
  123. r.HandleFunc("/stats/counter", ms.guard.WhiteList(statsCounterHandler))
  124. r.HandleFunc("/stats/memory", ms.guard.WhiteList(statsMemoryHandler))
  125. */
  126. r.HandleFunc("/{fileId}", ms.redirectHandler)
  127. }
  128. ms.Topo.StartRefreshWritableVolumes(
  129. ms.grpcDialOption,
  130. ms.option.GarbageThreshold,
  131. topology.VolumeGrowStrategy.Threshold,
  132. ms.preallocateSize,
  133. )
  134. ms.ProcessGrowRequest()
  135. if !option.IsFollower {
  136. ms.startAdminScripts()
  137. }
  138. return ms
  139. }
  140. func (ms *MasterServer) SetRaftServer(raftServer *RaftServer) {
  141. var raftServerName string
  142. ms.Topo.RaftServerAccessLock.Lock()
  143. if raftServer.raftServer != nil {
  144. ms.Topo.RaftServer = raftServer.raftServer
  145. ms.Topo.RaftServer.AddEventListener(raft.LeaderChangeEventType, func(e raft.Event) {
  146. glog.V(0).Infof("leader change event: %+v => %+v", e.PrevValue(), e.Value())
  147. stats.MasterLeaderChangeCounter.WithLabelValues(fmt.Sprintf("%+v", e.Value())).Inc()
  148. if ms.Topo.RaftServer.Leader() != "" {
  149. glog.V(0).Infof("[%s] %s becomes leader.", ms.Topo.RaftServer.Name(), ms.Topo.RaftServer.Leader())
  150. }
  151. })
  152. raftServerName = fmt.Sprintf("[%s]", ms.Topo.RaftServer.Name())
  153. } else if raftServer.RaftHashicorp != nil {
  154. ms.Topo.HashicorpRaft = raftServer.RaftHashicorp
  155. leaderCh := raftServer.RaftHashicorp.LeaderCh()
  156. prevLeader, _ := ms.Topo.HashicorpRaft.LeaderWithID()
  157. raftServerName = ms.Topo.HashicorpRaft.String()
  158. go func() {
  159. for {
  160. select {
  161. case isLeader := <-leaderCh:
  162. ms.Topo.RaftServerAccessLock.RLock()
  163. leader, _ := ms.Topo.HashicorpRaft.LeaderWithID()
  164. ms.Topo.RaftServerAccessLock.RUnlock()
  165. glog.V(0).Infof("is leader %+v change event: %+v => %+v", isLeader, prevLeader, leader)
  166. stats.MasterLeaderChangeCounter.WithLabelValues(fmt.Sprintf("%+v", leader)).Inc()
  167. prevLeader = leader
  168. }
  169. }
  170. }()
  171. }
  172. ms.Topo.RaftServerAccessLock.Unlock()
  173. if ms.Topo.IsLeader() {
  174. glog.V(0).Infof("%s I am the leader!", raftServerName)
  175. } else {
  176. var raftServerLeader string
  177. ms.Topo.RaftServerAccessLock.RLock()
  178. if ms.Topo.RaftServer != nil {
  179. raftServerLeader = ms.Topo.RaftServer.Leader()
  180. } else if ms.Topo.HashicorpRaft != nil {
  181. raftServerName = ms.Topo.HashicorpRaft.String()
  182. raftServerLeaderAddr, _ := ms.Topo.HashicorpRaft.LeaderWithID()
  183. raftServerLeader = string(raftServerLeaderAddr)
  184. }
  185. ms.Topo.RaftServerAccessLock.RUnlock()
  186. glog.V(0).Infof("%s %s - is the leader.", raftServerName, raftServerLeader)
  187. }
  188. }
  189. func (ms *MasterServer) proxyToLeader(f http.HandlerFunc) http.HandlerFunc {
  190. return func(w http.ResponseWriter, r *http.Request) {
  191. if ms.Topo.IsLeader() {
  192. f(w, r)
  193. return
  194. }
  195. // get the current raft leader
  196. leaderAddr, _ := ms.Topo.MaybeLeader()
  197. raftServerLeader := leaderAddr.ToHttpAddress()
  198. if raftServerLeader == "" {
  199. f(w, r)
  200. return
  201. }
  202. ms.boundedLeaderChan <- 1
  203. defer func() { <-ms.boundedLeaderChan }()
  204. targetUrl, err := url.Parse("http://" + raftServerLeader)
  205. if err != nil {
  206. writeJsonError(w, r, http.StatusInternalServerError,
  207. fmt.Errorf("Leader URL http://%s Parse Error: %v", raftServerLeader, err))
  208. return
  209. }
  210. // proxy to leader
  211. glog.V(4).Infoln("proxying to leader", raftServerLeader)
  212. proxy := httputil.NewSingleHostReverseProxy(targetUrl)
  213. director := proxy.Director
  214. proxy.Director = func(req *http.Request) {
  215. actualHost, err := security.GetActualRemoteHost(req)
  216. if err == nil {
  217. req.Header.Set("HTTP_X_FORWARDED_FOR", actualHost)
  218. }
  219. director(req)
  220. }
  221. proxy.Transport = util.Transport
  222. proxy.ServeHTTP(w, r)
  223. }
  224. }
  225. func (ms *MasterServer) startAdminScripts() {
  226. v := util.GetViper()
  227. adminScripts := v.GetString("master.maintenance.scripts")
  228. if adminScripts == "" {
  229. return
  230. }
  231. glog.V(0).Infof("adminScripts: %v", adminScripts)
  232. v.SetDefault("master.maintenance.sleep_minutes", 17)
  233. sleepMinutes := v.GetInt("master.maintenance.sleep_minutes")
  234. scriptLines := strings.Split(adminScripts, "\n")
  235. if !strings.Contains(adminScripts, "lock") {
  236. scriptLines = append(append([]string{}, "lock"), scriptLines...)
  237. scriptLines = append(scriptLines, "unlock")
  238. }
  239. masterAddress := string(ms.option.Master)
  240. var shellOptions shell.ShellOptions
  241. shellOptions.GrpcDialOption = security.LoadClientTLS(v, "grpc.master")
  242. shellOptions.Masters = &masterAddress
  243. shellOptions.Directory = "/"
  244. emptyFilerGroup := ""
  245. shellOptions.FilerGroup = &emptyFilerGroup
  246. commandEnv := shell.NewCommandEnv(&shellOptions)
  247. reg, _ := regexp.Compile(`'.*?'|".*?"|\S+`)
  248. go commandEnv.MasterClient.KeepConnectedToMaster()
  249. go func() {
  250. for {
  251. time.Sleep(time.Duration(sleepMinutes) * time.Minute)
  252. if ms.Topo.IsLeader() && ms.MasterClient.GetMaster() != "" {
  253. shellOptions.FilerAddress = ms.GetOneFiler(cluster.FilerGroupName(*shellOptions.FilerGroup))
  254. if shellOptions.FilerAddress == "" {
  255. continue
  256. }
  257. for _, line := range scriptLines {
  258. for _, c := range strings.Split(line, ";") {
  259. processEachCmd(reg, c, commandEnv)
  260. }
  261. }
  262. }
  263. }
  264. }()
  265. }
  266. func processEachCmd(reg *regexp.Regexp, line string, commandEnv *shell.CommandEnv) {
  267. cmds := reg.FindAllString(line, -1)
  268. if len(cmds) == 0 {
  269. return
  270. }
  271. args := make([]string, len(cmds[1:]))
  272. for i := range args {
  273. args[i] = strings.Trim(string(cmds[1+i]), "\"'")
  274. }
  275. cmd := cmds[0]
  276. for _, c := range shell.Commands {
  277. if c.Name() == cmd {
  278. glog.V(0).Infof("executing: %s %v", cmd, args)
  279. if err := c.Do(args, commandEnv, os.Stdout); err != nil {
  280. glog.V(0).Infof("error: %v", err)
  281. }
  282. }
  283. }
  284. }
  285. func (ms *MasterServer) createSequencer(option *MasterOption) sequence.Sequencer {
  286. var seq sequence.Sequencer
  287. v := util.GetViper()
  288. seqType := strings.ToLower(v.GetString(SequencerType))
  289. glog.V(1).Infof("[%s] : [%s]", SequencerType, seqType)
  290. switch strings.ToLower(seqType) {
  291. case "snowflake":
  292. var err error
  293. snowflakeId := v.GetInt(SequencerSnowflakeId)
  294. seq, err = sequence.NewSnowflakeSequencer(string(option.Master), snowflakeId)
  295. if err != nil {
  296. glog.Error(err)
  297. seq = nil
  298. }
  299. case "raft":
  300. fallthrough
  301. default:
  302. seq = sequence.NewMemorySequencer()
  303. }
  304. return seq
  305. }
  306. func (ms *MasterServer) OnPeerUpdate(update *master_pb.ClusterNodeUpdate, startFrom time.Time) {
  307. ms.Topo.RaftServerAccessLock.RLock()
  308. defer ms.Topo.RaftServerAccessLock.RUnlock()
  309. if update.NodeType != cluster.MasterType || ms.Topo.HashicorpRaft == nil {
  310. return
  311. }
  312. glog.V(4).Infof("OnPeerUpdate: %+v", update)
  313. peerAddress := pb.ServerAddress(update.Address)
  314. peerName := string(peerAddress)
  315. if ms.Topo.HashicorpRaft.State() != hashicorpRaft.Leader {
  316. return
  317. }
  318. if update.IsAdd {
  319. raftServerFound := false
  320. for _, server := range ms.Topo.HashicorpRaft.GetConfiguration().Configuration().Servers {
  321. if string(server.ID) == peerName {
  322. raftServerFound = true
  323. }
  324. }
  325. if !raftServerFound {
  326. glog.V(0).Infof("adding new raft server: %s", peerName)
  327. ms.Topo.HashicorpRaft.AddVoter(
  328. hashicorpRaft.ServerID(peerName),
  329. hashicorpRaft.ServerAddress(peerAddress.ToGrpcAddress()), 0, 0)
  330. }
  331. } else {
  332. pb.WithMasterClient(false, peerAddress, ms.grpcDialOption, true, func(client master_pb.SeaweedClient) error {
  333. ctx, cancel := context.WithTimeout(context.TODO(), 15*time.Second)
  334. defer cancel()
  335. if _, err := client.Ping(ctx, &master_pb.PingRequest{Target: string(peerAddress), TargetType: cluster.MasterType}); err != nil {
  336. glog.V(0).Infof("master %s didn't respond to pings. remove raft server", peerName)
  337. if err := ms.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
  338. _, err := client.RaftRemoveServer(context.Background(), &master_pb.RaftRemoveServerRequest{
  339. Id: peerName,
  340. Force: false,
  341. })
  342. return err
  343. }); err != nil {
  344. glog.Warningf("failed removing old raft server: %v", err)
  345. return err
  346. }
  347. } else {
  348. glog.V(0).Infof("master %s successfully responded to ping", peerName)
  349. }
  350. return nil
  351. })
  352. }
  353. }
  354. func (ms *MasterServer) Shutdown() {
  355. if ms.Topo == nil || ms.Topo.HashicorpRaft == nil {
  356. return
  357. }
  358. if ms.Topo.HashicorpRaft.State() == hashicorpRaft.Leader {
  359. ms.Topo.HashicorpRaft.LeadershipTransfer()
  360. }
  361. ms.Topo.HashicorpRaft.Shutdown()
  362. }