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.

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