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.

242 lines
6.8 KiB

9 years ago
7 years ago
9 years ago
7 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
  1. package weed_server
  2. import (
  3. "encoding/json"
  4. "errors"
  5. "io/ioutil"
  6. "net/http"
  7. "net/url"
  8. "strconv"
  9. "strings"
  10. "time"
  11. "github.com/chrislusf/seaweedfs/weed/filer2"
  12. "github.com/chrislusf/seaweedfs/weed/glog"
  13. "github.com/chrislusf/seaweedfs/weed/operation"
  14. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  15. "github.com/chrislusf/seaweedfs/weed/util"
  16. "os"
  17. )
  18. var (
  19. OS_UID = uint32(os.Getuid())
  20. OS_GID = uint32(os.Getgid())
  21. )
  22. type FilerPostResult struct {
  23. Name string `json:"name,omitempty"`
  24. Size uint32 `json:"size,omitempty"`
  25. Error string `json:"error,omitempty"`
  26. Fid string `json:"fid,omitempty"`
  27. Url string `json:"url,omitempty"`
  28. }
  29. func (fs *FilerServer) queryFileInfoByPath(w http.ResponseWriter, r *http.Request, path string) (fileId, urlLocation string, err error) {
  30. var entry *filer2.Entry
  31. entry, err = fs.filer.FindEntry(filer2.FullPath(path))
  32. if err == filer2.ErrNotFound {
  33. return "", "", nil
  34. }
  35. if err != nil {
  36. glog.V(0).Infoln("failing to find path in filer store", path, err.Error())
  37. writeJsonError(w, r, http.StatusInternalServerError, err)
  38. return
  39. }
  40. if len(entry.Chunks) == 0 {
  41. glog.V(1).Infof("empty entry: %s", path)
  42. w.WriteHeader(http.StatusNoContent)
  43. } else {
  44. fileId = entry.Chunks[0].FileId
  45. urlLocation, err = fs.filer.MasterClient.LookupFileId(fileId)
  46. if err != nil {
  47. glog.V(1).Infof("operation LookupFileId %s failed, err is %s", fileId, err.Error())
  48. w.WriteHeader(http.StatusNotFound)
  49. }
  50. }
  51. return
  52. }
  53. func (fs *FilerServer) assignNewFileInfo(w http.ResponseWriter, r *http.Request, replication, collection string, dataCenter string) (fileId, urlLocation string, err error) {
  54. ar := &operation.VolumeAssignRequest{
  55. Count: 1,
  56. Replication: replication,
  57. Collection: collection,
  58. Ttl: r.URL.Query().Get("ttl"),
  59. DataCenter: dataCenter,
  60. }
  61. var altRequest *operation.VolumeAssignRequest
  62. if dataCenter != "" {
  63. altRequest = &operation.VolumeAssignRequest{
  64. Count: 1,
  65. Replication: replication,
  66. Collection: collection,
  67. Ttl: r.URL.Query().Get("ttl"),
  68. DataCenter: "",
  69. }
  70. }
  71. assignResult, ae := operation.Assign(fs.filer.GetMaster(), ar, altRequest)
  72. if ae != nil {
  73. glog.Errorf("failing to assign a file id: %v", ae)
  74. writeJsonError(w, r, http.StatusInternalServerError, ae)
  75. err = ae
  76. return
  77. }
  78. fileId = assignResult.Fid
  79. urlLocation = "http://" + assignResult.Url + "/" + assignResult.Fid
  80. return
  81. }
  82. func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
  83. query := r.URL.Query()
  84. replication := query.Get("replication")
  85. if replication == "" {
  86. replication = fs.option.DefaultReplication
  87. }
  88. collection := query.Get("collection")
  89. if collection == "" {
  90. collection = fs.option.Collection
  91. }
  92. dataCenter := query.Get("dataCenter")
  93. if dataCenter == "" {
  94. dataCenter = fs.option.DataCenter
  95. }
  96. if autoChunked := fs.autoChunk(w, r, replication, collection, dataCenter); autoChunked {
  97. return
  98. }
  99. fileId, urlLocation, err := fs.queryFileInfoByPath(w, r, r.URL.Path)
  100. if err == nil && fileId == "" {
  101. fileId, urlLocation, err = fs.assignNewFileInfo(w, r, replication, collection, dataCenter)
  102. }
  103. if err != nil || fileId == "" || urlLocation == "" {
  104. return
  105. }
  106. glog.V(4).Infof("write %s to %v", r.URL.Path, urlLocation)
  107. u, _ := url.Parse(urlLocation)
  108. // This allows a client to generate a chunk manifest and submit it to the filer -- it is a little off
  109. // because they need to provide FIDs instead of file paths...
  110. cm, _ := strconv.ParseBool(query.Get("cm"))
  111. if cm {
  112. q := u.Query()
  113. q.Set("cm", "true")
  114. u.RawQuery = q.Encode()
  115. }
  116. glog.V(4).Infoln("post to", u)
  117. // send request to volume server
  118. request := &http.Request{
  119. Method: r.Method,
  120. URL: u,
  121. Proto: r.Proto,
  122. ProtoMajor: r.ProtoMajor,
  123. ProtoMinor: r.ProtoMinor,
  124. Header: r.Header,
  125. Body: r.Body,
  126. Host: r.Host,
  127. ContentLength: r.ContentLength,
  128. }
  129. resp, do_err := util.Do(request)
  130. if do_err != nil {
  131. glog.Errorf("failing to connect to volume server %s: %v, %+v", r.RequestURI, do_err, r.Method)
  132. writeJsonError(w, r, http.StatusInternalServerError, do_err)
  133. return
  134. }
  135. defer resp.Body.Close()
  136. etag := resp.Header.Get("ETag")
  137. resp_body, ra_err := ioutil.ReadAll(resp.Body)
  138. if ra_err != nil {
  139. glog.V(0).Infoln("failing to upload to volume server", r.RequestURI, ra_err.Error())
  140. writeJsonError(w, r, http.StatusInternalServerError, ra_err)
  141. return
  142. }
  143. glog.V(4).Infoln("post result", string(resp_body))
  144. var ret operation.UploadResult
  145. unmarshal_err := json.Unmarshal(resp_body, &ret)
  146. if unmarshal_err != nil {
  147. glog.V(0).Infoln("failing to read upload resonse", r.RequestURI, string(resp_body))
  148. writeJsonError(w, r, http.StatusInternalServerError, unmarshal_err)
  149. return
  150. }
  151. if ret.Error != "" {
  152. glog.V(0).Infoln("failing to post to volume server", r.RequestURI, ret.Error)
  153. writeJsonError(w, r, http.StatusInternalServerError, errors.New(ret.Error))
  154. return
  155. }
  156. // find correct final path
  157. path := r.URL.Path
  158. if strings.HasSuffix(path, "/") {
  159. if ret.Name != "" {
  160. path += ret.Name
  161. } else {
  162. fs.filer.DeleteFileByFileId(fileId)
  163. glog.V(0).Infoln("Can not to write to folder", path, "without a file name!")
  164. writeJsonError(w, r, http.StatusInternalServerError,
  165. errors.New("Can not to write to folder "+path+" without a file name"))
  166. return
  167. }
  168. }
  169. // update metadata in filer store
  170. glog.V(4).Infoln("saving", path, "=>", fileId)
  171. entry := &filer2.Entry{
  172. FullPath: filer2.FullPath(path),
  173. Attr: filer2.Attr{
  174. Mtime: time.Now(),
  175. Crtime: time.Now(),
  176. Mode: 0660,
  177. Uid: OS_UID,
  178. Gid: OS_GID,
  179. Replication: replication,
  180. Collection: collection,
  181. TtlSec: int32(util.ParseInt(r.URL.Query().Get("ttl"), 0)),
  182. },
  183. Chunks: []*filer_pb.FileChunk{{
  184. FileId: fileId,
  185. Size: uint64(ret.Size),
  186. Mtime: time.Now().UnixNano(),
  187. ETag: etag,
  188. }},
  189. }
  190. if db_err := fs.filer.CreateEntry(entry); db_err != nil {
  191. fs.filer.DeleteFileByFileId(fileId)
  192. glog.V(0).Infof("failing to write %s to filer server : %v", path, db_err)
  193. writeJsonError(w, r, http.StatusInternalServerError, db_err)
  194. return
  195. }
  196. // send back post result
  197. reply := FilerPostResult{
  198. Name: ret.Name,
  199. Size: ret.Size,
  200. Error: ret.Error,
  201. Fid: fileId,
  202. Url: urlLocation,
  203. }
  204. setEtag(w, etag)
  205. writeJsonQuiet(w, r, http.StatusCreated, reply)
  206. }
  207. // curl -X DELETE http://localhost:8888/path/to
  208. // curl -X DELETE http://localhost:8888/path/to?recursive=true
  209. func (fs *FilerServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
  210. isRecursive := r.FormValue("recursive") == "true"
  211. err := fs.filer.DeleteEntryMetaAndData(filer2.FullPath(r.URL.Path), isRecursive, true)
  212. if err != nil {
  213. glog.V(1).Infoln("deleting", r.URL.Path, ":", err.Error())
  214. writeJsonError(w, r, http.StatusInternalServerError, err)
  215. return
  216. }
  217. w.WriteHeader(http.StatusNoContent)
  218. }