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.

273 lines
8.4 KiB

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