filer_sync.go 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337
  1. package command
  2. import (
  3. "context"
  4. "errors"
  5. "fmt"
  6. "github.com/chrislusf/seaweedfs/weed/util/log"
  7. "github.com/chrislusf/seaweedfs/weed/pb"
  8. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  9. "github.com/chrislusf/seaweedfs/weed/replication"
  10. "github.com/chrislusf/seaweedfs/weed/replication/sink/filersink"
  11. "github.com/chrislusf/seaweedfs/weed/replication/source"
  12. "github.com/chrislusf/seaweedfs/weed/security"
  13. "github.com/chrislusf/seaweedfs/weed/util"
  14. "github.com/chrislusf/seaweedfs/weed/util/grace"
  15. "google.golang.org/grpc"
  16. "io"
  17. "strings"
  18. "time"
  19. )
  20. type SyncOptions struct {
  21. isActivePassive *bool
  22. filerA *string
  23. filerB *string
  24. aPath *string
  25. bPath *string
  26. aReplication *string
  27. bReplication *string
  28. aCollection *string
  29. bCollection *string
  30. aTtlSec *int
  31. bTtlSec *int
  32. aDebug *bool
  33. bDebug *bool
  34. }
  35. var (
  36. syncOptions SyncOptions
  37. syncCpuProfile *string
  38. syncMemProfile *string
  39. )
  40. func init() {
  41. cmdFilerSynchronize.Run = runFilerSynchronize // break init cycle
  42. syncOptions.isActivePassive = cmdFilerSynchronize.Flag.Bool("isActivePassive", false, "one directional follow if true")
  43. syncOptions.filerA = cmdFilerSynchronize.Flag.String("a", "", "filer A in one SeaweedFS cluster")
  44. syncOptions.filerB = cmdFilerSynchronize.Flag.String("b", "", "filer B in the other SeaweedFS cluster")
  45. syncOptions.aPath = cmdFilerSynchronize.Flag.String("a.path", "/", "directory to sync on filer A")
  46. syncOptions.bPath = cmdFilerSynchronize.Flag.String("b.path", "/", "directory to sync on filer B")
  47. syncOptions.aReplication = cmdFilerSynchronize.Flag.String("a.replication", "", "replication on filer A")
  48. syncOptions.bReplication = cmdFilerSynchronize.Flag.String("b.replication", "", "replication on filer B")
  49. syncOptions.aCollection = cmdFilerSynchronize.Flag.String("a.collection", "", "collection on filer A")
  50. syncOptions.bCollection = cmdFilerSynchronize.Flag.String("b.collection", "", "collection on filer B")
  51. syncOptions.aTtlSec = cmdFilerSynchronize.Flag.Int("a.ttlSec", 0, "ttl in seconds on filer A")
  52. syncOptions.bTtlSec = cmdFilerSynchronize.Flag.Int("b.ttlSec", 0, "ttl in seconds on filer B")
  53. syncOptions.aDebug = cmdFilerSynchronize.Flag.Bool("a.debug", false, "debug mode to print out filer A received files")
  54. syncOptions.bDebug = cmdFilerSynchronize.Flag.Bool("b.debug", false, "debug mode to print out filer B received files")
  55. syncCpuProfile = cmdFilerSynchronize.Flag.String("cpuprofile", "", "cpu profile output file")
  56. syncMemProfile = cmdFilerSynchronize.Flag.String("memprofile", "", "memory profile output file")
  57. }
  58. var cmdFilerSynchronize = &Command{
  59. UsageLine: "filer.sync -a=<oneFilerHost>:<oneFilerPort> -b=<otherFilerHost>:<otherFilerPort>",
  60. Short: "continuously synchronize between two active-active or active-passive SeaweedFS clusters",
  61. Long: `continuously synchronize file changes between two active-active or active-passive filers
  62. filer.sync listens on filer notifications. If any file is updated, it will fetch the updated content,
  63. and write to the other destination. Different from filer.replicate:
  64. * filer.sync only works between two filers.
  65. * filer.sync does not need any special message queue setup.
  66. * filer.sync supports both active-active and active-passive modes.
  67. If restarted, the synchronization will resume from the previous checkpoints, persisted every minute.
  68. A fresh sync will start from the earliest metadata logs.
  69. `,
  70. }
  71. func runFilerSynchronize(cmd *Command, args []string) bool {
  72. grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
  73. grace.SetupProfiling(*syncCpuProfile, *syncMemProfile)
  74. go func() {
  75. for {
  76. err := doSubscribeFilerMetaChanges(grpcDialOption, *syncOptions.filerA, *syncOptions.aPath, *syncOptions.filerB,
  77. *syncOptions.bPath, *syncOptions.bReplication, *syncOptions.bCollection, *syncOptions.bTtlSec, *syncOptions.bDebug)
  78. if err != nil {
  79. log.Errorf("sync from %s to %s: %v", *syncOptions.filerA, *syncOptions.filerB, err)
  80. time.Sleep(1747 * time.Millisecond)
  81. }
  82. }
  83. }()
  84. if !*syncOptions.isActivePassive {
  85. go func() {
  86. for {
  87. err := doSubscribeFilerMetaChanges(grpcDialOption, *syncOptions.filerB, *syncOptions.bPath, *syncOptions.filerA,
  88. *syncOptions.aPath, *syncOptions.aReplication, *syncOptions.aCollection, *syncOptions.aTtlSec, *syncOptions.aDebug)
  89. if err != nil {
  90. log.Errorf("sync from %s to %s: %v", *syncOptions.filerB, *syncOptions.filerA, err)
  91. time.Sleep(2147 * time.Millisecond)
  92. }
  93. }
  94. }()
  95. }
  96. select {}
  97. return true
  98. }
  99. func doSubscribeFilerMetaChanges(grpcDialOption grpc.DialOption, sourceFiler, sourcePath, targetFiler, targetPath string,
  100. replicationStr, collection string, ttlSec int, debug bool) error {
  101. // read source filer signature
  102. sourceFilerSignature, sourceErr := replication.ReadFilerSignature(grpcDialOption, sourceFiler)
  103. if sourceErr != nil {
  104. return sourceErr
  105. }
  106. // read target filer signature
  107. targetFilerSignature, targetErr := replication.ReadFilerSignature(grpcDialOption, targetFiler)
  108. if targetErr != nil {
  109. return targetErr
  110. }
  111. // if first time, start from now
  112. // if has previously synced, resume from that point of time
  113. sourceFilerOffsetTsNs, err := readSyncOffset(grpcDialOption, targetFiler, sourceFilerSignature)
  114. if err != nil {
  115. return err
  116. }
  117. log.Infof("start sync %s(%d) => %s(%d) from %v(%d)", sourceFiler, sourceFilerSignature, targetFiler, targetFilerSignature, time.Unix(0, sourceFilerOffsetTsNs), sourceFilerOffsetTsNs)
  118. // create filer sink
  119. filerSource := &source.FilerSource{}
  120. filerSource.DoInitialize(pb.ServerToGrpcAddress(sourceFiler), sourcePath)
  121. filerSink := &filersink.FilerSink{}
  122. filerSink.DoInitialize(pb.ServerToGrpcAddress(targetFiler), targetPath, replicationStr, collection, ttlSec, grpcDialOption)
  123. filerSink.SetSourceFiler(filerSource)
  124. processEventFn := func(resp *filer_pb.SubscribeMetadataResponse) error {
  125. message := resp.EventNotification
  126. var sourceOldKey, sourceNewKey util.FullPath
  127. if message.OldEntry != nil {
  128. sourceOldKey = util.FullPath(resp.Directory).Child(message.OldEntry.Name)
  129. }
  130. if message.NewEntry != nil {
  131. sourceNewKey = util.FullPath(message.NewParentPath).Child(message.NewEntry.Name)
  132. }
  133. for _, sig := range message.Signatures {
  134. if sig == targetFilerSignature && targetFilerSignature != 0 {
  135. fmt.Printf("%s skipping %s change to %v\n", targetFiler, sourceFiler, message)
  136. return nil
  137. }
  138. }
  139. if debug {
  140. fmt.Printf("%s check %s change %s,%s sig %v, target sig: %v\n", targetFiler, sourceFiler, sourceOldKey, sourceNewKey, message.Signatures, targetFilerSignature)
  141. }
  142. if !strings.HasPrefix(resp.Directory, sourcePath) {
  143. return nil
  144. }
  145. // handle deletions
  146. if message.OldEntry != nil && message.NewEntry == nil {
  147. if !strings.HasPrefix(string(sourceOldKey), sourcePath) {
  148. return nil
  149. }
  150. key := util.Join(targetPath, string(sourceOldKey)[len(sourcePath):])
  151. return filerSink.DeleteEntry(key, message.OldEntry.IsDirectory, message.DeleteChunks, message.Signatures)
  152. }
  153. // handle new entries
  154. if message.OldEntry == nil && message.NewEntry != nil {
  155. if !strings.HasPrefix(string(sourceNewKey), sourcePath) {
  156. return nil
  157. }
  158. key := util.Join(targetPath, string(sourceNewKey)[len(sourcePath):])
  159. return filerSink.CreateEntry(key, message.NewEntry, message.Signatures)
  160. }
  161. // this is something special?
  162. if message.OldEntry == nil && message.NewEntry == nil {
  163. return nil
  164. }
  165. // handle updates
  166. if strings.HasPrefix(string(sourceOldKey), sourcePath) {
  167. // old key is in the watched directory
  168. if strings.HasPrefix(string(sourceNewKey), sourcePath) {
  169. // new key is also in the watched directory
  170. oldKey := util.Join(targetPath, string(sourceOldKey)[len(sourcePath):])
  171. message.NewParentPath = util.Join(targetPath, message.NewParentPath[len(sourcePath):])
  172. foundExisting, err := filerSink.UpdateEntry(string(oldKey), message.OldEntry, message.NewParentPath, message.NewEntry, message.DeleteChunks, message.Signatures)
  173. if foundExisting {
  174. return err
  175. }
  176. // not able to find old entry
  177. if err = filerSink.DeleteEntry(string(oldKey), message.OldEntry.IsDirectory, false, message.Signatures); err != nil {
  178. return fmt.Errorf("delete old entry %v: %v", oldKey, err)
  179. }
  180. // create the new entry
  181. newKey := util.Join(targetPath, string(sourceNewKey)[len(sourcePath):])
  182. return filerSink.CreateEntry(newKey, message.NewEntry, message.Signatures)
  183. } else {
  184. // new key is outside of the watched directory
  185. key := util.Join(targetPath, string(sourceOldKey)[len(sourcePath):])
  186. return filerSink.DeleteEntry(key, message.OldEntry.IsDirectory, message.DeleteChunks, message.Signatures)
  187. }
  188. } else {
  189. // old key is outside of the watched directory
  190. if strings.HasPrefix(string(sourceNewKey), sourcePath) {
  191. // new key is in the watched directory
  192. key := util.Join(targetPath, string(sourceNewKey)[len(sourcePath):])
  193. return filerSink.CreateEntry(key, message.NewEntry, message.Signatures)
  194. } else {
  195. // new key is also outside of the watched directory
  196. // skip
  197. }
  198. }
  199. return nil
  200. }
  201. return pb.WithFilerClient(sourceFiler, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  202. ctx, cancel := context.WithCancel(context.Background())
  203. defer cancel()
  204. stream, err := client.SubscribeMetadata(ctx, &filer_pb.SubscribeMetadataRequest{
  205. ClientName: "syncTo_" + targetFiler,
  206. PathPrefix: sourcePath,
  207. SinceNs: sourceFilerOffsetTsNs,
  208. Signature: targetFilerSignature,
  209. })
  210. if err != nil {
  211. return fmt.Errorf("listen: %v", err)
  212. }
  213. var counter int64
  214. var lastWriteTime time.Time
  215. for {
  216. resp, listenErr := stream.Recv()
  217. if listenErr == io.EOF {
  218. return nil
  219. }
  220. if listenErr != nil {
  221. return listenErr
  222. }
  223. if err := processEventFn(resp); err != nil {
  224. return err
  225. }
  226. counter++
  227. if lastWriteTime.Add(3 * time.Second).Before(time.Now()) {
  228. log.Infof("sync %s => %s progressed to %v %0.2f/sec", sourceFiler, targetFiler, time.Unix(0, resp.TsNs), float64(counter)/float64(3))
  229. counter = 0
  230. lastWriteTime = time.Now()
  231. if err := writeSyncOffset(grpcDialOption, targetFiler, sourceFilerSignature, resp.TsNs); err != nil {
  232. return err
  233. }
  234. }
  235. }
  236. })
  237. }
  238. const (
  239. SyncKeyPrefix = "sync."
  240. )
  241. func readSyncOffset(grpcDialOption grpc.DialOption, filer string, filerSignature int32) (lastOffsetTsNs int64, readErr error) {
  242. readErr = pb.WithFilerClient(filer, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  243. syncKey := []byte(SyncKeyPrefix + "____")
  244. util.Uint32toBytes(syncKey[len(SyncKeyPrefix):len(SyncKeyPrefix)+4], uint32(filerSignature))
  245. resp, err := client.KvGet(context.Background(), &filer_pb.KvGetRequest{Key: syncKey})
  246. if err != nil {
  247. return err
  248. }
  249. if len(resp.Error) != 0 {
  250. return errors.New(resp.Error)
  251. }
  252. if len(resp.Value) < 8 {
  253. return nil
  254. }
  255. lastOffsetTsNs = int64(util.BytesToUint64(resp.Value))
  256. return nil
  257. })
  258. return
  259. }
  260. func writeSyncOffset(grpcDialOption grpc.DialOption, filer string, filerSignature int32, offsetTsNs int64) error {
  261. return pb.WithFilerClient(filer, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  262. syncKey := []byte(SyncKeyPrefix + "____")
  263. util.Uint32toBytes(syncKey[len(SyncKeyPrefix):len(SyncKeyPrefix)+4], uint32(filerSignature))
  264. valueBuf := make([]byte, 8)
  265. util.Uint64toBytes(valueBuf, uint64(offsetTsNs))
  266. resp, err := client.KvPut(context.Background(), &filer_pb.KvPutRequest{
  267. Key: syncKey,
  268. Value: valueBuf,
  269. })
  270. if err != nil {
  271. return err
  272. }
  273. if len(resp.Error) != 0 {
  274. return errors.New(resp.Error)
  275. }
  276. return nil
  277. })
  278. }