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.

256 lines
6.7 KiB

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