filer_sync.go 14 KB

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