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.

212 lines
5.8 KiB

4 years ago
4 years ago
4 years ago
4 years ago
5 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/util"
  6. "io"
  7. "sync"
  8. "time"
  9. "github.com/golang/protobuf/proto"
  10. "google.golang.org/grpc"
  11. "github.com/chrislusf/seaweedfs/weed/util/log"
  12. "github.com/chrislusf/seaweedfs/weed/pb"
  13. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  14. "github.com/chrislusf/seaweedfs/weed/util/log_buffer"
  15. )
  16. type MetaAggregator struct {
  17. filers []string
  18. grpcDialOption grpc.DialOption
  19. MetaLogBuffer *log_buffer.LogBuffer
  20. // notifying clients
  21. ListenersLock sync.Mutex
  22. ListenersCond *sync.Cond
  23. }
  24. // MetaAggregator only aggregates data "on the fly". The logs are not re-persisted to disk.
  25. // The old data comes from what each LocalMetadata persisted on disk.
  26. func NewMetaAggregator(filers []string, grpcDialOption grpc.DialOption) *MetaAggregator {
  27. t := &MetaAggregator{
  28. filers: filers,
  29. grpcDialOption: grpcDialOption,
  30. }
  31. t.ListenersCond = sync.NewCond(&t.ListenersLock)
  32. t.MetaLogBuffer = log_buffer.NewLogBuffer(LogFlushInterval, nil, func() {
  33. t.ListenersCond.Broadcast()
  34. })
  35. return t
  36. }
  37. func (ma *MetaAggregator) StartLoopSubscribe(f *Filer, self string) {
  38. for _, filer := range ma.filers {
  39. go ma.subscribeToOneFiler(f, self, filer)
  40. }
  41. }
  42. func (ma *MetaAggregator) subscribeToOneFiler(f *Filer, self string, peer string) {
  43. /*
  44. Each filer reads the "filer.store.id", which is the store's signature when filer starts.
  45. When reading from other filers' local meta changes:
  46. * if the received change does not contain signature from self, apply the change to current filer store.
  47. Upon connecting to other filers, need to remember their signature and their offsets.
  48. */
  49. var maybeReplicateMetadataChange func(*filer_pb.SubscribeMetadataResponse)
  50. lastPersistTime := time.Now()
  51. lastTsNs := time.Now().Add(-LogFlushInterval).UnixNano()
  52. peerSignature, err := ma.readFilerStoreSignature(peer)
  53. for err != nil {
  54. log.Infof("connecting to peer filer %s: %v", peer, err)
  55. time.Sleep(1357 * time.Millisecond)
  56. peerSignature, err = ma.readFilerStoreSignature(peer)
  57. }
  58. if peerSignature != f.Signature {
  59. if prevTsNs, err := ma.readOffset(f, peer, peerSignature); err == nil {
  60. lastTsNs = prevTsNs
  61. }
  62. log.Infof("follow peer: %v, last %v (%d)", peer, time.Unix(0, lastTsNs), lastTsNs)
  63. var counter int64
  64. var synced bool
  65. maybeReplicateMetadataChange = func(event *filer_pb.SubscribeMetadataResponse) {
  66. if err := Replay(f.Store, event); err != nil {
  67. log.Errorf("failed to reply metadata change from %v: %v", peer, err)
  68. return
  69. }
  70. counter++
  71. if lastPersistTime.Add(time.Minute).Before(time.Now()) {
  72. if err := ma.updateOffset(f, peer, peerSignature, event.TsNs); err == nil {
  73. if event.TsNs < time.Now().Add(-2*time.Minute).UnixNano() {
  74. log.Infof("sync with %s progressed to: %v %0.2f/sec", peer, time.Unix(0, event.TsNs), float64(counter)/60.0)
  75. } else if !synced {
  76. synced = true
  77. log.Infof("synced with %s", peer)
  78. }
  79. lastPersistTime = time.Now()
  80. counter = 0
  81. } else {
  82. log.Infof("failed to update offset for %v: %v", peer, err)
  83. }
  84. }
  85. }
  86. }
  87. processEventFn := func(event *filer_pb.SubscribeMetadataResponse) error {
  88. data, err := proto.Marshal(event)
  89. if err != nil {
  90. log.Errorf("failed to marshal subscribed filer_pb.SubscribeMetadataResponse %+v: %v", event, err)
  91. return err
  92. }
  93. dir := event.Directory
  94. // println("received meta change", dir, "size", len(data))
  95. ma.MetaLogBuffer.AddToBuffer([]byte(dir), data, 0)
  96. if maybeReplicateMetadataChange != nil {
  97. maybeReplicateMetadataChange(event)
  98. }
  99. return nil
  100. }
  101. for {
  102. err := pb.WithFilerClient(peer, ma.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  103. ctx, cancel := context.WithCancel(context.Background())
  104. defer cancel()
  105. stream, err := client.SubscribeLocalMetadata(ctx, &filer_pb.SubscribeMetadataRequest{
  106. ClientName: "filer:" + self,
  107. PathPrefix: "/",
  108. SinceNs: lastTsNs,
  109. })
  110. if err != nil {
  111. return fmt.Errorf("subscribe: %v", err)
  112. }
  113. for {
  114. resp, listenErr := stream.Recv()
  115. if listenErr == io.EOF {
  116. return nil
  117. }
  118. if listenErr != nil {
  119. return listenErr
  120. }
  121. if err := processEventFn(resp); err != nil {
  122. return fmt.Errorf("process %v: %v", resp, err)
  123. }
  124. lastTsNs = resp.TsNs
  125. f.onMetadataChangeEvent(resp)
  126. }
  127. })
  128. if err != nil {
  129. log.Infof("subscribing remote %s meta change: %v", peer, err)
  130. time.Sleep(1733 * time.Millisecond)
  131. }
  132. }
  133. }
  134. func (ma *MetaAggregator) readFilerStoreSignature(peer string) (sig int32, err error) {
  135. err = pb.WithFilerClient(peer, ma.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  136. resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
  137. if err != nil {
  138. return err
  139. }
  140. sig = resp.Signature
  141. return nil
  142. })
  143. return
  144. }
  145. const (
  146. MetaOffsetPrefix = "Meta"
  147. )
  148. func (ma *MetaAggregator) readOffset(f *Filer, peer string, peerSignature int32) (lastTsNs int64, err error) {
  149. key := []byte(MetaOffsetPrefix + "xxxx")
  150. util.Uint32toBytes(key[len(MetaOffsetPrefix):], uint32(peerSignature))
  151. value, err := f.Store.KvGet(context.Background(), key)
  152. if err == ErrKvNotFound {
  153. log.Warnf("readOffset %s not found", peer)
  154. return 0, nil
  155. }
  156. if err != nil {
  157. return 0, fmt.Errorf("readOffset %s : %v", peer, err)
  158. }
  159. lastTsNs = int64(util.BytesToUint64(value))
  160. log.Infof("readOffset %s : %d", peer, lastTsNs)
  161. return
  162. }
  163. func (ma *MetaAggregator) updateOffset(f *Filer, peer string, peerSignature int32, lastTsNs int64) (err error) {
  164. key := []byte(MetaOffsetPrefix + "xxxx")
  165. util.Uint32toBytes(key[len(MetaOffsetPrefix):], uint32(peerSignature))
  166. value := make([]byte, 8)
  167. util.Uint64toBytes(value, uint64(lastTsNs))
  168. err = f.Store.KvPut(context.Background(), key, value)
  169. if err != nil {
  170. return fmt.Errorf("updateOffset %s : %v", peer, err)
  171. }
  172. log.Tracef("updateOffset %s : %d", peer, lastTsNs)
  173. return
  174. }