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.

213 lines
6.0 KiB

6 years ago
6 years ago
6 years ago
6 years ago
6 years ago
  1. package filersink
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/joeslay/seaweedfs/weed/security"
  6. "github.com/spf13/viper"
  7. "google.golang.org/grpc"
  8. "github.com/joeslay/seaweedfs/weed/filer2"
  9. "github.com/joeslay/seaweedfs/weed/glog"
  10. "github.com/joeslay/seaweedfs/weed/pb/filer_pb"
  11. "github.com/joeslay/seaweedfs/weed/replication/sink"
  12. "github.com/joeslay/seaweedfs/weed/replication/source"
  13. "github.com/joeslay/seaweedfs/weed/util"
  14. )
  15. type FilerSink struct {
  16. filerSource *source.FilerSource
  17. grpcAddress string
  18. dir string
  19. replication string
  20. collection string
  21. ttlSec int32
  22. dataCenter string
  23. grpcDialOption grpc.DialOption
  24. }
  25. func init() {
  26. sink.Sinks = append(sink.Sinks, &FilerSink{})
  27. }
  28. func (fs *FilerSink) GetName() string {
  29. return "filer"
  30. }
  31. func (fs *FilerSink) GetSinkToDirectory() string {
  32. return fs.dir
  33. }
  34. func (fs *FilerSink) Initialize(configuration util.Configuration) error {
  35. return fs.initialize(
  36. configuration.GetString("grpcAddress"),
  37. configuration.GetString("directory"),
  38. configuration.GetString("replication"),
  39. configuration.GetString("collection"),
  40. configuration.GetInt("ttlSec"),
  41. )
  42. }
  43. func (fs *FilerSink) SetSourceFiler(s *source.FilerSource) {
  44. fs.filerSource = s
  45. }
  46. func (fs *FilerSink) initialize(grpcAddress string, dir string,
  47. replication string, collection string, ttlSec int) (err error) {
  48. fs.grpcAddress = grpcAddress
  49. fs.dir = dir
  50. fs.replication = replication
  51. fs.collection = collection
  52. fs.ttlSec = int32(ttlSec)
  53. fs.grpcDialOption = security.LoadClientTLS(viper.Sub("grpc"), "client")
  54. return nil
  55. }
  56. func (fs *FilerSink) DeleteEntry(ctx context.Context, key string, isDirectory, deleteIncludeChunks bool) error {
  57. return fs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
  58. dir, name := filer2.FullPath(key).DirAndName()
  59. request := &filer_pb.DeleteEntryRequest{
  60. Directory: dir,
  61. Name: name,
  62. IsDeleteData: deleteIncludeChunks,
  63. }
  64. glog.V(1).Infof("delete entry: %v", request)
  65. _, err := client.DeleteEntry(ctx, request)
  66. if err != nil {
  67. glog.V(0).Infof("delete entry %s: %v", key, err)
  68. return fmt.Errorf("delete entry %s: %v", key, err)
  69. }
  70. return nil
  71. })
  72. }
  73. func (fs *FilerSink) CreateEntry(ctx context.Context, key string, entry *filer_pb.Entry) error {
  74. return fs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
  75. dir, name := filer2.FullPath(key).DirAndName()
  76. // look up existing entry
  77. lookupRequest := &filer_pb.LookupDirectoryEntryRequest{
  78. Directory: dir,
  79. Name: name,
  80. }
  81. glog.V(1).Infof("lookup: %v", lookupRequest)
  82. if resp, err := client.LookupDirectoryEntry(ctx, lookupRequest); err == nil {
  83. if filer2.ETag(resp.Entry.Chunks) == filer2.ETag(entry.Chunks) {
  84. glog.V(0).Infof("already replicated %s", key)
  85. return nil
  86. }
  87. }
  88. replicatedChunks, err := fs.replicateChunks(ctx, entry.Chunks)
  89. if err != nil {
  90. glog.V(0).Infof("replicate entry chunks %s: %v", key, err)
  91. return fmt.Errorf("replicate entry chunks %s: %v", key, err)
  92. }
  93. glog.V(0).Infof("replicated %s %+v ===> %+v", key, entry.Chunks, replicatedChunks)
  94. request := &filer_pb.CreateEntryRequest{
  95. Directory: dir,
  96. Entry: &filer_pb.Entry{
  97. Name: name,
  98. IsDirectory: entry.IsDirectory,
  99. Attributes: entry.Attributes,
  100. Chunks: replicatedChunks,
  101. },
  102. }
  103. glog.V(1).Infof("create: %v", request)
  104. if _, err := client.CreateEntry(ctx, request); err != nil {
  105. glog.V(0).Infof("create entry %s: %v", key, err)
  106. return fmt.Errorf("create entry %s: %v", key, err)
  107. }
  108. return nil
  109. })
  110. }
  111. func (fs *FilerSink) UpdateEntry(ctx context.Context, key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool) (foundExistingEntry bool, err error) {
  112. dir, name := filer2.FullPath(key).DirAndName()
  113. // read existing entry
  114. var existingEntry *filer_pb.Entry
  115. err = fs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
  116. request := &filer_pb.LookupDirectoryEntryRequest{
  117. Directory: dir,
  118. Name: name,
  119. }
  120. glog.V(4).Infof("lookup entry: %v", request)
  121. resp, err := client.LookupDirectoryEntry(ctx, request)
  122. if err != nil {
  123. glog.V(0).Infof("lookup %s: %v", key, err)
  124. return err
  125. }
  126. existingEntry = resp.Entry
  127. return nil
  128. })
  129. if err != nil {
  130. return false, fmt.Errorf("lookup %s: %v", key, err)
  131. }
  132. glog.V(0).Infof("oldEntry %+v, newEntry %+v, existingEntry: %+v", oldEntry, newEntry, existingEntry)
  133. if existingEntry.Attributes.Mtime > newEntry.Attributes.Mtime {
  134. // skip if already changed
  135. // this usually happens when the messages are not ordered
  136. glog.V(0).Infof("late updates %s", key)
  137. } else if filer2.ETag(newEntry.Chunks) == filer2.ETag(existingEntry.Chunks) {
  138. // skip if no change
  139. // this usually happens when retrying the replication
  140. glog.V(0).Infof("already replicated %s", key)
  141. } else {
  142. // find out what changed
  143. deletedChunks, newChunks := compareChunks(oldEntry, newEntry)
  144. // delete the chunks that are deleted from the source
  145. if deleteIncludeChunks {
  146. // remove the deleted chunks. Actual data deletion happens in filer UpdateEntry FindUnusedFileChunks
  147. existingEntry.Chunks = filer2.MinusChunks(existingEntry.Chunks, deletedChunks)
  148. }
  149. // replicate the chunks that are new in the source
  150. replicatedChunks, err := fs.replicateChunks(ctx, newChunks)
  151. if err != nil {
  152. return true, fmt.Errorf("replicte %s chunks error: %v", key, err)
  153. }
  154. existingEntry.Chunks = append(existingEntry.Chunks, replicatedChunks...)
  155. }
  156. // save updated meta data
  157. return true, fs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
  158. request := &filer_pb.UpdateEntryRequest{
  159. Directory: newParentPath,
  160. Entry: existingEntry,
  161. }
  162. if _, err := client.UpdateEntry(ctx, request); err != nil {
  163. return fmt.Errorf("update existingEntry %s: %v", key, err)
  164. }
  165. return nil
  166. })
  167. }
  168. func compareChunks(oldEntry, newEntry *filer_pb.Entry) (deletedChunks, newChunks []*filer_pb.FileChunk) {
  169. deletedChunks = filer2.MinusChunks(oldEntry.Chunks, newEntry.Chunks)
  170. newChunks = filer2.MinusChunks(newEntry.Chunks, oldEntry.Chunks)
  171. return
  172. }