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.

146 lines
4.1 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. masterNodes *storage.MasterNodes
  29. }
  30. func NewFilerServer(r *http.ServeMux, ip string, port int, master string, dir string, collection string,
  31. replication string, redirectOnRead bool, disableDirListing bool,
  32. secret string,
  33. cassandra_server string, cassandra_keyspace string,
  34. redis_server string, redis_password string, redis_database int,
  35. ) (fs *FilerServer, err error) {
  36. fs = &FilerServer{
  37. master: master,
  38. collection: collection,
  39. defaultReplication: replication,
  40. redirectOnRead: redirectOnRead,
  41. disableDirListing: disableDirListing,
  42. port: ip + ":" + strconv.Itoa(port),
  43. }
  44. if cassandra_server != "" {
  45. cassandra_store, err := cassandra_store.NewCassandraStore(cassandra_keyspace, cassandra_server)
  46. if err != nil {
  47. glog.Fatalf("Can not connect to cassandra server %s with keyspace %s: %v", cassandra_server, cassandra_keyspace, err)
  48. }
  49. fs.filer = flat_namespace.NewFlatNamespaceFiler(master, cassandra_store)
  50. } else if redis_server != "" {
  51. redis_store := redis_store.NewRedisStore(redis_server, redis_password, redis_database)
  52. fs.filer = flat_namespace.NewFlatNamespaceFiler(master, redis_store)
  53. } else {
  54. if fs.filer, err = embedded_filer.NewFilerEmbedded(master, dir); err != nil {
  55. glog.Fatalf("Can not start filer in dir %s : %v", dir, err)
  56. return
  57. }
  58. r.HandleFunc("/admin/mv", fs.moveHandler)
  59. }
  60. r.HandleFunc("/", fs.filerHandler)
  61. go func() {
  62. connected := true
  63. fs.masterNodes = storage.NewMasterNodes(fs.master)
  64. glog.V(0).Infof("Filer server bootstraps with master %s", fs.getMasterNode())
  65. //force initialize with all available master nodes
  66. fs.masterNodes.FindMaster()
  67. for {
  68. glog.V(4).Infof("Filer server sending to master %s", fs.getMasterNode())
  69. master, err := fs.detectHealthyMaster(fs.getMasterNode())
  70. if err == nil {
  71. if !connected {
  72. connected = true
  73. if fs.getMasterNode() != master {
  74. fs.setMasterNode(master)
  75. }
  76. glog.V(0).Infoln("Filer Server Connected with master at", master)
  77. }
  78. } else {
  79. glog.V(1).Infof("Filer Server Failed to talk with master %s: %v", fs.getMasterNode(), err)
  80. if connected {
  81. connected = false
  82. }
  83. }
  84. if connected {
  85. time.Sleep(time.Duration(float32(10*1e3)*(1+rand.Float32())) * time.Millisecond)
  86. } else {
  87. time.Sleep(time.Duration(float32(10*1e3)*0.25) * time.Millisecond)
  88. }
  89. }
  90. }()
  91. return fs, nil
  92. }
  93. func (fs *FilerServer) jwt(fileId string) security.EncodedJwt {
  94. return security.GenJwt(fs.secret, fileId)
  95. }
  96. func (fs *FilerServer) getMasterNode() string {
  97. fs.mnLock.RLock()
  98. defer fs.mnLock.RUnlock()
  99. return fs.master
  100. }
  101. func (fs *FilerServer) setMasterNode(masterNode string) {
  102. fs.mnLock.Lock()
  103. defer fs.mnLock.Unlock()
  104. fs.master = masterNode
  105. }
  106. func (fs *FilerServer) detectHealthyMaster(masterNode string) (master string, e error) {
  107. if e = checkMaster(masterNode); e != nil {
  108. fs.masterNodes.Reset()
  109. for i := 0; i <= 3; i++ {
  110. master, e = fs.masterNodes.FindMaster()
  111. if e != nil {
  112. continue
  113. } else {
  114. if e = checkMaster(masterNode); e == nil {
  115. break
  116. }
  117. }
  118. }
  119. } else {
  120. master = masterNode
  121. }
  122. return
  123. }
  124. func checkMaster(masterNode string) error {
  125. statUrl := "http://" + masterNode + "/stats"
  126. glog.V(4).Infof("Connecting to %s ...", statUrl)
  127. _, e := util.Get(statUrl)
  128. return e
  129. }