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.

255 lines
7.8 KiB

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