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.

200 lines
5.6 KiB

5 years ago
5 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/glog"
  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. glog.V(0).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, file, peerSignature); err == nil {
  60. lastTsNs = prevTsNs
  61. }
  62. glog.V(0).Infof("follow peer: %v, last %v (%d)", peer, time.Unix(0, lastTsNs), lastTsNs)
  63. maybeReplicateMetadataChange = func(event *filer_pb.SubscribeMetadataResponse) {
  64. if err := Replay(f.Store.ActualStore, event); err != nil {
  65. glog.Errorf("failed to reply metadata change from %v: %v", peer, err)
  66. return
  67. }
  68. if lastPersistTime.Add(time.Minute).Before(time.Now()) {
  69. if err := ma.updateOffset(f, peer, peerSignature, event.TsNs); err == nil {
  70. if event.TsNs < time.Now().Add(-2*time.Minute).UnixNano() {
  71. glog.V(0).Infof("sync with %s progressed to: %v", peer, time.Unix(0, event.TsNs))
  72. }
  73. lastPersistTime = time.Now()
  74. } else {
  75. glog.V(0).Infof("failed to update offset for %v: %v", peer, err)
  76. }
  77. }
  78. }
  79. }
  80. processEventFn := func(event *filer_pb.SubscribeMetadataResponse) error {
  81. data, err := proto.Marshal(event)
  82. if err != nil {
  83. glog.Errorf("failed to marshal subscribed filer_pb.SubscribeMetadataResponse %+v: %v", event, err)
  84. return err
  85. }
  86. dir := event.Directory
  87. // println("received meta change", dir, "size", len(data))
  88. ma.MetaLogBuffer.AddToBuffer([]byte(dir), data, event.TsNs)
  89. if maybeReplicateMetadataChange != nil {
  90. maybeReplicateMetadataChange(event)
  91. }
  92. return nil
  93. }
  94. for {
  95. err := pb.WithFilerClient(peer, ma.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  96. stream, err := client.SubscribeLocalMetadata(context.Background(), &filer_pb.SubscribeMetadataRequest{
  97. ClientName: "filer:" + self,
  98. PathPrefix: "/",
  99. SinceNs: lastTsNs,
  100. })
  101. if err != nil {
  102. return fmt.Errorf("subscribe: %v", err)
  103. }
  104. for {
  105. resp, listenErr := stream.Recv()
  106. if listenErr == io.EOF {
  107. return nil
  108. }
  109. if listenErr != nil {
  110. return listenErr
  111. }
  112. if err := processEventFn(resp); err != nil {
  113. return fmt.Errorf("process %v: %v", resp, err)
  114. }
  115. lastTsNs = resp.TsNs
  116. }
  117. })
  118. if err != nil {
  119. glog.V(0).Infof("subscribing remote %s meta change: %v", peer, err)
  120. time.Sleep(1733 * time.Millisecond)
  121. }
  122. }
  123. }
  124. func (ma *MetaAggregator) readFilerStoreSignature(peer string) (sig int32, err error) {
  125. err = pb.WithFilerClient(peer, ma.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  126. resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
  127. if err != nil {
  128. return err
  129. }
  130. sig = resp.Signature
  131. return nil
  132. })
  133. return
  134. }
  135. const(
  136. MetaOffsetPrefix = "Meta"
  137. )
  138. func (ma *MetaAggregator) readOffset(f *Filer, peer string, peerSignature int32) (lastTsNs int64, err error) {
  139. key := []byte(MetaOffsetPrefix+"xxxx")
  140. util.Uint32toBytes(key[len(MetaOffsetPrefix):], uint32(peerSignature))
  141. value, err := f.Store.KvGet(context.Background(), key)
  142. if err == ErrKvNotFound {
  143. glog.Warningf("readOffset %s not found", peer)
  144. return 0, nil
  145. }
  146. if err != nil {
  147. return 0, fmt.Errorf("readOffset %s : %v", peer, err)
  148. }
  149. lastTsNs = int64(util.BytesToUint64(value))
  150. glog.V(0).Infof("readOffset %s : %d", peer, lastTsNs)
  151. return
  152. }
  153. func (ma *MetaAggregator) updateOffset(f *Filer, peer string, peerSignature int32, lastTsNs int64) (err error) {
  154. key := []byte(MetaOffsetPrefix+"xxxx")
  155. util.Uint32toBytes(key[len(MetaOffsetPrefix):], uint32(peerSignature))
  156. value := make([]byte, 8)
  157. util.Uint64toBytes(value, uint64(lastTsNs))
  158. err = f.Store.KvPut(context.Background(), key, value)
  159. if err != nil {
  160. return fmt.Errorf("updateOffset %s : %v", peer, err)
  161. }
  162. glog.V(4).Infof("updateOffset %s : %d", peer, lastTsNs)
  163. return
  164. }