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.

292 lines
9.1 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/seaweedfs/seaweedfs/weed/stats"
  5. "strings"
  6. "time"
  7. "github.com/golang/protobuf/proto"
  8. "github.com/seaweedfs/seaweedfs/weed/filer"
  9. "github.com/seaweedfs/seaweedfs/weed/glog"
  10. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  11. "github.com/seaweedfs/seaweedfs/weed/util"
  12. "github.com/seaweedfs/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, req.ClientEpoch)
  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, req.ClientEpoch)
  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. // use negative client id to differentiate from addClient()/deleteClient() used in SubscribeMetadata()
  71. req.ClientId = -req.ClientId
  72. alreadyKnown, clientName := fs.addClient(req.ClientName, peerAddress, req.ClientId, req.ClientEpoch)
  73. if alreadyKnown {
  74. return fmt.Errorf("duplicated local subscription detected for client %s clientId:%d", clientName, req.ClientId)
  75. }
  76. defer func() {
  77. glog.V(0).Infof(" - %v local subscribe %s clientId:%d", clientName, req.PathPrefix, req.ClientId)
  78. fs.deleteClient(clientName, req.ClientId, req.ClientEpoch)
  79. }()
  80. lastReadTime := time.Unix(0, req.SinceNs)
  81. glog.V(0).Infof(" + %v local subscribe %s from %+v clientId:%d", clientName, req.PathPrefix, lastReadTime, req.ClientId)
  82. eachEventNotificationFn := fs.eachEventNotificationFn(req, stream, clientName)
  83. eachLogEntryFn := eachLogEntryFn(eachEventNotificationFn)
  84. var processedTsNs int64
  85. var readPersistedLogErr error
  86. var readInMemoryLogErr error
  87. var isDone bool
  88. for {
  89. // println("reading from persisted logs ...")
  90. glog.V(0).Infof("read on disk %v local subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
  91. processedTsNs, isDone, readPersistedLogErr = fs.filer.ReadPersistedLogBuffer(lastReadTime, req.UntilNs, eachLogEntryFn)
  92. if readPersistedLogErr != nil {
  93. glog.V(0).Infof("read on disk %v local subscribe %s from %+v: %v", clientName, req.PathPrefix, lastReadTime, readPersistedLogErr)
  94. return fmt.Errorf("reading from persisted logs: %v", readPersistedLogErr)
  95. }
  96. if isDone {
  97. return nil
  98. }
  99. if processedTsNs != 0 {
  100. lastReadTime = time.Unix(0, processedTsNs)
  101. } else {
  102. if readInMemoryLogErr == log_buffer.ResumeFromDiskError {
  103. time.Sleep(1127 * time.Millisecond)
  104. continue
  105. }
  106. }
  107. glog.V(0).Infof("read in memory %v local subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
  108. lastReadTime, isDone, readInMemoryLogErr = fs.filer.LocalMetaLogBuffer.LoopProcessLogData("localMeta:"+clientName, lastReadTime, req.UntilNs, func() bool {
  109. fs.listenersLock.Lock()
  110. fs.listenersCond.Wait()
  111. fs.listenersLock.Unlock()
  112. return true
  113. }, eachLogEntryFn)
  114. if readInMemoryLogErr != nil {
  115. if readInMemoryLogErr == log_buffer.ResumeFromDiskError {
  116. continue
  117. }
  118. glog.Errorf("processed to %v: %v", lastReadTime, readInMemoryLogErr)
  119. if readInMemoryLogErr != log_buffer.ResumeError {
  120. break
  121. }
  122. }
  123. if isDone {
  124. return nil
  125. }
  126. }
  127. return readInMemoryLogErr
  128. }
  129. func eachLogEntryFn(eachEventNotificationFn func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error) func(logEntry *filer_pb.LogEntry) error {
  130. return func(logEntry *filer_pb.LogEntry) error {
  131. event := &filer_pb.SubscribeMetadataResponse{}
  132. if err := proto.Unmarshal(logEntry.Data, event); err != nil {
  133. glog.Errorf("unexpected unmarshal filer_pb.SubscribeMetadataResponse: %v", err)
  134. return fmt.Errorf("unexpected unmarshal filer_pb.SubscribeMetadataResponse: %v", err)
  135. }
  136. if err := eachEventNotificationFn(event.Directory, event.EventNotification, event.TsNs); err != nil {
  137. return err
  138. }
  139. return nil
  140. }
  141. }
  142. 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 {
  143. filtered := 0
  144. return func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error {
  145. defer func() {
  146. if filtered > MaxUnsyncedEvents {
  147. if err := stream.Send(&filer_pb.SubscribeMetadataResponse{
  148. EventNotification: &filer_pb.EventNotification{},
  149. TsNs: tsNs,
  150. }); err == nil {
  151. filtered = 0
  152. }
  153. }
  154. }()
  155. filtered++
  156. foundSelf := false
  157. for _, sig := range eventNotification.Signatures {
  158. if sig == req.Signature && req.Signature != 0 {
  159. return nil
  160. }
  161. if sig == fs.filer.Signature {
  162. foundSelf = true
  163. }
  164. }
  165. if !foundSelf {
  166. eventNotification.Signatures = append(eventNotification.Signatures, fs.filer.Signature)
  167. }
  168. // get complete path to the file or directory
  169. var entryName string
  170. if eventNotification.OldEntry != nil {
  171. entryName = eventNotification.OldEntry.Name
  172. } else if eventNotification.NewEntry != nil {
  173. entryName = eventNotification.NewEntry.Name
  174. }
  175. fullpath := util.Join(dirPath, entryName)
  176. // skip on filer internal meta logs
  177. if strings.HasPrefix(fullpath, filer.SystemLogDir) {
  178. return nil
  179. }
  180. if hasPrefixIn(fullpath, req.PathPrefixes) {
  181. // good
  182. } else {
  183. if !strings.HasPrefix(fullpath, req.PathPrefix) {
  184. if eventNotification.NewParentPath != "" {
  185. newFullPath := util.Join(eventNotification.NewParentPath, entryName)
  186. if !strings.HasPrefix(newFullPath, req.PathPrefix) {
  187. return nil
  188. }
  189. } else {
  190. return nil
  191. }
  192. }
  193. }
  194. // collect timestamps for path
  195. stats.FilerServerLastSendTsOfSubscribeGauge.WithLabelValues(fs.option.Host.String(), req.ClientName, req.PathPrefix).Set(float64(tsNs))
  196. message := &filer_pb.SubscribeMetadataResponse{
  197. Directory: dirPath,
  198. EventNotification: eventNotification,
  199. TsNs: tsNs,
  200. }
  201. // println("sending", dirPath, entryName)
  202. if err := stream.Send(message); err != nil {
  203. glog.V(0).Infof("=> client %v: %+v", clientName, err)
  204. return err
  205. }
  206. filtered = 0
  207. return nil
  208. }
  209. }
  210. func hasPrefixIn(text string, prefixes []string) bool {
  211. for _, p := range prefixes {
  212. if strings.HasPrefix(text, p) {
  213. return true
  214. }
  215. }
  216. return false
  217. }
  218. func (fs *FilerServer) addClient(clientType string, clientAddress string, clientId int32, clientEpoch int32) (alreadyKnown bool, clientName string) {
  219. clientName = clientType + "@" + clientAddress
  220. glog.V(0).Infof("+ listener %v", clientName)
  221. if clientId != 0 {
  222. fs.knownListenersLock.Lock()
  223. defer fs.knownListenersLock.Unlock()
  224. epoch, found := fs.knownListeners[clientId]
  225. if !found || epoch < clientEpoch {
  226. fs.knownListeners[clientId] = clientEpoch
  227. } else {
  228. alreadyKnown = true
  229. }
  230. }
  231. return
  232. }
  233. func (fs *FilerServer) deleteClient(clientName string, clientId int32, clientEpoch int32) {
  234. glog.V(0).Infof("- listener %v", clientName)
  235. if clientId != 0 {
  236. fs.knownListenersLock.Lock()
  237. defer fs.knownListenersLock.Unlock()
  238. epoch, found := fs.knownListeners[clientId]
  239. if found && epoch <= clientEpoch {
  240. delete(fs.knownListeners, clientId)
  241. }
  242. }
  243. }