b2_sink.go 2.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130
  1. package B2Sink
  2. import (
  3. "context"
  4. "github.com/chrislusf/seaweedfs/weed/replication/repl_util"
  5. "strings"
  6. "github.com/chrislusf/seaweedfs/weed/filer"
  7. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  8. "github.com/chrislusf/seaweedfs/weed/replication/sink"
  9. "github.com/chrislusf/seaweedfs/weed/replication/source"
  10. "github.com/chrislusf/seaweedfs/weed/util"
  11. "github.com/kurin/blazer/b2"
  12. )
  13. type B2Sink struct {
  14. client *b2.Client
  15. bucket string
  16. dir string
  17. filerSource *source.FilerSource
  18. isIncremental bool
  19. }
  20. func init() {
  21. sink.Sinks = append(sink.Sinks, &B2Sink{})
  22. }
  23. func (g *B2Sink) GetName() string {
  24. return "backblaze"
  25. }
  26. func (g *B2Sink) GetSinkToDirectory() string {
  27. return g.dir
  28. }
  29. func (g *B2Sink) IsIncremental() bool {
  30. return g.isIncremental
  31. }
  32. func (g *B2Sink) Initialize(configuration util.Configuration, prefix string) error {
  33. g.isIncremental = configuration.GetBool(prefix + "is_incremental")
  34. return g.initialize(
  35. configuration.GetString(prefix+"b2_account_id"),
  36. configuration.GetString(prefix+"b2_master_application_key"),
  37. configuration.GetString(prefix+"bucket"),
  38. configuration.GetString(prefix+"directory"),
  39. )
  40. }
  41. func (g *B2Sink) SetSourceFiler(s *source.FilerSource) {
  42. g.filerSource = s
  43. }
  44. func (g *B2Sink) initialize(accountId, accountKey, bucket, dir string) error {
  45. client, err := b2.NewClient(context.Background(), accountId, accountKey)
  46. if err != nil {
  47. return err
  48. }
  49. g.client = client
  50. g.bucket = bucket
  51. g.dir = dir
  52. return nil
  53. }
  54. func (g *B2Sink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error {
  55. key = cleanKey(key)
  56. if isDirectory {
  57. key = key + "/"
  58. }
  59. bucket, err := g.client.Bucket(context.Background(), g.bucket)
  60. if err != nil {
  61. return err
  62. }
  63. targetObject := bucket.Object(key)
  64. return targetObject.Delete(context.Background())
  65. }
  66. func (g *B2Sink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error {
  67. key = cleanKey(key)
  68. if entry.IsDirectory {
  69. return nil
  70. }
  71. totalSize := filer.FileSize(entry)
  72. chunkViews := filer.ViewFromChunks(g.filerSource.LookupFileId, entry.Chunks, 0, int64(totalSize))
  73. bucket, err := g.client.Bucket(context.Background(), g.bucket)
  74. if err != nil {
  75. return err
  76. }
  77. targetObject := bucket.Object(key)
  78. writer := targetObject.NewWriter(context.Background())
  79. writeFunc := func(data []byte) error {
  80. _, writeErr := writer.Write(data)
  81. return writeErr
  82. }
  83. defer writer.Close()
  84. if err := repl_util.CopyFromChunkViews(chunkViews, g.filerSource, writeFunc); err != nil {
  85. return err
  86. }
  87. return nil
  88. }
  89. func (g *B2Sink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error) {
  90. key = cleanKey(key)
  91. return true, g.CreateEntry(key, newEntry, signatures)
  92. }
  93. func cleanKey(key string) string {
  94. if strings.HasPrefix(key, "/") {
  95. key = key[1:]
  96. }
  97. return key
  98. }