fetch_write.go 3.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134
  1. package filersink
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/chrislusf/seaweedfs/weed/util"
  6. "sync"
  7. "google.golang.org/grpc"
  8. "github.com/chrislusf/seaweedfs/weed/glog"
  9. "github.com/chrislusf/seaweedfs/weed/operation"
  10. "github.com/chrislusf/seaweedfs/weed/pb"
  11. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  12. "github.com/chrislusf/seaweedfs/weed/security"
  13. )
  14. func (fs *FilerSink) replicateChunks(sourceChunks []*filer_pb.FileChunk, path string) (replicatedChunks []*filer_pb.FileChunk, err error) {
  15. if len(sourceChunks) == 0 {
  16. return
  17. }
  18. replicatedChunks = make([]*filer_pb.FileChunk, len(sourceChunks))
  19. var wg sync.WaitGroup
  20. for chunkIndex, sourceChunk := range sourceChunks {
  21. wg.Add(1)
  22. go func(chunk *filer_pb.FileChunk, index int) {
  23. defer wg.Done()
  24. replicatedChunk, e := fs.replicateOneChunk(chunk, path)
  25. if e != nil {
  26. err = e
  27. return
  28. }
  29. replicatedChunks[index] = replicatedChunk
  30. }(sourceChunk, chunkIndex)
  31. }
  32. wg.Wait()
  33. return
  34. }
  35. func (fs *FilerSink) replicateOneChunk(sourceChunk *filer_pb.FileChunk, path string) (*filer_pb.FileChunk, error) {
  36. fileId, err := fs.fetchAndWrite(sourceChunk, path)
  37. if err != nil {
  38. return nil, fmt.Errorf("copy %s: %v", sourceChunk.GetFileIdString(), err)
  39. }
  40. return &filer_pb.FileChunk{
  41. FileId: fileId,
  42. Offset: sourceChunk.Offset,
  43. Size: sourceChunk.Size,
  44. Mtime: sourceChunk.Mtime,
  45. ETag: sourceChunk.ETag,
  46. SourceFileId: sourceChunk.GetFileIdString(),
  47. CipherKey: sourceChunk.CipherKey,
  48. IsCompressed: sourceChunk.IsCompressed,
  49. }, nil
  50. }
  51. func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string) (fileId string, err error) {
  52. filename, header, resp, err := fs.filerSource.ReadPart(sourceChunk.GetFileIdString())
  53. if err != nil {
  54. return "", fmt.Errorf("read part %s: %v", sourceChunk.GetFileIdString(), err)
  55. }
  56. defer util.CloseResponse(resp)
  57. var host string
  58. var auth security.EncodedJwt
  59. if err := fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  60. request := &filer_pb.AssignVolumeRequest{
  61. Count: 1,
  62. Replication: fs.replication,
  63. Collection: fs.collection,
  64. TtlSec: fs.ttlSec,
  65. DataCenter: fs.dataCenter,
  66. DiskType: fs.diskType,
  67. Path: path,
  68. }
  69. resp, err := client.AssignVolume(context.Background(), request)
  70. if err != nil {
  71. glog.V(0).Infof("assign volume failure %v: %v", request, err)
  72. return err
  73. }
  74. if resp.Error != "" {
  75. return fmt.Errorf("assign volume failure %v: %v", request, resp.Error)
  76. }
  77. fileId, host, auth = resp.FileId, resp.Url, security.EncodedJwt(resp.Auth)
  78. return nil
  79. }); err != nil {
  80. return "", fmt.Errorf("filerGrpcAddress assign volume: %v", err)
  81. }
  82. fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)
  83. if fs.writeChunkByFiler {
  84. fileUrl = fmt.Sprintf("http://%s/?proxyChunkId=%s", fs.address, fileId)
  85. }
  86. glog.V(4).Infof("replicating %s to %s header:%+v", filename, fileUrl, header)
  87. // fetch data as is, regardless whether it is encrypted or not
  88. uploadResult, err, _ := operation.Upload(fileUrl, filename, false, resp.Body, "gzip" == header.Get("Content-Encoding"), header.Get("Content-Type"), nil, auth)
  89. if err != nil {
  90. glog.V(0).Infof("upload source data %v to %s: %v", sourceChunk.GetFileIdString(), fileUrl, err)
  91. return "", fmt.Errorf("upload data: %v", err)
  92. }
  93. if uploadResult.Error != "" {
  94. glog.V(0).Infof("upload failure %v to %s: %v", filename, fileUrl, err)
  95. return "", fmt.Errorf("upload result: %v", uploadResult.Error)
  96. }
  97. return
  98. }
  99. var _ = filer_pb.FilerClient(&FilerSink{})
  100. func (fs *FilerSink) WithFilerClient(fn func(filer_pb.SeaweedFilerClient) error) error {
  101. return pb.WithCachedGrpcClient(func(grpcConnection *grpc.ClientConn) error {
  102. client := filer_pb.NewSeaweedFilerClient(grpcConnection)
  103. return fn(client)
  104. }, fs.grpcAddress, fs.grpcDialOption)
  105. }
  106. func (fs *FilerSink) AdjustedUrl(location *filer_pb.Location) string {
  107. return location.Url
  108. }