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.

199 lines
5.5 KiB

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