aboutsummaryrefslogtreecommitdiff
path: root/weed/shell/command_ec_encode.go
diff options
context:
space:
mode:
Diffstat (limited to 'weed/shell/command_ec_encode.go')
-rw-r--r--weed/shell/command_ec_encode.go161
1 files changed, 19 insertions, 142 deletions
diff --git a/weed/shell/command_ec_encode.go b/weed/shell/command_ec_encode.go
index 14d1ae96b..94265a874 100644
--- a/weed/shell/command_ec_encode.go
+++ b/weed/shell/command_ec_encode.go
@@ -5,11 +5,9 @@ import (
"flag"
"fmt"
"io"
- "sort"
"sync"
"time"
- "github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/operation"
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
@@ -73,6 +71,7 @@ func (c *commandEcEncode) Do(args []string, commandEnv *commandEnv, writer io.Wr
if err != nil {
return err
}
+ fmt.Printf("ec encode volumes: %v\n", volumeIds)
for _, vid := range volumeIds {
if err = doEcEncode(ctx, commandEnv, *collection, vid); err != nil {
return err
@@ -96,9 +95,9 @@ func doEcEncode(ctx context.Context, commandEnv *commandEnv, collection string,
}
// balance the ec shards to current cluster
- err = balanceEcShards(ctx, commandEnv, vid, collection, locations)
+ err = spreadEcShards(ctx, commandEnv, vid, collection, locations)
if err != nil {
- return fmt.Errorf("balance ec shards for volume %d on %s: %v", vid, locations[0].Url, err)
+ return fmt.Errorf("spread ec shards for volume %d to %s: %v", vid, locations[0].Url, err)
}
return nil
@@ -118,7 +117,7 @@ func generateEcShards(ctx context.Context, grpcDialOption grpc.DialOption, volum
}
-func balanceEcShards(ctx context.Context, commandEnv *commandEnv, volumeId needle.VolumeId, collection string, existingLocations []wdclient.Location) (err error) {
+func spreadEcShards(ctx context.Context, commandEnv *commandEnv, volumeId needle.VolumeId, collection string, existingLocations []wdclient.Location) (err error) {
allEcNodes, totalFreeEcSlots, err := collectEcNodes(ctx, commandEnv)
if err != nil {
@@ -139,13 +138,19 @@ func balanceEcShards(ctx context.Context, commandEnv *commandEnv, volumeId needl
// ask the data nodes to copy from the source volume server
copiedShardIds, err := parallelCopyEcShardsFromSource(ctx, commandEnv.option.GrpcDialOption, allocatedDataNodes, allocated, volumeId, collection, existingLocations[0])
if err != nil {
- return nil
+ return err
+ }
+
+ // unmount the to be deleted shards
+ err = unmountEcShards(ctx, commandEnv.option.GrpcDialOption, volumeId, existingLocations[0].Url, copiedShardIds)
+ if err != nil {
+ return err
}
// ask the source volume server to clean up copied ec shards
- err = sourceServerDeleteEcShards(ctx, commandEnv.option.GrpcDialOption, volumeId, existingLocations[0].Url, copiedShardIds)
+ err = sourceServerDeleteEcShards(ctx, commandEnv.option.GrpcDialOption, collection, volumeId, existingLocations[0].Url, copiedShardIds)
if err != nil {
- return fmt.Errorf("sourceServerDeleteEcShards %s %d.%v: %v", existingLocations[0], volumeId, copiedShardIds, err)
+ return fmt.Errorf("source delete copied ecShards %s %d.%v: %v", existingLocations[0].Url, volumeId, copiedShardIds, err)
}
// ask the source volume server to delete the original volume
@@ -176,7 +181,7 @@ func parallelCopyEcShardsFromSource(ctx context.Context, grpcDialOption grpc.Dia
wg.Add(1)
go func(server *EcNode, startFromShardId uint32, shardCount int) {
defer wg.Done()
- copiedShardIds, copyErr := oneServerCopyEcShardsFromSource(ctx, grpcDialOption, server,
+ copiedShardIds, copyErr := oneServerCopyAndMountEcShardsFromSource(ctx, grpcDialOption, server,
startFromShardId, shardCount, volumeId, collection, existingLocation.Url)
if copyErr != nil {
err = copyErr
@@ -201,81 +206,12 @@ func parallelCopyEcShardsFromSource(ctx context.Context, grpcDialOption grpc.Dia
return
}
-func oneServerCopyEcShardsFromSource(ctx context.Context, grpcDialOption grpc.DialOption,
- targetServer *EcNode, startFromShardId uint32, shardCount int,
- volumeId needle.VolumeId, collection string, existingLocation string) (copiedShardIds []uint32, err error) {
-
- var shardIdsToCopy []uint32
- for shardId := startFromShardId; shardId < startFromShardId+uint32(shardCount); shardId++ {
- fmt.Printf("allocate %d.%d %s => %s\n", volumeId, shardId, existingLocation, targetServer.info.Id)
- shardIdsToCopy = append(shardIdsToCopy, shardId)
- }
-
- err = operation.WithVolumeServerClient(targetServer.info.Id, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
-
- if targetServer.info.Id != existingLocation {
-
- _, copyErr := volumeServerClient.VolumeEcShardsCopy(ctx, &volume_server_pb.VolumeEcShardsCopyRequest{
- VolumeId: uint32(volumeId),
- Collection: collection,
- ShardIds: shardIdsToCopy,
- SourceDataNode: existingLocation,
- })
- if copyErr != nil {
- return copyErr
- }
- }
-
- _, mountErr := volumeServerClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{
- VolumeId: uint32(volumeId),
- Collection: collection,
- ShardIds: shardIdsToCopy,
- })
- if mountErr != nil {
- return mountErr
- }
-
- if targetServer.info.Id != existingLocation {
- copiedShardIds = shardIdsToCopy
- glog.V(0).Infof("%s ec volume %d deletes shards %+v", existingLocation, volumeId, copiedShardIds)
- }
-
- return nil
- })
-
- if err != nil {
- return
- }
-
- return
-}
-
-func sourceServerDeleteEcShards(ctx context.Context, grpcDialOption grpc.DialOption,
- volumeId needle.VolumeId, sourceLocation string, toBeDeletedShardIds []uint32) error {
-
- shouldDeleteEcx := len(toBeDeletedShardIds) == erasure_coding.TotalShardsCount
-
- return operation.WithVolumeServerClient(sourceLocation, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
- _, deleteErr := volumeServerClient.VolumeEcShardsDelete(ctx, &volume_server_pb.VolumeEcShardsDeleteRequest{
- VolumeId: uint32(volumeId),
- ShardIds: toBeDeletedShardIds,
- ShouldDeleteEcx: shouldDeleteEcx,
- })
- return deleteErr
- })
-
-}
-
func balancedEcDistribution(servers []*EcNode) (allocated []int) {
- freeSlots := make([]int, len(servers))
allocated = make([]int, len(servers))
- for i, server := range servers {
- freeSlots[i] = countFreeShardSlots(server.info)
- }
allocatedCount := 0
for allocatedCount < erasure_coding.TotalShardsCount {
- for i, _ := range servers {
- if freeSlots[i]-allocated[i] > 0 {
+ for i, server := range servers {
+ if server.freeEcSlot-allocated[i] > 0 {
allocated[i] += 1
allocatedCount += 1
}
@@ -288,67 +224,6 @@ func balancedEcDistribution(servers []*EcNode) (allocated []int) {
return allocated
}
-func eachDataNode(topo *master_pb.TopologyInfo, fn func(*master_pb.DataNodeInfo)) {
- for _, dc := range topo.DataCenterInfos {
- for _, rack := range dc.RackInfos {
- for _, dn := range rack.DataNodeInfos {
- fn(dn)
- }
- }
- }
-}
-
-func countShards(ecShardInfos []*master_pb.VolumeEcShardInformationMessage) (count int) {
- for _, ecShardInfo := range ecShardInfos {
- shardBits := erasure_coding.ShardBits(ecShardInfo.EcIndexBits)
- count += shardBits.ShardIdCount()
- }
- return
-}
-
-func countFreeShardSlots(dn *master_pb.DataNodeInfo) (count int) {
- return int(dn.FreeVolumeCount)*10 - countShards(dn.EcShardInfos)
-}
-
-type EcNode struct {
- info *master_pb.DataNodeInfo
- freeEcSlot int
-}
-
-func sortEcNodes(ecNodes []*EcNode) {
- sort.Slice(ecNodes, func(i, j int) bool {
- return ecNodes[i].freeEcSlot > ecNodes[j].freeEcSlot
- })
-}
-
-func collectEcNodes(ctx context.Context, commandEnv *commandEnv) (ecNodes []*EcNode, totalFreeEcSlots int, err error) {
-
- // list all possible locations
- var resp *master_pb.VolumeListResponse
- err = commandEnv.masterClient.WithClient(ctx, func(client master_pb.SeaweedClient) error {
- resp, err = client.VolumeList(ctx, &master_pb.VolumeListRequest{})
- return err
- })
- if err != nil {
- return nil, 0, err
- }
-
- // find out all volume servers with one slot left.
- eachDataNode(resp.TopologyInfo, func(dn *master_pb.DataNodeInfo) {
- if freeEcSlots := countFreeShardSlots(dn); freeEcSlots > 0 {
- ecNodes = append(ecNodes, &EcNode{
- info: dn,
- freeEcSlot: int(freeEcSlots),
- })
- totalFreeEcSlots += freeEcSlots
- }
- })
-
- sortEcNodes(ecNodes)
-
- return
-}
-
func collectVolumeIdsForEcEncode(ctx context.Context, commandEnv *commandEnv, selectedCollection string, quietPeriod time.Duration) (vids []needle.VolumeId, err error) {
var resp *master_pb.VolumeListResponse
@@ -360,9 +235,11 @@ func collectVolumeIdsForEcEncode(ctx context.Context, commandEnv *commandEnv, se
return
}
- quietSeconds := int64((quietPeriod * time.Second).Seconds())
+ quietSeconds := int64(quietPeriod / time.Second)
nowUnixSeconds := time.Now().Unix()
+ fmt.Printf("ec encode volumes quiet for: %d seconds\n", quietSeconds)
+
vidMap := make(map[uint32]bool)
for _, dc := range resp.TopologyInfo.DataCenterInfos {
for _, r := range dc.RackInfos {
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-21 14:28:46 +0000