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.

273 lines
8.2 KiB

3 years ago
3 years ago
2 years ago
2 years ago
2 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
5 years ago
4 years ago
4 years ago
4 years ago
  1. package filer
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/chrislusf/seaweedfs/weed/cluster"
  6. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  7. "github.com/chrislusf/seaweedfs/weed/util"
  8. "io"
  9. "sync"
  10. "time"
  11. "github.com/golang/protobuf/proto"
  12. "google.golang.org/grpc"
  13. "github.com/chrislusf/seaweedfs/weed/glog"
  14. "github.com/chrislusf/seaweedfs/weed/pb"
  15. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  16. "github.com/chrislusf/seaweedfs/weed/util/log_buffer"
  17. )
  18. type MetaAggregator struct {
  19. filer *Filer
  20. self pb.ServerAddress
  21. isLeader bool
  22. grpcDialOption grpc.DialOption
  23. MetaLogBuffer *log_buffer.LogBuffer
  24. peerStatues map[pb.ServerAddress]int
  25. peerStatuesLock sync.Mutex
  26. // notifying clients
  27. ListenersLock sync.Mutex
  28. ListenersCond *sync.Cond
  29. }
  30. // MetaAggregator only aggregates data "on the fly". The logs are not re-persisted to disk.
  31. // The old data comes from what each LocalMetadata persisted on disk.
  32. func NewMetaAggregator(filer *Filer, self pb.ServerAddress, grpcDialOption grpc.DialOption) *MetaAggregator {
  33. t := &MetaAggregator{
  34. filer: filer,
  35. self: self,
  36. grpcDialOption: grpcDialOption,
  37. peerStatues: make(map[pb.ServerAddress]int),
  38. }
  39. t.ListenersCond = sync.NewCond(&t.ListenersLock)
  40. t.MetaLogBuffer = log_buffer.NewLogBuffer("aggr", LogFlushInterval, nil, func() {
  41. t.ListenersCond.Broadcast()
  42. })
  43. return t
  44. }
  45. func (ma *MetaAggregator) OnPeerUpdate(update *master_pb.ClusterNodeUpdate, startFrom time.Time) {
  46. if update.NodeType != cluster.FilerType {
  47. return
  48. }
  49. address := pb.ServerAddress(update.Address)
  50. if update.IsAdd {
  51. // every filer should subscribe to a new filer
  52. if ma.setActive(address, true) {
  53. go ma.loopSubscribeToOneFiler(ma.filer, ma.self, address, startFrom)
  54. }
  55. } else {
  56. ma.setActive(address, false)
  57. }
  58. }
  59. func (ma *MetaAggregator) setActive(address pb.ServerAddress, isActive bool) (notDuplicated bool) {
  60. ma.peerStatuesLock.Lock()
  61. defer ma.peerStatuesLock.Unlock()
  62. if isActive {
  63. if _, found := ma.peerStatues[address]; found {
  64. ma.peerStatues[address] += 1
  65. } else {
  66. ma.peerStatues[address] = 1
  67. notDuplicated = true
  68. }
  69. } else {
  70. if _, found := ma.peerStatues[address]; found {
  71. delete(ma.peerStatues, address)
  72. }
  73. }
  74. return
  75. }
  76. func (ma *MetaAggregator) isActive(address pb.ServerAddress) (isActive bool) {
  77. ma.peerStatuesLock.Lock()
  78. defer ma.peerStatuesLock.Unlock()
  79. var count int
  80. count, isActive = ma.peerStatues[address]
  81. return count > 0 && isActive
  82. }
  83. func (ma *MetaAggregator) loopSubscribeToOneFiler(f *Filer, self pb.ServerAddress, peer pb.ServerAddress, startFrom time.Time) {
  84. lastTsNs := startFrom.UnixNano()
  85. for {
  86. glog.V(0).Infof("loopSubscribeToOneFiler read %s start from %v %d", peer, time.Unix(0, lastTsNs), lastTsNs)
  87. nextLastTsNs, err := ma.doSubscribeToOneFiler(f, self, peer, lastTsNs)
  88. if !ma.isActive(peer) {
  89. glog.V(0).Infof("stop subscribing remote %s meta change", peer)
  90. return
  91. }
  92. if err != nil {
  93. glog.V(0).Infof("subscribing remote %s meta change: %v", peer, err)
  94. }
  95. if lastTsNs < nextLastTsNs {
  96. lastTsNs = nextLastTsNs
  97. }
  98. time.Sleep(1733 * time.Millisecond)
  99. }
  100. }
  101. func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress, peer pb.ServerAddress, startFrom int64) (int64, error) {
  102. /*
  103. Each filer reads the "filer.store.id", which is the store's signature when filer starts.
  104. When reading from other filers' local meta changes:
  105. * if the received change does not contain signature from self, apply the change to current filer store.
  106. Upon connecting to other filers, need to remember their signature and their offsets.
  107. */
  108. var maybeReplicateMetadataChange func(*filer_pb.SubscribeMetadataResponse)
  109. lastPersistTime := time.Now()
  110. lastTsNs := startFrom
  111. peerSignature, err := ma.readFilerStoreSignature(peer)
  112. if err != nil {
  113. return lastTsNs, fmt.Errorf("connecting to peer filer %s: %v", peer, err)
  114. }
  115. // when filer store is not shared by multiple filers
  116. if peerSignature != f.Signature {
  117. if prevTsNs, err := ma.readOffset(f, peer, peerSignature); err == nil {
  118. lastTsNs = prevTsNs
  119. defer func(prevTsNs int64) {
  120. if lastTsNs != prevTsNs && lastTsNs != lastPersistTime.UnixNano() {
  121. if err := ma.updateOffset(f, peer, peerSignature, lastTsNs); err == nil {
  122. glog.V(0).Infof("last sync time with %s at %v (%d)", peer, time.Unix(0, lastTsNs), lastTsNs)
  123. } else {
  124. glog.Errorf("failed to save last sync time with %s at %v (%d)", peer, time.Unix(0, lastTsNs), lastTsNs)
  125. }
  126. }
  127. }(prevTsNs)
  128. }
  129. glog.V(0).Infof("follow peer: %v, last %v (%d)", peer, time.Unix(0, lastTsNs), lastTsNs)
  130. var counter int64
  131. var synced bool
  132. maybeReplicateMetadataChange = func(event *filer_pb.SubscribeMetadataResponse) {
  133. if err := Replay(f.Store, event); err != nil {
  134. glog.Errorf("failed to reply metadata change from %v: %v", peer, err)
  135. return
  136. }
  137. counter++
  138. if lastPersistTime.Add(time.Minute).Before(time.Now()) {
  139. if err := ma.updateOffset(f, peer, peerSignature, event.TsNs); err == nil {
  140. if event.TsNs < time.Now().Add(-2*time.Minute).UnixNano() {
  141. glog.V(0).Infof("sync with %s progressed to: %v %0.2f/sec", peer, time.Unix(0, event.TsNs), float64(counter)/60.0)
  142. } else if !synced {
  143. synced = true
  144. glog.V(0).Infof("synced with %s", peer)
  145. }
  146. lastPersistTime = time.Now()
  147. counter = 0
  148. } else {
  149. glog.V(0).Infof("failed to update offset for %v: %v", peer, err)
  150. }
  151. }
  152. }
  153. }
  154. processEventFn := func(event *filer_pb.SubscribeMetadataResponse) error {
  155. data, err := proto.Marshal(event)
  156. if err != nil {
  157. glog.Errorf("failed to marshal subscribed filer_pb.SubscribeMetadataResponse %+v: %v", event, err)
  158. return err
  159. }
  160. dir := event.Directory
  161. // println("received meta change", dir, "size", len(data))
  162. ma.MetaLogBuffer.AddToBuffer([]byte(dir), data, event.TsNs)
  163. if maybeReplicateMetadataChange != nil {
  164. maybeReplicateMetadataChange(event)
  165. }
  166. return nil
  167. }
  168. glog.V(0).Infof("subscribing remote %s meta change: %v, clientId:%d", peer, time.Unix(0, lastTsNs), ma.filer.UniqueFileId)
  169. err = pb.WithFilerClient(true, peer, ma.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  170. ctx, cancel := context.WithCancel(context.Background())
  171. defer cancel()
  172. stream, err := client.SubscribeLocalMetadata(ctx, &filer_pb.SubscribeMetadataRequest{
  173. ClientName: "filer:" + string(self),
  174. PathPrefix: "/",
  175. SinceNs: lastTsNs,
  176. ClientId: int32(ma.filer.UniqueFileId),
  177. })
  178. if err != nil {
  179. return fmt.Errorf("subscribe: %v", err)
  180. }
  181. for {
  182. resp, listenErr := stream.Recv()
  183. if listenErr == io.EOF {
  184. return nil
  185. }
  186. if listenErr != nil {
  187. return listenErr
  188. }
  189. if err := processEventFn(resp); err != nil {
  190. return fmt.Errorf("process %v: %v", resp, err)
  191. }
  192. f.onMetadataChangeEvent(resp)
  193. lastTsNs = resp.TsNs
  194. }
  195. })
  196. return lastTsNs, err
  197. }
  198. func (ma *MetaAggregator) readFilerStoreSignature(peer pb.ServerAddress) (sig int32, err error) {
  199. err = pb.WithFilerClient(false, peer, ma.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  200. resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
  201. if err != nil {
  202. return err
  203. }
  204. sig = resp.Signature
  205. return nil
  206. })
  207. return
  208. }
  209. const (
  210. MetaOffsetPrefix = "Meta"
  211. )
  212. func (ma *MetaAggregator) readOffset(f *Filer, peer pb.ServerAddress, peerSignature int32) (lastTsNs int64, err error) {
  213. key := []byte(MetaOffsetPrefix + "xxxx")
  214. util.Uint32toBytes(key[len(MetaOffsetPrefix):], uint32(peerSignature))
  215. value, err := f.Store.KvGet(context.Background(), key)
  216. if err != nil {
  217. return 0, fmt.Errorf("readOffset %s : %v", peer, err)
  218. }
  219. lastTsNs = int64(util.BytesToUint64(value))
  220. glog.V(0).Infof("readOffset %s : %d", peer, lastTsNs)
  221. return
  222. }
  223. func (ma *MetaAggregator) updateOffset(f *Filer, peer pb.ServerAddress, peerSignature int32, lastTsNs int64) (err error) {
  224. key := []byte(MetaOffsetPrefix + "xxxx")
  225. util.Uint32toBytes(key[len(MetaOffsetPrefix):], uint32(peerSignature))
  226. value := make([]byte, 8)
  227. util.Uint64toBytes(value, uint64(lastTsNs))
  228. err = f.Store.KvPut(context.Background(), key, value)
  229. if err != nil {
  230. return fmt.Errorf("updateOffset %s : %v", peer, err)
  231. }
  232. glog.V(4).Infof("updateOffset %s : %d", peer, lastTsNs)
  233. return
  234. }