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.

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