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.

192 lines
5.3 KiB

5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
  1. package s3_backend
  2. import (
  3. "fmt"
  4. "io"
  5. "os"
  6. "strings"
  7. "time"
  8. "github.com/aws/aws-sdk-go/service/s3"
  9. "github.com/aws/aws-sdk-go/service/s3/s3iface"
  10. "github.com/google/uuid"
  11. "github.com/seaweedfs/seaweedfs/weed/glog"
  12. "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
  13. "github.com/seaweedfs/seaweedfs/weed/storage/backend"
  14. )
  15. func init() {
  16. backend.BackendStorageFactories["s3"] = &S3BackendFactory{}
  17. }
  18. type S3BackendFactory struct {
  19. }
  20. func (factory *S3BackendFactory) StorageType() backend.StorageType {
  21. return backend.StorageType("s3")
  22. }
  23. func (factory *S3BackendFactory) BuildStorage(configuration backend.StringProperties, configPrefix string, id string) (backend.BackendStorage, error) {
  24. return newS3BackendStorage(configuration, configPrefix, id)
  25. }
  26. type S3BackendStorage struct {
  27. id string
  28. aws_access_key_id string
  29. aws_secret_access_key string
  30. region string
  31. bucket string
  32. endpoint string
  33. storageClass string
  34. conn s3iface.S3API
  35. }
  36. func newS3BackendStorage(configuration backend.StringProperties, configPrefix string, id string) (s *S3BackendStorage, err error) {
  37. s = &S3BackendStorage{}
  38. s.id = id
  39. s.aws_access_key_id = configuration.GetString(configPrefix + "aws_access_key_id")
  40. s.aws_secret_access_key = configuration.GetString(configPrefix + "aws_secret_access_key")
  41. s.region = configuration.GetString(configPrefix + "region")
  42. s.bucket = configuration.GetString(configPrefix + "bucket")
  43. s.endpoint = configuration.GetString(configPrefix + "endpoint")
  44. s.storageClass = configuration.GetString(configPrefix + "storage_class")
  45. if s.storageClass == "" {
  46. s.storageClass = "STANDARD_IA"
  47. }
  48. s.conn, err = createSession(s.aws_access_key_id, s.aws_secret_access_key, s.region, s.endpoint)
  49. glog.V(0).Infof("created backend storage s3.%s for region %s bucket %s", s.id, s.region, s.bucket)
  50. return
  51. }
  52. func (s *S3BackendStorage) ToProperties() map[string]string {
  53. m := make(map[string]string)
  54. m["aws_access_key_id"] = s.aws_access_key_id
  55. m["aws_secret_access_key"] = s.aws_secret_access_key
  56. m["region"] = s.region
  57. m["bucket"] = s.bucket
  58. m["endpoint"] = s.endpoint
  59. m["storage_class"] = s.storageClass
  60. return m
  61. }
  62. func (s *S3BackendStorage) NewStorageFile(key string, tierInfo *volume_server_pb.VolumeInfo) backend.BackendStorageFile {
  63. if strings.HasPrefix(key, "/") {
  64. key = key[1:]
  65. }
  66. f := &S3BackendStorageFile{
  67. backendStorage: s,
  68. key: key,
  69. tierInfo: tierInfo,
  70. }
  71. return f
  72. }
  73. func (s *S3BackendStorage) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error) (key string, size int64, err error) {
  74. randomUuid, _ := uuid.NewRandom()
  75. key = randomUuid.String()
  76. glog.V(1).Infof("copying dat file of %s to remote s3.%s as %s", f.Name(), s.id, key)
  77. size, err = uploadToS3(s.conn, f.Name(), s.bucket, key, s.storageClass, fn)
  78. return
  79. }
  80. func (s *S3BackendStorage) DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error) (size int64, err error) {
  81. glog.V(1).Infof("download dat file of %s from remote s3.%s as %s", fileName, s.id, key)
  82. size, err = downloadFromS3(s.conn, fileName, s.bucket, key, fn)
  83. return
  84. }
  85. func (s *S3BackendStorage) DeleteFile(key string) (err error) {
  86. glog.V(1).Infof("delete dat file %s from remote", key)
  87. err = deleteFromS3(s.conn, s.bucket, key)
  88. return
  89. }
  90. type S3BackendStorageFile struct {
  91. backendStorage *S3BackendStorage
  92. key string
  93. tierInfo *volume_server_pb.VolumeInfo
  94. }
  95. func (s3backendStorageFile S3BackendStorageFile) ReadAt(p []byte, off int64) (n int, err error) {
  96. bytesRange := fmt.Sprintf("bytes=%d-%d", off, off+int64(len(p))-1)
  97. getObjectOutput, getObjectErr := s3backendStorageFile.backendStorage.conn.GetObject(&s3.GetObjectInput{
  98. Bucket: &s3backendStorageFile.backendStorage.bucket,
  99. Key: &s3backendStorageFile.key,
  100. Range: &bytesRange,
  101. })
  102. if getObjectErr != nil {
  103. return 0, fmt.Errorf("bucket %s GetObject %s: %v", s3backendStorageFile.backendStorage.bucket, s3backendStorageFile.key, getObjectErr)
  104. }
  105. defer getObjectOutput.Body.Close()
  106. // glog.V(3).Infof("read %s %s", s3backendStorageFile.key, bytesRange)
  107. // glog.V(3).Infof("content range: %s, contentLength: %d", *getObjectOutput.ContentRange, *getObjectOutput.ContentLength)
  108. var readCount int
  109. for {
  110. p = p[readCount:]
  111. readCount, err = getObjectOutput.Body.Read(p)
  112. n += readCount
  113. if err != nil {
  114. break
  115. }
  116. }
  117. if err == io.EOF {
  118. err = nil
  119. }
  120. return
  121. }
  122. func (s3backendStorageFile S3BackendStorageFile) WriteAt(p []byte, off int64) (n int, err error) {
  123. panic("not implemented")
  124. }
  125. func (s3backendStorageFile S3BackendStorageFile) Truncate(off int64) error {
  126. panic("not implemented")
  127. }
  128. func (s3backendStorageFile S3BackendStorageFile) Close() error {
  129. return nil
  130. }
  131. func (s3backendStorageFile S3BackendStorageFile) GetStat() (datSize int64, modTime time.Time, err error) {
  132. files := s3backendStorageFile.tierInfo.GetFiles()
  133. if len(files) == 0 {
  134. err = fmt.Errorf("remote file info not found")
  135. return
  136. }
  137. datSize = int64(files[0].FileSize)
  138. modTime = time.Unix(int64(files[0].ModifiedTime), 0)
  139. return
  140. }
  141. func (s3backendStorageFile S3BackendStorageFile) Name() string {
  142. return s3backendStorageFile.key
  143. }
  144. func (s3backendStorageFile S3BackendStorageFile) Sync() error {
  145. return nil
  146. }