seaweedfs/weed/server/volume_grpc_erasure_coding.go

414 lines
13 KiB
Go
Raw Normal View History

package weed_server
import (
"context"
"fmt"
2021-08-26 22:18:34 +00:00
"github.com/chrislusf/seaweedfs/weed/storage/volume_info"
2019-05-27 18:59:03 +00:00
"io"
2019-06-03 09:26:31 +00:00
"io/ioutil"
"math"
"os"
2019-06-03 09:26:31 +00:00
"path"
"strings"
2019-05-26 06:23:19 +00:00
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/operation"
"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
"github.com/chrislusf/seaweedfs/weed/storage"
"github.com/chrislusf/seaweedfs/weed/storage/erasure_coding"
"github.com/chrislusf/seaweedfs/weed/storage/needle"
2019-06-20 07:17:11 +00:00
"github.com/chrislusf/seaweedfs/weed/storage/types"
2019-06-03 09:26:31 +00:00
"github.com/chrislusf/seaweedfs/weed/util"
)
/*
Steps to apply erasure coding to .dat .idx files
0. ensure the volume is readonly
2019-12-23 20:48:20 +00:00
1. client call VolumeEcShardsGenerate to generate the .ecx and .ec00 ~ .ec13 files
2020-11-28 11:01:03 +00:00
2. client ask master for possible servers to hold the ec files
3. client call VolumeEcShardsCopy on above target servers to copy ec files from the source server
4. target servers report the new ec files to the master
5. master stores vid -> [14]*DataNode
6. client checks master. If all 14 slices are ready, delete the original .idx, .idx files
*/
2019-12-23 20:48:20 +00:00
// VolumeEcShardsGenerate generates the .ecx and .ec00 ~ .ec13 files
func (vs *VolumeServer) VolumeEcShardsGenerate(ctx context.Context, req *volume_server_pb.VolumeEcShardsGenerateRequest) (*volume_server_pb.VolumeEcShardsGenerateResponse, error) {
2020-06-25 16:43:38 +00:00
glog.V(0).Infof("VolumeEcShardsGenerate: %v", req)
v := vs.store.GetVolume(needle.VolumeId(req.VolumeId))
if v == nil {
return nil, fmt.Errorf("volume %d not found", req.VolumeId)
}
baseFileName := v.DataFileName()
if v.Collection != req.Collection {
return nil, fmt.Errorf("existing collection:%v unexpected input: %v", v.Collection, req.Collection)
}
2021-09-11 09:07:14 +00:00
shouldCleanup := true
defer func() {
if !shouldCleanup {
return
}
for i := 0; i < erasure_coding.TotalShardsCount; i++ {
os.Remove(fmt.Sprintf("%s.ec%2d", baseFileName, i))
}
os.Remove(v.IndexFileName() + ".ecx")
}()
2019-12-23 20:48:20 +00:00
// write .ec00 ~ .ec13 files
if err := erasure_coding.WriteEcFiles(baseFileName); err != nil {
return nil, fmt.Errorf("WriteEcFiles %s: %v", baseFileName, err)
}
// write .ecx file
if err := erasure_coding.WriteSortedFileFromIdx(v.IndexFileName(), ".ecx"); err != nil {
return nil, fmt.Errorf("WriteSortedFileFromIdx %s: %v", v.IndexFileName(), err)
}
2019-12-28 20:44:59 +00:00
// write .vif files
2021-08-26 22:18:34 +00:00
if err := volume_info.SaveVolumeInfo(baseFileName+".vif", &volume_server_pb.VolumeInfo{Version: uint32(v.Version())}); err != nil {
2019-12-28 20:44:59 +00:00
return nil, fmt.Errorf("WriteEcFiles %s: %v", baseFileName, err)
}
2021-09-11 09:07:14 +00:00
shouldCleanup = false
return &volume_server_pb.VolumeEcShardsGenerateResponse{}, nil
}
2019-12-23 20:48:20 +00:00
// VolumeEcShardsRebuild generates the any of the missing .ec00 ~ .ec13 files
2019-06-03 09:26:31 +00:00
func (vs *VolumeServer) VolumeEcShardsRebuild(ctx context.Context, req *volume_server_pb.VolumeEcShardsRebuildRequest) (*volume_server_pb.VolumeEcShardsRebuildResponse, error) {
2020-06-25 16:43:38 +00:00
glog.V(0).Infof("VolumeEcShardsRebuild: %v", req)
2019-06-03 09:26:31 +00:00
baseFileName := erasure_coding.EcShardBaseFileName(req.Collection, int(req.VolumeId))
var rebuiltShardIds []uint32
for _, location := range vs.store.Locations {
if util.FileExists(path.Join(location.IdxDirectory, baseFileName+".ecx")) {
2019-12-23 20:48:20 +00:00
// write .ec00 ~ .ec13 files
dataBaseFileName := path.Join(location.Directory, baseFileName)
if generatedShardIds, err := erasure_coding.RebuildEcFiles(dataBaseFileName); err != nil {
return nil, fmt.Errorf("RebuildEcFiles %s: %v", dataBaseFileName, err)
2019-06-03 09:26:31 +00:00
} else {
rebuiltShardIds = generatedShardIds
}
2019-06-20 05:57:14 +00:00
indexBaseFileName := path.Join(location.IdxDirectory, baseFileName)
if err := erasure_coding.RebuildEcxFile(indexBaseFileName); err != nil {
return nil, fmt.Errorf("RebuildEcxFile %s: %v", dataBaseFileName, err)
2019-06-20 05:57:14 +00:00
}
2019-06-03 09:26:31 +00:00
break
}
}
return &volume_server_pb.VolumeEcShardsRebuildResponse{
RebuiltShardIds: rebuiltShardIds,
}, nil
}
// VolumeEcShardsCopy copy the .ecx and some ec data slices
func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_server_pb.VolumeEcShardsCopyRequest) (*volume_server_pb.VolumeEcShardsCopyResponse, error) {
2020-06-25 16:43:38 +00:00
glog.V(0).Infof("VolumeEcShardsCopy: %v", req)
2021-02-16 10:47:02 +00:00
location := vs.store.FindFreeLocation(types.HardDriveType)
if location == nil {
return nil, fmt.Errorf("no space left")
}
dataBaseFileName := storage.VolumeFileName(location.Directory, req.Collection, int(req.VolumeId))
2020-11-28 10:21:16 +00:00
indexBaseFileName := storage.VolumeFileName(location.IdxDirectory, req.Collection, int(req.VolumeId))
err := operation.WithVolumeServerClient(req.SourceDataNode, vs.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
// copy ec data slices
2019-05-27 18:59:03 +00:00
for _, shardId := range req.ShardIds {
if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, dataBaseFileName, erasure_coding.ToExt(int(shardId)), false, false); err != nil {
return err
}
}
2019-12-23 20:48:20 +00:00
if req.CopyEcxFile {
2019-06-03 09:26:31 +00:00
2019-12-23 20:48:20 +00:00
// copy ecx file
if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, indexBaseFileName, ".ecx", false, false); err != nil {
2019-12-23 20:48:20 +00:00
return err
}
return nil
2019-06-20 05:57:14 +00:00
}
2019-12-23 20:48:20 +00:00
if req.CopyEcjFile {
// copy ecj file
if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, indexBaseFileName, ".ecj", true, true); err != nil {
2019-12-23 20:48:20 +00:00
return err
}
2019-06-03 09:26:31 +00:00
}
2019-12-28 20:44:59 +00:00
if req.CopyVifFile {
// copy vif file
if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, dataBaseFileName, ".vif", false, true); err != nil {
2019-12-28 20:44:59 +00:00
return err
}
}
return nil
})
if err != nil {
return nil, fmt.Errorf("VolumeEcShardsCopy volume %d: %v", req.VolumeId, err)
}
return &volume_server_pb.VolumeEcShardsCopyResponse{}, nil
}
2019-06-03 09:26:31 +00:00
// VolumeEcShardsDelete local delete the .ecx and some ec data slices if not needed
// the shard should not be mounted before calling this.
func (vs *VolumeServer) VolumeEcShardsDelete(ctx context.Context, req *volume_server_pb.VolumeEcShardsDeleteRequest) (*volume_server_pb.VolumeEcShardsDeleteResponse, error) {
bName := erasure_coding.EcShardBaseFileName(req.Collection, int(req.VolumeId))
2019-12-23 20:48:20 +00:00
glog.V(0).Infof("ec volume %d shard delete %v", req.VolumeId, req.ShardIds)
2019-06-04 03:25:02 +00:00
found := false
var indexBaseFilename, dataBaseFilename string
2019-06-04 03:25:02 +00:00
for _, location := range vs.store.Locations {
if util.FileExists(path.Join(location.IdxDirectory, bName+".ecx")) {
2019-06-04 03:25:02 +00:00
found = true
indexBaseFilename = path.Join(location.IdxDirectory, bName)
dataBaseFilename = path.Join(location.Directory, bName)
2019-06-04 03:25:02 +00:00
for _, shardId := range req.ShardIds {
os.Remove(dataBaseFilename + erasure_coding.ToExt(int(shardId)))
2019-06-04 03:25:02 +00:00
}
break
}
}
if !found {
return nil, nil
}
2019-12-23 20:48:20 +00:00
// check whether to delete the .ecx and .ecj file also
2019-06-03 09:26:31 +00:00
hasEcxFile := false
2020-01-03 20:46:39 +00:00
hasIdxFile := false
2019-06-03 09:26:31 +00:00
existingShardCount := 0
2019-06-03 09:26:31 +00:00
for _, location := range vs.store.Locations {
fileInfos, err := ioutil.ReadDir(location.Directory)
if err != nil {
continue
}
if location.IdxDirectory != location.Directory {
idxFileInfos, err := ioutil.ReadDir(location.IdxDirectory)
if err != nil {
continue
}
fileInfos = append(fileInfos, idxFileInfos...)
}
2019-06-03 09:26:31 +00:00
for _, fileInfo := range fileInfos {
2019-12-23 20:48:53 +00:00
if fileInfo.Name() == bName+".ecx" || fileInfo.Name() == bName+".ecj" {
2019-06-03 09:26:31 +00:00
hasEcxFile = true
continue
}
2020-01-03 20:46:39 +00:00
if fileInfo.Name() == bName+".idx" {
hasIdxFile = true
continue
}
2019-12-23 20:48:20 +00:00
if strings.HasPrefix(fileInfo.Name(), bName+".ec") {
2019-06-03 09:26:31 +00:00
existingShardCount++
}
}
}
2019-06-03 09:26:31 +00:00
if hasEcxFile && existingShardCount == 0 {
if err := os.Remove(indexBaseFilename + ".ecx"); err != nil {
2019-06-03 09:26:31 +00:00
return nil, err
}
os.Remove(indexBaseFilename + ".ecj")
}
2020-01-03 20:46:39 +00:00
if !hasIdxFile {
// .vif is used for ec volumes and normal volumes
os.Remove(dataBaseFilename + ".vif")
2020-01-03 20:46:39 +00:00
}
2019-06-03 09:26:31 +00:00
return &volume_server_pb.VolumeEcShardsDeleteResponse{}, nil
}
2019-05-26 06:23:19 +00:00
func (vs *VolumeServer) VolumeEcShardsMount(ctx context.Context, req *volume_server_pb.VolumeEcShardsMountRequest) (*volume_server_pb.VolumeEcShardsMountResponse, error) {
2020-06-25 16:43:38 +00:00
glog.V(0).Infof("VolumeEcShardsMount: %v", req)
2019-05-27 18:59:03 +00:00
for _, shardId := range req.ShardIds {
2019-05-26 06:23:19 +00:00
err := vs.store.MountEcShards(req.Collection, needle.VolumeId(req.VolumeId), erasure_coding.ShardId(shardId))
if err != nil {
glog.Errorf("ec shard mount %v: %v", req, err)
} else {
glog.V(2).Infof("ec shard mount %v", req)
}
if err != nil {
return nil, fmt.Errorf("mount %d.%d: %v", req.VolumeId, shardId, err)
}
}
return &volume_server_pb.VolumeEcShardsMountResponse{}, nil
}
func (vs *VolumeServer) VolumeEcShardsUnmount(ctx context.Context, req *volume_server_pb.VolumeEcShardsUnmountRequest) (*volume_server_pb.VolumeEcShardsUnmountResponse, error) {
2020-06-25 16:43:38 +00:00
glog.V(0).Infof("VolumeEcShardsUnmount: %v", req)
2019-05-27 18:59:03 +00:00
for _, shardId := range req.ShardIds {
2019-05-26 06:23:19 +00:00
err := vs.store.UnmountEcShards(needle.VolumeId(req.VolumeId), erasure_coding.ShardId(shardId))
if err != nil {
glog.Errorf("ec shard unmount %v: %v", req, err)
} else {
glog.V(2).Infof("ec shard unmount %v", req)
}
if err != nil {
return nil, fmt.Errorf("unmount %d.%d: %v", req.VolumeId, shardId, err)
}
}
return &volume_server_pb.VolumeEcShardsUnmountResponse{}, nil
}
2019-05-27 18:59:03 +00:00
func (vs *VolumeServer) VolumeEcShardRead(req *volume_server_pb.VolumeEcShardReadRequest, stream volume_server_pb.VolumeServer_VolumeEcShardReadServer) error {
2019-05-28 04:40:51 +00:00
ecVolume, found := vs.store.FindEcVolume(needle.VolumeId(req.VolumeId))
2019-05-27 18:59:03 +00:00
if !found {
return fmt.Errorf("VolumeEcShardRead not found ec volume id %d", req.VolumeId)
2019-05-27 18:59:03 +00:00
}
2019-05-28 04:40:51 +00:00
ecShard, found := ecVolume.FindEcVolumeShard(erasure_coding.ShardId(req.ShardId))
2019-05-27 18:59:03 +00:00
if !found {
return fmt.Errorf("not found ec shard %d.%d", req.VolumeId, req.ShardId)
}
2019-06-21 08:14:10 +00:00
if req.FileKey != 0 {
_, size, _ := ecVolume.FindNeedleFromEcx(types.Uint64ToNeedleId(req.FileKey))
2020-08-19 00:39:29 +00:00
if size.IsDeleted() {
2019-06-21 08:14:10 +00:00
return stream.Send(&volume_server_pb.VolumeEcShardReadResponse{
IsDeleted: true,
})
}
}
2019-05-29 04:29:07 +00:00
bufSize := req.Size
if bufSize > BufferSizeLimit {
bufSize = BufferSizeLimit
}
buffer := make([]byte, bufSize)
2019-05-27 18:59:03 +00:00
startOffset, bytesToRead := req.Offset, req.Size
for bytesToRead > 0 {
// min of bytesToRead and bufSize
bufferSize := bufSize
if bufferSize > bytesToRead {
bufferSize = bytesToRead
}
bytesread, err := ecShard.ReadAt(buffer[0:bufferSize], startOffset)
2019-05-27 18:59:03 +00:00
// println("read", ecShard.FileName(), "startOffset", startOffset, bytesread, "bytes, with target", bufferSize)
2019-05-29 04:29:07 +00:00
if bytesread > 0 {
2019-05-27 18:59:03 +00:00
2019-05-29 04:29:07 +00:00
if int64(bytesread) > bytesToRead {
bytesread = int(bytesToRead)
}
err = stream.Send(&volume_server_pb.VolumeEcShardReadResponse{
Data: buffer[:bytesread],
})
if err != nil {
// println("sending", bytesread, "bytes err", err.Error())
2019-05-27 18:59:03 +00:00
return err
}
startOffset += int64(bytesread)
2019-05-29 04:29:07 +00:00
bytesToRead -= int64(bytesread)
2019-05-27 18:59:03 +00:00
}
2019-05-29 04:29:07 +00:00
2019-05-27 18:59:03 +00:00
if err != nil {
2019-05-29 04:29:07 +00:00
if err != io.EOF {
return err
}
return nil
2019-05-27 18:59:03 +00:00
}
}
return nil
}
2019-06-20 07:17:11 +00:00
func (vs *VolumeServer) VolumeEcBlobDelete(ctx context.Context, req *volume_server_pb.VolumeEcBlobDeleteRequest) (*volume_server_pb.VolumeEcBlobDeleteResponse, error) {
2020-06-25 16:43:38 +00:00
glog.V(0).Infof("VolumeEcBlobDelete: %v", req)
2019-06-20 07:17:11 +00:00
resp := &volume_server_pb.VolumeEcBlobDeleteResponse{}
for _, location := range vs.store.Locations {
if localEcVolume, found := location.FindEcVolume(needle.VolumeId(req.VolumeId)); found {
_, size, _, err := localEcVolume.LocateEcShardNeedle(types.NeedleId(req.FileKey), needle.Version(req.Version))
if err != nil {
return nil, fmt.Errorf("locate in local ec volume: %v", err)
}
2020-08-19 00:39:29 +00:00
if size.IsDeleted() {
2019-06-20 07:17:11 +00:00
return resp, nil
}
err = localEcVolume.DeleteNeedleFromEcx(types.NeedleId(req.FileKey))
if err != nil {
return nil, err
}
break
}
}
return resp, nil
}
2019-12-23 20:48:20 +00:00
// VolumeEcShardsToVolume generates the .idx, .dat files from .ecx, .ecj and .ec01 ~ .ec14 files
func (vs *VolumeServer) VolumeEcShardsToVolume(ctx context.Context, req *volume_server_pb.VolumeEcShardsToVolumeRequest) (*volume_server_pb.VolumeEcShardsToVolumeResponse, error) {
2020-06-25 16:43:38 +00:00
glog.V(0).Infof("VolumeEcShardsToVolume: %v", req)
2019-12-23 20:48:20 +00:00
v, found := vs.store.FindEcVolume(needle.VolumeId(req.VolumeId))
if !found {
return nil, fmt.Errorf("ec volume %d not found", req.VolumeId)
}
if v.Collection != req.Collection {
return nil, fmt.Errorf("existing collection:%v unexpected input: %v", v.Collection, req.Collection)
}
dataBaseFileName, indexBaseFileName := v.DataBaseFileName(), v.IndexBaseFileName()
2019-12-23 20:48:20 +00:00
// calculate .dat file size
datFileSize, err := erasure_coding.FindDatFileSize(dataBaseFileName, indexBaseFileName)
2019-12-23 20:48:20 +00:00
if err != nil {
return nil, fmt.Errorf("FindDatFileSize %s: %v", dataBaseFileName, err)
2019-12-23 20:48:20 +00:00
}
// write .dat file from .ec00 ~ .ec09 files
if err := erasure_coding.WriteDatFile(dataBaseFileName, datFileSize); err != nil {
return nil, fmt.Errorf("WriteEcFiles %s: %v", dataBaseFileName, err)
2019-12-23 20:48:20 +00:00
}
// write .idx file from .ecx and .ecj files
if err := erasure_coding.WriteIdxFileFromEcIndex(indexBaseFileName); err != nil {
return nil, fmt.Errorf("WriteIdxFileFromEcIndex %s: %v", v.IndexBaseFileName(), err)
2019-12-23 20:48:20 +00:00
}
return &volume_server_pb.VolumeEcShardsToVolumeResponse{}, nil
}