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.

384 lines
12 KiB

12 years ago
12 years ago
12 years ago
12 years ago
12 years ago
12 years ago
  1. package main
  2. import (
  3. "code.google.com/p/weed-fs/go/glog"
  4. "code.google.com/p/weed-fs/go/operation"
  5. "code.google.com/p/weed-fs/go/replication"
  6. "code.google.com/p/weed-fs/go/storage"
  7. "math/rand"
  8. "mime"
  9. "net/http"
  10. "os"
  11. "path/filepath"
  12. "runtime"
  13. "strconv"
  14. "strings"
  15. "time"
  16. )
  17. func init() {
  18. cmdVolume.Run = runVolume // break init cycle
  19. cmdVolume.IsDebug = cmdVolume.Flag.Bool("debug", false, "enable debug mode")
  20. }
  21. var cmdVolume = &Command{
  22. UsageLine: "volume -port=8080 -dir=/tmp -max=5 -ip=server_name -mserver=localhost:9333",
  23. Short: "start a volume server",
  24. Long: `start a volume server to provide storage spaces
  25. `,
  26. }
  27. var (
  28. vport = cmdVolume.Flag.Int("port", 8080, "http listen port")
  29. volumeFolders = cmdVolume.Flag.String("dir", os.TempDir(), "directories to store data files. dir[,dir]...")
  30. maxVolumeCounts = cmdVolume.Flag.String("max", "7", "maximum numbers of volumes, count[,count]...")
  31. ip = cmdVolume.Flag.String("ip", "localhost", "ip or server name")
  32. publicUrl = cmdVolume.Flag.String("publicUrl", "", "Publicly accessible <ip|server_name>:<port>")
  33. masterNode = cmdVolume.Flag.String("mserver", "localhost:9333", "master server location")
  34. vpulse = cmdVolume.Flag.Int("pulseSeconds", 5, "number of seconds between heartbeats, must be smaller than the master's setting")
  35. vReadTimeout = cmdVolume.Flag.Int("readTimeout", 3, "connection read timeout in seconds. Increase this if uploading large files.")
  36. vMaxCpu = cmdVolume.Flag.Int("maxCpu", 0, "maximum number of CPUs. 0 means all available CPUs")
  37. dataCenter = cmdVolume.Flag.String("dataCenter", "", "current volume server's data center name")
  38. rack = cmdVolume.Flag.String("rack", "", "current volume server's rack name")
  39. volumeWhiteListOption = cmdVolume.Flag.String("whiteList", "", "comma separated Ip addresses having write permission. No limit if empty.")
  40. store *storage.Store
  41. volumeWhiteList []string
  42. )
  43. var fileNameEscaper = strings.NewReplacer("\\", "\\\\", "\"", "\\\"")
  44. func statusHandler(w http.ResponseWriter, r *http.Request) {
  45. m := make(map[string]interface{})
  46. m["Version"] = VERSION
  47. m["Volumes"] = store.Status()
  48. writeJsonQuiet(w, r, m)
  49. }
  50. func assignVolumeHandler(w http.ResponseWriter, r *http.Request) {
  51. err := store.AddVolume(r.FormValue("volume"), r.FormValue("replicationType"))
  52. if err == nil {
  53. writeJsonQuiet(w, r, map[string]string{"error": ""})
  54. } else {
  55. writeJsonQuiet(w, r, map[string]string{"error": err.Error()})
  56. }
  57. debug("assign volume =", r.FormValue("volume"), ", replicationType =", r.FormValue("replicationType"), ", error =", err)
  58. }
  59. func vacuumVolumeCheckHandler(w http.ResponseWriter, r *http.Request) {
  60. err, ret := store.CheckCompactVolume(r.FormValue("volume"), r.FormValue("garbageThreshold"))
  61. if err == nil {
  62. writeJsonQuiet(w, r, map[string]interface{}{"error": "", "result": ret})
  63. } else {
  64. writeJsonQuiet(w, r, map[string]interface{}{"error": err.Error(), "result": false})
  65. }
  66. debug("checked compacting volume =", r.FormValue("volume"), "garbageThreshold =", r.FormValue("garbageThreshold"), "vacuum =", ret)
  67. }
  68. func vacuumVolumeCompactHandler(w http.ResponseWriter, r *http.Request) {
  69. err := store.CompactVolume(r.FormValue("volume"))
  70. if err == nil {
  71. writeJsonQuiet(w, r, map[string]string{"error": ""})
  72. } else {
  73. writeJsonQuiet(w, r, map[string]string{"error": err.Error()})
  74. }
  75. debug("compacted volume =", r.FormValue("volume"), ", error =", err)
  76. }
  77. func vacuumVolumeCommitHandler(w http.ResponseWriter, r *http.Request) {
  78. err := store.CommitCompactVolume(r.FormValue("volume"))
  79. if err == nil {
  80. writeJsonQuiet(w, r, map[string]interface{}{"error": ""})
  81. } else {
  82. writeJsonQuiet(w, r, map[string]string{"error": err.Error()})
  83. }
  84. debug("commit compact volume =", r.FormValue("volume"), ", error =", err)
  85. }
  86. func freezeVolumeHandler(w http.ResponseWriter, r *http.Request) {
  87. //TODO: notify master that this volume will be read-only
  88. err := store.FreezeVolume(r.FormValue("volume"))
  89. if err == nil {
  90. writeJsonQuiet(w, r, map[string]interface{}{"error": ""})
  91. } else {
  92. writeJsonQuiet(w, r, map[string]string{"error": err.Error()})
  93. }
  94. debug("freeze volume =", r.FormValue("volume"), ", error =", err)
  95. }
  96. func submitFromVolumeServerHandler(w http.ResponseWriter, r *http.Request) {
  97. submitForClientHandler(w, r, *masterNode)
  98. }
  99. func storeHandler(w http.ResponseWriter, r *http.Request) {
  100. switch r.Method {
  101. case "GET":
  102. GetOrHeadHandler(w, r, true)
  103. case "HEAD":
  104. GetOrHeadHandler(w, r, false)
  105. case "DELETE":
  106. secure(volumeWhiteList, DeleteHandler)(w, r)
  107. case "POST":
  108. secure(volumeWhiteList, PostHandler)(w, r)
  109. }
  110. }
  111. func GetOrHeadHandler(w http.ResponseWriter, r *http.Request, isGetMethod bool) {
  112. n := new(storage.Needle)
  113. vid, fid, filename, ext, _ := parseURLPath(r.URL.Path)
  114. volumeId, err := storage.NewVolumeId(vid)
  115. if err != nil {
  116. debug("parsing error:", err, r.URL.Path)
  117. return
  118. }
  119. n.ParsePath(fid)
  120. debug("volume", volumeId, "reading", n)
  121. if !store.HasVolume(volumeId) {
  122. lookupResult, err := operation.Lookup(*masterNode, volumeId)
  123. debug("volume", volumeId, "found on", lookupResult, "error", err)
  124. if err == nil {
  125. http.Redirect(w, r, "http://"+lookupResult.Locations[0].PublicUrl+r.URL.Path, http.StatusMovedPermanently)
  126. } else {
  127. debug("lookup error:", err, r.URL.Path)
  128. w.WriteHeader(http.StatusNotFound)
  129. }
  130. return
  131. }
  132. cookie := n.Cookie
  133. count, e := store.Read(volumeId, n)
  134. debug("read bytes", count, "error", e)
  135. if e != nil || count <= 0 {
  136. debug("read error:", e, r.URL.Path)
  137. w.WriteHeader(http.StatusNotFound)
  138. return
  139. }
  140. if n.Cookie != cookie {
  141. glog.V(0).Infoln("request with unmaching cookie from ", r.RemoteAddr, "agent", r.UserAgent())
  142. w.WriteHeader(http.StatusNotFound)
  143. return
  144. }
  145. if n.LastModified != 0 {
  146. w.Header().Set("Last-Modified", time.Unix(int64(n.LastModified), 0).UTC().Format(http.TimeFormat))
  147. if r.Header.Get("If-Modified-Since") != "" {
  148. if t, parseError := time.Parse(http.TimeFormat, r.Header.Get("If-Modified-Since")); parseError == nil {
  149. if t.Unix() >= int64(n.LastModified) {
  150. w.WriteHeader(http.StatusNotModified)
  151. return
  152. }
  153. }
  154. }
  155. }
  156. if n.NameSize > 0 && filename == "" {
  157. filename := string(n.Name)
  158. dotIndex := strings.LastIndex(filename, ".")
  159. if dotIndex > 0 {
  160. ext = filename[dotIndex:]
  161. }
  162. }
  163. mtype := ""
  164. if ext != "" {
  165. mtype = mime.TypeByExtension(ext)
  166. }
  167. if n.MimeSize > 0 {
  168. mtype = string(n.Mime)
  169. }
  170. if mtype != "" {
  171. w.Header().Set("Content-Type", mtype)
  172. }
  173. if filename != "" {
  174. w.Header().Set("Content-Disposition", "filename="+fileNameEscaper.Replace(filename))
  175. }
  176. if ext != ".gz" {
  177. if n.IsGzipped() {
  178. if strings.Contains(r.Header.Get("Accept-Encoding"), "gzip") {
  179. w.Header().Set("Content-Encoding", "gzip")
  180. } else {
  181. if n.Data, err = storage.UnGzipData(n.Data); err != nil {
  182. debug("lookup error:", err, r.URL.Path)
  183. }
  184. }
  185. }
  186. }
  187. w.Header().Set("Content-Length", strconv.Itoa(len(n.Data)))
  188. if isGetMethod {
  189. if _, e = w.Write(n.Data); e != nil {
  190. debug("response write error:", e)
  191. }
  192. }
  193. }
  194. func PostHandler(w http.ResponseWriter, r *http.Request) {
  195. m := make(map[string]interface{})
  196. if e := r.ParseForm(); e != nil {
  197. debug("form parse error:", e)
  198. writeJsonError(w, r, e)
  199. return
  200. }
  201. vid, _, _, _, _ := parseURLPath(r.URL.Path)
  202. volumeId, ve := storage.NewVolumeId(vid)
  203. if ve != nil {
  204. debug("NewVolumeId error:", ve)
  205. writeJsonError(w, r, ve)
  206. return
  207. }
  208. needle, ne := storage.NewNeedle(r)
  209. if ne != nil {
  210. writeJsonError(w, r, ne)
  211. return
  212. }
  213. ret, errorStatus := replication.ReplicatedWrite(*masterNode, store, volumeId, needle, r)
  214. if errorStatus == "" {
  215. w.WriteHeader(http.StatusCreated)
  216. } else {
  217. w.WriteHeader(http.StatusInternalServerError)
  218. m["error"] = errorStatus
  219. }
  220. m["size"] = ret
  221. writeJsonQuiet(w, r, m)
  222. }
  223. func DeleteHandler(w http.ResponseWriter, r *http.Request) {
  224. n := new(storage.Needle)
  225. vid, fid, _, _, _ := parseURLPath(r.URL.Path)
  226. volumeId, _ := storage.NewVolumeId(vid)
  227. n.ParsePath(fid)
  228. debug("deleting", n)
  229. cookie := n.Cookie
  230. count, ok := store.Read(volumeId, n)
  231. if ok != nil {
  232. m := make(map[string]uint32)
  233. m["size"] = 0
  234. writeJsonQuiet(w, r, m)
  235. return
  236. }
  237. if n.Cookie != cookie {
  238. glog.V(0).Infoln("delete with unmaching cookie from ", r.RemoteAddr, "agent", r.UserAgent())
  239. return
  240. }
  241. n.Size = 0
  242. ret := replication.ReplicatedDelete(*masterNode, store, volumeId, n, r)
  243. if ret != 0 {
  244. w.WriteHeader(http.StatusAccepted)
  245. } else {
  246. w.WriteHeader(http.StatusInternalServerError)
  247. }
  248. m := make(map[string]uint32)
  249. m["size"] = uint32(count)
  250. writeJsonQuiet(w, r, m)
  251. }
  252. func parseURLPath(path string) (vid, fid, filename, ext string, isVolumeIdOnly bool) {
  253. switch strings.Count(path, "/") {
  254. case 3:
  255. parts := strings.Split(path, "/")
  256. vid, fid, filename = parts[1], parts[2], parts[3]
  257. ext = filepath.Ext(filename)
  258. case 2:
  259. parts := strings.Split(path, "/")
  260. vid, fid = parts[1], parts[2]
  261. dotIndex := strings.LastIndex(fid, ".")
  262. if dotIndex > 0 {
  263. ext = fid[dotIndex:]
  264. fid = fid[0:dotIndex]
  265. }
  266. default:
  267. sepIndex := strings.LastIndex(path, "/")
  268. commaIndex := strings.LastIndex(path[sepIndex:], ",")
  269. if commaIndex <= 0 {
  270. vid, isVolumeIdOnly = path[sepIndex+1:], true
  271. return
  272. }
  273. dotIndex := strings.LastIndex(path[sepIndex:], ".")
  274. vid = path[sepIndex+1 : commaIndex]
  275. fid = path[commaIndex+1:]
  276. ext = ""
  277. if dotIndex > 0 {
  278. fid = path[commaIndex+1 : dotIndex]
  279. ext = path[dotIndex:]
  280. }
  281. }
  282. return
  283. }
  284. func runVolume(cmd *Command, args []string) bool {
  285. if *vMaxCpu < 1 {
  286. *vMaxCpu = runtime.NumCPU()
  287. }
  288. runtime.GOMAXPROCS(*vMaxCpu)
  289. folders := strings.Split(*volumeFolders, ",")
  290. maxCountStrings := strings.Split(*maxVolumeCounts, ",")
  291. maxCounts := make([]int, 0)
  292. for _, maxString := range maxCountStrings {
  293. if max, e := strconv.Atoi(maxString); e == nil {
  294. maxCounts = append(maxCounts, max)
  295. } else {
  296. glog.Fatalf("The max specified in -max not a valid number %s", max)
  297. }
  298. }
  299. if len(folders) != len(maxCounts) {
  300. glog.Fatalf("%d directories by -dir, but only %d max is set by -max", len(folders), len(maxCounts))
  301. }
  302. for _, folder := range folders {
  303. fileInfo, err := os.Stat(folder)
  304. if err != nil {
  305. glog.Fatalf("No Existing Folder:%s", folder)
  306. }
  307. if !fileInfo.IsDir() {
  308. glog.Fatalf("Volume Folder should not be a file:%s", folder)
  309. }
  310. perm := fileInfo.Mode().Perm()
  311. glog.V(0).Infoln("Volume Folder", folder)
  312. glog.V(0).Infoln("Permission:", perm)
  313. }
  314. if *publicUrl == "" {
  315. *publicUrl = *ip + ":" + strconv.Itoa(*vport)
  316. }
  317. if *volumeWhiteListOption != "" {
  318. volumeWhiteList = strings.Split(*volumeWhiteListOption, ",")
  319. }
  320. store = storage.NewStore(*vport, *ip, *publicUrl, folders, maxCounts)
  321. defer store.Close()
  322. http.HandleFunc("/", storeHandler)
  323. http.HandleFunc("/submit", secure(volumeWhiteList, submitFromVolumeServerHandler))
  324. http.HandleFunc("/status", secure(volumeWhiteList, statusHandler))
  325. http.HandleFunc("/admin/assign_volume", secure(volumeWhiteList, assignVolumeHandler))
  326. http.HandleFunc("/admin/vacuum_volume_check", secure(volumeWhiteList, vacuumVolumeCheckHandler))
  327. http.HandleFunc("/admin/vacuum_volume_compact", secure(volumeWhiteList, vacuumVolumeCompactHandler))
  328. http.HandleFunc("/admin/vacuum_volume_commit", secure(volumeWhiteList, vacuumVolumeCommitHandler))
  329. http.HandleFunc("/admin/freeze_volume", secure(volumeWhiteList, freezeVolumeHandler))
  330. go func() {
  331. connected := true
  332. store.SetMaster(*masterNode)
  333. store.SetDataCenter(*dataCenter)
  334. store.SetRack(*rack)
  335. for {
  336. err := store.Join()
  337. if err == nil {
  338. if !connected {
  339. connected = true
  340. glog.V(0).Infoln("Reconnected with master")
  341. }
  342. } else {
  343. if connected {
  344. connected = false
  345. }
  346. }
  347. time.Sleep(time.Duration(float32(*vpulse*1e3)*(1+rand.Float32())) * time.Millisecond)
  348. }
  349. }()
  350. glog.V(0).Infoln("store joined at", *masterNode)
  351. glog.V(0).Infoln("Start Weed volume server", VERSION, "at http://"+*ip+":"+strconv.Itoa(*vport))
  352. srv := &http.Server{
  353. Addr: ":" + strconv.Itoa(*vport),
  354. Handler: http.DefaultServeMux,
  355. ReadTimeout: (time.Duration(*vReadTimeout) * time.Second),
  356. }
  357. e := srv.ListenAndServe()
  358. if e != nil {
  359. glog.Fatalf("Fail to start:%s", e.Error())
  360. }
  361. return true
  362. }