volume_backup.go 6.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250
  1. package storage
  2. import (
  3. "context"
  4. "fmt"
  5. "io"
  6. "os"
  7. "google.golang.org/grpc"
  8. "github.com/chrislusf/seaweedfs/weed/operation"
  9. "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
  10. "github.com/chrislusf/seaweedfs/weed/storage/idx"
  11. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  12. "github.com/chrislusf/seaweedfs/weed/storage/super_block"
  13. . "github.com/chrislusf/seaweedfs/weed/storage/types"
  14. )
  15. func (v *Volume) GetVolumeSyncStatus() *volume_server_pb.VolumeSyncStatusResponse {
  16. v.dataFileAccessLock.RLock()
  17. defer v.dataFileAccessLock.RUnlock()
  18. var syncStatus = &volume_server_pb.VolumeSyncStatusResponse{}
  19. if datSize, _, err := v.DataBackend.GetStat(); err == nil {
  20. syncStatus.TailOffset = uint64(datSize)
  21. }
  22. syncStatus.Collection = v.Collection
  23. syncStatus.IdxFileSize = v.nm.IndexFileSize()
  24. syncStatus.CompactRevision = uint32(v.SuperBlock.CompactionRevision)
  25. syncStatus.Ttl = v.SuperBlock.Ttl.String()
  26. syncStatus.Replication = v.SuperBlock.ReplicaPlacement.String()
  27. return syncStatus
  28. }
  29. // The volume sync with a master volume via 2 steps:
  30. // 1. The slave checks master side to find subscription checkpoint
  31. // to setup the replication.
  32. // 2. The slave receives the updates from master
  33. /*
  34. Assume the slave volume needs to follow the master volume.
  35. The master volume could be compacted, and could be many files ahead of
  36. slave volume.
  37. Step 0: // implemented in command/backup.go, to avoid dat file size overflow.
  38. 0.1 If slave compact version is less than the master, do a local compaction, and set
  39. local compact version the same as the master.
  40. 0.2 If the slave size is still bigger than the master, discard local copy and do a full copy.
  41. Step 1:
  42. The slave volume ask the master by the last modification time t.
  43. The master do a binary search in volume (use .idx as an array, and check the appendAtNs in .dat file),
  44. to find the first entry with appendAtNs > t.
  45. Step 2:
  46. The master send content bytes to the slave. The bytes are not chunked by needle.
  47. Step 3:
  48. The slave generate the needle map for the new bytes. (This may be optimized to incrementally
  49. update needle map when receiving new .dat bytes. But seems not necessary now.)
  50. */
  51. func (v *Volume) IncrementalBackup(volumeServer string, grpcDialOption grpc.DialOption) error {
  52. startFromOffset, _, _ := v.FileStat()
  53. appendAtNs, err := v.findLastAppendAtNs()
  54. if err != nil {
  55. return err
  56. }
  57. writeOffset := int64(startFromOffset)
  58. err = operation.WithVolumeServerClient(volumeServer, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
  59. stream, err := client.VolumeIncrementalCopy(context.Background(), &volume_server_pb.VolumeIncrementalCopyRequest{
  60. VolumeId: uint32(v.Id),
  61. SinceNs: appendAtNs,
  62. })
  63. if err != nil {
  64. return err
  65. }
  66. for {
  67. resp, recvErr := stream.Recv()
  68. if recvErr != nil {
  69. if recvErr == io.EOF {
  70. break
  71. } else {
  72. return recvErr
  73. }
  74. }
  75. n, writeErr := v.DataBackend.WriteAt(resp.FileContent, writeOffset)
  76. if writeErr != nil {
  77. return writeErr
  78. }
  79. writeOffset += int64(n)
  80. }
  81. return nil
  82. })
  83. if err != nil {
  84. return err
  85. }
  86. // add to needle map
  87. return ScanVolumeFileFrom(v.Version(), v.DataBackend, int64(startFromOffset), &VolumeFileScanner4GenIdx{v: v})
  88. }
  89. func (v *Volume) findLastAppendAtNs() (uint64, error) {
  90. offset, err := v.locateLastAppendEntry()
  91. if err != nil {
  92. return 0, err
  93. }
  94. if offset.IsZero() {
  95. return 0, nil
  96. }
  97. return v.readAppendAtNs(offset)
  98. }
  99. func (v *Volume) locateLastAppendEntry() (Offset, error) {
  100. indexFile, e := os.OpenFile(v.FileName(".idx"), os.O_RDONLY, 0644)
  101. if e != nil {
  102. return Offset{}, fmt.Errorf("cannot read %s: %v", v.FileName(".idx"), e)
  103. }
  104. defer indexFile.Close()
  105. fi, err := indexFile.Stat()
  106. if err != nil {
  107. return Offset{}, fmt.Errorf("file %s stat error: %v", indexFile.Name(), err)
  108. }
  109. fileSize := fi.Size()
  110. if fileSize%NeedleMapEntrySize != 0 {
  111. return Offset{}, fmt.Errorf("unexpected file %s size: %d", indexFile.Name(), fileSize)
  112. }
  113. if fileSize == 0 {
  114. return Offset{}, nil
  115. }
  116. bytes := make([]byte, NeedleMapEntrySize)
  117. n, e := indexFile.ReadAt(bytes, fileSize-NeedleMapEntrySize)
  118. if n != NeedleMapEntrySize {
  119. return Offset{}, fmt.Errorf("file %s read error: %v", indexFile.Name(), e)
  120. }
  121. _, offset, _ := idx.IdxFileEntry(bytes)
  122. return offset, nil
  123. }
  124. func (v *Volume) readAppendAtNs(offset Offset) (uint64, error) {
  125. n, _, bodyLength, err := needle.ReadNeedleHeader(v.DataBackend, v.SuperBlock.Version, offset.ToActualOffset())
  126. if err != nil {
  127. return 0, fmt.Errorf("ReadNeedleHeader %s [%d,%d): %v", v.DataBackend.Name(), offset.ToActualOffset(), offset.ToActualOffset()+NeedleHeaderSize, err)
  128. }
  129. _, err = n.ReadNeedleBody(v.DataBackend, v.SuperBlock.Version, offset.ToActualOffset()+NeedleHeaderSize, bodyLength)
  130. if err != nil {
  131. return 0, fmt.Errorf("ReadNeedleBody offset %d, bodyLength %d: %v", offset.ToActualOffset(), bodyLength, err)
  132. }
  133. return n.AppendAtNs, nil
  134. }
  135. // on server side
  136. func (v *Volume) BinarySearchByAppendAtNs(sinceNs uint64) (offset Offset, isLast bool, err error) {
  137. fileSize := int64(v.IndexFileSize())
  138. if fileSize%NeedleMapEntrySize != 0 {
  139. err = fmt.Errorf("unexpected file %s.idx size: %d", v.IndexFileName(), fileSize)
  140. return
  141. }
  142. entryCount := fileSize / NeedleMapEntrySize
  143. l := int64(0)
  144. h := entryCount
  145. for l < h {
  146. m := (l + h) / 2
  147. if m == entryCount {
  148. return Offset{}, true, nil
  149. }
  150. // read the appendAtNs for entry m
  151. offset, err = v.readOffsetFromIndex(m)
  152. if err != nil {
  153. return
  154. }
  155. mNs, nsReadErr := v.readAppendAtNs(offset)
  156. if nsReadErr != nil {
  157. err = nsReadErr
  158. return
  159. }
  160. // move the boundary
  161. if mNs <= sinceNs {
  162. l = m + 1
  163. } else {
  164. h = m
  165. }
  166. }
  167. if l == entryCount {
  168. return Offset{}, true, nil
  169. }
  170. offset, err = v.readOffsetFromIndex(l)
  171. return offset, false, err
  172. }
  173. // bytes is of size NeedleMapEntrySize
  174. func (v *Volume) readOffsetFromIndex(m int64) (Offset, error) {
  175. v.dataFileAccessLock.RLock()
  176. defer v.dataFileAccessLock.RUnlock()
  177. if v.nm == nil {
  178. return Offset{}, io.EOF
  179. }
  180. _, offset, _, err := v.nm.ReadIndexEntry(m)
  181. return offset, err
  182. }
  183. // generate the volume idx
  184. type VolumeFileScanner4GenIdx struct {
  185. v *Volume
  186. }
  187. func (scanner *VolumeFileScanner4GenIdx) VisitSuperBlock(superBlock super_block.SuperBlock) error {
  188. return nil
  189. }
  190. func (scanner *VolumeFileScanner4GenIdx) ReadNeedleBody() bool {
  191. return false
  192. }
  193. func (scanner *VolumeFileScanner4GenIdx) VisitNeedle(n *needle.Needle, offset int64, needleHeader, needleBody []byte) error {
  194. if n.Size > 0 && n.Size.IsValid() {
  195. return scanner.v.nm.Put(n.Id, ToOffset(offset), n.Size)
  196. }
  197. return scanner.v.nm.Delete(n.Id, ToOffset(offset))
  198. }