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.

267 lines
8.4 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
5 years ago
  1. package weed_server
  2. import (
  3. "context"
  4. "fmt"
  5. "io"
  6. "math"
  7. "os"
  8. "path"
  9. "time"
  10. "github.com/chrislusf/seaweedfs/weed/glog"
  11. "github.com/chrislusf/seaweedfs/weed/operation"
  12. "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
  13. "github.com/chrislusf/seaweedfs/weed/storage"
  14. "github.com/chrislusf/seaweedfs/weed/storage/erasure_coding"
  15. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  16. "github.com/chrislusf/seaweedfs/weed/util"
  17. )
  18. const BufferSizeLimit = 1024 * 1024 * 2
  19. // VolumeCopy copy the .idx .dat .vif files, and mount the volume
  20. func (vs *VolumeServer) VolumeCopy(ctx context.Context, req *volume_server_pb.VolumeCopyRequest) (*volume_server_pb.VolumeCopyResponse, error) {
  21. v := vs.store.GetVolume(needle.VolumeId(req.VolumeId))
  22. if v != nil {
  23. return nil, fmt.Errorf("volume %d already exists", req.VolumeId)
  24. }
  25. location := vs.store.FindFreeLocation()
  26. if location == nil {
  27. return nil, fmt.Errorf("no space left")
  28. }
  29. // the master will not start compaction for read-only volumes, so it is safe to just copy files directly
  30. // copy .dat and .idx files
  31. // read .idx .dat file size and timestamp
  32. // send .idx file
  33. // send .dat file
  34. // confirm size and timestamp
  35. var volFileInfoResp *volume_server_pb.ReadVolumeFileStatusResponse
  36. var volumeFileName, idxFileName, datFileName string
  37. err := operation.WithVolumeServerClient(req.SourceDataNode, vs.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
  38. var err error
  39. volFileInfoResp, err = client.ReadVolumeFileStatus(ctx,
  40. &volume_server_pb.ReadVolumeFileStatusRequest{
  41. VolumeId: req.VolumeId,
  42. })
  43. if nil != err {
  44. return fmt.Errorf("read volume file status failed, %v", err)
  45. }
  46. volumeFileName = storage.VolumeFileName(location.Directory, volFileInfoResp.Collection, int(req.VolumeId))
  47. // println("source:", volFileInfoResp.String())
  48. // copy ecx file
  49. if err := vs.doCopyFile(ctx, client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.IdxFileSize, volumeFileName, ".idx", false, false); err != nil {
  50. return err
  51. }
  52. if err := vs.doCopyFile(ctx, client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.DatFileSize, volumeFileName, ".dat", false, true); err != nil {
  53. return err
  54. }
  55. if err := vs.doCopyFile(ctx, client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.DatFileSize, volumeFileName, ".vif", false, true); err != nil {
  56. return err
  57. }
  58. return nil
  59. })
  60. idxFileName = volumeFileName + ".idx"
  61. datFileName = volumeFileName + ".dat"
  62. if err != nil && volumeFileName != "" {
  63. os.Remove(idxFileName)
  64. os.Remove(datFileName)
  65. os.Remove(volumeFileName + ".vif")
  66. return nil, err
  67. }
  68. if err = checkCopyFiles(volFileInfoResp, idxFileName, datFileName); err != nil { // added by panyc16
  69. return nil, err
  70. }
  71. // mount the volume
  72. err = vs.store.MountVolume(needle.VolumeId(req.VolumeId))
  73. if err != nil {
  74. return nil, fmt.Errorf("failed to mount volume %d: %v", req.VolumeId, err)
  75. }
  76. return &volume_server_pb.VolumeCopyResponse{
  77. LastAppendAtNs: volFileInfoResp.DatFileTimestampSeconds * uint64(time.Second),
  78. }, err
  79. }
  80. func (vs *VolumeServer) doCopyFile(ctx context.Context, client volume_server_pb.VolumeServerClient, isEcVolume bool, collection string, vid uint32,
  81. compactRevision uint32, stopOffset uint64, baseFileName, ext string, isAppend bool, ignoreSourceFileNotFound bool) error {
  82. copyFileClient, err := client.CopyFile(ctx, &volume_server_pb.CopyFileRequest{
  83. VolumeId: vid,
  84. Ext: ext,
  85. CompactionRevision: compactRevision,
  86. StopOffset: stopOffset,
  87. Collection: collection,
  88. IsEcVolume: isEcVolume,
  89. IgnoreSourceFileNotFound: ignoreSourceFileNotFound,
  90. })
  91. if err != nil {
  92. return fmt.Errorf("failed to start copying volume %d %s file: %v", vid, ext, err)
  93. }
  94. err = writeToFile(copyFileClient, baseFileName+ext, util.NewWriteThrottler(vs.compactionBytePerSecond), isAppend)
  95. if err != nil {
  96. return fmt.Errorf("failed to copy %s file: %v", baseFileName+ext, err)
  97. }
  98. return nil
  99. }
  100. /**
  101. only check the the differ of the file size
  102. todo: maybe should check the received count and deleted count of the volume
  103. */
  104. func checkCopyFiles(originFileInf *volume_server_pb.ReadVolumeFileStatusResponse, idxFileName, datFileName string) error {
  105. stat, err := os.Stat(idxFileName)
  106. if err != nil {
  107. return fmt.Errorf("stat idx file %s failed, %v", idxFileName, err)
  108. }
  109. if originFileInf.IdxFileSize != uint64(stat.Size()) {
  110. return fmt.Errorf("idx file %s size [%v] is not same as origin file size [%v]",
  111. idxFileName, stat.Size(), originFileInf.IdxFileSize)
  112. }
  113. stat, err = os.Stat(datFileName)
  114. if err != nil {
  115. return fmt.Errorf("get dat file info failed, %v", err)
  116. }
  117. if originFileInf.DatFileSize != uint64(stat.Size()) {
  118. return fmt.Errorf("the dat file size [%v] is not same as origin file size [%v]",
  119. stat.Size(), originFileInf.DatFileSize)
  120. }
  121. return nil
  122. }
  123. func writeToFile(client volume_server_pb.VolumeServer_CopyFileClient, fileName string, wt *util.WriteThrottler, isAppend bool) error {
  124. glog.V(4).Infof("writing to %s", fileName)
  125. flags := os.O_WRONLY | os.O_CREATE | os.O_TRUNC
  126. if isAppend {
  127. flags = os.O_WRONLY | os.O_CREATE
  128. }
  129. dst, err := os.OpenFile(fileName, flags, 0644)
  130. if err != nil {
  131. return nil
  132. }
  133. defer dst.Close()
  134. for {
  135. resp, receiveErr := client.Recv()
  136. if receiveErr == io.EOF {
  137. break
  138. }
  139. if receiveErr != nil {
  140. return fmt.Errorf("receiving %s: %v", fileName, receiveErr)
  141. }
  142. dst.Write(resp.FileContent)
  143. wt.MaybeSlowdown(int64(len(resp.FileContent)))
  144. }
  145. return nil
  146. }
  147. func (vs *VolumeServer) ReadVolumeFileStatus(ctx context.Context, req *volume_server_pb.ReadVolumeFileStatusRequest) (*volume_server_pb.ReadVolumeFileStatusResponse, error) {
  148. resp := &volume_server_pb.ReadVolumeFileStatusResponse{}
  149. v := vs.store.GetVolume(needle.VolumeId(req.VolumeId))
  150. if v == nil {
  151. return nil, fmt.Errorf("not found volume id %d", req.VolumeId)
  152. }
  153. resp.VolumeId = req.VolumeId
  154. datSize, idxSize, modTime := v.FileStat()
  155. resp.DatFileSize = datSize
  156. resp.IdxFileSize = idxSize
  157. resp.DatFileTimestampSeconds = uint64(modTime.Unix())
  158. resp.IdxFileTimestampSeconds = uint64(modTime.Unix())
  159. resp.FileCount = v.FileCount()
  160. resp.CompactionRevision = uint32(v.CompactionRevision)
  161. resp.Collection = v.Collection
  162. return resp, nil
  163. }
  164. // CopyFile client pulls the volume related file from the source server.
  165. // if req.CompactionRevision != math.MaxUint32, it ensures the compact revision is as expected
  166. // The copying still stop at req.StopOffset, but you can set it to math.MaxUint64 in order to read all data.
  167. func (vs *VolumeServer) CopyFile(req *volume_server_pb.CopyFileRequest, stream volume_server_pb.VolumeServer_CopyFileServer) error {
  168. var fileName string
  169. if !req.IsEcVolume {
  170. v := vs.store.GetVolume(needle.VolumeId(req.VolumeId))
  171. if v == nil {
  172. return fmt.Errorf("not found volume id %d", req.VolumeId)
  173. }
  174. if uint32(v.CompactionRevision) != req.CompactionRevision && req.CompactionRevision != math.MaxUint32 {
  175. return fmt.Errorf("volume %d is compacted", req.VolumeId)
  176. }
  177. fileName = v.FileName() + req.Ext
  178. } else {
  179. baseFileName := erasure_coding.EcShardBaseFileName(req.Collection, int(req.VolumeId)) + req.Ext
  180. for _, location := range vs.store.Locations {
  181. tName := path.Join(location.Directory, baseFileName)
  182. if util.FileExists(tName) {
  183. fileName = tName
  184. }
  185. }
  186. if fileName == "" {
  187. if req.IgnoreSourceFileNotFound {
  188. return nil
  189. }
  190. return fmt.Errorf("CopyFile not found ec volume id %d", req.VolumeId)
  191. }
  192. }
  193. bytesToRead := int64(req.StopOffset)
  194. file, err := os.Open(fileName)
  195. if err != nil {
  196. if req.IgnoreSourceFileNotFound && err == os.ErrNotExist {
  197. return nil
  198. }
  199. return err
  200. }
  201. defer file.Close()
  202. buffer := make([]byte, BufferSizeLimit)
  203. for bytesToRead > 0 {
  204. bytesread, err := file.Read(buffer)
  205. // println(fileName, "read", bytesread, "bytes, with target", bytesToRead)
  206. if err != nil {
  207. if err != io.EOF {
  208. return err
  209. }
  210. // println(fileName, "read", bytesread, "bytes, with target", bytesToRead, "err", err.Error())
  211. break
  212. }
  213. if int64(bytesread) > bytesToRead {
  214. bytesread = int(bytesToRead)
  215. }
  216. err = stream.Send(&volume_server_pb.CopyFileResponse{
  217. FileContent: buffer[:bytesread],
  218. })
  219. if err != nil {
  220. // println("sending", bytesread, "bytes err", err.Error())
  221. return err
  222. }
  223. bytesToRead -= int64(bytesread)
  224. }
  225. return nil
  226. }