diff --git a/weed/worker/tasks/balance/plugin_handler.go b/weed/worker/tasks/balance/plugin_handler.go index a253a3e7b..89df412e8 100644 --- a/weed/worker/tasks/balance/plugin_handler.go +++ b/weed/worker/tasks/balance/plugin_handler.go @@ -13,9 +13,11 @@ import ( "github.com/seaweedfs/seaweedfs/weed/admin/topology" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" pluginworker "github.com/seaweedfs/seaweedfs/weed/plugin/worker" + "github.com/seaweedfs/seaweedfs/weed/storage/types" "github.com/seaweedfs/seaweedfs/weed/util/wildcard" workertypes "github.com/seaweedfs/seaweedfs/weed/worker/types" "google.golang.org/grpc" @@ -739,7 +741,9 @@ func (h *VolumeBalanceHandler) executeSingleMove( return err } - execErr := h.checkMoveStillValid(execCtx, request.GetClusterContext().GetMasterGrpcAddresses(), params.VolumeId, params.Sources[0].Node, params.Targets[0].Node) + masterAddresses := request.GetClusterContext().GetMasterGrpcAddresses() + moveTopology := h.fetchMoveTopology(execCtx, masterAddresses) + execErr := h.checkMoveStillValid(execCtx, masterAddresses, moveTopology, params.VolumeId, params.Sources[0].Node, params.Targets[0].Node) if execErr == nil { execErr = task.Execute(execCtx, params) } @@ -793,7 +797,10 @@ func (h *VolumeBalanceHandler) executeSingleMove( // would overwrite and the source delete would then reduce to a single copy. // Without master addresses (older admin) the check is skipped and the move // relies on the task's own execution-time guards. -func (h *VolumeBalanceHandler) checkMoveStillValid(ctx context.Context, masterAddresses []string, volumeID uint32, sourceNode, targetNode string) error { +func (h *VolumeBalanceHandler) checkMoveStillValid(ctx context.Context, masterAddresses []string, topologyInfo *master_pb.TopologyInfo, volumeID uint32, sourceNode, targetNode string) error { + if err := checkMoveDiskType(topologyInfo, volumeID, sourceNode, targetNode); err != nil { + return err + } if len(masterAddresses) == 0 { glog.Warningf("volume balance: no master addresses in cluster context, skipping pre-move check for volume %d", volumeID) return nil @@ -805,6 +812,72 @@ func (h *VolumeBalanceHandler) checkMoveStillValid(ctx context.Context, masterAd return checkMovePreconditions(locations, volumeID, sourceNode, targetNode) } +// fetchMoveTopology returns the master's volume list topology so each move +// can be checked for disk-type compatibility before copying. Returns nil if +// the master cannot be reached; checks that need it are then skipped. +func (h *VolumeBalanceHandler) fetchMoveTopology(ctx context.Context, masterAddresses []string) *master_pb.TopologyInfo { + for _, address := range masterAddresses { + resp, err := pluginworker.FetchVolumeList(ctx, address, h.grpcDialOption) + if err == nil && resp != nil && resp.TopologyInfo != nil { + return resp.TopologyInfo + } + } + return nil +} + +// checkMoveDiskType rejects a move whose target lacks a disk matching the +// volume's disk type on the source. Detection buckets candidates by disk +// type, but jobs can be stale or submitted out-of-band; without this check +// the copy is doomed to be rejected by the target after the source disk was +// already read. Unknown pieces (volume or target missing from the topology) +// skip the check and let the server's own rejection decide. +func checkMoveDiskType(topologyInfo *master_pb.TopologyInfo, volumeID uint32, sourceNode, targetNode string) error { + if topologyInfo == nil { + return nil + } + sourceAddress := pb.ServerAddress(strings.TrimSpace(sourceNode)) + targetAddress := pb.ServerAddress(strings.TrimSpace(targetNode)) + volumeFound := false + volumeDiskType := "" + targetFound := false + targetDiskTypes := map[string]bool{} + for _, dc := range topologyInfo.DataCenterInfos { + for _, rack := range dc.RackInfos { + for _, node := range rack.DataNodeInfos { + // NewServerAddressFromDataNode covers older masters whose + // topology omits Address: the move nodes carry the grpc + // suffix derived from the plain node Id and GrpcPort. + nodeAddress := pb.NewServerAddressFromDataNode(node) + isSource := node.Id == sourceNode || pb.ServerAddress(node.Address).Equals(sourceAddress) || nodeAddress.Equals(sourceAddress) + isTarget := node.Id == targetNode || pb.ServerAddress(node.Address).Equals(targetAddress) || nodeAddress.Equals(targetAddress) + if isTarget { + targetFound = true + for diskType := range node.DiskInfos { + targetDiskTypes[string(types.ToDiskType(diskType))] = true + } + } + if isSource && !volumeFound { + for diskType, diskInfo := range node.DiskInfos { + for _, v := range diskInfo.VolumeInfos { + if v.Id == volumeID { + volumeFound = true + volumeDiskType = diskType + } + } + } + } + } + } + } + if !volumeFound || !targetFound { + return nil + } + if !targetDiskTypes[string(types.ToDiskType(volumeDiskType))] { + return fmt.Errorf("target %s has no %s disk for volume %d", targetNode, volumeDiskType, volumeID) + } + return nil +} + func checkMovePreconditions(locations []string, volumeID uint32, sourceNode, targetNode string) error { // Proposal nodes carry the grpc suffix (host:port.grpcPort) while the // master reports plain host:port urls, so compare in http form. @@ -942,6 +1015,7 @@ func (h *VolumeBalanceHandler) executeBatchMoves( sem := make(chan struct{}, maxConcurrent) results := make(chan moveResult, totalMoves) masterAddresses := request.GetClusterContext().GetMasterGrpcAddresses() + moveTopology := h.fetchMoveTopology(batchCtx, masterAddresses) for i, move := range moves { sem <- struct{}{} // acquire slot @@ -960,7 +1034,7 @@ func (h *VolumeBalanceHandler) executeBatchMoves( }) moveParams := buildMoveTaskParams(m, bp) - err := h.checkMoveStillValid(batchCtx, masterAddresses, m.VolumeId, m.SourceNode, m.TargetNode) + err := h.checkMoveStillValid(batchCtx, masterAddresses, moveTopology, m.VolumeId, m.SourceNode, m.TargetNode) if err == nil { err = task.Execute(batchCtx, moveParams) } diff --git a/weed/worker/tasks/balance/plugin_handler_test.go b/weed/worker/tasks/balance/plugin_handler_test.go index 2067623fa..123a0a279 100644 --- a/weed/worker/tasks/balance/plugin_handler_test.go +++ b/weed/worker/tasks/balance/plugin_handler_test.go @@ -7,6 +7,7 @@ import ( "sync" "testing" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" pluginworker "github.com/seaweedfs/seaweedfs/weed/plugin/worker" @@ -826,3 +827,106 @@ func TestCheckMovePreconditions(t *testing.T) { }) } } + +func TestCheckMoveDiskType(t *testing.T) { + volumeOnDisk := func(id uint32) []*master_pb.VolumeInformationMessage { + return []*master_pb.VolumeInformationMessage{{Id: id}} + } + topologyWith := func(sourceDisks, targetDisks map[string]*master_pb.DiskInfo) *master_pb.TopologyInfo { + return &master_pb.TopologyInfo{DataCenterInfos: []*master_pb.DataCenterInfo{ + {RackInfos: []*master_pb.RackInfo{{DataNodeInfos: []*master_pb.DataNodeInfo{ + {Id: "10.0.0.1:8080.18080", Address: "10.0.0.1:8080", DiskInfos: sourceDisks}, + {Id: "10.0.0.2:8080.18080", Address: "10.0.0.2:8080", DiskInfos: targetDisks}, + }}}}, + }} + } + oldTopologyWith := func(sourceDisks, targetDisks map[string]*master_pb.DiskInfo) *master_pb.TopologyInfo { + return &master_pb.TopologyInfo{DataCenterInfos: []*master_pb.DataCenterInfo{ + {RackInfos: []*master_pb.RackInfo{{DataNodeInfos: []*master_pb.DataNodeInfo{ + {Id: "10.0.0.1:8080", GrpcPort: 18080, DiskInfos: sourceDisks}, + {Id: "10.0.0.2:8080", GrpcPort: 18080, DiskInfos: targetDisks}, + }}}}, + }} + } + + tests := []struct { + name string + topology *master_pb.TopologyInfo + volumeID uint32 + sourceNode string + targetNode string + wantErr string + }{ + { + name: "same disk type allowed", + topology: topologyWith( + map[string]*master_pb.DiskInfo{"backup-hdd": {VolumeInfos: volumeOnDisk(5)}}, + map[string]*master_pb.DiskInfo{"backup-hdd": {}}), + volumeID: 5, sourceNode: "10.0.0.1:8080.18080", targetNode: "10.0.0.2:8080.18080", + }, + { + name: "target without matching disk type rejected", + topology: topologyWith( + map[string]*master_pb.DiskInfo{"backup-hdd": {VolumeInfos: volumeOnDisk(5)}}, + map[string]*master_pb.DiskInfo{"ssd": {}}), + volumeID: 5, sourceNode: "10.0.0.1:8080.18080", targetNode: "10.0.0.2:8080.18080", + wantErr: "has no backup-hdd disk", + }, + { + name: "unknown volume skips check", + topology: topologyWith( + map[string]*master_pb.DiskInfo{"backup-hdd": {}}, + map[string]*master_pb.DiskInfo{"ssd": {}}), + volumeID: 7, sourceNode: "10.0.0.1:8080.18080", targetNode: "10.0.0.2:8080.18080", + }, + { + name: "unknown target skips check", + topology: topologyWith( + map[string]*master_pb.DiskInfo{"backup-hdd": {VolumeInfos: volumeOnDisk(5)}}, + nil), + volumeID: 5, sourceNode: "10.0.0.1:8080.18080", targetNode: "10.0.0.9:8080.18080", + }, + { + name: "nil topology skips check", + topology: nil, + volumeID: 5, sourceNode: "10.0.0.1:8080.18080", targetNode: "10.0.0.2:8080.18080", + }, + { + name: "nodes matched by address without grpc suffix", + topology: topologyWith( + map[string]*master_pb.DiskInfo{"backup-hdd": {VolumeInfos: volumeOnDisk(5)}}, + map[string]*master_pb.DiskInfo{"ssd": {}}), + volumeID: 5, sourceNode: "10.0.0.1:8080", targetNode: "10.0.0.2:8080", + wantErr: "has no backup-hdd disk", + }, + { + name: "old topology nodes matched via id and grpc port", + topology: oldTopologyWith( + map[string]*master_pb.DiskInfo{"backup-hdd": {VolumeInfos: volumeOnDisk(5)}}, + map[string]*master_pb.DiskInfo{"ssd": {}}), + volumeID: 5, sourceNode: "10.0.0.1:8080.18080", targetNode: "10.0.0.2:8080.18080", + wantErr: "has no backup-hdd disk", + }, + { + name: "old topology allows matching disk type", + topology: oldTopologyWith( + map[string]*master_pb.DiskInfo{"backup-hdd": {VolumeInfos: volumeOnDisk(5)}}, + map[string]*master_pb.DiskInfo{"backup-hdd": {}}), + volumeID: 5, sourceNode: "10.0.0.1:8080.18080", targetNode: "10.0.0.2:8080.18080", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := checkMoveDiskType(tt.topology, tt.volumeID, tt.sourceNode, tt.targetNode) + if tt.wantErr == "" { + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + return + } + if err == nil || !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("expected error containing %q, got %v", tt.wantErr, err) + } + }) + } +}