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.

194 lines
5.6 KiB

6 years ago
6 years ago
6 years ago
  1. package weed_server
  2. import (
  3. "bytes"
  4. "context"
  5. "io"
  6. "io/ioutil"
  7. "net/http"
  8. "path"
  9. "strconv"
  10. "strings"
  11. "time"
  12. "github.com/chrislusf/seaweedfs/weed/filer2"
  13. "github.com/chrislusf/seaweedfs/weed/glog"
  14. "github.com/chrislusf/seaweedfs/weed/operation"
  15. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  16. "github.com/chrislusf/seaweedfs/weed/security"
  17. "github.com/chrislusf/seaweedfs/weed/util"
  18. )
  19. func (fs *FilerServer) autoChunk(ctx context.Context, w http.ResponseWriter, r *http.Request,
  20. replication string, collection string, dataCenter string) bool {
  21. if r.Method != "POST" {
  22. glog.V(4).Infoln("AutoChunking not supported for method", r.Method)
  23. return false
  24. }
  25. // autoChunking can be set at the command-line level or as a query param. Query param overrides command-line
  26. query := r.URL.Query()
  27. parsedMaxMB, _ := strconv.ParseInt(query.Get("maxMB"), 10, 32)
  28. maxMB := int32(parsedMaxMB)
  29. if maxMB <= 0 && fs.option.MaxMB > 0 {
  30. maxMB = int32(fs.option.MaxMB)
  31. }
  32. if maxMB <= 0 {
  33. glog.V(4).Infoln("AutoChunking not enabled")
  34. return false
  35. }
  36. glog.V(4).Infoln("AutoChunking level set to", maxMB, "(MB)")
  37. chunkSize := 1024 * 1024 * maxMB
  38. contentLength := int64(0)
  39. if contentLengthHeader := r.Header["Content-Length"]; len(contentLengthHeader) == 1 {
  40. contentLength, _ = strconv.ParseInt(contentLengthHeader[0], 10, 64)
  41. if contentLength <= int64(chunkSize) {
  42. glog.V(4).Infoln("Content-Length of", contentLength, "is less than the chunk size of", chunkSize, "so autoChunking will be skipped.")
  43. return false
  44. }
  45. }
  46. if contentLength <= 0 {
  47. glog.V(4).Infoln("Content-Length value is missing or unexpected so autoChunking will be skipped.")
  48. return false
  49. }
  50. reply, err := fs.doAutoChunk(ctx, w, r, contentLength, chunkSize, replication, collection, dataCenter)
  51. if err != nil {
  52. writeJsonError(w, r, http.StatusInternalServerError, err)
  53. } else if reply != nil {
  54. writeJsonQuiet(w, r, http.StatusCreated, reply)
  55. }
  56. return true
  57. }
  58. func (fs *FilerServer) doAutoChunk(ctx context.Context, w http.ResponseWriter, r *http.Request,
  59. contentLength int64, chunkSize int32, replication string, collection string, dataCenter string) (filerResult *FilerPostResult, replyerr error) {
  60. multipartReader, multipartReaderErr := r.MultipartReader()
  61. if multipartReaderErr != nil {
  62. return nil, multipartReaderErr
  63. }
  64. part1, part1Err := multipartReader.NextPart()
  65. if part1Err != nil {
  66. return nil, part1Err
  67. }
  68. fileName := part1.FileName()
  69. if fileName != "" {
  70. fileName = path.Base(fileName)
  71. }
  72. var fileChunks []*filer_pb.FileChunk
  73. totalBytesRead := int64(0)
  74. tmpBufferSize := int32(1024 * 1024)
  75. tmpBuffer := bytes.NewBuffer(make([]byte, 0, tmpBufferSize))
  76. chunkBuf := make([]byte, chunkSize+tmpBufferSize, chunkSize+tmpBufferSize) // chunk size plus a little overflow
  77. chunkBufOffset := int32(0)
  78. chunkOffset := int64(0)
  79. writtenChunks := 0
  80. filerResult = &FilerPostResult{
  81. Name: fileName,
  82. }
  83. for totalBytesRead < contentLength {
  84. tmpBuffer.Reset()
  85. bytesRead, readErr := io.CopyN(tmpBuffer, part1, int64(tmpBufferSize))
  86. readFully := readErr != nil && readErr == io.EOF
  87. tmpBuf := tmpBuffer.Bytes()
  88. bytesToCopy := tmpBuf[0:int(bytesRead)]
  89. copy(chunkBuf[chunkBufOffset:chunkBufOffset+int32(bytesRead)], bytesToCopy)
  90. chunkBufOffset = chunkBufOffset + int32(bytesRead)
  91. if chunkBufOffset >= chunkSize || readFully || (chunkBufOffset > 0 && bytesRead == 0) {
  92. writtenChunks = writtenChunks + 1
  93. fileId, urlLocation, auth, assignErr := fs.assignNewFileInfo(w, r, replication, collection, dataCenter)
  94. if assignErr != nil {
  95. return nil, assignErr
  96. }
  97. // upload the chunk to the volume server
  98. chunkName := fileName + "_chunk_" + strconv.FormatInt(int64(len(fileChunks)+1), 10)
  99. uploadErr := fs.doUpload(urlLocation, w, r, chunkBuf[0:chunkBufOffset], chunkName, "application/octet-stream", fileId, auth)
  100. if uploadErr != nil {
  101. return nil, uploadErr
  102. }
  103. // Save to chunk manifest structure
  104. fileChunks = append(fileChunks,
  105. &filer_pb.FileChunk{
  106. FileId: fileId,
  107. Offset: chunkOffset,
  108. Size: uint64(chunkBufOffset),
  109. Mtime: time.Now().UnixNano(),
  110. },
  111. )
  112. // reset variables for the next chunk
  113. chunkBufOffset = 0
  114. chunkOffset = totalBytesRead + int64(bytesRead)
  115. }
  116. totalBytesRead = totalBytesRead + int64(bytesRead)
  117. if bytesRead == 0 || readFully {
  118. break
  119. }
  120. if readErr != nil {
  121. return nil, readErr
  122. }
  123. }
  124. path := r.URL.Path
  125. if strings.HasSuffix(path, "/") {
  126. if fileName != "" {
  127. path += fileName
  128. }
  129. }
  130. glog.V(4).Infoln("saving", path)
  131. entry := &filer2.Entry{
  132. FullPath: filer2.FullPath(path),
  133. Attr: filer2.Attr{
  134. Mtime: time.Now(),
  135. Crtime: time.Now(),
  136. Mode: 0660,
  137. Uid: OS_UID,
  138. Gid: OS_GID,
  139. Replication: replication,
  140. Collection: collection,
  141. TtlSec: int32(util.ParseInt(r.URL.Query().Get("ttl"), 0)),
  142. },
  143. Chunks: fileChunks,
  144. }
  145. if db_err := fs.filer.CreateEntry(ctx, entry); db_err != nil {
  146. replyerr = db_err
  147. filerResult.Error = db_err.Error()
  148. glog.V(0).Infof("failing to write %s to filer server : %v", path, db_err)
  149. return
  150. }
  151. return
  152. }
  153. func (fs *FilerServer) doUpload(urlLocation string, w http.ResponseWriter, r *http.Request,
  154. chunkBuf []byte, fileName string, contentType string, fileId string, auth security.EncodedJwt) (err error) {
  155. ioReader := ioutil.NopCloser(bytes.NewBuffer(chunkBuf))
  156. uploadResult, uploadError := operation.Upload(urlLocation, fileName, ioReader, false, contentType, nil, auth)
  157. if uploadResult != nil {
  158. glog.V(0).Infoln("Chunk upload result. Name:", uploadResult.Name, "Fid:", fileId, "Size:", uploadResult.Size)
  159. }
  160. if uploadError != nil {
  161. err = uploadError
  162. }
  163. return
  164. }