|
|
@ -20,10 +20,17 @@ func init() { |
|
|
|
Commands = append(Commands, &commandVolumeTierMove{}) |
|
|
|
} |
|
|
|
|
|
|
|
type volumeTierMoveJob struct { |
|
|
|
src pb.ServerAddress |
|
|
|
vid needle.VolumeId |
|
|
|
} |
|
|
|
|
|
|
|
type commandVolumeTierMove struct { |
|
|
|
activeServers map[pb.ServerAddress]struct{} |
|
|
|
activeServersLock sync.Mutex |
|
|
|
activeServersCond *sync.Cond |
|
|
|
activeServers sync.Map |
|
|
|
queues map[pb.ServerAddress]chan volumeTierMoveJob |
|
|
|
//activeServers map[pb.ServerAddress]struct{}
|
|
|
|
//activeServersLock sync.Mutex
|
|
|
|
//activeServersCond *sync.Cond
|
|
|
|
} |
|
|
|
|
|
|
|
func (c *commandVolumeTierMove) Name() string { |
|
|
@ -43,15 +50,13 @@ func (c *commandVolumeTierMove) Help() string { |
|
|
|
|
|
|
|
func (c *commandVolumeTierMove) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) { |
|
|
|
|
|
|
|
c.activeServers = make(map[pb.ServerAddress]struct{}) |
|
|
|
c.activeServersCond = sync.NewCond(new(sync.Mutex)) |
|
|
|
|
|
|
|
tierCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) |
|
|
|
collectionPattern := tierCommand.String("collectionPattern", "", "match with wildcard characters '*' and '?'") |
|
|
|
fullPercentage := tierCommand.Float64("fullPercent", 95, "the volume reaches the percentage of max volume size") |
|
|
|
quietPeriod := tierCommand.Duration("quietFor", 24*time.Hour, "select volumes without no writes for this period") |
|
|
|
source := tierCommand.String("fromDiskType", "", "the source disk type") |
|
|
|
target := tierCommand.String("toDiskType", "", "the target disk type") |
|
|
|
limitWorkers := tierCommand.Int("limitWorkers", 0, "limit the number of active copying workers") |
|
|
|
applyChange := tierCommand.Bool("force", false, "actually apply the changes") |
|
|
|
if err = tierCommand.Parse(args); err != nil { |
|
|
|
return nil |
|
|
@ -82,15 +87,87 @@ func (c *commandVolumeTierMove) Do(args []string, commandEnv *CommandEnv, writer |
|
|
|
fmt.Printf("tier move volumes: %v\n", volumeIds) |
|
|
|
|
|
|
|
_, allLocations := collectVolumeReplicaLocations(topologyInfo) |
|
|
|
allLocations = filterLocationsByDiskType(allLocations, toDiskType) |
|
|
|
keepDataNodesSorted(allLocations, toDiskType) |
|
|
|
|
|
|
|
if len(allLocations) > 0 && *limitWorkers > 0 && *limitWorkers < len(allLocations) { |
|
|
|
allLocations = allLocations[:*limitWorkers] |
|
|
|
} |
|
|
|
|
|
|
|
wg := sync.WaitGroup{} |
|
|
|
bufferLen := len(allLocations) |
|
|
|
c.queues = make(map[pb.ServerAddress]chan volumeTierMoveJob) |
|
|
|
|
|
|
|
for _, dst := range allLocations { |
|
|
|
destServerAddress := pb.NewServerAddressFromDataNode(dst.dataNode) |
|
|
|
c.queues[destServerAddress] = make(chan volumeTierMoveJob, bufferLen) |
|
|
|
|
|
|
|
wg.Add(1) |
|
|
|
go func (dst location, jobs <-chan volumeTierMoveJob, applyChanges bool) { |
|
|
|
for job := range jobs { |
|
|
|
fmt.Fprintf(writer, "moving volume %d from %s to %s with disk type %s ...\n", job.vid, job.src, dst.dataNode.Id, toDiskType.ReadableString()) |
|
|
|
if !applyChanges { |
|
|
|
continue |
|
|
|
} |
|
|
|
|
|
|
|
locations, found := commandEnv.MasterClient.GetLocations(uint32(job.vid)) |
|
|
|
if !found { |
|
|
|
fmt.Printf("volume %d not found", job.vid) |
|
|
|
continue |
|
|
|
} |
|
|
|
|
|
|
|
unlock := c.Lock(job.src) |
|
|
|
|
|
|
|
if err := c.doMoveOneVolume(commandEnv, writer, job.vid, toDiskType, locations, job.src, dst); err != nil { |
|
|
|
fmt.Fprintf(writer, "move volume %d %s => %s: %v\n", job.vid, job.src, dst.dataNode.Id, err) |
|
|
|
} |
|
|
|
unlock() |
|
|
|
} |
|
|
|
wg.Done() |
|
|
|
}(dst, c.queues[destServerAddress], *applyChange) |
|
|
|
} |
|
|
|
|
|
|
|
for _, vid := range volumeIds { |
|
|
|
if err = c.doVolumeTierMove(commandEnv, writer, vid, toDiskType, allLocations, *applyChange); err != nil { |
|
|
|
if err = c.doVolumeTierMove(commandEnv, writer, vid, toDiskType, allLocations); err != nil { |
|
|
|
fmt.Printf("tier move volume %d: %v\n", vid, err) |
|
|
|
} |
|
|
|
allLocations = rotateDataNodes(allLocations) |
|
|
|
} |
|
|
|
for key, _ := range c.queues { |
|
|
|
close(c.queues[key]) |
|
|
|
} |
|
|
|
|
|
|
|
wg.Wait() |
|
|
|
|
|
|
|
return nil |
|
|
|
} |
|
|
|
|
|
|
|
func (c *commandVolumeTierMove) Lock(key pb.ServerAddress) func() { |
|
|
|
value, _ := c.activeServers.LoadOrStore(key, &sync.Mutex{}) |
|
|
|
mtx := value.(*sync.Mutex) |
|
|
|
mtx.Lock() |
|
|
|
|
|
|
|
return func() { mtx.Unlock() } |
|
|
|
} |
|
|
|
|
|
|
|
func filterLocationsByDiskType(dataNodes []location, diskType types.DiskType) (ret []location) { |
|
|
|
for _, loc := range dataNodes { |
|
|
|
_, found := loc.dataNode.DiskInfos[string(diskType)] |
|
|
|
if found { |
|
|
|
ret = append(ret, loc) |
|
|
|
} |
|
|
|
} |
|
|
|
return |
|
|
|
} |
|
|
|
|
|
|
|
func rotateDataNodes(dataNodes []location) []location { |
|
|
|
if len(dataNodes) > 0 { |
|
|
|
return append(dataNodes[1:], dataNodes[0]) |
|
|
|
} else { |
|
|
|
return dataNodes |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
func isOneOf(server string, locations []wdclient.Location) bool { |
|
|
|
for _, loc := range locations { |
|
|
|
if server == loc.Url { |
|
|
@ -100,7 +177,7 @@ func isOneOf(server string, locations []wdclient.Location) bool { |
|
|
|
return false |
|
|
|
} |
|
|
|
|
|
|
|
func (c *commandVolumeTierMove) doVolumeTierMove(commandEnv *CommandEnv, writer io.Writer, vid needle.VolumeId, toDiskType types.DiskType, allLocations []location, applyChanges bool) (err error) { |
|
|
|
func (c *commandVolumeTierMove) doVolumeTierMove(commandEnv *CommandEnv, writer io.Writer, vid needle.VolumeId, toDiskType types.DiskType, allLocations []location) (err error) { |
|
|
|
// find volume location
|
|
|
|
locations, found := commandEnv.MasterClient.GetLocations(uint32(vid)) |
|
|
|
if !found { |
|
|
@ -109,9 +186,7 @@ func (c *commandVolumeTierMove) doVolumeTierMove(commandEnv *CommandEnv, writer |
|
|
|
|
|
|
|
// find one server with the most empty volume slots with target disk type
|
|
|
|
hasFoundTarget := false |
|
|
|
keepDataNodesSorted(allLocations, toDiskType) |
|
|
|
fn := capacityByFreeVolumeCount(toDiskType) |
|
|
|
wg := sync.WaitGroup{} |
|
|
|
for _, dst := range allLocations { |
|
|
|
if fn(dst.dataNode) > 0 && !hasFoundTarget { |
|
|
|
// ask the volume server to replicate the volume
|
|
|
@ -127,44 +202,16 @@ func (c *commandVolumeTierMove) doVolumeTierMove(commandEnv *CommandEnv, writer |
|
|
|
if sourceVolumeServer == "" { |
|
|
|
continue |
|
|
|
} |
|
|
|
fmt.Fprintf(writer, "moving volume %d from %s to %s with disk type %s ...\n", vid, sourceVolumeServer, dst.dataNode.Id, toDiskType.ReadableString()) |
|
|
|
hasFoundTarget = true |
|
|
|
|
|
|
|
if !applyChanges { |
|
|
|
// adjust volume count
|
|
|
|
dst.dataNode.DiskInfos[string(toDiskType)].VolumeCount++ |
|
|
|
break |
|
|
|
} |
|
|
|
|
|
|
|
destServerAddress := pb.NewServerAddressFromDataNode(dst.dataNode) |
|
|
|
c.activeServersCond.L.Lock() |
|
|
|
_, isSourceActive := c.activeServers[sourceVolumeServer] |
|
|
|
_, isDestActive := c.activeServers[destServerAddress] |
|
|
|
for isSourceActive || isDestActive { |
|
|
|
c.activeServersCond.Wait() |
|
|
|
_, isSourceActive = c.activeServers[sourceVolumeServer] |
|
|
|
_, isDestActive = c.activeServers[destServerAddress] |
|
|
|
} |
|
|
|
c.activeServers[sourceVolumeServer] = struct{}{} |
|
|
|
c.activeServers[destServerAddress] = struct{}{} |
|
|
|
c.activeServersCond.L.Unlock() |
|
|
|
|
|
|
|
wg.Add(1) |
|
|
|
go func(dst location) { |
|
|
|
if err := c.doMoveOneVolume(commandEnv, writer, vid, toDiskType, locations, sourceVolumeServer, dst); err != nil { |
|
|
|
fmt.Fprintf(writer, "move volume %d %s => %s: %v\n", vid, sourceVolumeServer, dst.dataNode.Id, err) |
|
|
|
} |
|
|
|
delete(c.activeServers, sourceVolumeServer) |
|
|
|
delete(c.activeServers, destServerAddress) |
|
|
|
c.activeServersCond.Signal() |
|
|
|
wg.Done() |
|
|
|
}(dst) |
|
|
|
|
|
|
|
c.queues[destServerAddress] <- volumeTierMoveJob{sourceVolumeServer, vid} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
wg.Wait() |
|
|
|
|
|
|
|
if !hasFoundTarget { |
|
|
|
fmt.Fprintf(writer, "can not find disk type %s for volume %d\n", toDiskType.ReadableString(), vid) |
|
|
|
} |
|
|
|