submit.go 5.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230
  1. package operation
  2. import (
  3. "io"
  4. "mime"
  5. "net/url"
  6. "os"
  7. "path"
  8. "strconv"
  9. "strings"
  10. "google.golang.org/grpc"
  11. "github.com/chrislusf/seaweedfs/weed/glog"
  12. "github.com/chrislusf/seaweedfs/weed/security"
  13. )
  14. type FilePart struct {
  15. Reader io.Reader
  16. FileName string
  17. FileSize int64
  18. MimeType string
  19. ModTime int64 //in seconds
  20. Replication string
  21. Collection string
  22. DataCenter string
  23. Ttl string
  24. Server string //this comes from assign result
  25. Fid string //this comes from assign result, but customizable
  26. Fsync bool
  27. }
  28. type SubmitResult struct {
  29. FileName string `json:"fileName,omitempty"`
  30. FileUrl string `json:"fileUrl,omitempty"`
  31. Fid string `json:"fid,omitempty"`
  32. Size uint32 `json:"size,omitempty"`
  33. Error string `json:"error,omitempty"`
  34. }
  35. func SubmitFiles(master string, grpcDialOption grpc.DialOption, files []FilePart, replication string, collection string, dataCenter string, ttl string, maxMB int, usePublicUrl bool) ([]SubmitResult, error) {
  36. results := make([]SubmitResult, len(files))
  37. for index, file := range files {
  38. results[index].FileName = file.FileName
  39. }
  40. ar := &VolumeAssignRequest{
  41. Count: uint64(len(files)),
  42. Replication: replication,
  43. Collection: collection,
  44. DataCenter: dataCenter,
  45. Ttl: ttl,
  46. }
  47. ret, err := Assign(master, grpcDialOption, ar)
  48. if err != nil {
  49. for index := range files {
  50. results[index].Error = err.Error()
  51. }
  52. return results, err
  53. }
  54. for index, file := range files {
  55. file.Fid = ret.Fid
  56. if index > 0 {
  57. file.Fid = file.Fid + "_" + strconv.Itoa(index)
  58. }
  59. file.Server = ret.Url
  60. if usePublicUrl {
  61. file.Server = ret.PublicUrl
  62. }
  63. file.Replication = replication
  64. file.Collection = collection
  65. file.DataCenter = dataCenter
  66. results[index].Size, err = file.Upload(maxMB, master, usePublicUrl, ret.Auth, grpcDialOption)
  67. if err != nil {
  68. results[index].Error = err.Error()
  69. }
  70. results[index].Fid = file.Fid
  71. results[index].FileUrl = ret.PublicUrl + "/" + file.Fid
  72. }
  73. return results, nil
  74. }
  75. func NewFileParts(fullPathFilenames []string) (ret []FilePart, err error) {
  76. ret = make([]FilePart, len(fullPathFilenames))
  77. for index, file := range fullPathFilenames {
  78. if ret[index], err = newFilePart(file); err != nil {
  79. return
  80. }
  81. }
  82. return
  83. }
  84. func newFilePart(fullPathFilename string) (ret FilePart, err error) {
  85. fh, openErr := os.Open(fullPathFilename)
  86. if openErr != nil {
  87. glog.V(0).Info("Failed to open file: ", fullPathFilename)
  88. return ret, openErr
  89. }
  90. ret.Reader = fh
  91. fi, fiErr := fh.Stat()
  92. if fiErr != nil {
  93. glog.V(0).Info("Failed to stat file:", fullPathFilename)
  94. return ret, fiErr
  95. }
  96. ret.ModTime = fi.ModTime().UTC().Unix()
  97. ret.FileSize = fi.Size()
  98. ext := strings.ToLower(path.Ext(fullPathFilename))
  99. ret.FileName = fi.Name()
  100. if ext != "" {
  101. ret.MimeType = mime.TypeByExtension(ext)
  102. }
  103. return ret, nil
  104. }
  105. func (fi FilePart) Upload(maxMB int, master string, usePublicUrl bool, jwt security.EncodedJwt, grpcDialOption grpc.DialOption) (retSize uint32, err error) {
  106. fileUrl := "http://" + fi.Server + "/" + fi.Fid
  107. if fi.ModTime != 0 {
  108. fileUrl += "?ts=" + strconv.Itoa(int(fi.ModTime))
  109. }
  110. if fi.Fsync {
  111. fileUrl += "?fsync=true"
  112. }
  113. if closer, ok := fi.Reader.(io.Closer); ok {
  114. defer closer.Close()
  115. }
  116. baseName := path.Base(fi.FileName)
  117. if maxMB > 0 && fi.FileSize > int64(maxMB*1024*1024) {
  118. chunkSize := int64(maxMB * 1024 * 1024)
  119. chunks := fi.FileSize/chunkSize + 1
  120. cm := ChunkManifest{
  121. Name: baseName,
  122. Size: fi.FileSize,
  123. Mime: fi.MimeType,
  124. Chunks: make([]*ChunkInfo, 0, chunks),
  125. }
  126. var ret *AssignResult
  127. var id string
  128. if fi.DataCenter != "" {
  129. ar := &VolumeAssignRequest{
  130. Count: uint64(chunks),
  131. Replication: fi.Replication,
  132. Collection: fi.Collection,
  133. Ttl: fi.Ttl,
  134. }
  135. ret, err = Assign(master, grpcDialOption, ar)
  136. if err != nil {
  137. return
  138. }
  139. }
  140. for i := int64(0); i < chunks; i++ {
  141. if fi.DataCenter == "" {
  142. ar := &VolumeAssignRequest{
  143. Count: 1,
  144. Replication: fi.Replication,
  145. Collection: fi.Collection,
  146. Ttl: fi.Ttl,
  147. }
  148. ret, err = Assign(master, grpcDialOption, ar)
  149. if err != nil {
  150. // delete all uploaded chunks
  151. cm.DeleteChunks(master, usePublicUrl, grpcDialOption)
  152. return
  153. }
  154. id = ret.Fid
  155. } else {
  156. id = ret.Fid
  157. if i > 0 {
  158. id += "_" + strconv.FormatInt(i, 10)
  159. }
  160. }
  161. fileUrl := "http://" + ret.Url + "/" + id
  162. count, e := upload_one_chunk(
  163. baseName+"-"+strconv.FormatInt(i+1, 10),
  164. io.LimitReader(fi.Reader, chunkSize),
  165. master, fileUrl,
  166. ret.Auth)
  167. if e != nil {
  168. // delete all uploaded chunks
  169. cm.DeleteChunks(master, usePublicUrl, grpcDialOption)
  170. return 0, e
  171. }
  172. cm.Chunks = append(cm.Chunks,
  173. &ChunkInfo{
  174. Offset: i * chunkSize,
  175. Size: int64(count),
  176. Fid: id,
  177. },
  178. )
  179. retSize += count
  180. }
  181. err = upload_chunked_file_manifest(fileUrl, &cm, jwt)
  182. if err != nil {
  183. // delete all uploaded chunks
  184. cm.DeleteChunks(master, usePublicUrl, grpcDialOption)
  185. }
  186. } else {
  187. ret, e, _ := Upload(fileUrl, baseName, false, fi.Reader, false, fi.MimeType, nil, jwt)
  188. if e != nil {
  189. return 0, e
  190. }
  191. return ret.Size, e
  192. }
  193. return
  194. }
  195. func upload_one_chunk(filename string, reader io.Reader, master,
  196. fileUrl string, jwt security.EncodedJwt,
  197. ) (size uint32, e error) {
  198. glog.V(4).Info("Uploading part ", filename, " to ", fileUrl, "...")
  199. uploadResult, uploadError, _ := Upload(fileUrl, filename, false, reader, false, "", nil, jwt)
  200. if uploadError != nil {
  201. return 0, uploadError
  202. }
  203. return uploadResult.Size, nil
  204. }
  205. func upload_chunked_file_manifest(fileUrl string, manifest *ChunkManifest, jwt security.EncodedJwt) error {
  206. buf, e := manifest.Marshal()
  207. if e != nil {
  208. return e
  209. }
  210. glog.V(4).Info("Uploading chunks manifest ", manifest.Name, " to ", fileUrl, "...")
  211. u, _ := url.Parse(fileUrl)
  212. q := u.Query()
  213. q.Set("cm", "true")
  214. u.RawQuery = q.Encode()
  215. _, e = UploadData(u.String(), manifest.Name, false, buf, false, "application/json", nil, jwt)
  216. return e
  217. }