filer_server_handlers_write_upload.go 6.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218
  1. package weed_server
  2. import (
  3. "bytes"
  4. "crypto/md5"
  5. "fmt"
  6. "hash"
  7. "io"
  8. "net/http"
  9. "strconv"
  10. "sync"
  11. "time"
  12. "golang.org/x/exp/slices"
  13. "github.com/seaweedfs/seaweedfs/weed/glog"
  14. "github.com/seaweedfs/seaweedfs/weed/operation"
  15. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  16. "github.com/seaweedfs/seaweedfs/weed/security"
  17. "github.com/seaweedfs/seaweedfs/weed/stats"
  18. "github.com/seaweedfs/seaweedfs/weed/util"
  19. )
  20. var bufPool = sync.Pool{
  21. New: func() interface{} {
  22. return new(bytes.Buffer)
  23. },
  24. }
  25. func (fs *FilerServer) uploadReaderToChunks(w http.ResponseWriter, r *http.Request, reader io.Reader, chunkSize int32, fileName, contentType string, contentLength int64, so *operation.StorageOption) (fileChunks []*filer_pb.FileChunk, md5Hash hash.Hash, chunkOffset int64, uploadErr error, smallContent []byte) {
  26. query := r.URL.Query()
  27. isAppend := isAppend(r)
  28. if query.Has("offset") {
  29. offset := query.Get("offset")
  30. offsetInt, err := strconv.ParseInt(offset, 10, 64)
  31. if err != nil || offsetInt < 0 {
  32. err = fmt.Errorf("invalid 'offset': '%s'", offset)
  33. return nil, nil, 0, err, nil
  34. }
  35. if isAppend && offsetInt > 0 {
  36. err = fmt.Errorf("cannot set offset when op=append")
  37. return nil, nil, 0, err, nil
  38. }
  39. chunkOffset = offsetInt
  40. }
  41. md5Hash = md5.New()
  42. var partReader = io.NopCloser(io.TeeReader(reader, md5Hash))
  43. var wg sync.WaitGroup
  44. var bytesBufferCounter int64 = 4
  45. bytesBufferLimitChan := make(chan struct{}, bytesBufferCounter)
  46. var fileChunksLock sync.Mutex
  47. var uploadErrLock sync.Mutex
  48. for {
  49. // need to throttle used byte buffer
  50. bytesBufferLimitChan <- struct{}{}
  51. bytesBuffer := bufPool.Get().(*bytes.Buffer)
  52. limitedReader := io.LimitReader(partReader, int64(chunkSize))
  53. bytesBuffer.Reset()
  54. dataSize, err := bytesBuffer.ReadFrom(limitedReader)
  55. // data, err := io.ReadAll(limitedReader)
  56. if err != nil || dataSize == 0 {
  57. bufPool.Put(bytesBuffer)
  58. <-bytesBufferLimitChan
  59. if err != nil {
  60. uploadErrLock.Lock()
  61. if uploadErr == nil {
  62. uploadErr = err
  63. }
  64. uploadErrLock.Unlock()
  65. }
  66. break
  67. }
  68. if chunkOffset == 0 && !isAppend {
  69. if dataSize < fs.option.SaveToFilerLimit {
  70. chunkOffset += dataSize
  71. smallContent = make([]byte, dataSize)
  72. bytesBuffer.Read(smallContent)
  73. bufPool.Put(bytesBuffer)
  74. <-bytesBufferLimitChan
  75. stats.FilerHandlerCounter.WithLabelValues(stats.ContentSaveToFiler).Inc()
  76. break
  77. }
  78. } else {
  79. stats.FilerHandlerCounter.WithLabelValues(stats.AutoChunk).Inc()
  80. }
  81. wg.Add(1)
  82. go func(offset int64) {
  83. defer func() {
  84. bufPool.Put(bytesBuffer)
  85. <-bytesBufferLimitChan
  86. wg.Done()
  87. }()
  88. chunks, toChunkErr := fs.dataToChunk(fileName, contentType, bytesBuffer.Bytes(), offset, so)
  89. if toChunkErr != nil {
  90. uploadErrLock.Lock()
  91. if uploadErr == nil {
  92. uploadErr = toChunkErr
  93. }
  94. uploadErrLock.Unlock()
  95. }
  96. if chunks != nil {
  97. fileChunksLock.Lock()
  98. fileChunksSize := len(fileChunks) + len(chunks)
  99. for _, chunk := range chunks {
  100. fileChunks = append(fileChunks, chunk)
  101. glog.V(4).Infof("uploaded %s chunk %d to %s [%d,%d)", fileName, fileChunksSize, chunk.FileId, offset, offset+int64(chunk.Size))
  102. }
  103. fileChunksLock.Unlock()
  104. }
  105. }(chunkOffset)
  106. // reset variables for the next chunk
  107. chunkOffset = chunkOffset + dataSize
  108. // if last chunk was not at full chunk size, but already exhausted the reader
  109. if dataSize < int64(chunkSize) {
  110. break
  111. }
  112. }
  113. wg.Wait()
  114. if uploadErr != nil {
  115. fs.filer.DeleteUncommittedChunks(fileChunks)
  116. return nil, md5Hash, 0, uploadErr, nil
  117. }
  118. slices.SortFunc(fileChunks, func(a, b *filer_pb.FileChunk) int {
  119. return int(a.Offset - b.Offset)
  120. })
  121. return fileChunks, md5Hash, chunkOffset, nil, smallContent
  122. }
  123. func (fs *FilerServer) doUpload(urlLocation string, limitedReader io.Reader, fileName string, contentType string, pairMap map[string]string, auth security.EncodedJwt) (*operation.UploadResult, error, []byte) {
  124. stats.FilerHandlerCounter.WithLabelValues(stats.ChunkUpload).Inc()
  125. start := time.Now()
  126. defer func() {
  127. stats.FilerRequestHistogram.WithLabelValues(stats.ChunkUpload).Observe(time.Since(start).Seconds())
  128. }()
  129. uploadOption := &operation.UploadOption{
  130. UploadUrl: urlLocation,
  131. Filename: fileName,
  132. Cipher: fs.option.Cipher,
  133. IsInputCompressed: false,
  134. MimeType: contentType,
  135. PairMap: pairMap,
  136. Jwt: auth,
  137. }
  138. uploader, err := operation.NewUploader()
  139. if err != nil {
  140. return nil, err, []byte{}
  141. }
  142. uploadResult, err, data := uploader.Upload(limitedReader, uploadOption)
  143. if uploadResult != nil && uploadResult.RetryCount > 0 {
  144. stats.FilerHandlerCounter.WithLabelValues(stats.ChunkUploadRetry).Add(float64(uploadResult.RetryCount))
  145. }
  146. return uploadResult, err, data
  147. }
  148. func (fs *FilerServer) dataToChunk(fileName, contentType string, data []byte, chunkOffset int64, so *operation.StorageOption) ([]*filer_pb.FileChunk, error) {
  149. dataReader := util.NewBytesReader(data)
  150. // retry to assign a different file id
  151. var fileId, urlLocation string
  152. var auth security.EncodedJwt
  153. var uploadErr error
  154. var uploadResult *operation.UploadResult
  155. var failedFileChunks []*filer_pb.FileChunk
  156. err := util.Retry("filerDataToChunk", func() error {
  157. // assign one file id for one chunk
  158. fileId, urlLocation, auth, uploadErr = fs.assignNewFileInfo(so)
  159. if uploadErr != nil {
  160. glog.V(4).Infof("retry later due to assign error: %v", uploadErr)
  161. stats.FilerHandlerCounter.WithLabelValues(stats.ChunkAssignRetry).Inc()
  162. return uploadErr
  163. }
  164. // upload the chunk to the volume server
  165. uploadResult, uploadErr, _ = fs.doUpload(urlLocation, dataReader, fileName, contentType, nil, auth)
  166. if uploadErr != nil {
  167. glog.V(4).Infof("retry later due to upload error: %v", uploadErr)
  168. stats.FilerHandlerCounter.WithLabelValues(stats.ChunkDoUploadRetry).Inc()
  169. fid, _ := filer_pb.ToFileIdObject(fileId)
  170. fileChunk := filer_pb.FileChunk{
  171. FileId: fileId,
  172. Offset: chunkOffset,
  173. Fid: fid,
  174. }
  175. failedFileChunks = append(failedFileChunks, &fileChunk)
  176. return uploadErr
  177. }
  178. return nil
  179. })
  180. if err != nil {
  181. glog.Errorf("upload error: %v", err)
  182. return failedFileChunks, err
  183. }
  184. // if last chunk exhausted the reader exactly at the border
  185. if uploadResult.Size == 0 {
  186. return nil, nil
  187. }
  188. return []*filer_pb.FileChunk{uploadResult.ToPbFileChunk(fileId, chunkOffset, time.Now().UnixNano())}, nil
  189. }