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.

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