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.

469 lines
15 KiB

4 years ago
2 years ago
2 years ago
6 years ago
6 years ago
4 years ago
4 years ago
  1. package weed_server
  2. import (
  3. "context"
  4. "fmt"
  5. "io"
  6. "math"
  7. "os"
  8. "path"
  9. "strings"
  10. "time"
  11. "github.com/seaweedfs/seaweedfs/weed/glog"
  12. "github.com/seaweedfs/seaweedfs/weed/operation"
  13. "github.com/seaweedfs/seaweedfs/weed/pb"
  14. "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
  15. "github.com/seaweedfs/seaweedfs/weed/storage"
  16. "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
  17. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  18. "github.com/seaweedfs/seaweedfs/weed/storage/types"
  19. "github.com/seaweedfs/seaweedfs/weed/storage/volume_info"
  20. "github.com/seaweedfs/seaweedfs/weed/util"
  21. )
  22. /*
  23. Steps to apply erasure coding to .dat .idx files
  24. 0. ensure the volume is readonly
  25. 1. client call VolumeEcShardsGenerate to generate the .ecx and .ec00 ~ .ec13 files
  26. 2. client ask master for possible servers to hold the ec files
  27. 3. client call VolumeEcShardsCopy on above target servers to copy ec files from the source server
  28. 4. target servers report the new ec files to the master
  29. 5. master stores vid -> [14]*DataNode
  30. 6. client checks master. If all 14 slices are ready, delete the original .idx, .idx files
  31. */
  32. // VolumeEcShardsGenerate generates the .ecx and .ec00 ~ .ec13 files
  33. func (vs *VolumeServer) VolumeEcShardsGenerate(ctx context.Context, req *volume_server_pb.VolumeEcShardsGenerateRequest) (*volume_server_pb.VolumeEcShardsGenerateResponse, error) {
  34. glog.V(0).Infof("VolumeEcShardsGenerate: %v", req)
  35. v := vs.store.GetVolume(needle.VolumeId(req.VolumeId))
  36. if v == nil {
  37. return nil, fmt.Errorf("volume %d not found", req.VolumeId)
  38. }
  39. baseFileName := v.DataFileName()
  40. if v.Collection != req.Collection {
  41. return nil, fmt.Errorf("existing collection:%v unexpected input: %v", v.Collection, req.Collection)
  42. }
  43. shouldCleanup := true
  44. defer func() {
  45. if !shouldCleanup {
  46. return
  47. }
  48. for i := 0; i < erasure_coding.TotalShardsCount; i++ {
  49. os.Remove(fmt.Sprintf("%s.ec%2d", baseFileName, i))
  50. }
  51. os.Remove(v.IndexFileName() + ".ecx")
  52. }()
  53. // write .ec00 ~ .ec13 files
  54. if err := erasure_coding.WriteEcFiles(baseFileName); err != nil {
  55. return nil, fmt.Errorf("WriteEcFiles %s: %v", baseFileName, err)
  56. }
  57. // write .ecx file
  58. if err := erasure_coding.WriteSortedFileFromIdx(v.IndexFileName(), ".ecx"); err != nil {
  59. return nil, fmt.Errorf("WriteSortedFileFromIdx %s: %v", v.IndexFileName(), err)
  60. }
  61. // write .vif files
  62. var expireAtSec uint64
  63. if v.Ttl != nil {
  64. ttlSecond := v.Ttl.ToSeconds()
  65. if ttlSecond > 0 {
  66. expireAtSec = uint64(time.Now().Unix()) + ttlSecond //calculated expiration time
  67. }
  68. }
  69. volumeInfo := &volume_server_pb.VolumeInfo{Version: uint32(v.Version())}
  70. volumeInfo.ExpireAtSec = expireAtSec
  71. datSize, _, _ := v.FileStat()
  72. volumeInfo.DatFileSize = int64(datSize)
  73. if err := volume_info.SaveVolumeInfo(baseFileName+".vif", volumeInfo); err != nil {
  74. return nil, fmt.Errorf("SaveVolumeInfo %s: %v", baseFileName, err)
  75. }
  76. shouldCleanup = false
  77. return &volume_server_pb.VolumeEcShardsGenerateResponse{}, nil
  78. }
  79. // VolumeEcShardsRebuild generates the any of the missing .ec00 ~ .ec13 files
  80. func (vs *VolumeServer) VolumeEcShardsRebuild(ctx context.Context, req *volume_server_pb.VolumeEcShardsRebuildRequest) (*volume_server_pb.VolumeEcShardsRebuildResponse, error) {
  81. glog.V(0).Infof("VolumeEcShardsRebuild: %v", req)
  82. baseFileName := erasure_coding.EcShardBaseFileName(req.Collection, int(req.VolumeId))
  83. var rebuiltShardIds []uint32
  84. for _, location := range vs.store.Locations {
  85. _, _, existingShardCount, err := checkEcVolumeStatus(baseFileName, location)
  86. if err != nil {
  87. return nil, err
  88. }
  89. if existingShardCount == 0 {
  90. continue
  91. }
  92. if util.FileExists(path.Join(location.IdxDirectory, baseFileName+".ecx")) {
  93. // write .ec00 ~ .ec13 files
  94. dataBaseFileName := path.Join(location.Directory, baseFileName)
  95. if generatedShardIds, err := erasure_coding.RebuildEcFiles(dataBaseFileName); err != nil {
  96. return nil, fmt.Errorf("RebuildEcFiles %s: %v", dataBaseFileName, err)
  97. } else {
  98. rebuiltShardIds = generatedShardIds
  99. }
  100. indexBaseFileName := path.Join(location.IdxDirectory, baseFileName)
  101. if err := erasure_coding.RebuildEcxFile(indexBaseFileName); err != nil {
  102. return nil, fmt.Errorf("RebuildEcxFile %s: %v", dataBaseFileName, err)
  103. }
  104. break
  105. }
  106. }
  107. return &volume_server_pb.VolumeEcShardsRebuildResponse{
  108. RebuiltShardIds: rebuiltShardIds,
  109. }, nil
  110. }
  111. // VolumeEcShardsCopy copy the .ecx and some ec data slices
  112. func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_server_pb.VolumeEcShardsCopyRequest) (*volume_server_pb.VolumeEcShardsCopyResponse, error) {
  113. glog.V(0).Infof("VolumeEcShardsCopy: %v", req)
  114. var location *storage.DiskLocation
  115. if req.CopyEcxFile {
  116. location = vs.store.FindFreeLocation(func(location *storage.DiskLocation) bool {
  117. return location.DiskType == types.HardDriveType
  118. })
  119. } else {
  120. location = vs.store.FindFreeLocation(func(location *storage.DiskLocation) bool {
  121. //(location.FindEcVolume) This method is error, will cause location is nil, redundant judgment
  122. // _, found := location.FindEcVolume(needle.VolumeId(req.VolumeId))
  123. // return found
  124. return true
  125. })
  126. }
  127. if location == nil {
  128. return nil, fmt.Errorf("no space left")
  129. }
  130. dataBaseFileName := storage.VolumeFileName(location.Directory, req.Collection, int(req.VolumeId))
  131. indexBaseFileName := storage.VolumeFileName(location.IdxDirectory, req.Collection, int(req.VolumeId))
  132. err := operation.WithVolumeServerClient(true, pb.ServerAddress(req.SourceDataNode), vs.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
  133. // copy ec data slices
  134. for _, shardId := range req.ShardIds {
  135. if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, dataBaseFileName, erasure_coding.ToExt(int(shardId)), false, false, nil); err != nil {
  136. return err
  137. }
  138. }
  139. if req.CopyEcxFile {
  140. // copy ecx file
  141. if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, indexBaseFileName, ".ecx", false, false, nil); err != nil {
  142. return err
  143. }
  144. }
  145. if req.CopyEcjFile {
  146. // copy ecj file
  147. if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, indexBaseFileName, ".ecj", true, true, nil); err != nil {
  148. return err
  149. }
  150. }
  151. if req.CopyVifFile {
  152. // copy vif file
  153. if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, dataBaseFileName, ".vif", false, true, nil); err != nil {
  154. return err
  155. }
  156. }
  157. return nil
  158. })
  159. if err != nil {
  160. return nil, fmt.Errorf("VolumeEcShardsCopy volume %d: %v", req.VolumeId, err)
  161. }
  162. return &volume_server_pb.VolumeEcShardsCopyResponse{}, nil
  163. }
  164. // VolumeEcShardsDelete local delete the .ecx and some ec data slices if not needed
  165. // the shard should not be mounted before calling this.
  166. func (vs *VolumeServer) VolumeEcShardsDelete(ctx context.Context, req *volume_server_pb.VolumeEcShardsDeleteRequest) (*volume_server_pb.VolumeEcShardsDeleteResponse, error) {
  167. bName := erasure_coding.EcShardBaseFileName(req.Collection, int(req.VolumeId))
  168. glog.V(0).Infof("ec volume %s shard delete %v", bName, req.ShardIds)
  169. for _, location := range vs.store.Locations {
  170. if err := deleteEcShardIdsForEachLocation(bName, location, req.ShardIds); err != nil {
  171. glog.Errorf("deleteEcShards from %s %s.%v: %v", location.Directory, bName, req.ShardIds, err)
  172. return nil, err
  173. }
  174. }
  175. return &volume_server_pb.VolumeEcShardsDeleteResponse{}, nil
  176. }
  177. func deleteEcShardIdsForEachLocation(bName string, location *storage.DiskLocation, shardIds []uint32) error {
  178. found := false
  179. indexBaseFilename := path.Join(location.IdxDirectory, bName)
  180. dataBaseFilename := path.Join(location.Directory, bName)
  181. if util.FileExists(path.Join(location.IdxDirectory, bName+".ecx")) {
  182. for _, shardId := range shardIds {
  183. shardFileName := dataBaseFilename + erasure_coding.ToExt(int(shardId))
  184. if util.FileExists(shardFileName) {
  185. found = true
  186. os.Remove(shardFileName)
  187. }
  188. }
  189. }
  190. if !found {
  191. return nil
  192. }
  193. hasEcxFile, hasIdxFile, existingShardCount, err := checkEcVolumeStatus(bName, location)
  194. if err != nil {
  195. return err
  196. }
  197. if hasEcxFile && existingShardCount == 0 {
  198. if err := os.Remove(indexBaseFilename + ".ecx"); err != nil {
  199. return err
  200. }
  201. os.Remove(indexBaseFilename + ".ecj")
  202. if !hasIdxFile {
  203. // .vif is used for ec volumes and normal volumes
  204. os.Remove(dataBaseFilename + ".vif")
  205. }
  206. }
  207. return nil
  208. }
  209. func checkEcVolumeStatus(bName string, location *storage.DiskLocation) (hasEcxFile bool, hasIdxFile bool, existingShardCount int, err error) {
  210. // check whether to delete the .ecx and .ecj file also
  211. fileInfos, err := os.ReadDir(location.Directory)
  212. if err != nil {
  213. return false, false, 0, err
  214. }
  215. if location.IdxDirectory != location.Directory {
  216. idxFileInfos, err := os.ReadDir(location.IdxDirectory)
  217. if err != nil {
  218. return false, false, 0, err
  219. }
  220. fileInfos = append(fileInfos, idxFileInfos...)
  221. }
  222. for _, fileInfo := range fileInfos {
  223. if fileInfo.Name() == bName+".ecx" || fileInfo.Name() == bName+".ecj" {
  224. hasEcxFile = true
  225. continue
  226. }
  227. if fileInfo.Name() == bName+".idx" {
  228. hasIdxFile = true
  229. continue
  230. }
  231. if strings.HasPrefix(fileInfo.Name(), bName+".ec") {
  232. existingShardCount++
  233. }
  234. }
  235. return hasEcxFile, hasIdxFile, existingShardCount, nil
  236. }
  237. func (vs *VolumeServer) VolumeEcShardsMount(ctx context.Context, req *volume_server_pb.VolumeEcShardsMountRequest) (*volume_server_pb.VolumeEcShardsMountResponse, error) {
  238. glog.V(0).Infof("VolumeEcShardsMount: %v", req)
  239. for _, shardId := range req.ShardIds {
  240. err := vs.store.MountEcShards(req.Collection, needle.VolumeId(req.VolumeId), erasure_coding.ShardId(shardId))
  241. if err != nil {
  242. glog.Errorf("ec shard mount %v: %v", req, err)
  243. } else {
  244. glog.V(2).Infof("ec shard mount %v", req)
  245. }
  246. if err != nil {
  247. return nil, fmt.Errorf("mount %d.%d: %v", req.VolumeId, shardId, err)
  248. }
  249. }
  250. return &volume_server_pb.VolumeEcShardsMountResponse{}, nil
  251. }
  252. func (vs *VolumeServer) VolumeEcShardsUnmount(ctx context.Context, req *volume_server_pb.VolumeEcShardsUnmountRequest) (*volume_server_pb.VolumeEcShardsUnmountResponse, error) {
  253. glog.V(0).Infof("VolumeEcShardsUnmount: %v", req)
  254. for _, shardId := range req.ShardIds {
  255. err := vs.store.UnmountEcShards(needle.VolumeId(req.VolumeId), erasure_coding.ShardId(shardId))
  256. if err != nil {
  257. glog.Errorf("ec shard unmount %v: %v", req, err)
  258. } else {
  259. glog.V(2).Infof("ec shard unmount %v", req)
  260. }
  261. if err != nil {
  262. return nil, fmt.Errorf("unmount %d.%d: %v", req.VolumeId, shardId, err)
  263. }
  264. }
  265. return &volume_server_pb.VolumeEcShardsUnmountResponse{}, nil
  266. }
  267. func (vs *VolumeServer) VolumeEcShardRead(req *volume_server_pb.VolumeEcShardReadRequest, stream volume_server_pb.VolumeServer_VolumeEcShardReadServer) error {
  268. ecVolume, found := vs.store.FindEcVolume(needle.VolumeId(req.VolumeId))
  269. if !found {
  270. return fmt.Errorf("VolumeEcShardRead not found ec volume id %d", req.VolumeId)
  271. }
  272. ecShard, found := ecVolume.FindEcVolumeShard(erasure_coding.ShardId(req.ShardId))
  273. if !found {
  274. return fmt.Errorf("not found ec shard %d.%d", req.VolumeId, req.ShardId)
  275. }
  276. if req.FileKey != 0 {
  277. _, size, _ := ecVolume.FindNeedleFromEcx(types.Uint64ToNeedleId(req.FileKey))
  278. if size.IsDeleted() {
  279. return stream.Send(&volume_server_pb.VolumeEcShardReadResponse{
  280. IsDeleted: true,
  281. })
  282. }
  283. }
  284. bufSize := req.Size
  285. if bufSize > BufferSizeLimit {
  286. bufSize = BufferSizeLimit
  287. }
  288. buffer := make([]byte, bufSize)
  289. startOffset, bytesToRead := req.Offset, req.Size
  290. for bytesToRead > 0 {
  291. // min of bytesToRead and bufSize
  292. bufferSize := bufSize
  293. if bufferSize > bytesToRead {
  294. bufferSize = bytesToRead
  295. }
  296. bytesread, err := ecShard.ReadAt(buffer[0:bufferSize], startOffset)
  297. // println("read", ecShard.FileName(), "startOffset", startOffset, bytesread, "bytes, with target", bufferSize)
  298. if bytesread > 0 {
  299. if int64(bytesread) > bytesToRead {
  300. bytesread = int(bytesToRead)
  301. }
  302. err = stream.Send(&volume_server_pb.VolumeEcShardReadResponse{
  303. Data: buffer[:bytesread],
  304. })
  305. if err != nil {
  306. // println("sending", bytesread, "bytes err", err.Error())
  307. return err
  308. }
  309. startOffset += int64(bytesread)
  310. bytesToRead -= int64(bytesread)
  311. }
  312. if err != nil {
  313. if err != io.EOF {
  314. return err
  315. }
  316. return nil
  317. }
  318. }
  319. return nil
  320. }
  321. func (vs *VolumeServer) VolumeEcBlobDelete(ctx context.Context, req *volume_server_pb.VolumeEcBlobDeleteRequest) (*volume_server_pb.VolumeEcBlobDeleteResponse, error) {
  322. glog.V(0).Infof("VolumeEcBlobDelete: %v", req)
  323. resp := &volume_server_pb.VolumeEcBlobDeleteResponse{}
  324. for _, location := range vs.store.Locations {
  325. if localEcVolume, found := location.FindEcVolume(needle.VolumeId(req.VolumeId)); found {
  326. _, size, _, err := localEcVolume.LocateEcShardNeedle(types.NeedleId(req.FileKey), needle.Version(req.Version))
  327. if err != nil {
  328. return nil, fmt.Errorf("locate in local ec volume: %v", err)
  329. }
  330. if size.IsDeleted() {
  331. return resp, nil
  332. }
  333. err = localEcVolume.DeleteNeedleFromEcx(types.NeedleId(req.FileKey))
  334. if err != nil {
  335. return nil, err
  336. }
  337. break
  338. }
  339. }
  340. return resp, nil
  341. }
  342. // VolumeEcShardsToVolume generates the .idx, .dat files from .ecx, .ecj and .ec01 ~ .ec14 files
  343. func (vs *VolumeServer) VolumeEcShardsToVolume(ctx context.Context, req *volume_server_pb.VolumeEcShardsToVolumeRequest) (*volume_server_pb.VolumeEcShardsToVolumeResponse, error) {
  344. glog.V(0).Infof("VolumeEcShardsToVolume: %v", req)
  345. // collect .ec00 ~ .ec09 files
  346. shardFileNames := make([]string, erasure_coding.DataShardsCount)
  347. v, found := vs.store.CollectEcShards(needle.VolumeId(req.VolumeId), shardFileNames)
  348. if !found {
  349. return nil, fmt.Errorf("ec volume %d not found", req.VolumeId)
  350. }
  351. if v.Collection != req.Collection {
  352. return nil, fmt.Errorf("existing collection:%v unexpected input: %v", v.Collection, req.Collection)
  353. }
  354. for shardId := 0; shardId < erasure_coding.DataShardsCount; shardId++ {
  355. if shardFileNames[shardId] == "" {
  356. return nil, fmt.Errorf("ec volume %d missing shard %d", req.VolumeId, shardId)
  357. }
  358. }
  359. dataBaseFileName, indexBaseFileName := v.DataBaseFileName(), v.IndexBaseFileName()
  360. // calculate .dat file size
  361. datFileSize, err := erasure_coding.FindDatFileSize(dataBaseFileName, indexBaseFileName)
  362. if err != nil {
  363. return nil, fmt.Errorf("FindDatFileSize %s: %v", dataBaseFileName, err)
  364. }
  365. // write .dat file from .ec00 ~ .ec09 files
  366. if err := erasure_coding.WriteDatFile(dataBaseFileName, datFileSize, shardFileNames); err != nil {
  367. return nil, fmt.Errorf("WriteDatFile %s: %v", dataBaseFileName, err)
  368. }
  369. // write .idx file from .ecx and .ecj files
  370. if err := erasure_coding.WriteIdxFileFromEcIndex(indexBaseFileName); err != nil {
  371. return nil, fmt.Errorf("WriteIdxFileFromEcIndex %s: %v", v.IndexBaseFileName(), err)
  372. }
  373. return &volume_server_pb.VolumeEcShardsToVolumeResponse{}, nil
  374. }