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.

360 lines
11 KiB

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