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.

287 lines
8.9 KiB

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