filer_server_handlers_write_cipher.go 2.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107
  1. package weed_server
  2. import (
  3. "bytes"
  4. "context"
  5. "fmt"
  6. "net/http"
  7. "strings"
  8. "time"
  9. "github.com/seaweedfs/seaweedfs/weed/filer"
  10. "github.com/seaweedfs/seaweedfs/weed/glog"
  11. "github.com/seaweedfs/seaweedfs/weed/operation"
  12. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  13. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  14. "github.com/seaweedfs/seaweedfs/weed/util"
  15. )
  16. // handling single chunk POST or PUT upload
  17. func (fs *FilerServer) encrypt(ctx context.Context, w http.ResponseWriter, r *http.Request, so *operation.StorageOption) (filerResult *FilerPostResult, err error) {
  18. fileId, urlLocation, auth, err := fs.assignNewFileInfo(so)
  19. if err != nil || fileId == "" || urlLocation == "" {
  20. return nil, fmt.Errorf("fail to allocate volume for %s, collection:%s, datacenter:%s", r.URL.Path, so.Collection, so.DataCenter)
  21. }
  22. glog.V(4).Infof("write %s to %v", r.URL.Path, urlLocation)
  23. // Note: encrypt(gzip(data)), encrypt data first, then gzip
  24. sizeLimit := int64(fs.option.MaxMB) * 1024 * 1024
  25. bytesBuffer := bufPool.Get().(*bytes.Buffer)
  26. defer bufPool.Put(bytesBuffer)
  27. pu, err := needle.ParseUpload(r, sizeLimit, bytesBuffer)
  28. uncompressedData := pu.Data
  29. if pu.IsGzipped {
  30. uncompressedData = pu.UncompressedData
  31. }
  32. if pu.MimeType == "" {
  33. pu.MimeType = http.DetectContentType(uncompressedData)
  34. // println("detect2 mimetype to", pu.MimeType)
  35. }
  36. uploadOption := &operation.UploadOption{
  37. UploadUrl: urlLocation,
  38. Filename: pu.FileName,
  39. Cipher: true,
  40. IsInputCompressed: false,
  41. MimeType: pu.MimeType,
  42. PairMap: pu.PairMap,
  43. Jwt: auth,
  44. }
  45. uploader, uploaderErr := operation.NewUploader()
  46. if uploaderErr != nil {
  47. return nil, fmt.Errorf("uploader initialization error: %v", uploaderErr)
  48. }
  49. uploadResult, uploadError := uploader.UploadData(uncompressedData, uploadOption)
  50. if uploadError != nil {
  51. return nil, fmt.Errorf("upload to volume server: %v", uploadError)
  52. }
  53. // Save to chunk manifest structure
  54. fileChunks := []*filer_pb.FileChunk{uploadResult.ToPbFileChunk(fileId, 0, time.Now().UnixNano())}
  55. // fmt.Printf("uploaded: %+v\n", uploadResult)
  56. path := r.URL.Path
  57. if strings.HasSuffix(path, "/") {
  58. if pu.FileName != "" {
  59. path += pu.FileName
  60. }
  61. }
  62. entry := &filer.Entry{
  63. FullPath: util.FullPath(path),
  64. Attr: filer.Attr{
  65. Mtime: time.Now(),
  66. Crtime: time.Now(),
  67. Mode: 0660,
  68. Uid: OS_UID,
  69. Gid: OS_GID,
  70. TtlSec: so.TtlSeconds,
  71. Mime: pu.MimeType,
  72. Md5: util.Base64Md5ToBytes(pu.ContentMd5),
  73. },
  74. Chunks: fileChunks,
  75. }
  76. filerResult = &FilerPostResult{
  77. Name: pu.FileName,
  78. Size: int64(pu.OriginalDataSize),
  79. }
  80. if dbErr := fs.filer.CreateEntry(ctx, entry, false, false, nil, false, so.MaxFileNameLength); dbErr != nil {
  81. fs.filer.DeleteUncommittedChunks(entry.GetChunks())
  82. err = dbErr
  83. filerResult.Error = dbErr.Error()
  84. return
  85. }
  86. return
  87. }