seaweedfs/weed/shell/command_volume_configure_replication.go

package shell

import (
	"context"
	"errors"
	"flag"
	"fmt"
	"io"
	"sync"

	"github.com/seaweedfs/seaweedfs/weed/pb"

	"github.com/seaweedfs/seaweedfs/weed/operation"
	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
)

func init() {
	Commands = append(Commands, &commandVolumeConfigureReplication{})
}

type commandVolumeConfigureReplication struct {
}

func (c *commandVolumeConfigureReplication) Name() string {
	return "volume.configure.replication"
}

func (c *commandVolumeConfigureReplication) Help() string {
	return `change volume replication value

	This command changes a volume replication value. It should be followed by "volume.fix.replication".

	Up to -maxParallelization volume servers are configured at a time, one volume at a time on each,
	and replicas of the same volume on different servers take turns.

`
}

func (c *commandVolumeConfigureReplication) HasTag(CommandTag) bool {
	return false
}

func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *CommandEnv, _ io.Writer) (err error) {

	configureReplicationCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
	volumeIdInt := configureReplicationCommand.Int("volumeId", 0, "the volume id")
	replicationString := configureReplicationCommand.String("replication", "", "the intended replication value")
	collectionPattern := configureReplicationCommand.String("collectionPattern", "", "comma-separated collection names, with '*' and '?' wildcards; empty matches all")
	maxParallelization := configureReplicationCommand.Int("maxParallelization", DefaultMaxParallelization, "run up to X tasks in parallel, whenever possible")
	if err = configureReplicationCommand.Parse(args); err != nil {
		return err
	}

	if err = commandEnv.confirmIsLocked(args); err != nil {
		return
	}

	if *replicationString == "" {
		return fmt.Errorf("empty replication value")
	}

	replicaPlacement, err := super_block.NewReplicaPlacementFromString(*replicationString)
	if err != nil {
		return fmt.Errorf("replication format: %w", err)
	}

	// collect topology information
	topologyInfo, _, err := collectTopologyInfo(commandEnv, 0)
	if err != nil {
		return err
	}

	vid := needle.VolumeId(*volumeIdInt)
	volumeFilter, err := getVolumeFilter(replicaPlacement, uint32(vid), *collectionPattern)
	if err != nil {
		return err
	}

	return configureReplicationOnVolumeServers(commandEnv, topologyInfo, volumeFilter, replicaPlacement.String(), *maxParallelization)
}

func configureReplicationOnVolumeServers(commandEnv *CommandEnv, topologyInfo *master_pb.TopologyInfo, volumeFilter func(*master_pb.VolumeInformationMessage) bool, replication string, maxParallelization int) error {
	type serverVolumes struct {
		addr      pb.ServerAddress
		volumeIds []uint32
	}
	var servers []serverVolumes
	// VolumeConfigure unmounts and remounts the volume, so replicas take turns
	// to keep at least one of them serving.
	volumeLocks := make(map[uint32]*sync.Mutex)

	// find all data nodes with volumes that needs replication change
	eachDataNode(topologyInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
		var targetVolumeIds []uint32
		for _, diskInfo := range dn.DiskInfos {
			for _, v := range diskInfo.VolumeInfos {
				if volumeFilter(v) {
					targetVolumeIds = append(targetVolumeIds, v.Id)
					if volumeLocks[v.Id] == nil {
						volumeLocks[v.Id] = &sync.Mutex{}
					}
				}
			}
		}
		if len(targetVolumeIds) > 0 {
			servers = append(servers, serverVolumes{pb.NewServerAddressFromDataNode(dn), targetVolumeIds})
		}
	})

	ewg := NewErrorWaitGroup(maxParallelization)
	for _, server := range servers {
		ewg.Add(func() error {
			err := operation.WithVolumeServerClient(false, server.addr, commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
				for _, targetVolumeId := range server.volumeIds {
					volumeLocks[targetVolumeId].Lock()
					resp, configureErr := volumeServerClient.VolumeConfigure(context.Background(), &volume_server_pb.VolumeConfigureRequest{
						VolumeId:    targetVolumeId,
						Replication: replication,
					})
					volumeLocks[targetVolumeId].Unlock()
					if configureErr != nil {
						return configureErr
					}
					if resp.Error != "" {
						return errors.New(resp.Error)
					}
				}
				return nil
			})
			if err != nil {
				return fmt.Errorf("volume server %s: %w", server.addr, err)
			}
			return nil
		})
	}

	return ewg.Wait()
}

func getVolumeFilter(replicaPlacement *super_block.ReplicaPlacement, volumeId uint32, collectionPattern string) (func(message *master_pb.VolumeInformationMessage) bool, error) {
	replicaPlacementInt32 := uint32(replicaPlacement.Byte())
	if volumeId > 0 {
		return func(v *master_pb.VolumeInformationMessage) bool {
			return v.Id == volumeId && v.ReplicaPlacement != replicaPlacementInt32
		}, nil
	}
	collectionMatcher, err := wildcard.CompileCollectionMatcher(collectionPattern)
	if err != nil {
		return nil, err
	}
	return func(v *master_pb.VolumeInformationMessage) bool {
		return collectionMatcher.Matches(v.Collection) && v.ReplicaPlacement != replicaPlacementInt32
	}, nil
}