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.7 KiB

11 years ago
11 years ago
11 years ago
11 years ago
11 years ago
11 years ago
11 years ago
11 years ago
11 years ago
  1. package weed_server
  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/util"
  6. "encoding/json"
  7. "errors"
  8. "github.com/syndtr/goleveldb/leveldb"
  9. "io"
  10. "io/ioutil"
  11. "math/rand"
  12. "net/http"
  13. "net/url"
  14. "strconv"
  15. "strings"
  16. )
  17. func (fs *FilerServer) filerHandler(w http.ResponseWriter, r *http.Request) {
  18. switch r.Method {
  19. case "GET":
  20. fs.GetOrHeadHandler(w, r, true)
  21. case "HEAD":
  22. fs.GetOrHeadHandler(w, r, false)
  23. case "DELETE":
  24. fs.DeleteHandler(w, r)
  25. case "PUT":
  26. fs.PostHandler(w, r)
  27. case "POST":
  28. fs.PostHandler(w, r)
  29. }
  30. }
  31. func (fs *FilerServer) listDirectoryHandler(w http.ResponseWriter, r *http.Request) {
  32. if !strings.HasSuffix(r.URL.Path, "/") {
  33. return
  34. }
  35. dirlist, err := fs.filer.ListDirectories(r.URL.Path)
  36. if err == leveldb.ErrNotFound {
  37. glog.V(3).Infoln("Directory Not Found in db", r.URL.Path)
  38. w.WriteHeader(http.StatusNotFound)
  39. return
  40. }
  41. m := make(map[string]interface{})
  42. m["Directory"] = r.URL.Path
  43. m["Subdirectories"] = dirlist
  44. lastFileName := r.FormValue("lastFileName")
  45. limit, limit_err := strconv.Atoi(r.FormValue("limit"))
  46. if limit_err != nil {
  47. limit = 100
  48. }
  49. m["Files"], _ = fs.filer.ListFiles(r.URL.Path, lastFileName, limit)
  50. writeJsonQuiet(w, r, m)
  51. }
  52. func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request, isGetMethod bool) {
  53. if strings.HasSuffix(r.URL.Path, "/") {
  54. fs.listDirectoryHandler(w, r)
  55. return
  56. }
  57. fileId, err := fs.filer.FindFile(r.URL.Path)
  58. if err == leveldb.ErrNotFound {
  59. glog.V(3).Infoln("Not found in db", r.URL.Path)
  60. w.WriteHeader(http.StatusNotFound)
  61. return
  62. }
  63. parts := strings.Split(fileId, ",")
  64. if len(parts) != 2 {
  65. glog.V(1).Infoln("Invalid fileId", fileId)
  66. w.WriteHeader(http.StatusNotFound)
  67. return
  68. }
  69. lookup, lookupError := operation.Lookup(fs.master, parts[0])
  70. if lookupError != nil {
  71. glog.V(1).Infoln("Invalid lookup", lookupError.Error())
  72. w.WriteHeader(http.StatusNotFound)
  73. return
  74. }
  75. if len(lookup.Locations) == 0 {
  76. glog.V(1).Infoln("Can not find location for volume", parts[0])
  77. w.WriteHeader(http.StatusNotFound)
  78. return
  79. }
  80. urlLocation := lookup.Locations[rand.Intn(len(lookup.Locations))].PublicUrl
  81. u, _ := url.Parse("http://" + urlLocation + "/" + fileId)
  82. request := &http.Request{
  83. Method: r.Method,
  84. URL: u,
  85. Proto: r.Proto,
  86. ProtoMajor: r.ProtoMajor,
  87. ProtoMinor: r.ProtoMinor,
  88. Header: r.Header,
  89. Body: r.Body,
  90. Host: r.Host,
  91. ContentLength: r.ContentLength,
  92. }
  93. glog.V(3).Infoln("retrieving from", u)
  94. resp, do_err := util.Do(request)
  95. if do_err != nil {
  96. glog.V(0).Infoln("failing to connect to volume server", do_err.Error())
  97. writeJsonError(w, r, do_err)
  98. return
  99. }
  100. defer resp.Body.Close()
  101. for k, v := range resp.Header {
  102. w.Header()[k] = v
  103. }
  104. io.Copy(w, resp.Body)
  105. }
  106. func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
  107. query := r.URL.Query()
  108. assignResult, ae := operation.Assign(fs.master, 1, query.Get("replication"), fs.collection)
  109. if ae != nil {
  110. glog.V(0).Infoln("failing to assign a file id", ae.Error())
  111. writeJsonError(w, r, ae)
  112. return
  113. }
  114. u, _ := url.Parse("http://" + assignResult.PublicUrl + "/" + assignResult.Fid)
  115. glog.V(4).Infoln("post to", u)
  116. request := &http.Request{
  117. Method: r.Method,
  118. URL: u,
  119. Proto: r.Proto,
  120. ProtoMajor: r.ProtoMajor,
  121. ProtoMinor: r.ProtoMinor,
  122. Header: r.Header,
  123. Body: r.Body,
  124. Host: r.Host,
  125. ContentLength: r.ContentLength,
  126. }
  127. resp, do_err := util.Do(request)
  128. if do_err != nil {
  129. glog.V(0).Infoln("failing to connect to volume server", do_err.Error())
  130. writeJsonError(w, r, do_err)
  131. return
  132. }
  133. defer resp.Body.Close()
  134. resp_body, ra_err := ioutil.ReadAll(resp.Body)
  135. if ra_err != nil {
  136. glog.V(0).Infoln("failing to upload to volume server", ra_err.Error())
  137. writeJsonError(w, r, ra_err)
  138. return
  139. }
  140. glog.V(4).Infoln("post result", string(resp_body))
  141. var ret operation.UploadResult
  142. unmarshal_err := json.Unmarshal(resp_body, &ret)
  143. if unmarshal_err != nil {
  144. glog.V(0).Infoln("failing to read upload resonse", string(resp_body))
  145. writeJsonError(w, r, unmarshal_err)
  146. return
  147. }
  148. if ret.Error != "" {
  149. glog.V(0).Infoln("failing to post to volume server", ret.Error)
  150. writeJsonError(w, r, errors.New(ret.Error))
  151. return
  152. }
  153. path := r.URL.Path
  154. if strings.HasSuffix(path, "/") {
  155. if ret.Name != "" {
  156. path += ret.Name
  157. } else {
  158. operation.DeleteFile(fs.master, assignResult.Fid) //clean up
  159. glog.V(0).Infoln("Can not to write to folder", path, "without a file name!")
  160. writeJsonError(w, r, errors.New("Can not to write to folder "+path+" without a file name"))
  161. return
  162. }
  163. }
  164. glog.V(4).Infoln("saving", path, "=>", assignResult.Fid)
  165. if db_err := fs.filer.CreateFile(path, assignResult.Fid); db_err != nil {
  166. operation.DeleteFile(fs.master, assignResult.Fid) //clean up
  167. glog.V(0).Infoln("failing to write to filer server", db_err.Error())
  168. writeJsonError(w, r, db_err)
  169. return
  170. }
  171. w.WriteHeader(http.StatusCreated)
  172. }
  173. // curl -X DELETE http://localhost:8888/path/to
  174. // curl -X DELETE http://localhost:8888/path/to?recursive=true
  175. func (fs *FilerServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
  176. var err error
  177. var fid string
  178. if strings.HasSuffix(r.URL.Path, "/") {
  179. isRecursive := r.FormValue("recursive") == "true"
  180. err = fs.filer.DeleteDirectory(r.URL.Path, isRecursive)
  181. } else {
  182. fid, err = fs.filer.DeleteFile(r.URL.Path)
  183. if err == nil {
  184. err = operation.DeleteFile(fs.master, fid)
  185. }
  186. }
  187. if err == nil {
  188. w.WriteHeader(http.StatusAccepted)
  189. writeJsonQuiet(w, r, map[string]string{"error": ""})
  190. } else {
  191. glog.V(4).Infoln("deleting", r.URL.Path, ":", err.Error())
  192. writeJsonError(w, r, err)
  193. }
  194. }