filer_replication.go 4.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138
  1. package command
  2. import (
  3. "context"
  4. "strings"
  5. "github.com/chrislusf/seaweedfs/weed/glog"
  6. "github.com/chrislusf/seaweedfs/weed/replication"
  7. "github.com/chrislusf/seaweedfs/weed/replication/sink"
  8. _ "github.com/chrislusf/seaweedfs/weed/replication/sink/azuresink"
  9. _ "github.com/chrislusf/seaweedfs/weed/replication/sink/b2sink"
  10. _ "github.com/chrislusf/seaweedfs/weed/replication/sink/filersink"
  11. _ "github.com/chrislusf/seaweedfs/weed/replication/sink/gcssink"
  12. _ "github.com/chrislusf/seaweedfs/weed/replication/sink/s3sink"
  13. "github.com/chrislusf/seaweedfs/weed/replication/sub"
  14. "github.com/chrislusf/seaweedfs/weed/util"
  15. "github.com/spf13/viper"
  16. )
  17. func init() {
  18. cmdFilerReplicate.Run = runFilerReplicate // break init cycle
  19. }
  20. var cmdFilerReplicate = &Command{
  21. UsageLine: "filer.replicate",
  22. Short: "replicate file changes to another destination",
  23. Long: `replicate file changes to another destination
  24. filer.replicate listens on filer notifications. If any file is updated, it will fetch the updated content,
  25. and write to the other destination.
  26. Run "weed scaffold -config=replication" to generate a replication.toml file and customize the parameters.
  27. `,
  28. }
  29. func runFilerReplicate(cmd *Command, args []string) bool {
  30. util.LoadConfiguration("security", false)
  31. util.LoadConfiguration("replication", true)
  32. util.LoadConfiguration("notification", true)
  33. config := util.GetViper()
  34. var notificationInput sub.NotificationInput
  35. validateOneEnabledInput(config)
  36. for _, input := range sub.NotificationInputs {
  37. if config.GetBool("notification." + input.GetName() + ".enabled") {
  38. if err := input.Initialize(config, "notification."+input.GetName()+"."); err != nil {
  39. glog.Fatalf("Failed to initialize notification input for %s: %+v",
  40. input.GetName(), err)
  41. }
  42. glog.V(0).Infof("Configure notification input to %s", input.GetName())
  43. notificationInput = input
  44. break
  45. }
  46. }
  47. if notificationInput == nil {
  48. println("No notification is defined in notification.toml file.")
  49. println("Please follow 'weed scaffold -config=notification' to see example notification configurations.")
  50. return true
  51. }
  52. // avoid recursive replication
  53. if config.GetBool("notification.source.filer.enabled") && config.GetBool("notification.sink.filer.enabled") {
  54. if config.GetString("source.filer.grpcAddress") == config.GetString("sink.filer.grpcAddress") {
  55. fromDir := config.GetString("source.filer.directory")
  56. toDir := config.GetString("sink.filer.directory")
  57. if strings.HasPrefix(toDir, fromDir) {
  58. glog.Fatalf("recursive replication! source directory %s includes the sink directory %s", fromDir, toDir)
  59. }
  60. }
  61. }
  62. var dataSink sink.ReplicationSink
  63. for _, sk := range sink.Sinks {
  64. if config.GetBool("sink." + sk.GetName() + ".enabled") {
  65. if err := sk.Initialize(config, "sink."+sk.GetName()+"."); err != nil {
  66. glog.Fatalf("Failed to initialize sink for %s: %+v",
  67. sk.GetName(), err)
  68. }
  69. glog.V(0).Infof("Configure sink to %s", sk.GetName())
  70. dataSink = sk
  71. break
  72. }
  73. }
  74. if dataSink == nil {
  75. println("no data sink configured in replication.toml:")
  76. for _, sk := range sink.Sinks {
  77. println(" " + sk.GetName())
  78. }
  79. return true
  80. }
  81. replicator := replication.NewReplicator(config, "source.filer.", dataSink)
  82. for {
  83. key, m, err := notificationInput.ReceiveMessage()
  84. if err != nil {
  85. glog.Errorf("receive %s: %+v", key, err)
  86. continue
  87. }
  88. if key == "" {
  89. // long poll received no messages
  90. continue
  91. }
  92. if m.OldEntry != nil && m.NewEntry == nil {
  93. glog.V(1).Infof("delete: %s", key)
  94. } else if m.OldEntry == nil && m.NewEntry != nil {
  95. glog.V(1).Infof(" add: %s", key)
  96. } else {
  97. glog.V(1).Infof("modify: %s", key)
  98. }
  99. if err = replicator.Replicate(context.Background(), key, m); err != nil {
  100. glog.Errorf("replicate %s: %+v", key, err)
  101. } else {
  102. glog.V(1).Infof("replicated %s", key)
  103. }
  104. }
  105. return true
  106. }
  107. func validateOneEnabledInput(config *viper.Viper) {
  108. enabledInput := ""
  109. for _, input := range sub.NotificationInputs {
  110. if config.GetBool("notification." + input.GetName() + ".enabled") {
  111. if enabledInput == "" {
  112. enabledInput = input.GetName()
  113. } else {
  114. glog.Fatalf("Notification input is enabled for both %s and %s", enabledInput, input.GetName())
  115. }
  116. }
  117. }
  118. }