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.

135 lines
4.1 KiB

7 years ago
7 years ago
6 years ago
7 years ago
7 years ago
7 years ago
6 years ago
7 years ago
6 years ago
6 years ago
  1. package weed_server
  2. import (
  3. "context"
  4. "fmt"
  5. "net/http"
  6. "os"
  7. "time"
  8. "google.golang.org/grpc"
  9. "github.com/chrislusf/seaweedfs/weed/operation"
  10. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  11. "github.com/chrislusf/seaweedfs/weed/stats"
  12. "github.com/chrislusf/seaweedfs/weed/util"
  13. "github.com/chrislusf/seaweedfs/weed/filer2"
  14. _ "github.com/chrislusf/seaweedfs/weed/filer2/cassandra"
  15. _ "github.com/chrislusf/seaweedfs/weed/filer2/etcd"
  16. _ "github.com/chrislusf/seaweedfs/weed/filer2/leveldb"
  17. _ "github.com/chrislusf/seaweedfs/weed/filer2/leveldb2"
  18. _ "github.com/chrislusf/seaweedfs/weed/filer2/mysql"
  19. _ "github.com/chrislusf/seaweedfs/weed/filer2/postgres"
  20. _ "github.com/chrislusf/seaweedfs/weed/filer2/redis"
  21. _ "github.com/chrislusf/seaweedfs/weed/filer2/tikv"
  22. "github.com/chrislusf/seaweedfs/weed/glog"
  23. "github.com/chrislusf/seaweedfs/weed/notification"
  24. _ "github.com/chrislusf/seaweedfs/weed/notification/aws_sqs"
  25. _ "github.com/chrislusf/seaweedfs/weed/notification/gocdk_pub_sub"
  26. _ "github.com/chrislusf/seaweedfs/weed/notification/google_pub_sub"
  27. _ "github.com/chrislusf/seaweedfs/weed/notification/kafka"
  28. _ "github.com/chrislusf/seaweedfs/weed/notification/log"
  29. "github.com/chrislusf/seaweedfs/weed/security"
  30. )
  31. type FilerOption struct {
  32. Masters []string
  33. Collection string
  34. DefaultReplication string
  35. RedirectOnRead bool
  36. DisableDirListing bool
  37. MaxMB int
  38. DirListingLimit int
  39. DataCenter string
  40. DefaultLevelDbDir string
  41. DisableHttp bool
  42. Port int
  43. recursiveDelete bool
  44. DirBucketsPath string
  45. }
  46. type FilerServer struct {
  47. option *FilerOption
  48. secret security.SigningKey
  49. filer *filer2.Filer
  50. grpcDialOption grpc.DialOption
  51. }
  52. func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) (fs *FilerServer, err error) {
  53. fs = &FilerServer{
  54. option: option,
  55. grpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.filer"),
  56. }
  57. if len(option.Masters) == 0 {
  58. glog.Fatal("master list is required!")
  59. }
  60. fs.filer = filer2.NewFiler(option.Masters, fs.grpcDialOption)
  61. go fs.filer.KeepConnectedToMaster()
  62. v := util.GetViper()
  63. if !util.LoadConfiguration("filer", false) {
  64. v.Set("leveldb2.enabled", true)
  65. v.Set("leveldb2.dir", option.DefaultLevelDbDir)
  66. _, err := os.Stat(option.DefaultLevelDbDir)
  67. if os.IsNotExist(err) {
  68. os.MkdirAll(option.DefaultLevelDbDir, 0755)
  69. }
  70. }
  71. util.LoadConfiguration("notification", false)
  72. fs.option.recursiveDelete = v.GetBool("filer.options.recursive_delete")
  73. fs.filer.LoadConfiguration(v)
  74. notification.LoadConfiguration(v, "notification.")
  75. handleStaticResources(defaultMux)
  76. if !option.DisableHttp {
  77. defaultMux.HandleFunc("/", fs.filerHandler)
  78. }
  79. if defaultMux != readonlyMux {
  80. readonlyMux.HandleFunc("/", fs.readonlyFilerHandler)
  81. }
  82. maybeStartMetrics(fs, option)
  83. return fs, nil
  84. }
  85. func maybeStartMetrics(fs *FilerServer, option *FilerOption) {
  86. isConnected := false
  87. var metricsAddress string
  88. var metricsIntervalSec int
  89. var readErr error
  90. for !isConnected {
  91. metricsAddress, metricsIntervalSec, readErr = readFilerConfiguration(fs.grpcDialOption, option.Masters[0])
  92. if readErr == nil {
  93. isConnected = true
  94. } else {
  95. time.Sleep(7 * time.Second)
  96. }
  97. }
  98. if metricsAddress == "" && metricsIntervalSec <= 0 {
  99. return
  100. }
  101. go stats.LoopPushingMetric("filer", stats.SourceName(option.Port), stats.FilerGather,
  102. func() (addr string, intervalSeconds int) {
  103. return metricsAddress, metricsIntervalSec
  104. })
  105. }
  106. func readFilerConfiguration(grpcDialOption grpc.DialOption, masterGrpcAddress string) (metricsAddress string, metricsIntervalSec int, err error) {
  107. err = operation.WithMasterServerClient(masterGrpcAddress, grpcDialOption, func(ctx context.Context, masterClient master_pb.SeaweedClient) error {
  108. resp, err := masterClient.GetMasterConfiguration(ctx, &master_pb.GetMasterConfigurationRequest{})
  109. if err != nil {
  110. return fmt.Errorf("get master %s configuration: %v", masterGrpcAddress, err)
  111. }
  112. metricsAddress, metricsIntervalSec = resp.MetricsAddress, int(resp.MetricsIntervalSeconds)
  113. return nil
  114. })
  115. return
  116. }