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.

128 lines
3.5 KiB

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