filer_backup.go 5.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143
  1. package command
  2. import (
  3. "fmt"
  4. "github.com/seaweedfs/seaweedfs/weed/glog"
  5. "github.com/seaweedfs/seaweedfs/weed/pb"
  6. "github.com/seaweedfs/seaweedfs/weed/replication/source"
  7. "github.com/seaweedfs/seaweedfs/weed/security"
  8. "github.com/seaweedfs/seaweedfs/weed/util"
  9. "google.golang.org/grpc"
  10. "time"
  11. )
  12. type FilerBackupOptions struct {
  13. isActivePassive *bool
  14. filer *string
  15. path *string
  16. excludePaths *string
  17. debug *bool
  18. proxyByFiler *bool
  19. timeAgo *time.Duration
  20. retentionDays *int
  21. }
  22. var (
  23. filerBackupOptions FilerBackupOptions
  24. )
  25. func init() {
  26. cmdFilerBackup.Run = runFilerBackup // break init cycle
  27. filerBackupOptions.filer = cmdFilerBackup.Flag.String("filer", "localhost:8888", "filer of one SeaweedFS cluster")
  28. filerBackupOptions.path = cmdFilerBackup.Flag.String("filerPath", "/", "directory to sync on filer")
  29. filerBackupOptions.excludePaths = cmdFilerBackup.Flag.String("filerExcludePaths", "", "exclude directories to sync on filer")
  30. filerBackupOptions.proxyByFiler = cmdFilerBackup.Flag.Bool("filerProxy", false, "read and write file chunks by filer instead of volume servers")
  31. filerBackupOptions.debug = cmdFilerBackup.Flag.Bool("debug", false, "debug mode to print out received files")
  32. filerBackupOptions.timeAgo = cmdFilerBackup.Flag.Duration("timeAgo", 0, "start time before now. \"300ms\", \"1.5h\" or \"2h45m\". Valid time units are \"ns\", \"us\" (or \"µs\"), \"ms\", \"s\", \"m\", \"h\"")
  33. filerBackupOptions.retentionDays = cmdFilerBackup.Flag.Int("retentionDays", 0, "incremental backup retention days")
  34. }
  35. var cmdFilerBackup = &Command{
  36. UsageLine: "filer.backup -filer=<filerHost>:<filerPort> ",
  37. Short: "resume-able continuously replicate files from a SeaweedFS cluster to another location defined in replication.toml",
  38. Long: `resume-able continuously replicate files from a SeaweedFS cluster to another location defined in replication.toml
  39. filer.backup listens on filer notifications. If any file is updated, it will fetch the updated content,
  40. and write to the destination. This is to replace filer.replicate command since additional message queue is not needed.
  41. If restarted and "-timeAgo" is not set, the synchronization will resume from the previous checkpoints, persisted every minute.
  42. A fresh sync will start from the earliest metadata logs. To reset the checkpoints, just set "-timeAgo" to a high value.
  43. `,
  44. }
  45. func runFilerBackup(cmd *Command, args []string) bool {
  46. util.LoadConfiguration("security", false)
  47. util.LoadConfiguration("replication", true)
  48. grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
  49. clientId := util.RandomInt32()
  50. var clientEpoch int32
  51. for {
  52. clientEpoch++
  53. err := doFilerBackup(grpcDialOption, &filerBackupOptions, clientId, clientEpoch)
  54. if err != nil {
  55. glog.Errorf("backup from %s: %v", *filerBackupOptions.filer, err)
  56. time.Sleep(1747 * time.Millisecond)
  57. }
  58. }
  59. return true
  60. }
  61. const (
  62. BackupKeyPrefix = "backup."
  63. )
  64. func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOptions, clientId int32, clientEpoch int32) error {
  65. // find data sink
  66. config := util.GetViper()
  67. dataSink := findSink(config)
  68. if dataSink == nil {
  69. return fmt.Errorf("no data sink configured in replication.toml")
  70. }
  71. sourceFiler := pb.ServerAddress(*backupOption.filer)
  72. sourcePath := *backupOption.path
  73. excludePaths := util.StringSplit(*backupOption.excludePaths, ",")
  74. timeAgo := *backupOption.timeAgo
  75. targetPath := dataSink.GetSinkToDirectory()
  76. debug := *backupOption.debug
  77. // get start time for the data sink
  78. startFrom := time.Unix(0, 0)
  79. sinkId := util.HashStringToLong(dataSink.GetName() + dataSink.GetSinkToDirectory())
  80. if timeAgo.Milliseconds() == 0 {
  81. lastOffsetTsNs, err := getOffset(grpcDialOption, sourceFiler, BackupKeyPrefix, int32(sinkId))
  82. if err != nil {
  83. glog.V(0).Infof("starting from %v", startFrom)
  84. } else {
  85. startFrom = time.Unix(0, lastOffsetTsNs)
  86. glog.V(0).Infof("resuming from %v", startFrom)
  87. }
  88. } else {
  89. startFrom = time.Now().Add(-timeAgo)
  90. glog.V(0).Infof("start time is set to %v", startFrom)
  91. }
  92. // create filer sink
  93. filerSource := &source.FilerSource{}
  94. filerSource.DoInitialize(
  95. sourceFiler.ToHttpAddress(),
  96. sourceFiler.ToGrpcAddress(),
  97. sourcePath,
  98. *backupOption.proxyByFiler)
  99. dataSink.SetSourceFiler(filerSource)
  100. processEventFn := genProcessFunction(sourcePath, targetPath, excludePaths, dataSink, debug)
  101. processEventFnWithOffset := pb.AddOffsetFunc(processEventFn, 3*time.Second, func(counter int64, lastTsNs int64) error {
  102. glog.V(0).Infof("backup %s progressed to %v %0.2f/sec", sourceFiler, time.Unix(0, lastTsNs), float64(counter)/float64(3))
  103. return setOffset(grpcDialOption, sourceFiler, BackupKeyPrefix, int32(sinkId), lastTsNs)
  104. })
  105. if dataSink.IsIncremental() && *filerBackupOptions.retentionDays > 0 {
  106. go func() {
  107. for {
  108. now := time.Now()
  109. time.Sleep(time.Hour * 24)
  110. key := util.Join(targetPath, now.Add(-1*time.Hour*24*time.Duration(*filerBackupOptions.retentionDays)).Format("2006-01-02"))
  111. _ = dataSink.DeleteEntry(util.Join(targetPath, key), true, true, nil)
  112. glog.V(0).Infof("incremental backup delete directory:%s", key)
  113. }
  114. }()
  115. }
  116. return pb.FollowMetadata(sourceFiler, grpcDialOption, "backup_"+dataSink.GetName(), clientId, clientEpoch, sourcePath, nil, startFrom.UnixNano(), 0, 0, processEventFnWithOffset, pb.TrivialOnError)
  117. }