volume.go 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405
  1. package command
  2. import (
  3. "fmt"
  4. "net/http"
  5. httppprof "net/http/pprof"
  6. "os"
  7. "runtime/pprof"
  8. "strconv"
  9. "strings"
  10. "time"
  11. "github.com/seaweedfs/seaweedfs/weed/storage/types"
  12. "github.com/spf13/viper"
  13. "google.golang.org/grpc"
  14. "github.com/seaweedfs/seaweedfs/weed/util/grace"
  15. "github.com/seaweedfs/seaweedfs/weed/pb"
  16. "github.com/seaweedfs/seaweedfs/weed/security"
  17. "github.com/seaweedfs/seaweedfs/weed/server/constants"
  18. "github.com/seaweedfs/seaweedfs/weed/util/httpdown"
  19. "google.golang.org/grpc/reflection"
  20. "github.com/seaweedfs/seaweedfs/weed/glog"
  21. "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
  22. weed_server "github.com/seaweedfs/seaweedfs/weed/server"
  23. stats_collect "github.com/seaweedfs/seaweedfs/weed/stats"
  24. "github.com/seaweedfs/seaweedfs/weed/storage"
  25. "github.com/seaweedfs/seaweedfs/weed/util"
  26. )
  27. var (
  28. v VolumeServerOptions
  29. )
  30. type VolumeServerOptions struct {
  31. port *int
  32. portGrpc *int
  33. publicPort *int
  34. folders []string
  35. folderMaxLimits []int32
  36. idxFolder *string
  37. ip *string
  38. publicUrl *string
  39. bindIp *string
  40. mastersString *string
  41. masters []pb.ServerAddress
  42. idleConnectionTimeout *int
  43. dataCenter *string
  44. rack *string
  45. whiteList []string
  46. indexType *string
  47. diskType *string
  48. fixJpgOrientation *bool
  49. readMode *string
  50. cpuProfile *string
  51. memProfile *string
  52. compactionMBPerSecond *int
  53. fileSizeLimitMB *int
  54. concurrentUploadLimitMB *int
  55. concurrentDownloadLimitMB *int
  56. pprof *bool
  57. preStopSeconds *int
  58. metricsHttpPort *int
  59. metricsHttpIp *string
  60. // pulseSeconds *int
  61. inflightUploadDataTimeout *time.Duration
  62. hasSlowRead *bool
  63. readBufferSizeMB *int
  64. ldbTimeout *int64
  65. }
  66. func init() {
  67. cmdVolume.Run = runVolume // break init cycle
  68. v.port = cmdVolume.Flag.Int("port", 8080, "http listen port")
  69. v.portGrpc = cmdVolume.Flag.Int("port.grpc", 0, "grpc listen port")
  70. v.publicPort = cmdVolume.Flag.Int("port.public", 0, "port opened to public")
  71. v.ip = cmdVolume.Flag.String("ip", util.DetectedHostAddress(), "ip or server name, also used as identifier")
  72. v.publicUrl = cmdVolume.Flag.String("publicUrl", "", "Publicly accessible address")
  73. v.bindIp = cmdVolume.Flag.String("ip.bind", "", "ip address to bind to. If empty, default to same as -ip option.")
  74. v.mastersString = cmdVolume.Flag.String("mserver", "localhost:9333", "comma-separated master servers")
  75. v.preStopSeconds = cmdVolume.Flag.Int("preStopSeconds", 10, "number of seconds between stop send heartbeats and stop volume server")
  76. // v.pulseSeconds = cmdVolume.Flag.Int("pulseSeconds", 5, "number of seconds between heartbeats, must be smaller than or equal to the master's setting")
  77. v.idleConnectionTimeout = cmdVolume.Flag.Int("idleTimeout", 30, "connection idle seconds")
  78. v.dataCenter = cmdVolume.Flag.String("dataCenter", "", "current volume server's data center name")
  79. v.rack = cmdVolume.Flag.String("rack", "", "current volume server's rack name")
  80. v.indexType = cmdVolume.Flag.String("index", "memory", "Choose [memory|leveldb|leveldbMedium|leveldbLarge] mode for memory~performance balance.")
  81. v.diskType = cmdVolume.Flag.String("disk", "", "[hdd|ssd|<tag>] hard drive or solid state drive or any tag")
  82. v.fixJpgOrientation = cmdVolume.Flag.Bool("images.fix.orientation", false, "Adjust jpg orientation when uploading.")
  83. v.readMode = cmdVolume.Flag.String("readMode", "proxy", "[local|proxy|redirect] how to deal with non-local volume: 'not found|proxy to remote node|redirect volume location'.")
  84. v.cpuProfile = cmdVolume.Flag.String("cpuprofile", "", "cpu profile output file")
  85. v.memProfile = cmdVolume.Flag.String("memprofile", "", "memory profile output file")
  86. v.compactionMBPerSecond = cmdVolume.Flag.Int("compactionMBps", 0, "limit background compaction or copying speed in mega bytes per second")
  87. v.fileSizeLimitMB = cmdVolume.Flag.Int("fileSizeLimitMB", 256, "limit file size to avoid out of memory")
  88. v.ldbTimeout = cmdVolume.Flag.Int64("index.leveldbTimeout", 0, "alive time for leveldb (default to 0). If leveldb of volume is not accessed in ldbTimeout hours, it will be off loaded to reduce opened files and memory consumption.")
  89. v.concurrentUploadLimitMB = cmdVolume.Flag.Int("concurrentUploadLimitMB", 256, "limit total concurrent upload size")
  90. v.concurrentDownloadLimitMB = cmdVolume.Flag.Int("concurrentDownloadLimitMB", 256, "limit total concurrent download size")
  91. v.pprof = cmdVolume.Flag.Bool("pprof", false, "enable pprof http handlers. precludes --memprofile and --cpuprofile")
  92. v.metricsHttpPort = cmdVolume.Flag.Int("metricsPort", 0, "Prometheus metrics listen port")
  93. v.metricsHttpIp = cmdVolume.Flag.String("metricsIp", "", "metrics listen ip. If empty, default to same as -ip.bind option.")
  94. v.idxFolder = cmdVolume.Flag.String("dir.idx", "", "directory to store .idx files")
  95. v.inflightUploadDataTimeout = cmdVolume.Flag.Duration("inflightUploadDataTimeout", 60*time.Second, "inflight upload data wait timeout of volume servers")
  96. v.hasSlowRead = cmdVolume.Flag.Bool("hasSlowRead", true, "<experimental> if true, this prevents slow reads from blocking other requests, but large file read P99 latency will increase.")
  97. v.readBufferSizeMB = cmdVolume.Flag.Int("readBufferSizeMB", 4, "<experimental> larger values can optimize query performance but will increase some memory usage,Use with hasSlowRead normally.")
  98. }
  99. var cmdVolume = &Command{
  100. UsageLine: "volume -port=8080 -dir=/tmp -max=5 -ip=server_name -mserver=localhost:9333",
  101. Short: "start a volume server",
  102. Long: `start a volume server to provide storage spaces
  103. `,
  104. }
  105. var (
  106. volumeFolders = cmdVolume.Flag.String("dir", os.TempDir(), "directories to store data files. dir[,dir]...")
  107. maxVolumeCounts = cmdVolume.Flag.String("max", "8", "maximum numbers of volumes, count[,count]... If set to zero, the limit will be auto configured as free disk space divided by volume size.")
  108. volumeWhiteListOption = cmdVolume.Flag.String("whiteList", "", "comma separated Ip addresses having write permission. No limit if empty.")
  109. minFreeSpacePercent = cmdVolume.Flag.String("minFreeSpacePercent", "1", "minimum free disk space (default to 1%). Low disk space will mark all volumes as ReadOnly (deprecated, use minFreeSpace instead).")
  110. minFreeSpace = cmdVolume.Flag.String("minFreeSpace", "", "min free disk space (value<=100 as percentage like 1, other as human readable bytes, like 10GiB). Low disk space will mark all volumes as ReadOnly.")
  111. )
  112. func runVolume(cmd *Command, args []string) bool {
  113. util.LoadSecurityConfiguration()
  114. // If --pprof is set we assume the caller wants to be able to collect
  115. // cpu and memory profiles via go tool pprof
  116. if !*v.pprof {
  117. grace.SetupProfiling(*v.cpuProfile, *v.memProfile)
  118. }
  119. switch {
  120. case *v.metricsHttpIp != "":
  121. // noting to do, use v.metricsHttpIp
  122. case *v.bindIp != "":
  123. *v.metricsHttpIp = *v.bindIp
  124. case *v.ip != "":
  125. *v.metricsHttpIp = *v.ip
  126. }
  127. go stats_collect.StartMetricsServer(*v.metricsHttpIp, *v.metricsHttpPort)
  128. minFreeSpaces := util.MustParseMinFreeSpace(*minFreeSpace, *minFreeSpacePercent)
  129. v.masters = pb.ServerAddresses(*v.mastersString).ToAddresses()
  130. v.startVolumeServer(*volumeFolders, *maxVolumeCounts, *volumeWhiteListOption, minFreeSpaces)
  131. return true
  132. }
  133. func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, volumeWhiteListOption string, minFreeSpaces []util.MinFreeSpace) {
  134. // Set multiple folders and each folder's max volume count limit'
  135. v.folders = strings.Split(volumeFolders, ",")
  136. for _, folder := range v.folders {
  137. if err := util.TestFolderWritable(util.ResolvePath(folder)); err != nil {
  138. glog.Fatalf("Check Data Folder(-dir) Writable %s : %s", folder, err)
  139. }
  140. }
  141. // set max
  142. maxCountStrings := strings.Split(maxVolumeCounts, ",")
  143. for _, maxString := range maxCountStrings {
  144. if max, e := strconv.ParseInt(maxString, 10, 64); e == nil {
  145. v.folderMaxLimits = append(v.folderMaxLimits, int32(max))
  146. } else {
  147. glog.Fatalf("The max specified in -max not a valid number %s", maxString)
  148. }
  149. }
  150. if len(v.folderMaxLimits) == 1 && len(v.folders) > 1 {
  151. for i := 0; i < len(v.folders)-1; i++ {
  152. v.folderMaxLimits = append(v.folderMaxLimits, v.folderMaxLimits[0])
  153. }
  154. }
  155. if len(v.folders) != len(v.folderMaxLimits) {
  156. glog.Fatalf("%d directories by -dir, but only %d max is set by -max", len(v.folders), len(v.folderMaxLimits))
  157. }
  158. if len(minFreeSpaces) == 1 && len(v.folders) > 1 {
  159. for i := 0; i < len(v.folders)-1; i++ {
  160. minFreeSpaces = append(minFreeSpaces, minFreeSpaces[0])
  161. }
  162. }
  163. if len(v.folders) != len(minFreeSpaces) {
  164. glog.Fatalf("%d directories by -dir, but only %d minFreeSpacePercent is set by -minFreeSpacePercent", len(v.folders), len(minFreeSpaces))
  165. }
  166. // set disk types
  167. var diskTypes []types.DiskType
  168. diskTypeStrings := strings.Split(*v.diskType, ",")
  169. for _, diskTypeString := range diskTypeStrings {
  170. diskTypes = append(diskTypes, types.ToDiskType(diskTypeString))
  171. }
  172. if len(diskTypes) == 1 && len(v.folders) > 1 {
  173. for i := 0; i < len(v.folders)-1; i++ {
  174. diskTypes = append(diskTypes, diskTypes[0])
  175. }
  176. }
  177. if len(v.folders) != len(diskTypes) {
  178. glog.Fatalf("%d directories by -dir, but only %d disk types is set by -disk", len(v.folders), len(diskTypes))
  179. }
  180. // security related white list configuration
  181. v.whiteList = util.StringSplit(volumeWhiteListOption, ",")
  182. if *v.ip == "" {
  183. *v.ip = util.DetectedHostAddress()
  184. glog.V(0).Infof("detected volume server ip address: %v", *v.ip)
  185. }
  186. if *v.bindIp == "" {
  187. *v.bindIp = *v.ip
  188. }
  189. if *v.publicPort == 0 {
  190. *v.publicPort = *v.port
  191. }
  192. if *v.portGrpc == 0 {
  193. *v.portGrpc = 10000 + *v.port
  194. }
  195. if *v.publicUrl == "" {
  196. *v.publicUrl = util.JoinHostPort(*v.ip, *v.publicPort)
  197. }
  198. volumeMux := http.NewServeMux()
  199. publicVolumeMux := volumeMux
  200. if v.isSeparatedPublicPort() {
  201. publicVolumeMux = http.NewServeMux()
  202. }
  203. if *v.pprof {
  204. volumeMux.HandleFunc("/debug/pprof/", httppprof.Index)
  205. volumeMux.HandleFunc("/debug/pprof/cmdline", httppprof.Cmdline)
  206. volumeMux.HandleFunc("/debug/pprof/profile", httppprof.Profile)
  207. volumeMux.HandleFunc("/debug/pprof/symbol", httppprof.Symbol)
  208. volumeMux.HandleFunc("/debug/pprof/trace", httppprof.Trace)
  209. }
  210. volumeNeedleMapKind := storage.NeedleMapInMemory
  211. switch *v.indexType {
  212. case "leveldb":
  213. volumeNeedleMapKind = storage.NeedleMapLevelDb
  214. case "leveldbMedium":
  215. volumeNeedleMapKind = storage.NeedleMapLevelDbMedium
  216. case "leveldbLarge":
  217. volumeNeedleMapKind = storage.NeedleMapLevelDbLarge
  218. }
  219. volumeServer := weed_server.NewVolumeServer(volumeMux, publicVolumeMux,
  220. *v.ip, *v.port, *v.portGrpc, *v.publicUrl,
  221. v.folders, v.folderMaxLimits, minFreeSpaces, diskTypes,
  222. *v.idxFolder,
  223. volumeNeedleMapKind,
  224. v.masters, constants.VolumePulseSeconds, *v.dataCenter, *v.rack,
  225. v.whiteList,
  226. *v.fixJpgOrientation, *v.readMode,
  227. *v.compactionMBPerSecond,
  228. *v.fileSizeLimitMB,
  229. int64(*v.concurrentUploadLimitMB)*1024*1024,
  230. int64(*v.concurrentDownloadLimitMB)*1024*1024,
  231. *v.inflightUploadDataTimeout,
  232. *v.hasSlowRead,
  233. *v.readBufferSizeMB,
  234. *v.ldbTimeout,
  235. )
  236. // starting grpc server
  237. grpcS := v.startGrpcService(volumeServer)
  238. // starting public http server
  239. var publicHttpDown httpdown.Server
  240. if v.isSeparatedPublicPort() {
  241. publicHttpDown = v.startPublicHttpService(publicVolumeMux)
  242. if nil == publicHttpDown {
  243. glog.Fatalf("start public http service failed")
  244. }
  245. }
  246. // starting the cluster http server
  247. clusterHttpServer := v.startClusterHttpService(volumeMux)
  248. grace.OnReload(volumeServer.LoadNewVolumes)
  249. stopChan := make(chan bool)
  250. grace.OnInterrupt(func() {
  251. fmt.Println("volume server has been killed")
  252. // Stop heartbeats
  253. if !volumeServer.StopHeartbeat() {
  254. volumeServer.SetStopping()
  255. glog.V(0).Infof("stop send heartbeat and wait %d seconds until shutdown ...", *v.preStopSeconds)
  256. time.Sleep(time.Duration(*v.preStopSeconds) * time.Second)
  257. }
  258. shutdown(publicHttpDown, clusterHttpServer, grpcS, volumeServer)
  259. stopChan <- true
  260. })
  261. select {
  262. case <-stopChan:
  263. }
  264. }
  265. func shutdown(publicHttpDown httpdown.Server, clusterHttpServer httpdown.Server, grpcS *grpc.Server, volumeServer *weed_server.VolumeServer) {
  266. // firstly, stop the public http service to prevent from receiving new user request
  267. if nil != publicHttpDown {
  268. glog.V(0).Infof("stop public http server ... ")
  269. if err := publicHttpDown.Stop(); err != nil {
  270. glog.Warningf("stop the public http server failed, %v", err)
  271. }
  272. }
  273. glog.V(0).Infof("graceful stop cluster http server ... ")
  274. if err := clusterHttpServer.Stop(); err != nil {
  275. glog.Warningf("stop the cluster http server failed, %v", err)
  276. }
  277. glog.V(0).Infof("graceful stop gRPC ...")
  278. grpcS.GracefulStop()
  279. volumeServer.Shutdown()
  280. pprof.StopCPUProfile()
  281. }
  282. // check whether configure the public port
  283. func (v VolumeServerOptions) isSeparatedPublicPort() bool {
  284. return *v.publicPort != *v.port
  285. }
  286. func (v VolumeServerOptions) startGrpcService(vs volume_server_pb.VolumeServerServer) *grpc.Server {
  287. grpcPort := *v.portGrpc
  288. grpcL, err := util.NewListener(util.JoinHostPort(*v.bindIp, grpcPort), 0)
  289. if err != nil {
  290. glog.Fatalf("failed to listen on grpc port %d: %v", grpcPort, err)
  291. }
  292. grpcS := pb.NewGrpcServer(security.LoadServerTLS(util.GetViper(), "grpc.volume"))
  293. volume_server_pb.RegisterVolumeServerServer(grpcS, vs)
  294. reflection.Register(grpcS)
  295. go func() {
  296. if err := grpcS.Serve(grpcL); err != nil {
  297. glog.Fatalf("start gRPC service failed, %s", err)
  298. }
  299. }()
  300. return grpcS
  301. }
  302. func (v VolumeServerOptions) startPublicHttpService(handler http.Handler) httpdown.Server {
  303. publicListeningAddress := util.JoinHostPort(*v.bindIp, *v.publicPort)
  304. glog.V(0).Infoln("Start Seaweed volume server", util.Version(), "public at", publicListeningAddress)
  305. publicListener, e := util.NewListener(publicListeningAddress, time.Duration(*v.idleConnectionTimeout)*time.Second)
  306. if e != nil {
  307. glog.Fatalf("Volume server listener error:%v", e)
  308. }
  309. pubHttp := httpdown.HTTP{StopTimeout: 5 * time.Minute, KillTimeout: 5 * time.Minute}
  310. publicHttpDown := pubHttp.Serve(&http.Server{Handler: handler}, publicListener)
  311. go func() {
  312. if err := publicHttpDown.Wait(); err != nil {
  313. glog.Errorf("public http down wait failed, %v", err)
  314. }
  315. }()
  316. return publicHttpDown
  317. }
  318. func (v VolumeServerOptions) startClusterHttpService(handler http.Handler) httpdown.Server {
  319. var (
  320. certFile, keyFile string
  321. )
  322. if viper.GetString("https.volume.key") != "" {
  323. certFile = viper.GetString("https.volume.cert")
  324. keyFile = viper.GetString("https.volume.key")
  325. }
  326. listeningAddress := util.JoinHostPort(*v.bindIp, *v.port)
  327. glog.V(0).Infof("Start Seaweed volume server %s at %s", util.Version(), listeningAddress)
  328. listener, e := util.NewListener(listeningAddress, time.Duration(*v.idleConnectionTimeout)*time.Second)
  329. if e != nil {
  330. glog.Fatalf("Volume server listener error:%v", e)
  331. }
  332. httpDown := httpdown.HTTP{
  333. KillTimeout: time.Minute,
  334. StopTimeout: 30 * time.Second,
  335. CertFile: certFile,
  336. KeyFile: keyFile}
  337. httpS := &http.Server{Handler: handler}
  338. if viper.GetString("https.volume.ca") != "" {
  339. clientCertFile := viper.GetString("https.volume.ca")
  340. httpS.TLSConfig = security.LoadClientTLSHTTP(clientCertFile)
  341. }
  342. clusterHttpServer := httpDown.Serve(httpS, listener)
  343. go func() {
  344. if e := clusterHttpServer.Wait(); e != nil {
  345. glog.Fatalf("Volume server fail to serve: %v", e)
  346. }
  347. }()
  348. return clusterHttpServer
  349. }