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.

223 lines
6.5 KiB

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