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.

279 lines
8.7 KiB

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