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.

187 lines
5.3 KiB

6 years ago
5 years ago
5 years ago
5 years ago
5 years ago
6 years ago
5 years ago
5 years ago
5 years ago
4 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
  1. package filer
  2. import (
  3. "context"
  4. "fmt"
  5. "io"
  6. "math"
  7. "strings"
  8. "time"
  9. "github.com/golang/protobuf/proto"
  10. "github.com/chrislusf/seaweedfs/weed/glog"
  11. "github.com/chrislusf/seaweedfs/weed/notification"
  12. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  13. "github.com/chrislusf/seaweedfs/weed/util"
  14. )
  15. func (f *Filer) NotifyUpdateEvent(ctx context.Context, oldEntry, newEntry *Entry, deleteChunks, isFromOtherCluster bool, signatures []int32) {
  16. var fullpath string
  17. if oldEntry != nil {
  18. fullpath = string(oldEntry.FullPath)
  19. } else if newEntry != nil {
  20. fullpath = string(newEntry.FullPath)
  21. } else {
  22. return
  23. }
  24. // println("fullpath:", fullpath)
  25. if strings.HasPrefix(fullpath, SystemLogDir) {
  26. return
  27. }
  28. foundSelf := false
  29. for _, sig := range signatures {
  30. if sig == f.Signature {
  31. foundSelf = true
  32. }
  33. }
  34. if !foundSelf {
  35. signatures = append(signatures, f.Signature)
  36. }
  37. newParentPath := ""
  38. if newEntry != nil {
  39. newParentPath, _ = newEntry.FullPath.DirAndName()
  40. }
  41. eventNotification := &filer_pb.EventNotification{
  42. OldEntry: oldEntry.ToProtoEntry(),
  43. NewEntry: newEntry.ToProtoEntry(),
  44. DeleteChunks: deleteChunks,
  45. NewParentPath: newParentPath,
  46. IsFromOtherCluster: isFromOtherCluster,
  47. Signatures: signatures,
  48. }
  49. if notification.Queue != nil {
  50. glog.V(3).Infof("notifying entry update %v", fullpath)
  51. if err := notification.Queue.SendMessage(fullpath, eventNotification); err != nil {
  52. // throw message
  53. glog.Error(err)
  54. }
  55. }
  56. f.logMetaEvent(ctx, fullpath, eventNotification)
  57. }
  58. func (f *Filer) logMetaEvent(ctx context.Context, fullpath string, eventNotification *filer_pb.EventNotification) {
  59. dir, _ := util.FullPath(fullpath).DirAndName()
  60. event := &filer_pb.SubscribeMetadataResponse{
  61. Directory: dir,
  62. EventNotification: eventNotification,
  63. TsNs: time.Now().UnixNano(),
  64. }
  65. data, err := proto.Marshal(event)
  66. if err != nil {
  67. glog.Errorf("failed to marshal filer_pb.SubscribeMetadataResponse %+v: %v", event, err)
  68. return
  69. }
  70. f.LocalMetaLogBuffer.AddToBuffer([]byte(dir), data, event.TsNs)
  71. }
  72. func (f *Filer) logFlushFunc(startTime, stopTime time.Time, buf []byte) {
  73. if len(buf) == 0 {
  74. return
  75. }
  76. startTime, stopTime = startTime.UTC(), stopTime.UTC()
  77. targetFile := fmt.Sprintf("%s/%04d-%02d-%02d/%02d-%02d.%08x", SystemLogDir,
  78. startTime.Year(), startTime.Month(), startTime.Day(), startTime.Hour(), startTime.Minute(), f.UniqueFileId,
  79. // startTime.Second(), startTime.Nanosecond(),
  80. )
  81. for {
  82. if err := f.appendToFile(targetFile, buf); err != nil {
  83. glog.V(0).Infof("metadata log write failed %s: %v", targetFile, err)
  84. time.Sleep(737 * time.Millisecond)
  85. } else {
  86. break
  87. }
  88. }
  89. }
  90. func (f *Filer) ReadPersistedLogBuffer(startTime time.Time, eachLogEntryFn func(logEntry *filer_pb.LogEntry) error) (lastTsNs int64, err error) {
  91. startTime = startTime.UTC()
  92. startDate := fmt.Sprintf("%04d-%02d-%02d", startTime.Year(), startTime.Month(), startTime.Day())
  93. startHourMinute := fmt.Sprintf("%02d-%02d", startTime.Hour(), startTime.Minute())
  94. sizeBuf := make([]byte, 4)
  95. startTsNs := startTime.UnixNano()
  96. dayEntries, _, listDayErr := f.ListDirectoryEntries(context.Background(), SystemLogDir, startDate, true, 366, "", "", "")
  97. if listDayErr != nil {
  98. return lastTsNs, fmt.Errorf("fail to list log by day: %v", listDayErr)
  99. }
  100. for _, dayEntry := range dayEntries {
  101. // println("checking day", dayEntry.FullPath)
  102. hourMinuteEntries, _, listHourMinuteErr := f.ListDirectoryEntries(context.Background(), util.NewFullPath(SystemLogDir, dayEntry.Name()), "", false, math.MaxInt32, "", "", "")
  103. if listHourMinuteErr != nil {
  104. return lastTsNs, fmt.Errorf("fail to list log %s by day: %v", dayEntry.Name(), listHourMinuteErr)
  105. }
  106. for _, hourMinuteEntry := range hourMinuteEntries {
  107. // println("checking hh-mm", hourMinuteEntry.FullPath)
  108. if dayEntry.Name() == startDate {
  109. hourMinute := util.FileNameBase(hourMinuteEntry.Name())
  110. if strings.Compare(hourMinute, startHourMinute) < 0 {
  111. continue
  112. }
  113. }
  114. // println("processing", hourMinuteEntry.FullPath)
  115. chunkedFileReader := NewChunkStreamReaderFromFiler(f.MasterClient, hourMinuteEntry.Chunks)
  116. if lastTsNs, err = ReadEachLogEntry(chunkedFileReader, sizeBuf, startTsNs, eachLogEntryFn); err != nil {
  117. chunkedFileReader.Close()
  118. if err == io.EOF {
  119. continue
  120. }
  121. return lastTsNs, fmt.Errorf("reading %s: %v", hourMinuteEntry.FullPath, err)
  122. }
  123. chunkedFileReader.Close()
  124. }
  125. }
  126. return lastTsNs, nil
  127. }
  128. func ReadEachLogEntry(r io.Reader, sizeBuf []byte, ns int64, eachLogEntryFn func(logEntry *filer_pb.LogEntry) error) (lastTsNs int64, err error) {
  129. for {
  130. n, err := r.Read(sizeBuf)
  131. if err != nil {
  132. return lastTsNs, err
  133. }
  134. if n != 4 {
  135. return lastTsNs, fmt.Errorf("size %d bytes, expected 4 bytes", n)
  136. }
  137. size := util.BytesToUint32(sizeBuf)
  138. // println("entry size", size)
  139. entryData := make([]byte, size)
  140. n, err = r.Read(entryData)
  141. if err != nil {
  142. return lastTsNs, err
  143. }
  144. if n != int(size) {
  145. return lastTsNs, fmt.Errorf("entry data %d bytes, expected %d bytes", n, size)
  146. }
  147. logEntry := &filer_pb.LogEntry{}
  148. if err = proto.Unmarshal(entryData, logEntry); err != nil {
  149. return lastTsNs, err
  150. }
  151. if logEntry.TsNs <= ns {
  152. continue
  153. }
  154. // println("each log: ", logEntry.TsNs)
  155. if err := eachLogEntryFn(logEntry); err != nil {
  156. return lastTsNs, err
  157. } else {
  158. lastTsNs = logEntry.TsNs
  159. }
  160. }
  161. }