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.

125 lines
3.3 KiB

6 years ago
6 years ago
  1. package filersink
  2. import (
  3. "context"
  4. "fmt"
  5. "strings"
  6. "sync"
  7. "github.com/chrislusf/seaweedfs/weed/glog"
  8. "github.com/chrislusf/seaweedfs/weed/operation"
  9. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  10. "github.com/chrislusf/seaweedfs/weed/security"
  11. "github.com/chrislusf/seaweedfs/weed/util"
  12. )
  13. func (fs *FilerSink) replicateChunks(sourceChunks []*filer_pb.FileChunk) (replicatedChunks []*filer_pb.FileChunk, err error) {
  14. if len(sourceChunks) == 0 {
  15. return
  16. }
  17. var wg sync.WaitGroup
  18. for _, sourceChunk := range sourceChunks {
  19. wg.Add(1)
  20. go func(chunk *filer_pb.FileChunk) {
  21. defer wg.Done()
  22. replicatedChunk, e := fs.replicateOneChunk(chunk)
  23. if e != nil {
  24. err = e
  25. }
  26. replicatedChunks = append(replicatedChunks, replicatedChunk)
  27. }(sourceChunk)
  28. }
  29. wg.Wait()
  30. return
  31. }
  32. func (fs *FilerSink) replicateOneChunk(sourceChunk *filer_pb.FileChunk) (*filer_pb.FileChunk, error) {
  33. fileId, err := fs.fetchAndWrite(sourceChunk)
  34. if err != nil {
  35. return nil, fmt.Errorf("copy %s: %v", sourceChunk.FileId, err)
  36. }
  37. return &filer_pb.FileChunk{
  38. FileId: fileId,
  39. Offset: sourceChunk.Offset,
  40. Size: sourceChunk.Size,
  41. Mtime: sourceChunk.Mtime,
  42. ETag: sourceChunk.ETag,
  43. SourceFileId: sourceChunk.FileId,
  44. }, nil
  45. }
  46. func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk) (fileId string, err error) {
  47. filename, header, readCloser, err := fs.filerSource.ReadPart(sourceChunk.FileId)
  48. if err != nil {
  49. return "", fmt.Errorf("read part %s: %v", sourceChunk.FileId, err)
  50. }
  51. defer readCloser.Close()
  52. var host string
  53. var auth security.EncodedJwt
  54. if err := fs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  55. request := &filer_pb.AssignVolumeRequest{
  56. Count: 1,
  57. Replication: fs.replication,
  58. Collection: fs.collection,
  59. TtlSec: fs.ttlSec,
  60. DataCenter: fs.dataCenter,
  61. }
  62. resp, err := client.AssignVolume(context.Background(), request)
  63. if err != nil {
  64. glog.V(0).Infof("assign volume failure %v: %v", request, err)
  65. return err
  66. }
  67. fileId, host, auth = resp.FileId, resp.Url, security.EncodedJwt(resp.Auth)
  68. return nil
  69. }); err != nil {
  70. return "", fmt.Errorf("filerGrpcAddress assign volume: %v", err)
  71. }
  72. fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)
  73. glog.V(4).Infof("replicating %s to %s header:%+v", filename, fileUrl, header)
  74. uploadResult, err := operation.Upload(fileUrl, filename, readCloser,
  75. "gzip" == header.Get("Content-Encoding"), header.Get("Content-Type"), nil, auth)
  76. if err != nil {
  77. glog.V(0).Infof("upload data %v to %s: %v", filename, fileUrl, err)
  78. return "", fmt.Errorf("upload data: %v", err)
  79. }
  80. if uploadResult.Error != "" {
  81. glog.V(0).Infof("upload failure %v to %s: %v", filename, fileUrl, err)
  82. return "", fmt.Errorf("upload result: %v", uploadResult.Error)
  83. }
  84. return
  85. }
  86. func (fs *FilerSink) withFilerClient(fn func(filer_pb.SeaweedFilerClient) error) error {
  87. grpcConnection, err := util.GrpcDial(fs.grpcAddress, fs.grpcDialOption)
  88. if err != nil {
  89. return fmt.Errorf("fail to dial %s: %v", fs.grpcAddress, err)
  90. }
  91. defer grpcConnection.Close()
  92. client := filer_pb.NewSeaweedFilerClient(grpcConnection)
  93. return fn(client)
  94. }
  95. func volumeId(fileId string) string {
  96. lastCommaIndex := strings.LastIndex(fileId, ",")
  97. if lastCommaIndex > 0 {
  98. return fileId[:lastCommaIndex]
  99. }
  100. return fileId
  101. }