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.

120 lines
4.2 KiB

  1. package weed_server
  2. import (
  3. "fmt"
  4. "net/http"
  5. "net/http/httputil"
  6. "net/url"
  7. "sync"
  8. "github.com/chrislusf/weed-fs/go/glog"
  9. "github.com/chrislusf/weed-fs/go/security"
  10. "github.com/chrislusf/weed-fs/go/sequence"
  11. "github.com/chrislusf/weed-fs/go/topology"
  12. "github.com/chrislusf/weed-fs/go/util"
  13. "github.com/goraft/raft"
  14. "github.com/gorilla/mux"
  15. )
  16. type MasterServer struct {
  17. port int
  18. metaFolder string
  19. volumeSizeLimitMB uint
  20. pulseSeconds int
  21. defaultReplicaPlacement string
  22. garbageThreshold string
  23. Topo *topology.Topology
  24. vg *topology.VolumeGrowth
  25. vgLock sync.Mutex
  26. bounedLeaderChan chan int
  27. }
  28. func NewMasterServer(r *mux.Router, port int, metaFolder string,
  29. volumeSizeLimitMB uint,
  30. pulseSeconds int,
  31. confFile string,
  32. defaultReplicaPlacement string,
  33. garbageThreshold string,
  34. whiteList []string,
  35. secureKey string,
  36. ) *MasterServer {
  37. ms := &MasterServer{
  38. port: port,
  39. volumeSizeLimitMB: volumeSizeLimitMB,
  40. pulseSeconds: pulseSeconds,
  41. defaultReplicaPlacement: defaultReplicaPlacement,
  42. garbageThreshold: garbageThreshold,
  43. }
  44. ms.bounedLeaderChan = make(chan int, 16)
  45. seq := sequence.NewMemorySequencer()
  46. var e error
  47. if ms.Topo, e = topology.NewTopology("topo", confFile, seq,
  48. uint64(volumeSizeLimitMB)*1024*1024, pulseSeconds); e != nil {
  49. glog.Fatalf("cannot create topology:%s", e)
  50. }
  51. ms.vg = topology.NewDefaultVolumeGrowth()
  52. glog.V(0).Infoln("Volume Size Limit is", volumeSizeLimitMB, "MB")
  53. guard := security.NewGuard(whiteList, secureKey)
  54. r.HandleFunc("/dir/assign", ms.proxyToLeader(guard.Secure(ms.dirAssignHandler)))
  55. r.HandleFunc("/dir/lookup", ms.proxyToLeader(guard.Secure(ms.dirLookupHandler)))
  56. r.HandleFunc("/dir/join", ms.proxyToLeader(guard.Secure(ms.dirJoinHandler)))
  57. r.HandleFunc("/dir/status", ms.proxyToLeader(guard.Secure(ms.dirStatusHandler)))
  58. r.HandleFunc("/col/delete", ms.proxyToLeader(guard.Secure(ms.collectionDeleteHandler)))
  59. r.HandleFunc("/vol/lookup", ms.proxyToLeader(guard.Secure(ms.volumeLookupHandler)))
  60. r.HandleFunc("/vol/grow", ms.proxyToLeader(guard.Secure(ms.volumeGrowHandler)))
  61. r.HandleFunc("/vol/status", ms.proxyToLeader(guard.Secure(ms.volumeStatusHandler)))
  62. r.HandleFunc("/vol/vacuum", ms.proxyToLeader(guard.Secure(ms.volumeVacuumHandler)))
  63. r.HandleFunc("/submit", guard.Secure(ms.submitFromMasterServerHandler))
  64. r.HandleFunc("/delete", guard.Secure(ms.deleteFromMasterServerHandler))
  65. r.HandleFunc("/{fileId}", ms.redirectHandler)
  66. r.HandleFunc("/stats/counter", guard.Secure(statsCounterHandler))
  67. r.HandleFunc("/stats/memory", guard.Secure(statsMemoryHandler))
  68. ms.Topo.StartRefreshWritableVolumes(garbageThreshold)
  69. return ms
  70. }
  71. func (ms *MasterServer) SetRaftServer(raftServer *RaftServer) {
  72. ms.Topo.RaftServer = raftServer.raftServer
  73. ms.Topo.RaftServer.AddEventListener(raft.LeaderChangeEventType, func(e raft.Event) {
  74. if ms.Topo.RaftServer.Leader() != "" {
  75. glog.V(0).Infoln("[", ms.Topo.RaftServer.Name(), "]", ms.Topo.RaftServer.Leader(), "becomes leader.")
  76. }
  77. })
  78. if ms.Topo.IsLeader() {
  79. glog.V(0).Infoln("[", ms.Topo.RaftServer.Name(), "]", "I am the leader!")
  80. } else {
  81. if ms.Topo.RaftServer.Leader() != "" {
  82. glog.V(0).Infoln("[", ms.Topo.RaftServer.Name(), "]", ms.Topo.RaftServer.Leader(), "is the leader.")
  83. }
  84. }
  85. }
  86. func (ms *MasterServer) proxyToLeader(f func(w http.ResponseWriter, r *http.Request)) func(w http.ResponseWriter, r *http.Request) {
  87. return func(w http.ResponseWriter, r *http.Request) {
  88. if ms.Topo.IsLeader() {
  89. f(w, r)
  90. } else if ms.Topo.RaftServer != nil && ms.Topo.RaftServer.Leader() != "" {
  91. ms.bounedLeaderChan <- 1
  92. defer func() { <-ms.bounedLeaderChan }()
  93. targetUrl, err := url.Parse("http://" + ms.Topo.RaftServer.Leader())
  94. if err != nil {
  95. writeJsonError(w, r, http.StatusInternalServerError,
  96. fmt.Errorf("Leader URL http://%s Parse Error: %v", ms.Topo.RaftServer.Leader(), err))
  97. return
  98. }
  99. glog.V(4).Infoln("proxying to leader", ms.Topo.RaftServer.Leader())
  100. proxy := httputil.NewSingleHostReverseProxy(targetUrl)
  101. proxy.Transport = util.Transport
  102. proxy.ServeHTTP(w, r)
  103. } else {
  104. //drop it to the floor
  105. //writeJsonError(w, r, errors.New(ms.Topo.RaftServer.Name()+" does not know Leader yet:"+ms.Topo.RaftServer.Leader()))
  106. }
  107. }
  108. }