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.

259 lines
8.0 KiB

6 years ago
4 years ago
4 years ago
5 years ago
5 years ago
5 years ago
4 years ago
5 years ago
5 years ago
12 years ago
5 years ago
5 years ago
  1. package operation
  2. import (
  3. "bytes"
  4. "encoding/json"
  5. "errors"
  6. "fmt"
  7. "io"
  8. "io/ioutil"
  9. "mime"
  10. "mime/multipart"
  11. "net/http"
  12. "net/textproto"
  13. "path/filepath"
  14. "strings"
  15. "time"
  16. "github.com/chrislusf/seaweedfs/weed/glog"
  17. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  18. "github.com/chrislusf/seaweedfs/weed/security"
  19. "github.com/chrislusf/seaweedfs/weed/util"
  20. )
  21. type UploadResult struct {
  22. Name string `json:"name,omitempty"`
  23. Size uint32 `json:"size,omitempty"`
  24. Error string `json:"error,omitempty"`
  25. ETag string `json:"eTag,omitempty"`
  26. CipherKey []byte `json:"cipherKey,omitempty"`
  27. Mime string `json:"mime,omitempty"`
  28. Gzip uint32 `json:"gzip,omitempty"`
  29. ContentMd5 string `json:"contentMd5,omitempty"`
  30. }
  31. func (uploadResult *UploadResult) ToPbFileChunk(fileId string, offset int64) *filer_pb.FileChunk {
  32. fid, _ := filer_pb.ToFileIdObject(fileId)
  33. return &filer_pb.FileChunk{
  34. FileId: fileId,
  35. Offset: offset,
  36. Size: uint64(uploadResult.Size),
  37. Mtime: time.Now().UnixNano(),
  38. ETag: uploadResult.ETag,
  39. CipherKey: uploadResult.CipherKey,
  40. IsCompressed: uploadResult.Gzip > 0,
  41. Fid: fid,
  42. }
  43. }
  44. // HTTPClient interface for testing
  45. type HTTPClient interface {
  46. Do(req *http.Request) (*http.Response, error)
  47. }
  48. var (
  49. HttpClient HTTPClient
  50. )
  51. func init() {
  52. HttpClient = &http.Client{Transport: &http.Transport{
  53. MaxIdleConnsPerHost: 1024,
  54. }}
  55. }
  56. var fileNameEscaper = strings.NewReplacer("\\", "\\\\", "\"", "\\\"")
  57. // Upload sends a POST request to a volume server to upload the content with adjustable compression level
  58. func UploadData(uploadUrl string, filename string, cipher bool, data []byte, isInputCompressed bool, mtype string, pairMap map[string]string, jwt security.EncodedJwt) (uploadResult *UploadResult, err error) {
  59. uploadResult, err = retriedUploadData(uploadUrl, filename, cipher, data, isInputCompressed, mtype, pairMap, jwt)
  60. return
  61. }
  62. // Upload sends a POST request to a volume server to upload the content with fast compression
  63. func Upload(uploadUrl string, filename string, cipher bool, reader io.Reader, isInputCompressed bool, mtype string, pairMap map[string]string, jwt security.EncodedJwt) (uploadResult *UploadResult, err error, data []byte) {
  64. uploadResult, err, data = doUpload(uploadUrl, filename, cipher, reader, isInputCompressed, mtype, pairMap, jwt)
  65. return
  66. }
  67. func doUpload(uploadUrl string, filename string, cipher bool, reader io.Reader, isInputCompressed bool, mtype string, pairMap map[string]string, jwt security.EncodedJwt) (uploadResult *UploadResult, err error, data []byte) {
  68. data, err = ioutil.ReadAll(reader)
  69. if err != nil {
  70. err = fmt.Errorf("read input: %v", err)
  71. return
  72. }
  73. uploadResult, uploadErr := retriedUploadData(uploadUrl, filename, cipher, data, isInputCompressed, mtype, pairMap, jwt)
  74. return uploadResult, uploadErr, data
  75. }
  76. func retriedUploadData(uploadUrl string, filename string, cipher bool, data []byte, isInputCompressed bool, mtype string, pairMap map[string]string, jwt security.EncodedJwt) (uploadResult *UploadResult, err error) {
  77. for i := 0; i < 3; i++ {
  78. uploadResult, err = doUploadData(uploadUrl, filename, cipher, data, isInputCompressed, mtype, pairMap, jwt)
  79. if err == nil {
  80. return
  81. } else {
  82. glog.Warningf("uploading to %s: %v", uploadUrl, err)
  83. }
  84. }
  85. return
  86. }
  87. func doUploadData(uploadUrl string, filename string, cipher bool, data []byte, isInputCompressed bool, mtype string, pairMap map[string]string, jwt security.EncodedJwt) (uploadResult *UploadResult, err error) {
  88. contentIsGzipped := isInputCompressed
  89. shouldGzipNow := false
  90. if !isInputCompressed {
  91. if mtype == "" {
  92. mtype = http.DetectContentType(data)
  93. // println("detect1 mimetype to", mtype)
  94. if mtype == "application/octet-stream" {
  95. mtype = ""
  96. }
  97. }
  98. if shouldBeCompressed, iAmSure := util.IsCompressableFileType(filepath.Base(filename), mtype); iAmSure && shouldBeCompressed {
  99. shouldGzipNow = true
  100. } else if !iAmSure && mtype == "" && len(data) > 128 {
  101. var compressed []byte
  102. compressed, err = util.GzipData(data[0:128])
  103. shouldGzipNow = len(compressed)*10 < 128*9 // can not compress to less than 90%
  104. }
  105. }
  106. var clearDataLen int
  107. // gzip if possible
  108. // this could be double copying
  109. clearDataLen = len(data)
  110. if shouldGzipNow {
  111. compressed, compressErr := util.GzipData(data)
  112. // fmt.Printf("data is compressed from %d ==> %d\n", len(data), len(compressed))
  113. if compressErr == nil {
  114. data = compressed
  115. contentIsGzipped = true
  116. }
  117. } else if isInputCompressed {
  118. // just to get the clear data length
  119. clearData, err := util.DecompressData(data)
  120. if err == nil {
  121. clearDataLen = len(clearData)
  122. }
  123. }
  124. if cipher {
  125. // encrypt(gzip(data))
  126. // encrypt
  127. cipherKey := util.GenCipherKey()
  128. encryptedData, encryptionErr := util.Encrypt(data, cipherKey)
  129. if encryptionErr != nil {
  130. err = fmt.Errorf("encrypt input: %v", encryptionErr)
  131. return
  132. }
  133. // upload data
  134. uploadResult, err = upload_content(uploadUrl, func(w io.Writer) (err error) {
  135. _, err = w.Write(encryptedData)
  136. return
  137. }, "", false, len(encryptedData), "", nil, jwt)
  138. if uploadResult != nil {
  139. uploadResult.Name = filename
  140. uploadResult.Mime = mtype
  141. uploadResult.CipherKey = cipherKey
  142. }
  143. } else {
  144. // upload data
  145. uploadResult, err = upload_content(uploadUrl, func(w io.Writer) (err error) {
  146. _, err = w.Write(data)
  147. return
  148. }, filename, contentIsGzipped, 0, mtype, pairMap, jwt)
  149. }
  150. if uploadResult == nil {
  151. return
  152. }
  153. uploadResult.Size = uint32(clearDataLen)
  154. if contentIsGzipped {
  155. uploadResult.Gzip = 1
  156. }
  157. return uploadResult, err
  158. }
  159. func upload_content(uploadUrl string, fillBufferFunction func(w io.Writer) error, filename string, isGzipped bool, originalDataSize int, mtype string, pairMap map[string]string, jwt security.EncodedJwt) (*UploadResult, error) {
  160. body_buf := bytes.NewBufferString("")
  161. body_writer := multipart.NewWriter(body_buf)
  162. h := make(textproto.MIMEHeader)
  163. h.Set("Content-Disposition", fmt.Sprintf(`form-data; name="file"; filename="%s"`, fileNameEscaper.Replace(filename)))
  164. if mtype == "" {
  165. mtype = mime.TypeByExtension(strings.ToLower(filepath.Ext(filename)))
  166. }
  167. if mtype != "" {
  168. h.Set("Content-Type", mtype)
  169. }
  170. if isGzipped {
  171. h.Set("Content-Encoding", "gzip")
  172. }
  173. file_writer, cp_err := body_writer.CreatePart(h)
  174. if cp_err != nil {
  175. glog.V(0).Infoln("error creating form file", cp_err.Error())
  176. return nil, cp_err
  177. }
  178. if err := fillBufferFunction(file_writer); err != nil {
  179. glog.V(0).Infoln("error copying data", err)
  180. return nil, err
  181. }
  182. content_type := body_writer.FormDataContentType()
  183. if err := body_writer.Close(); err != nil {
  184. glog.V(0).Infoln("error closing body", err)
  185. return nil, err
  186. }
  187. req, postErr := http.NewRequest("POST", uploadUrl, body_buf)
  188. if postErr != nil {
  189. glog.V(1).Infof("failing to upload to %s: %v", uploadUrl, postErr)
  190. return nil, fmt.Errorf("failing to upload to %s: %v", uploadUrl, postErr)
  191. }
  192. req.Header.Set("Content-Type", content_type)
  193. for k, v := range pairMap {
  194. req.Header.Set(k, v)
  195. }
  196. if jwt != "" {
  197. req.Header.Set("Authorization", "BEARER "+string(jwt))
  198. }
  199. resp, post_err := HttpClient.Do(req)
  200. if post_err != nil {
  201. glog.V(1).Infof("failing to upload to %v: %v", uploadUrl, post_err)
  202. return nil, fmt.Errorf("failing to upload to %v: %v", uploadUrl, post_err)
  203. }
  204. defer resp.Body.Close()
  205. var ret UploadResult
  206. etag := getEtag(resp)
  207. if resp.StatusCode == http.StatusNoContent {
  208. ret.ETag = etag
  209. return &ret, nil
  210. }
  211. resp_body, ra_err := ioutil.ReadAll(resp.Body)
  212. if ra_err != nil {
  213. return nil, ra_err
  214. }
  215. unmarshal_err := json.Unmarshal(resp_body, &ret)
  216. if unmarshal_err != nil {
  217. glog.V(0).Infoln("failing to read upload response", uploadUrl, string(resp_body))
  218. return nil, unmarshal_err
  219. }
  220. if ret.Error != "" {
  221. return nil, errors.New(ret.Error)
  222. }
  223. ret.ETag = etag
  224. ret.ContentMd5 = resp.Header.Get("Content-MD5")
  225. return &ret, nil
  226. }
  227. func getEtag(r *http.Response) (etag string) {
  228. etag = r.Header.Get("ETag")
  229. if strings.HasPrefix(etag, "\"") && strings.HasSuffix(etag, "\"") {
  230. etag = etag[1 : len(etag)-1]
  231. }
  232. return
  233. }