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.

233 lines
7.6 KiB

6 years ago
10 years ago
9 years ago
  1. package weed_server
  2. import (
  3. "fmt"
  4. "github.com/chrislusf/seaweedfs/weed/shell"
  5. "google.golang.org/grpc"
  6. "net/http"
  7. "net/http/httputil"
  8. "net/url"
  9. "os"
  10. "regexp"
  11. "strconv"
  12. "strings"
  13. "sync"
  14. "time"
  15. "github.com/chrislusf/raft"
  16. "github.com/chrislusf/seaweedfs/weed/glog"
  17. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  18. "github.com/chrislusf/seaweedfs/weed/security"
  19. "github.com/chrislusf/seaweedfs/weed/sequence"
  20. "github.com/chrislusf/seaweedfs/weed/topology"
  21. "github.com/chrislusf/seaweedfs/weed/util"
  22. "github.com/gorilla/mux"
  23. "github.com/spf13/viper"
  24. )
  25. type MasterServer struct {
  26. port int
  27. metaFolder string
  28. volumeSizeLimitMB uint
  29. preallocate int64
  30. pulseSeconds int
  31. defaultReplicaPlacement string
  32. garbageThreshold float64
  33. guard *security.Guard
  34. metricsAddress string
  35. metricsIntervalSec int
  36. Topo *topology.Topology
  37. vg *topology.VolumeGrowth
  38. vgLock sync.Mutex
  39. bounedLeaderChan chan int
  40. // notifying clients
  41. clientChansLock sync.RWMutex
  42. clientChans map[string]chan *master_pb.VolumeLocation
  43. grpcDialOpiton grpc.DialOption
  44. }
  45. func NewMasterServer(r *mux.Router, port int, metaFolder string,
  46. volumeSizeLimitMB uint,
  47. preallocate bool,
  48. pulseSeconds int,
  49. defaultReplicaPlacement string,
  50. garbageThreshold float64,
  51. whiteList []string,
  52. disableHttp bool,
  53. metricsAddress string,
  54. metricsIntervalSec int,
  55. ) *MasterServer {
  56. v := viper.GetViper()
  57. signingKey := v.GetString("jwt.signing.key")
  58. v.SetDefault("jwt.signing.expires_after_seconds", 10)
  59. expiresAfterSec := v.GetInt("jwt.signing.expires_after_seconds")
  60. readSigningKey := v.GetString("jwt.signing.read.key")
  61. v.SetDefault("jwt.signing.read.expires_after_seconds", 60)
  62. readExpiresAfterSec := v.GetInt("jwt.signing.read.expires_after_seconds")
  63. var preallocateSize int64
  64. if preallocate {
  65. preallocateSize = int64(volumeSizeLimitMB) * (1 << 20)
  66. }
  67. ms := &MasterServer{
  68. port: port,
  69. volumeSizeLimitMB: volumeSizeLimitMB,
  70. preallocate: preallocateSize,
  71. pulseSeconds: pulseSeconds,
  72. defaultReplicaPlacement: defaultReplicaPlacement,
  73. garbageThreshold: garbageThreshold,
  74. clientChans: make(map[string]chan *master_pb.VolumeLocation),
  75. grpcDialOpiton: security.LoadClientTLS(v.Sub("grpc"), "master"),
  76. metricsAddress: metricsAddress,
  77. metricsIntervalSec: metricsIntervalSec,
  78. }
  79. ms.bounedLeaderChan = make(chan int, 16)
  80. seq := sequence.NewMemorySequencer()
  81. ms.Topo = topology.NewTopology("topo", seq, uint64(volumeSizeLimitMB)*1024*1024, pulseSeconds)
  82. ms.vg = topology.NewDefaultVolumeGrowth()
  83. glog.V(0).Infoln("Volume Size Limit is", volumeSizeLimitMB, "MB")
  84. ms.guard = security.NewGuard(whiteList, signingKey, expiresAfterSec, readSigningKey, readExpiresAfterSec)
  85. if !disableHttp {
  86. handleStaticResources2(r)
  87. r.HandleFunc("/", ms.proxyToLeader(ms.uiStatusHandler))
  88. r.HandleFunc("/ui/index.html", ms.uiStatusHandler)
  89. r.HandleFunc("/dir/assign", ms.proxyToLeader(ms.guard.WhiteList(ms.dirAssignHandler)))
  90. r.HandleFunc("/dir/lookup", ms.proxyToLeader(ms.guard.WhiteList(ms.dirLookupHandler)))
  91. r.HandleFunc("/dir/status", ms.proxyToLeader(ms.guard.WhiteList(ms.dirStatusHandler)))
  92. r.HandleFunc("/col/delete", ms.proxyToLeader(ms.guard.WhiteList(ms.collectionDeleteHandler)))
  93. r.HandleFunc("/vol/grow", ms.proxyToLeader(ms.guard.WhiteList(ms.volumeGrowHandler)))
  94. r.HandleFunc("/vol/status", ms.proxyToLeader(ms.guard.WhiteList(ms.volumeStatusHandler)))
  95. r.HandleFunc("/vol/vacuum", ms.proxyToLeader(ms.guard.WhiteList(ms.volumeVacuumHandler)))
  96. r.HandleFunc("/submit", ms.guard.WhiteList(ms.submitFromMasterServerHandler))
  97. r.HandleFunc("/stats/health", ms.guard.WhiteList(statsHealthHandler))
  98. r.HandleFunc("/stats/counter", ms.guard.WhiteList(statsCounterHandler))
  99. r.HandleFunc("/stats/memory", ms.guard.WhiteList(statsMemoryHandler))
  100. r.HandleFunc("/{fileId}", ms.proxyToLeader(ms.redirectHandler))
  101. }
  102. ms.Topo.StartRefreshWritableVolumes(ms.grpcDialOpiton, garbageThreshold, ms.preallocate)
  103. ms.startAdminScripts()
  104. return ms
  105. }
  106. func (ms *MasterServer) SetRaftServer(raftServer *RaftServer) {
  107. ms.Topo.RaftServer = raftServer.raftServer
  108. ms.Topo.RaftServer.AddEventListener(raft.LeaderChangeEventType, func(e raft.Event) {
  109. glog.V(0).Infof("event: %+v", e)
  110. if ms.Topo.RaftServer.Leader() != "" {
  111. glog.V(0).Infoln("[", ms.Topo.RaftServer.Name(), "]", ms.Topo.RaftServer.Leader(), "becomes leader.")
  112. }
  113. })
  114. ms.Topo.RaftServer.AddEventListener(raft.StateChangeEventType, func(e raft.Event) {
  115. glog.V(0).Infof("state change: %+v", e)
  116. })
  117. if ms.Topo.IsLeader() {
  118. glog.V(0).Infoln("[", ms.Topo.RaftServer.Name(), "]", "I am the leader!")
  119. } else {
  120. if ms.Topo.RaftServer.Leader() != "" {
  121. glog.V(0).Infoln("[", ms.Topo.RaftServer.Name(), "]", ms.Topo.RaftServer.Leader(), "is the leader.")
  122. }
  123. }
  124. }
  125. func (ms *MasterServer) proxyToLeader(f func(w http.ResponseWriter, r *http.Request)) func(w http.ResponseWriter, r *http.Request) {
  126. return func(w http.ResponseWriter, r *http.Request) {
  127. if ms.Topo.IsLeader() {
  128. f(w, r)
  129. } else if ms.Topo.RaftServer != nil && ms.Topo.RaftServer.Leader() != "" {
  130. ms.bounedLeaderChan <- 1
  131. defer func() { <-ms.bounedLeaderChan }()
  132. targetUrl, err := url.Parse("http://" + ms.Topo.RaftServer.Leader())
  133. if err != nil {
  134. writeJsonError(w, r, http.StatusInternalServerError,
  135. fmt.Errorf("Leader URL http://%s Parse Error: %v", ms.Topo.RaftServer.Leader(), err))
  136. return
  137. }
  138. glog.V(4).Infoln("proxying to leader", ms.Topo.RaftServer.Leader())
  139. proxy := httputil.NewSingleHostReverseProxy(targetUrl)
  140. director := proxy.Director
  141. proxy.Director = func(req *http.Request) {
  142. actualHost, err := security.GetActualRemoteHost(req)
  143. if err == nil {
  144. req.Header.Set("HTTP_X_FORWARDED_FOR", actualHost)
  145. }
  146. director(req)
  147. }
  148. proxy.Transport = util.Transport
  149. proxy.ServeHTTP(w, r)
  150. } else {
  151. //drop it to the floor
  152. //writeJsonError(w, r, errors.New(ms.Topo.RaftServer.Name()+" does not know Leader yet:"+ms.Topo.RaftServer.Leader()))
  153. }
  154. }
  155. }
  156. func (ms *MasterServer) startAdminScripts() {
  157. v := viper.GetViper()
  158. adminScripts := v.GetString("master.maintenance.scripts")
  159. v.SetDefault("master.maintenance.sleep_minutes", 17)
  160. sleepMinutes := v.GetInt("master.maintenance.sleep_minutes")
  161. glog.V(0).Infof("adminScripts:\n%v", adminScripts)
  162. if adminScripts == "" {
  163. return
  164. }
  165. scriptLines := strings.Split(adminScripts, "\n")
  166. masterAddress := "localhost:" + strconv.Itoa(ms.port)
  167. var shellOptions shell.ShellOptions
  168. shellOptions.GrpcDialOption = security.LoadClientTLS(viper.Sub("grpc"), "master")
  169. shellOptions.Masters = &masterAddress
  170. shellOptions.FilerHost = "localhost"
  171. shellOptions.FilerPort = 8888
  172. shellOptions.Directory = "/"
  173. commandEnv := shell.NewCommandEnv(shellOptions)
  174. reg, _ := regexp.Compile(`'.*?'|".*?"|\S+`)
  175. go commandEnv.MasterClient.KeepConnectedToMaster()
  176. go func() {
  177. commandEnv.MasterClient.WaitUntilConnected()
  178. c := time.Tick(time.Duration(sleepMinutes) * time.Minute)
  179. for _ = range c {
  180. if ms.Topo.IsLeader() {
  181. for _, line := range scriptLines {
  182. cmds := reg.FindAllString(line, -1)
  183. if len(cmds) == 0 {
  184. continue
  185. }
  186. args := make([]string, len(cmds[1:]))
  187. for i := range args {
  188. args[i] = strings.Trim(string(cmds[1+i]), "\"'")
  189. }
  190. cmd := strings.ToLower(cmds[0])
  191. for _, c := range shell.Commands {
  192. if c.Name() == cmd {
  193. glog.V(0).Infof("executing: %s %v", cmd, args)
  194. if err := c.Do(args, commandEnv, os.Stdout); err != nil {
  195. glog.V(0).Infof("error: %v", err)
  196. }
  197. }
  198. }
  199. }
  200. }
  201. }
  202. }()
  203. }