gsh-digital-services/seaweedFS
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix Change replication via volume.configure.replication by collection

Browse Source 
fix https://github.com/chrislusf/seaweedfs/issues/3346
This commit is contained in:
chrislu
2022-07-21 22:01:05 -07:00
parent fc8241fb5e
commit 7a6c559ab4
1 changed files with 20 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
weed/shell/command_volume_configure_replication.go
View File

@@ -68,27 +68,22 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman
volumeFilter := getVolumeFilter(replicaPlacement, uint32(vid), *collectionPattern) volumeFilter := getVolumeFilter(replicaPlacement, uint32(vid), *collectionPattern)
// find all data nodes with volumes that needs replication change // find all data nodes with volumes that needs replication change
var allLocations []location
eachDataNode(topologyInfo, func(dc string, rack RackId, dn *master_pb.DataNodeInfo) { eachDataNode(topologyInfo, func(dc string, rack RackId, dn *master_pb.DataNodeInfo) {
loc := newLocation(dc, string(rack), dn) var targetVolumeIds []uint32
for _, diskInfo := range dn.DiskInfos { for _, diskInfo := range dn.DiskInfos {
for _, v := range diskInfo.VolumeInfos { for _, v := range diskInfo.VolumeInfos {
if volumeFilter(v) { if volumeFilter(v) {
allLocations = append(allLocations, loc) targetVolumeIds = append(targetVolumeIds, v.Id)
continue
} }
} }
} }
}) if len(targetVolumeIds) == 0 {
return
if len(allLocations) == 0 {
return fmt.Errorf("no volume needs change")
} }
err = operation.WithVolumeServerClient(false, pb.NewServerAddressFromDataNode(dn), commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
for _, dst := range allLocations { for _, targetVolumeId := range targetVolumeIds {
err := operation.WithVolumeServerClient(false, pb.NewServerAddressFromDataNode(dst.dataNode), commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
resp, configureErr := volumeServerClient.VolumeConfigure(context.Background(), &volume_server_pb.VolumeConfigureRequest{ resp, configureErr := volumeServerClient.VolumeConfigure(context.Background(), &volume_server_pb.VolumeConfigureRequest{
VolumeId: uint32(vid), VolumeId: targetVolumeId,
Replication: replicaPlacement.String(), Replication: replicaPlacement.String(),
}) })
if configureErr != nil { if configureErr != nil {
@@ -97,18 +92,17 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman
if resp.Error != "" { if resp.Error != "" {
return errors.New(resp.Error) return errors.New(resp.Error)
} }
}
return nil return nil
}) })
if err != nil { if err != nil {
return
}
})
return err return err
} }
}
return nil
}
func getVolumeFilter(replicaPlacement *super_block.ReplicaPlacement, volumeId uint32, collectionPattern string) func(message *master_pb.VolumeInformationMessage) bool { func getVolumeFilter(replicaPlacement *super_block.ReplicaPlacement, volumeId uint32, collectionPattern string) func(message *master_pb.VolumeInformationMessage) bool {
replicaPlacementInt32 := uint32(replicaPlacement.Byte()) replicaPlacementInt32 := uint32(replicaPlacement.Byte())
if volumeId > 0 { if volumeId > 0 {