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.

483 lines
18 KiB

4 years ago
4 years ago
2 years ago
4 years ago
4 years ago
4 years ago
4 years ago
3 years ago
3 years ago
3 years ago
3 years ago
  1. package command
  2. import (
  3. "context"
  4. "errors"
  5. "fmt"
  6. "github.com/seaweedfs/seaweedfs/weed/glog"
  7. "github.com/seaweedfs/seaweedfs/weed/pb"
  8. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  9. "github.com/seaweedfs/seaweedfs/weed/replication"
  10. "github.com/seaweedfs/seaweedfs/weed/replication/sink"
  11. "github.com/seaweedfs/seaweedfs/weed/replication/sink/filersink"
  12. "github.com/seaweedfs/seaweedfs/weed/replication/source"
  13. "github.com/seaweedfs/seaweedfs/weed/security"
  14. statsCollect "github.com/seaweedfs/seaweedfs/weed/stats"
  15. "github.com/seaweedfs/seaweedfs/weed/util"
  16. "github.com/seaweedfs/seaweedfs/weed/util/grace"
  17. "google.golang.org/grpc"
  18. "os"
  19. "strings"
  20. "time"
  21. )
  22. type SyncOptions struct {
  23. isActivePassive *bool
  24. filerA *string
  25. filerB *string
  26. aPath *string
  27. aExcludePaths *string
  28. bPath *string
  29. bExcludePaths *string
  30. aReplication *string
  31. bReplication *string
  32. aCollection *string
  33. bCollection *string
  34. aTtlSec *int
  35. bTtlSec *int
  36. aDiskType *string
  37. bDiskType *string
  38. aDebug *bool
  39. bDebug *bool
  40. aFromTsMs *int64
  41. bFromTsMs *int64
  42. aProxyByFiler *bool
  43. bProxyByFiler *bool
  44. metricsHttpIp *string
  45. metricsHttpPort *int
  46. concurrency *int
  47. clientId int32
  48. clientEpoch int32
  49. }
  50. const (
  51. SyncKeyPrefix = "sync."
  52. DefaultConcurrencyLimit = 32
  53. )
  54. var (
  55. syncOptions SyncOptions
  56. syncCpuProfile *string
  57. syncMemProfile *string
  58. )
  59. func init() {
  60. cmdFilerSynchronize.Run = runFilerSynchronize // break init cycle
  61. syncOptions.isActivePassive = cmdFilerSynchronize.Flag.Bool("isActivePassive", false, "one directional follow from A to B if true")
  62. syncOptions.filerA = cmdFilerSynchronize.Flag.String("a", "", "filer A in one SeaweedFS cluster")
  63. syncOptions.filerB = cmdFilerSynchronize.Flag.String("b", "", "filer B in the other SeaweedFS cluster")
  64. syncOptions.aPath = cmdFilerSynchronize.Flag.String("a.path", "/", "directory to sync on filer A")
  65. syncOptions.aExcludePaths = cmdFilerSynchronize.Flag.String("a.excludePaths", "", "exclude directories to sync on filer A")
  66. syncOptions.bPath = cmdFilerSynchronize.Flag.String("b.path", "/", "directory to sync on filer B")
  67. syncOptions.bExcludePaths = cmdFilerSynchronize.Flag.String("b.excludePaths", "", "exclude directories to sync on filer B")
  68. syncOptions.aReplication = cmdFilerSynchronize.Flag.String("a.replication", "", "replication on filer A")
  69. syncOptions.bReplication = cmdFilerSynchronize.Flag.String("b.replication", "", "replication on filer B")
  70. syncOptions.aCollection = cmdFilerSynchronize.Flag.String("a.collection", "", "collection on filer A")
  71. syncOptions.bCollection = cmdFilerSynchronize.Flag.String("b.collection", "", "collection on filer B")
  72. syncOptions.aTtlSec = cmdFilerSynchronize.Flag.Int("a.ttlSec", 0, "ttl in seconds on filer A")
  73. syncOptions.bTtlSec = cmdFilerSynchronize.Flag.Int("b.ttlSec", 0, "ttl in seconds on filer B")
  74. syncOptions.aDiskType = cmdFilerSynchronize.Flag.String("a.disk", "", "[hdd|ssd|<tag>] hard drive or solid state drive or any tag on filer A")
  75. syncOptions.bDiskType = cmdFilerSynchronize.Flag.String("b.disk", "", "[hdd|ssd|<tag>] hard drive or solid state drive or any tag on filer B")
  76. syncOptions.aProxyByFiler = cmdFilerSynchronize.Flag.Bool("a.filerProxy", false, "read and write file chunks by filer A instead of volume servers")
  77. syncOptions.bProxyByFiler = cmdFilerSynchronize.Flag.Bool("b.filerProxy", false, "read and write file chunks by filer B instead of volume servers")
  78. syncOptions.aDebug = cmdFilerSynchronize.Flag.Bool("a.debug", false, "debug mode to print out filer A received files")
  79. syncOptions.bDebug = cmdFilerSynchronize.Flag.Bool("b.debug", false, "debug mode to print out filer B received files")
  80. syncOptions.aFromTsMs = cmdFilerSynchronize.Flag.Int64("a.fromTsMs", 0, "synchronization from timestamp on filer A. The unit is millisecond")
  81. syncOptions.bFromTsMs = cmdFilerSynchronize.Flag.Int64("b.fromTsMs", 0, "synchronization from timestamp on filer B. The unit is millisecond")
  82. syncOptions.concurrency = cmdFilerSynchronize.Flag.Int("concurrency", DefaultConcurrencyLimit, "The maximum number of files that will be synced concurrently.")
  83. syncCpuProfile = cmdFilerSynchronize.Flag.String("cpuprofile", "", "cpu profile output file")
  84. syncMemProfile = cmdFilerSynchronize.Flag.String("memprofile", "", "memory profile output file")
  85. syncOptions.metricsHttpIp = cmdFilerSynchronize.Flag.String("metricsIp", "", "metrics listen ip")
  86. syncOptions.metricsHttpPort = cmdFilerSynchronize.Flag.Int("metricsPort", 0, "metrics listen port")
  87. syncOptions.clientId = util.RandomInt32()
  88. }
  89. var cmdFilerSynchronize = &Command{
  90. UsageLine: "filer.sync -a=<oneFilerHost>:<oneFilerPort> -b=<otherFilerHost>:<otherFilerPort>",
  91. Short: "resumable continuous synchronization between two active-active or active-passive SeaweedFS clusters",
  92. Long: `resumable continuous synchronization for file changes between two active-active or active-passive filers
  93. filer.sync listens on filer notifications. If any file is updated, it will fetch the updated content,
  94. and write to the other destination. Different from filer.replicate:
  95. * filer.sync only works between two filers.
  96. * filer.sync does not need any special message queue setup.
  97. * filer.sync supports both active-active and active-passive modes.
  98. If restarted, the synchronization will resume from the previous checkpoints, persisted every minute.
  99. A fresh sync will start from the earliest metadata logs.
  100. `,
  101. }
  102. func runFilerSynchronize(cmd *Command, args []string) bool {
  103. util.LoadConfiguration("security", false)
  104. grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
  105. grace.SetupProfiling(*syncCpuProfile, *syncMemProfile)
  106. filerA := pb.ServerAddress(*syncOptions.filerA)
  107. filerB := pb.ServerAddress(*syncOptions.filerB)
  108. // start filer.sync metrics server
  109. go statsCollect.StartMetricsServer(*syncOptions.metricsHttpIp, *syncOptions.metricsHttpPort)
  110. // read a filer signature
  111. aFilerSignature, aFilerErr := replication.ReadFilerSignature(grpcDialOption, filerA)
  112. if aFilerErr != nil {
  113. glog.Errorf("get filer 'a' signature %d error from %s to %s: %v", aFilerSignature, *syncOptions.filerA, *syncOptions.filerB, aFilerErr)
  114. return true
  115. }
  116. // read b filer signature
  117. bFilerSignature, bFilerErr := replication.ReadFilerSignature(grpcDialOption, filerB)
  118. if bFilerErr != nil {
  119. glog.Errorf("get filer 'b' signature %d error from %s to %s: %v", bFilerSignature, *syncOptions.filerA, *syncOptions.filerB, bFilerErr)
  120. return true
  121. }
  122. go func() {
  123. // a->b
  124. // set synchronization start timestamp to offset
  125. initOffsetError := initOffsetFromTsMs(grpcDialOption, filerB, aFilerSignature, *syncOptions.bFromTsMs, getSignaturePrefixByPath(*syncOptions.aPath))
  126. if initOffsetError != nil {
  127. glog.Errorf("init offset from timestamp %d error from %s to %s: %v", *syncOptions.bFromTsMs, *syncOptions.filerA, *syncOptions.filerB, initOffsetError)
  128. os.Exit(2)
  129. }
  130. for {
  131. syncOptions.clientEpoch++
  132. err := doSubscribeFilerMetaChanges(
  133. syncOptions.clientId,
  134. syncOptions.clientEpoch,
  135. grpcDialOption,
  136. filerA,
  137. *syncOptions.aPath,
  138. util.StringSplit(*syncOptions.aExcludePaths, ","),
  139. *syncOptions.aProxyByFiler,
  140. filerB,
  141. *syncOptions.bPath,
  142. *syncOptions.bReplication,
  143. *syncOptions.bCollection,
  144. *syncOptions.bTtlSec,
  145. *syncOptions.bProxyByFiler,
  146. *syncOptions.bDiskType,
  147. *syncOptions.bDebug,
  148. *syncOptions.concurrency,
  149. aFilerSignature,
  150. bFilerSignature)
  151. if err != nil {
  152. glog.Errorf("sync from %s to %s: %v", *syncOptions.filerA, *syncOptions.filerB, err)
  153. time.Sleep(1747 * time.Millisecond)
  154. }
  155. }
  156. }()
  157. if !*syncOptions.isActivePassive {
  158. // b->a
  159. // set synchronization start timestamp to offset
  160. initOffsetError := initOffsetFromTsMs(grpcDialOption, filerA, bFilerSignature, *syncOptions.aFromTsMs, getSignaturePrefixByPath(*syncOptions.bPath))
  161. if initOffsetError != nil {
  162. glog.Errorf("init offset from timestamp %d error from %s to %s: %v", *syncOptions.aFromTsMs, *syncOptions.filerB, *syncOptions.filerA, initOffsetError)
  163. os.Exit(2)
  164. }
  165. go func() {
  166. for {
  167. syncOptions.clientEpoch++
  168. err := doSubscribeFilerMetaChanges(
  169. syncOptions.clientId,
  170. syncOptions.clientEpoch,
  171. grpcDialOption,
  172. filerB,
  173. *syncOptions.bPath,
  174. util.StringSplit(*syncOptions.bExcludePaths, ","),
  175. *syncOptions.bProxyByFiler,
  176. filerA,
  177. *syncOptions.aPath,
  178. *syncOptions.aReplication,
  179. *syncOptions.aCollection,
  180. *syncOptions.aTtlSec,
  181. *syncOptions.aProxyByFiler,
  182. *syncOptions.aDiskType,
  183. *syncOptions.aDebug,
  184. *syncOptions.concurrency,
  185. bFilerSignature,
  186. aFilerSignature)
  187. if err != nil {
  188. glog.Errorf("sync from %s to %s: %v", *syncOptions.filerB, *syncOptions.filerA, err)
  189. time.Sleep(2147 * time.Millisecond)
  190. }
  191. }
  192. }()
  193. }
  194. select {}
  195. return true
  196. }
  197. // initOffsetFromTsMs Initialize offset
  198. func initOffsetFromTsMs(grpcDialOption grpc.DialOption, targetFiler pb.ServerAddress, sourceFilerSignature int32, fromTsMs int64, signaturePrefix string) error {
  199. if fromTsMs <= 0 {
  200. return nil
  201. }
  202. // convert to nanosecond
  203. fromTsNs := fromTsMs * 1000_000
  204. // If not successful, exit the program.
  205. setOffsetErr := setOffset(grpcDialOption, targetFiler, signaturePrefix, sourceFilerSignature, fromTsNs)
  206. if setOffsetErr != nil {
  207. return setOffsetErr
  208. }
  209. glog.Infof("setOffset from timestamp ms success! start offset: %d from %s to %s", fromTsNs, *syncOptions.filerA, *syncOptions.filerB)
  210. return nil
  211. }
  212. func doSubscribeFilerMetaChanges(clientId int32, clientEpoch int32, grpcDialOption grpc.DialOption, sourceFiler pb.ServerAddress, sourcePath string, sourceExcludePaths []string, sourceReadChunkFromFiler bool, targetFiler pb.ServerAddress, targetPath string,
  213. replicationStr, collection string, ttlSec int, sinkWriteChunkByFiler bool, diskType string, debug bool, concurrency int, sourceFilerSignature int32, targetFilerSignature int32) error {
  214. // if first time, start from now
  215. // if has previously synced, resume from that point of time
  216. sourceFilerOffsetTsNs, err := getOffset(grpcDialOption, targetFiler, getSignaturePrefixByPath(sourcePath), sourceFilerSignature)
  217. if err != nil {
  218. return err
  219. }
  220. glog.V(0).Infof("start sync %s(%d) => %s(%d) from %v(%d)", sourceFiler, sourceFilerSignature, targetFiler, targetFilerSignature, time.Unix(0, sourceFilerOffsetTsNs), sourceFilerOffsetTsNs)
  221. // create filer sink
  222. filerSource := &source.FilerSource{}
  223. filerSource.DoInitialize(sourceFiler.ToHttpAddress(), sourceFiler.ToGrpcAddress(), sourcePath, sourceReadChunkFromFiler)
  224. filerSink := &filersink.FilerSink{}
  225. filerSink.DoInitialize(targetFiler.ToHttpAddress(), targetFiler.ToGrpcAddress(), targetPath, replicationStr, collection, ttlSec, diskType, grpcDialOption, sinkWriteChunkByFiler)
  226. filerSink.SetSourceFiler(filerSource)
  227. persistEventFn := genProcessFunction(sourcePath, targetPath, sourceExcludePaths, filerSink, debug)
  228. processEventFn := func(resp *filer_pb.SubscribeMetadataResponse) error {
  229. message := resp.EventNotification
  230. for _, sig := range message.Signatures {
  231. if sig == targetFilerSignature && targetFilerSignature != 0 {
  232. fmt.Printf("%s skipping %s change to %v\n", targetFiler, sourceFiler, message)
  233. return nil
  234. }
  235. }
  236. return persistEventFn(resp)
  237. }
  238. if concurrency < 0 || concurrency > 1024 {
  239. glog.Warningf("invalid concurrency value, using default: %d", DefaultConcurrencyLimit)
  240. concurrency = DefaultConcurrencyLimit
  241. }
  242. processor := NewMetadataProcessor(processEventFn, concurrency)
  243. var lastLogTsNs = time.Now().UnixNano()
  244. var clientName = fmt.Sprintf("syncFrom_%s_To_%s", string(sourceFiler), string(targetFiler))
  245. processEventFnWithOffset := pb.AddOffsetFunc(func(resp *filer_pb.SubscribeMetadataResponse) error {
  246. processor.AddSyncJob(resp)
  247. return nil
  248. }, 3*time.Second, func(counter int64, lastTsNs int64) error {
  249. if processor.processedTsWatermark == 0 {
  250. return nil
  251. }
  252. // use processor.processedTsWatermark instead of the lastTsNs from the most recent job
  253. now := time.Now().UnixNano()
  254. glog.V(0).Infof("sync %s to %s progressed to %v %0.2f/sec", sourceFiler, targetFiler, time.Unix(0, processor.processedTsWatermark), float64(counter)/(float64(now-lastLogTsNs)/1e9))
  255. lastLogTsNs = now
  256. // collect synchronous offset
  257. statsCollect.FilerSyncOffsetGauge.WithLabelValues(sourceFiler.String(), targetFiler.String(), clientName, sourcePath).Set(float64(processor.processedTsWatermark))
  258. return setOffset(grpcDialOption, targetFiler, getSignaturePrefixByPath(sourcePath), sourceFilerSignature, processor.processedTsWatermark)
  259. })
  260. return pb.FollowMetadata(sourceFiler, grpcDialOption, clientName, clientId, clientEpoch,
  261. sourcePath, nil, sourceFilerOffsetTsNs, 0, targetFilerSignature, processEventFnWithOffset, pb.RetryForeverOnError)
  262. }
  263. // When each business is distinguished according to path, and offsets need to be maintained separately.
  264. func getSignaturePrefixByPath(path string) string {
  265. // compatible historical version
  266. if path == "/" {
  267. return SyncKeyPrefix
  268. } else {
  269. return SyncKeyPrefix + path
  270. }
  271. }
  272. func getOffset(grpcDialOption grpc.DialOption, filer pb.ServerAddress, signaturePrefix string, signature int32) (lastOffsetTsNs int64, readErr error) {
  273. readErr = pb.WithFilerClient(false, filer, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  274. syncKey := []byte(signaturePrefix + "____")
  275. util.Uint32toBytes(syncKey[len(signaturePrefix):len(signaturePrefix)+4], uint32(signature))
  276. resp, err := client.KvGet(context.Background(), &filer_pb.KvGetRequest{Key: syncKey})
  277. if err != nil {
  278. return err
  279. }
  280. if len(resp.Error) != 0 {
  281. return errors.New(resp.Error)
  282. }
  283. if len(resp.Value) < 8 {
  284. return nil
  285. }
  286. lastOffsetTsNs = int64(util.BytesToUint64(resp.Value))
  287. return nil
  288. })
  289. return
  290. }
  291. func setOffset(grpcDialOption grpc.DialOption, filer pb.ServerAddress, signaturePrefix string, signature int32, offsetTsNs int64) error {
  292. return pb.WithFilerClient(false, filer, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  293. syncKey := []byte(signaturePrefix + "____")
  294. util.Uint32toBytes(syncKey[len(signaturePrefix):len(signaturePrefix)+4], uint32(signature))
  295. valueBuf := make([]byte, 8)
  296. util.Uint64toBytes(valueBuf, uint64(offsetTsNs))
  297. resp, err := client.KvPut(context.Background(), &filer_pb.KvPutRequest{
  298. Key: syncKey,
  299. Value: valueBuf,
  300. })
  301. if err != nil {
  302. return err
  303. }
  304. if len(resp.Error) != 0 {
  305. return errors.New(resp.Error)
  306. }
  307. return nil
  308. })
  309. }
  310. func genProcessFunction(sourcePath string, targetPath string, excludePaths []string, dataSink sink.ReplicationSink, debug bool) func(resp *filer_pb.SubscribeMetadataResponse) error {
  311. // process function
  312. processEventFn := func(resp *filer_pb.SubscribeMetadataResponse) error {
  313. message := resp.EventNotification
  314. var sourceOldKey, sourceNewKey util.FullPath
  315. if message.OldEntry != nil {
  316. sourceOldKey = util.FullPath(resp.Directory).Child(message.OldEntry.Name)
  317. }
  318. if message.NewEntry != nil {
  319. sourceNewKey = util.FullPath(message.NewParentPath).Child(message.NewEntry.Name)
  320. }
  321. if debug {
  322. glog.V(0).Infof("received %v", resp)
  323. }
  324. if !strings.HasPrefix(resp.Directory, sourcePath) {
  325. return nil
  326. }
  327. for _, excludePath := range excludePaths {
  328. if strings.HasPrefix(resp.Directory, excludePath) {
  329. return nil
  330. }
  331. }
  332. // handle deletions
  333. if filer_pb.IsDelete(resp) {
  334. if !strings.HasPrefix(string(sourceOldKey), sourcePath) {
  335. return nil
  336. }
  337. key := buildKey(dataSink, message, targetPath, sourceOldKey, sourcePath)
  338. if !dataSink.IsIncremental() {
  339. return dataSink.DeleteEntry(key, message.OldEntry.IsDirectory, message.DeleteChunks, message.Signatures)
  340. }
  341. return nil
  342. }
  343. // handle new entries
  344. if filer_pb.IsCreate(resp) {
  345. if !strings.HasPrefix(string(sourceNewKey), sourcePath) {
  346. return nil
  347. }
  348. key := buildKey(dataSink, message, targetPath, sourceNewKey, sourcePath)
  349. if err := dataSink.CreateEntry(key, message.NewEntry, message.Signatures); err != nil {
  350. return fmt.Errorf("create entry1 : %v", err)
  351. } else {
  352. return nil
  353. }
  354. }
  355. // this is something special?
  356. if filer_pb.IsEmpty(resp) {
  357. return nil
  358. }
  359. // handle updates
  360. if strings.HasPrefix(string(sourceOldKey), sourcePath) {
  361. // old key is in the watched directory
  362. if strings.HasPrefix(string(sourceNewKey), sourcePath) {
  363. // new key is also in the watched directory
  364. if !dataSink.IsIncremental() {
  365. oldKey := util.Join(targetPath, string(sourceOldKey)[len(sourcePath):])
  366. message.NewParentPath = util.Join(targetPath, message.NewParentPath[len(sourcePath):])
  367. foundExisting, err := dataSink.UpdateEntry(string(oldKey), message.OldEntry, message.NewParentPath, message.NewEntry, message.DeleteChunks, message.Signatures)
  368. if foundExisting {
  369. return err
  370. }
  371. // not able to find old entry
  372. if err = dataSink.DeleteEntry(string(oldKey), message.OldEntry.IsDirectory, false, message.Signatures); err != nil {
  373. return fmt.Errorf("delete old entry %v: %v", oldKey, err)
  374. }
  375. }
  376. // create the new entry
  377. newKey := buildKey(dataSink, message, targetPath, sourceNewKey, sourcePath)
  378. if err := dataSink.CreateEntry(newKey, message.NewEntry, message.Signatures); err != nil {
  379. return fmt.Errorf("create entry2 : %v", err)
  380. } else {
  381. return nil
  382. }
  383. } else {
  384. // new key is outside of the watched directory
  385. if !dataSink.IsIncremental() {
  386. key := buildKey(dataSink, message, targetPath, sourceOldKey, sourcePath)
  387. return dataSink.DeleteEntry(key, message.OldEntry.IsDirectory, message.DeleteChunks, message.Signatures)
  388. }
  389. }
  390. } else {
  391. // old key is outside of the watched directory
  392. if strings.HasPrefix(string(sourceNewKey), sourcePath) {
  393. // new key is in the watched directory
  394. key := buildKey(dataSink, message, targetPath, sourceNewKey, sourcePath)
  395. if err := dataSink.CreateEntry(key, message.NewEntry, message.Signatures); err != nil {
  396. return fmt.Errorf("create entry3 : %v", err)
  397. } else {
  398. return nil
  399. }
  400. } else {
  401. // new key is also outside of the watched directory
  402. // skip
  403. }
  404. }
  405. return nil
  406. }
  407. return processEventFn
  408. }
  409. func buildKey(dataSink sink.ReplicationSink, message *filer_pb.EventNotification, targetPath string, sourceKey util.FullPath, sourcePath string) (key string) {
  410. if !dataSink.IsIncremental() {
  411. key = util.Join(targetPath, string(sourceKey)[len(sourcePath):])
  412. } else {
  413. var mTime int64
  414. if message.NewEntry != nil {
  415. mTime = message.NewEntry.Attributes.Mtime
  416. } else if message.OldEntry != nil {
  417. mTime = message.OldEntry.Attributes.Mtime
  418. }
  419. dateKey := time.Unix(mTime, 0).Format("2006-01-02")
  420. key = util.Join(targetPath, dateKey, string(sourceKey)[len(sourcePath):])
  421. }
  422. return escapeKey(key)
  423. }