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.

157 lines
4.3 KiB

11 years ago
10 years ago
10 years ago
10 years ago
  1. package weed_server
  2. import (
  3. "math/rand"
  4. "net/http"
  5. "strconv"
  6. "sync"
  7. "time"
  8. "github.com/chrislusf/seaweedfs/weed/filer"
  9. "github.com/chrislusf/seaweedfs/weed/filer/cassandra_store"
  10. "github.com/chrislusf/seaweedfs/weed/filer/embedded_filer"
  11. "github.com/chrislusf/seaweedfs/weed/filer/flat_namespace"
  12. "github.com/chrislusf/seaweedfs/weed/filer/redis_store"
  13. "github.com/chrislusf/seaweedfs/weed/glog"
  14. "github.com/chrislusf/seaweedfs/weed/security"
  15. "github.com/chrislusf/seaweedfs/weed/storage"
  16. "github.com/chrislusf/seaweedfs/weed/util"
  17. )
  18. type FilerServer struct {
  19. port string
  20. master string
  21. mnLock sync.RWMutex
  22. collection string
  23. defaultReplication string
  24. redirectOnRead bool
  25. disableDirListing bool
  26. secret security.Secret
  27. filer filer.Filer
  28. maxMB int
  29. masterNodes *storage.MasterNodes
  30. }
  31. func NewFilerServer(r *http.ServeMux, ip string, port int, master string, dir string, collection string,
  32. replication string, redirectOnRead bool, disableDirListing bool,
  33. secret string,
  34. cassandra_server string, cassandra_keyspace string,
  35. redis_server string, redis_password string, redis_database int,
  36. ) (fs *FilerServer, err error) {
  37. fs = &FilerServer{
  38. master: master,
  39. collection: collection,
  40. defaultReplication: replication,
  41. redirectOnRead: redirectOnRead,
  42. disableDirListing: disableDirListing,
  43. maxMB: maxMB,
  44. port: ip + ":" + strconv.Itoa(port),
  45. }
  46. if cassandra_server != "" {
  47. cassandra_store, err := cassandra_store.NewCassandraStore(cassandra_keyspace, cassandra_server)
  48. if err != nil {
  49. glog.Fatalf("Can not connect to cassandra server %s with keyspace %s: %v", cassandra_server, cassandra_keyspace, err)
  50. }
  51. fs.filer = flat_namespace.NewFlatNamespaceFiler(master, cassandra_store)
  52. } else if redis_server != "" {
  53. redis_store := redis_store.NewRedisStore(redis_server, redis_password, redis_database)
  54. fs.filer = flat_namespace.NewFlatNamespaceFiler(master, redis_store)
  55. } else {
  56. if fs.filer, err = embedded_filer.NewFilerEmbedded(master, dir); err != nil {
  57. glog.Fatalf("Can not start filer in dir %s : %v", dir, err)
  58. return
  59. }
  60. r.HandleFunc("/admin/mv", fs.moveHandler)
  61. r.HandleFunc("/admin/register", fs.registerHandler)
  62. }
  63. r.HandleFunc("/", fs.filerHandler)
  64. go func() {
  65. connected := true
  66. fs.masterNodes = storage.NewMasterNodes(fs.master)
  67. glog.V(0).Infof("Filer server bootstraps with master %s", fs.getMasterNode())
  68. //force initialize with all available master nodes
  69. for {
  70. _, err := fs.masterNodes.FindMaster()
  71. if err != nil {
  72. glog.Infof("filer server failed to get master cluster info:%s", err.Error())
  73. time.Sleep(3 * time.Second)
  74. } else {
  75. break
  76. }
  77. }
  78. for {
  79. glog.V(4).Infof("Filer server sending to master %s", fs.getMasterNode())
  80. master, err := fs.detectHealthyMaster(fs.getMasterNode())
  81. if err == nil {
  82. if !connected {
  83. connected = true
  84. if fs.getMasterNode() != master {
  85. fs.setMasterNode(master)
  86. }
  87. glog.V(0).Infoln("Filer Server Connected with master at", master)
  88. }
  89. } else {
  90. glog.V(1).Infof("Filer Server Failed to talk with master %s: %v", fs.getMasterNode(), err)
  91. if connected {
  92. connected = false
  93. }
  94. }
  95. if connected {
  96. time.Sleep(time.Duration(float32(10*1e3)*(1+rand.Float32())) * time.Millisecond)
  97. } else {
  98. time.Sleep(time.Duration(float32(10*1e3)*0.25) * time.Millisecond)
  99. }
  100. }
  101. }()
  102. return fs, nil
  103. }
  104. func (fs *FilerServer) jwt(fileId string) security.EncodedJwt {
  105. return security.GenJwt(fs.secret, fileId)
  106. }
  107. func (fs *FilerServer) getMasterNode() string {
  108. fs.mnLock.RLock()
  109. defer fs.mnLock.RUnlock()
  110. return fs.master
  111. }
  112. func (fs *FilerServer) setMasterNode(masterNode string) {
  113. fs.mnLock.Lock()
  114. defer fs.mnLock.Unlock()
  115. fs.master = masterNode
  116. }
  117. func (fs *FilerServer) detectHealthyMaster(masterNode string) (master string, e error) {
  118. if e = checkMaster(masterNode); e != nil {
  119. fs.masterNodes.Reset()
  120. for i := 0; i <= 3; i++ {
  121. master, e = fs.masterNodes.FindMaster()
  122. if e != nil {
  123. continue
  124. } else {
  125. if e = checkMaster(master); e == nil {
  126. break
  127. }
  128. }
  129. }
  130. } else {
  131. master = masterNode
  132. }
  133. return
  134. }
  135. func checkMaster(masterNode string) error {
  136. statUrl := "http://" + masterNode + "/stats"
  137. glog.V(4).Infof("Connecting to %s ...", statUrl)
  138. _, e := util.Get(statUrl)
  139. return e
  140. }