filer_notify.go 3.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148
  1. package filer
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/seaweedfs/seaweedfs/weed/util/log_buffer"
  6. "io"
  7. "regexp"
  8. "strings"
  9. "time"
  10. "google.golang.org/protobuf/proto"
  11. "github.com/seaweedfs/seaweedfs/weed/glog"
  12. "github.com/seaweedfs/seaweedfs/weed/notification"
  13. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  14. "github.com/seaweedfs/seaweedfs/weed/util"
  15. )
  16. func (f *Filer) NotifyUpdateEvent(ctx context.Context, oldEntry, newEntry *Entry, deleteChunks, isFromOtherCluster bool, signatures []int32) {
  17. var fullpath string
  18. if oldEntry != nil {
  19. fullpath = string(oldEntry.FullPath)
  20. } else if newEntry != nil {
  21. fullpath = string(newEntry.FullPath)
  22. } else {
  23. return
  24. }
  25. // println("fullpath:", fullpath)
  26. if strings.HasPrefix(fullpath, SystemLogDir) {
  27. return
  28. }
  29. foundSelf := false
  30. for _, sig := range signatures {
  31. if sig == f.Signature {
  32. foundSelf = true
  33. }
  34. }
  35. if !foundSelf {
  36. signatures = append(signatures, f.Signature)
  37. }
  38. newParentPath := ""
  39. if newEntry != nil {
  40. newParentPath, _ = newEntry.FullPath.DirAndName()
  41. }
  42. eventNotification := &filer_pb.EventNotification{
  43. OldEntry: oldEntry.ToProtoEntry(),
  44. NewEntry: newEntry.ToProtoEntry(),
  45. DeleteChunks: deleteChunks,
  46. NewParentPath: newParentPath,
  47. IsFromOtherCluster: isFromOtherCluster,
  48. Signatures: signatures,
  49. }
  50. if notification.Queue != nil {
  51. glog.V(3).Infof("notifying entry update %v", fullpath)
  52. if err := notification.Queue.SendMessage(fullpath, eventNotification); err != nil {
  53. // throw message
  54. glog.Error(err)
  55. }
  56. }
  57. f.logMetaEvent(ctx, fullpath, eventNotification)
  58. }
  59. func (f *Filer) logMetaEvent(ctx context.Context, fullpath string, eventNotification *filer_pb.EventNotification) {
  60. dir, _ := util.FullPath(fullpath).DirAndName()
  61. event := &filer_pb.SubscribeMetadataResponse{
  62. Directory: dir,
  63. EventNotification: eventNotification,
  64. TsNs: time.Now().UnixNano(),
  65. }
  66. data, err := proto.Marshal(event)
  67. if err != nil {
  68. glog.Errorf("failed to marshal filer_pb.SubscribeMetadataResponse %+v: %v", event, err)
  69. return
  70. }
  71. f.LocalMetaLogBuffer.AddDataToBuffer([]byte(dir), data, event.TsNs)
  72. }
  73. func (f *Filer) logFlushFunc(logBuffer *log_buffer.LogBuffer, startTime, stopTime time.Time, buf []byte) {
  74. if len(buf) == 0 {
  75. return
  76. }
  77. startTime, stopTime = startTime.UTC(), stopTime.UTC()
  78. targetFile := fmt.Sprintf("%s/%04d-%02d-%02d/%02d-%02d.%08x", SystemLogDir,
  79. startTime.Year(), startTime.Month(), startTime.Day(), startTime.Hour(), startTime.Minute(), f.UniqueFilerId,
  80. // startTime.Second(), startTime.Nanosecond(),
  81. )
  82. for {
  83. if err := f.appendToFile(targetFile, buf); err != nil {
  84. glog.V(0).Infof("metadata log write failed %s: %v", targetFile, err)
  85. time.Sleep(737 * time.Millisecond)
  86. } else {
  87. break
  88. }
  89. }
  90. }
  91. var (
  92. VolumeNotFoundPattern = regexp.MustCompile(`volume \d+? not found`)
  93. )
  94. func (f *Filer) ReadPersistedLogBuffer(startPosition log_buffer.MessagePosition, stopTsNs int64, eachLogEntryFn log_buffer.EachLogEntryFuncType) (lastTsNs int64, isDone bool, err error) {
  95. visitor, visitErr := f.collectPersistedLogBuffer(startPosition, stopTsNs)
  96. if visitErr != nil {
  97. if visitErr == io.EOF {
  98. return
  99. }
  100. err = fmt.Errorf("reading from persisted logs: %v", visitErr)
  101. return
  102. }
  103. var logEntry *filer_pb.LogEntry
  104. for {
  105. logEntry, visitErr = visitor.GetNext()
  106. if visitErr != nil {
  107. if visitErr == io.EOF {
  108. break
  109. }
  110. err = fmt.Errorf("read next from persisted logs: %v", visitErr)
  111. return
  112. }
  113. isDone, visitErr = eachLogEntryFn(logEntry)
  114. if visitErr != nil {
  115. err = fmt.Errorf("process persisted log entry: %v", visitErr)
  116. return
  117. }
  118. lastTsNs = logEntry.TsNs
  119. if isDone {
  120. return
  121. }
  122. }
  123. return
  124. }