Files
seaweedfs/weed/shell/command_ec_balance.go
T

108 lines
4.0 KiB
Go

package shell
import (
"flag"
"fmt"
"io"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
"github.com/seaweedfs/seaweedfs/weed/storage/types"
"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
)
func init() {
Commands = append(Commands, &commandEcBalance{})
}
type commandEcBalance struct {
}
func (c *commandEcBalance) Name() string {
return "ec.balance"
}
func (c *commandEcBalance) Help() string {
return `balance all ec shards among all racks and volume servers
ec.balance [-c EACH_COLLECTION|<collection_name>] [-apply] [-dataCenter <data_center>] [-shardReplicaPlacement <replica_placement>] [-diskType <disk_type>] [-volumeIds <id>[,<id>...]]
Options:
-diskType: the disk type for EC shards (hdd, ssd, or empty for default hdd)
-volumeIds: only balance these ec volume ids, e.g. -volumeIds 123,456. The plan is
built as if no other ec volume existed, so nothing else is moved or deduplicated.
Ids without any ec shard in the selected collection, dataCenter and disk type are
an error.
Algorithm:
` + ecBalanceAlgorithmDescription
}
func (c *commandEcBalance) HasTag(CommandTag) bool {
return false
}
func (c *commandEcBalance) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
balanceCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
collection := balanceCommand.String("collection", string(wildcard.CollectionFilterEach), "comma-separated collection names, wildcards, or regex patterns, \""+CollectionDefault+"\" for the collection with no name, or \""+string(wildcard.CollectionFilterEach)+"\" for each collection")
dc := balanceCommand.String("dataCenter", "", "only apply the balancing for this dataCenter")
shardReplicaPlacement := balanceCommand.String("shardReplicaPlacement", "", "replica placement for EC shards, or master default if empty")
diskTypeStr := balanceCommand.String("diskType", "", "the disk type for EC shards (hdd, ssd, or empty for default hdd)")
volumeIdsStr := balanceCommand.String("volumeIds", "", "optional comma-separated list of ec volume ids to balance; defaults to all")
maxParallelization := balanceCommand.Int("maxParallelization", DefaultMaxParallelization, "run up to X tasks in parallel, whenever possible")
ioBytePerSecond := balanceCommand.Int64("ioBytePerSecond", 0, "limit the speed of each shard copy; 0 falls back to the volume server's maintenance rate")
applyBalancing := balanceCommand.Bool("apply", false, "apply the balancing plan")
// TODO: remove this alias
applyBalancingAlias := balanceCommand.Bool("force", false, "apply the balancing plan (alias for -apply)")
if err = balanceCommand.Parse(args); err != nil {
return nil
}
handleDeprecatedForceFlag(writer, balanceCommand, applyBalancingAlias, applyBalancing)
infoAboutSimulationMode(writer, *applyBalancing, "-apply")
var volumeIds []needle.VolumeId
if *volumeIdsStr != "" {
if volumeIds, err = parseVolumeIdsFlag(*volumeIdsStr); err != nil {
return err
}
}
if err = commandEnv.confirmIsLocked(args); err != nil {
return
}
collectionNames, err := ListCollectionNames(commandEnv, false, true)
if err != nil {
return err
}
var collections []string
if *collection == string(wildcard.CollectionFilterEach) {
collections = collectionNames
} else {
collectionMatcher, err := compileCollectionPattern(*collection)
if err != nil {
return fmt.Errorf("invalid collection pattern %q: %w", *collection, err)
}
for _, name := range collectionNames {
if collectionMatcher.Matches(name) {
collections = append(collections, name)
}
}
if len(collections) == 0 {
return fmt.Errorf("no collections match %q", *collection)
}
}
glog.V(1).Infof("balanceEcVolumes collections %+v\n", len(collections))
rp, err := parseReplicaPlacementArg(commandEnv, *shardReplicaPlacement)
if err != nil {
return err
}
diskType := types.ToDiskType(*diskTypeStr)
return EcBalance(commandEnv, collections, *dc, rp, diskType, *maxParallelization, *ioBytePerSecond, *applyBalancing, nil, volumeIds)
}