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.

274 lines
8.3 KiB

5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
4 years ago
4 years ago
5 years ago
5 years ago
4 years ago
3 years ago
4 years ago
5 years ago
5 years ago
5 years ago
5 years ago
3 years ago
5 years ago
5 years ago
5 years ago
5 years ago
  1. package weed_server
  2. import (
  3. "fmt"
  4. "strings"
  5. "time"
  6. "github.com/golang/protobuf/proto"
  7. "github.com/chrislusf/seaweedfs/weed/filer"
  8. "github.com/chrislusf/seaweedfs/weed/glog"
  9. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  10. "github.com/chrislusf/seaweedfs/weed/util"
  11. "github.com/chrislusf/seaweedfs/weed/util/log_buffer"
  12. )
  13. const (
  14. // MaxUnsyncedEvents send empty notification with timestamp when certain amount of events have been filtered
  15. MaxUnsyncedEvents = 1e3
  16. )
  17. func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, stream filer_pb.SeaweedFiler_SubscribeMetadataServer) error {
  18. peerAddress := findClientAddress(stream.Context(), 0)
  19. alreadyKnown, clientName := fs.addClient(req.ClientName, peerAddress, req.ClientId)
  20. if alreadyKnown {
  21. return fmt.Errorf("duplicated subscription detected for client %s id %d", clientName, req.ClientId)
  22. }
  23. defer fs.deleteClient(clientName, req.ClientId)
  24. lastReadTime := time.Unix(0, req.SinceNs)
  25. glog.V(0).Infof(" %v starts to subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
  26. eachEventNotificationFn := fs.eachEventNotificationFn(req, stream, clientName)
  27. eachLogEntryFn := eachLogEntryFn(eachEventNotificationFn)
  28. var processedTsNs int64
  29. var readPersistedLogErr error
  30. var readInMemoryLogErr error
  31. var isDone bool
  32. for {
  33. glog.V(4).Infof("read on disk %v aggregated subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
  34. processedTsNs, isDone, readPersistedLogErr = fs.filer.ReadPersistedLogBuffer(lastReadTime, req.UntilNs, eachLogEntryFn)
  35. if readPersistedLogErr != nil {
  36. return fmt.Errorf("reading from persisted logs: %v", readPersistedLogErr)
  37. }
  38. if isDone {
  39. return nil
  40. }
  41. if processedTsNs != 0 {
  42. lastReadTime = time.Unix(0, processedTsNs)
  43. }
  44. glog.V(4).Infof("read in memory %v aggregated subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
  45. lastReadTime, isDone, readInMemoryLogErr = fs.filer.MetaAggregator.MetaLogBuffer.LoopProcessLogData("aggMeta:"+clientName, lastReadTime, req.UntilNs, func() bool {
  46. fs.filer.MetaAggregator.ListenersLock.Lock()
  47. fs.filer.MetaAggregator.ListenersCond.Wait()
  48. fs.filer.MetaAggregator.ListenersLock.Unlock()
  49. return true
  50. }, eachLogEntryFn)
  51. if readInMemoryLogErr != nil {
  52. if readInMemoryLogErr == log_buffer.ResumeFromDiskError {
  53. continue
  54. }
  55. glog.Errorf("processed to %v: %v", lastReadTime, readInMemoryLogErr)
  56. if readInMemoryLogErr != log_buffer.ResumeError {
  57. break
  58. }
  59. }
  60. if isDone {
  61. return nil
  62. }
  63. time.Sleep(1127 * time.Millisecond)
  64. }
  65. return readInMemoryLogErr
  66. }
  67. func (fs *FilerServer) SubscribeLocalMetadata(req *filer_pb.SubscribeMetadataRequest, stream filer_pb.SeaweedFiler_SubscribeLocalMetadataServer) error {
  68. peerAddress := findClientAddress(stream.Context(), 0)
  69. alreadyKnown, clientName := fs.addClient(req.ClientName, peerAddress, req.ClientId)
  70. if alreadyKnown {
  71. return fmt.Errorf("duplicated local subscription detected for client %s id %d", clientName, req.ClientId)
  72. }
  73. defer fs.deleteClient(clientName, req.ClientId)
  74. lastReadTime := time.Unix(0, req.SinceNs)
  75. glog.V(0).Infof(" %v local subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
  76. eachEventNotificationFn := fs.eachEventNotificationFn(req, stream, clientName)
  77. eachLogEntryFn := eachLogEntryFn(eachEventNotificationFn)
  78. var processedTsNs int64
  79. var readPersistedLogErr error
  80. var readInMemoryLogErr error
  81. var isDone bool
  82. for {
  83. // println("reading from persisted logs ...")
  84. glog.V(0).Infof("read on disk %v local subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
  85. processedTsNs, isDone, readPersistedLogErr = fs.filer.ReadPersistedLogBuffer(lastReadTime, req.UntilNs, eachLogEntryFn)
  86. if readPersistedLogErr != nil {
  87. glog.V(0).Infof("read on disk %v local subscribe %s from %+v: %v", clientName, req.PathPrefix, lastReadTime, readPersistedLogErr)
  88. return fmt.Errorf("reading from persisted logs: %v", readPersistedLogErr)
  89. }
  90. if isDone {
  91. return nil
  92. }
  93. if processedTsNs != 0 {
  94. lastReadTime = time.Unix(0, processedTsNs)
  95. } else {
  96. if readInMemoryLogErr == log_buffer.ResumeFromDiskError {
  97. time.Sleep(1127 * time.Millisecond)
  98. continue
  99. }
  100. }
  101. glog.V(0).Infof("read in memory %v local subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
  102. lastReadTime, isDone, readInMemoryLogErr = fs.filer.LocalMetaLogBuffer.LoopProcessLogData("localMeta:"+clientName, lastReadTime, req.UntilNs, func() bool {
  103. fs.listenersLock.Lock()
  104. fs.listenersCond.Wait()
  105. fs.listenersLock.Unlock()
  106. return true
  107. }, eachLogEntryFn)
  108. if readInMemoryLogErr != nil {
  109. if readInMemoryLogErr == log_buffer.ResumeFromDiskError {
  110. continue
  111. }
  112. glog.Errorf("processed to %v: %v", lastReadTime, readInMemoryLogErr)
  113. if readInMemoryLogErr != log_buffer.ResumeError {
  114. break
  115. }
  116. }
  117. if isDone {
  118. return nil
  119. }
  120. }
  121. return readInMemoryLogErr
  122. }
  123. func eachLogEntryFn(eachEventNotificationFn func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error) func(logEntry *filer_pb.LogEntry) error {
  124. return func(logEntry *filer_pb.LogEntry) error {
  125. event := &filer_pb.SubscribeMetadataResponse{}
  126. if err := proto.Unmarshal(logEntry.Data, event); err != nil {
  127. glog.Errorf("unexpected unmarshal filer_pb.SubscribeMetadataResponse: %v", err)
  128. return fmt.Errorf("unexpected unmarshal filer_pb.SubscribeMetadataResponse: %v", err)
  129. }
  130. if err := eachEventNotificationFn(event.Directory, event.EventNotification, event.TsNs); err != nil {
  131. return err
  132. }
  133. return nil
  134. }
  135. }
  136. func (fs *FilerServer) eachEventNotificationFn(req *filer_pb.SubscribeMetadataRequest, stream filer_pb.SeaweedFiler_SubscribeMetadataServer, clientName string) func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error {
  137. filtered := 0
  138. return func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error {
  139. defer func() {
  140. if filtered > MaxUnsyncedEvents {
  141. if err := stream.Send(&filer_pb.SubscribeMetadataResponse{
  142. EventNotification: &filer_pb.EventNotification{},
  143. TsNs: tsNs,
  144. }); err == nil {
  145. filtered = 0
  146. }
  147. }
  148. }()
  149. filtered++
  150. foundSelf := false
  151. for _, sig := range eventNotification.Signatures {
  152. if sig == req.Signature && req.Signature != 0 {
  153. return nil
  154. }
  155. if sig == fs.filer.Signature {
  156. foundSelf = true
  157. }
  158. }
  159. if !foundSelf {
  160. eventNotification.Signatures = append(eventNotification.Signatures, fs.filer.Signature)
  161. }
  162. // get complete path to the file or directory
  163. var entryName string
  164. if eventNotification.OldEntry != nil {
  165. entryName = eventNotification.OldEntry.Name
  166. } else if eventNotification.NewEntry != nil {
  167. entryName = eventNotification.NewEntry.Name
  168. }
  169. fullpath := util.Join(dirPath, entryName)
  170. // skip on filer internal meta logs
  171. if strings.HasPrefix(fullpath, filer.SystemLogDir) {
  172. return nil
  173. }
  174. if hasPrefixIn(fullpath, req.PathPrefixes) {
  175. // good
  176. } else {
  177. if !strings.HasPrefix(fullpath, req.PathPrefix) {
  178. if eventNotification.NewParentPath != "" {
  179. newFullPath := util.Join(eventNotification.NewParentPath, entryName)
  180. if !strings.HasPrefix(newFullPath, req.PathPrefix) {
  181. return nil
  182. }
  183. } else {
  184. return nil
  185. }
  186. }
  187. }
  188. message := &filer_pb.SubscribeMetadataResponse{
  189. Directory: dirPath,
  190. EventNotification: eventNotification,
  191. TsNs: tsNs,
  192. }
  193. // println("sending", dirPath, entryName)
  194. if err := stream.Send(message); err != nil {
  195. glog.V(0).Infof("=> client %v: %+v", clientName, err)
  196. return err
  197. }
  198. filtered = 0
  199. return nil
  200. }
  201. }
  202. func hasPrefixIn(text string, prefixes []string) bool {
  203. for _, p := range prefixes {
  204. if strings.HasPrefix(text, p) {
  205. return true
  206. }
  207. }
  208. return false
  209. }
  210. func (fs *FilerServer) addClient(clientType string, clientAddress string, clientId int32) (alreadyKnown bool, clientName string) {
  211. clientName = clientType + "@" + clientAddress
  212. glog.V(0).Infof("+ listener %v", clientName)
  213. if clientId != 0 {
  214. fs.knownListenersLock.Lock()
  215. _, alreadyKnown = fs.knownListeners[clientId]
  216. fs.knownListenersLock.Unlock()
  217. }
  218. return
  219. }
  220. func (fs *FilerServer) deleteClient(clientName string, clientId int32) {
  221. glog.V(0).Infof("- listener %v", clientName)
  222. if clientId != 0 {
  223. fs.knownListenersLock.Lock()
  224. delete(fs.knownListeners, clientId)
  225. fs.knownListenersLock.Unlock()
  226. }
  227. }