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.

163 lines
4.6 KiB

6 years ago
6 years ago
6 years ago
6 years ago
3 years ago
5 years ago
6 years ago
6 years ago
5 years ago
5 years ago
5 years ago
6 years ago
6 years ago
  1. package S3Sink
  2. import (
  3. "context"
  4. "fmt"
  5. "strings"
  6. "sync"
  7. "github.com/aws/aws-sdk-go/aws"
  8. "github.com/aws/aws-sdk-go/aws/credentials"
  9. "github.com/aws/aws-sdk-go/aws/session"
  10. "github.com/aws/aws-sdk-go/service/s3"
  11. "github.com/aws/aws-sdk-go/service/s3/s3iface"
  12. "github.com/seaweedfs/seaweedfs/weed/filer"
  13. "github.com/seaweedfs/seaweedfs/weed/glog"
  14. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  15. "github.com/seaweedfs/seaweedfs/weed/replication/sink"
  16. "github.com/seaweedfs/seaweedfs/weed/replication/source"
  17. "github.com/seaweedfs/seaweedfs/weed/util"
  18. )
  19. type S3Sink struct {
  20. conn s3iface.S3API
  21. region string
  22. bucket string
  23. dir string
  24. endpoint string
  25. acl string
  26. filerSource *source.FilerSource
  27. isIncremental bool
  28. }
  29. func init() {
  30. sink.Sinks = append(sink.Sinks, &S3Sink{})
  31. }
  32. func (s3sink *S3Sink) GetName() string {
  33. return "s3"
  34. }
  35. func (s3sink *S3Sink) GetSinkToDirectory() string {
  36. return s3sink.dir
  37. }
  38. func (s3sink *S3Sink) IsIncremental() bool {
  39. return s3sink.isIncremental
  40. }
  41. func (s3sink *S3Sink) Initialize(configuration util.Configuration, prefix string) error {
  42. glog.V(0).Infof("sink.s3.region: %v", configuration.GetString(prefix+"region"))
  43. glog.V(0).Infof("sink.s3.bucket: %v", configuration.GetString(prefix+"bucket"))
  44. glog.V(0).Infof("sink.s3.directory: %v", configuration.GetString(prefix+"directory"))
  45. glog.V(0).Infof("sink.s3.endpoint: %v", configuration.GetString(prefix+"endpoint"))
  46. glog.V(0).Infof("sink.s3.acl: %v", configuration.GetString(prefix+"acl"))
  47. glog.V(0).Infof("sink.s3.is_incremental: %v", configuration.GetString(prefix+"is_incremental"))
  48. s3sink.isIncremental = configuration.GetBool(prefix + "is_incremental")
  49. return s3sink.initialize(
  50. configuration.GetString(prefix+"aws_access_key_id"),
  51. configuration.GetString(prefix+"aws_secret_access_key"),
  52. configuration.GetString(prefix+"region"),
  53. configuration.GetString(prefix+"bucket"),
  54. configuration.GetString(prefix+"directory"),
  55. configuration.GetString(prefix+"endpoint"),
  56. configuration.GetString(prefix+"acl"),
  57. )
  58. }
  59. func (s3sink *S3Sink) SetSourceFiler(s *source.FilerSource) {
  60. s3sink.filerSource = s
  61. }
  62. func (s3sink *S3Sink) initialize(awsAccessKeyId, awsSecretAccessKey, region, bucket, dir, endpoint, acl string) error {
  63. s3sink.region = region
  64. s3sink.bucket = bucket
  65. s3sink.dir = dir
  66. s3sink.endpoint = endpoint
  67. s3sink.acl = acl
  68. config := &aws.Config{
  69. Region: aws.String(s3sink.region),
  70. Endpoint: aws.String(s3sink.endpoint),
  71. S3ForcePathStyle: aws.Bool(true),
  72. S3DisableContentMD5Validation: aws.Bool(true),
  73. }
  74. if awsAccessKeyId != "" && awsSecretAccessKey != "" {
  75. config.Credentials = credentials.NewStaticCredentials(awsAccessKeyId, awsSecretAccessKey, "")
  76. }
  77. sess, err := session.NewSession(config)
  78. if err != nil {
  79. return fmt.Errorf("create aws session: %v", err)
  80. }
  81. s3sink.conn = s3.New(sess)
  82. return nil
  83. }
  84. func (s3sink *S3Sink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error {
  85. key = cleanKey(key)
  86. if isDirectory {
  87. key = key + "/"
  88. }
  89. return s3sink.deleteObject(key)
  90. }
  91. func (s3sink *S3Sink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error {
  92. key = cleanKey(key)
  93. if entry.IsDirectory {
  94. return nil
  95. }
  96. uploadId, err := s3sink.createMultipartUpload(key, entry)
  97. if err != nil {
  98. return fmt.Errorf("createMultipartUpload: %v", err)
  99. }
  100. totalSize := filer.FileSize(entry)
  101. chunkViews := filer.ViewFromChunks(s3sink.filerSource.LookupFileId, entry.Chunks, 0, int64(totalSize))
  102. parts := make([]*s3.CompletedPart, len(chunkViews))
  103. var wg sync.WaitGroup
  104. for chunkIndex, chunk := range chunkViews {
  105. partId := chunkIndex + 1
  106. wg.Add(1)
  107. go func(chunk *filer.ChunkView, index int) {
  108. defer wg.Done()
  109. if part, uploadErr := s3sink.uploadPart(key, uploadId, partId, chunk); uploadErr != nil {
  110. err = uploadErr
  111. glog.Errorf("uploadPart: %v", uploadErr)
  112. } else {
  113. parts[index] = part
  114. }
  115. }(chunk, chunkIndex)
  116. }
  117. wg.Wait()
  118. if err != nil {
  119. s3sink.abortMultipartUpload(key, uploadId)
  120. return fmt.Errorf("uploadPart: %v", err)
  121. }
  122. return s3sink.completeMultipartUpload(context.Background(), key, uploadId, parts)
  123. }
  124. func (s3sink *S3Sink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error) {
  125. key = cleanKey(key)
  126. return true, s3sink.CreateEntry(key, newEntry, signatures)
  127. }
  128. func cleanKey(key string) string {
  129. if strings.HasPrefix(key, "/") {
  130. key = key[1:]
  131. }
  132. return key
  133. }