diff --git a/weed/shell/command_fs_distribute_chunks.go b/weed/shell/command_fs_distribute_chunks.go new file mode 100644 index 000000000..a9b41db8c --- /dev/null +++ b/weed/shell/command_fs_distribute_chunks.go @@ -0,0 +1,901 @@ +package shell + +import ( + "bytes" + "context" + "flag" + "fmt" + "io" + "math" + "net/http" + "maps" + "slices" + "sort" + "strings" + "sync" + "time" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/operation" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/security" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/util" + util_http "github.com/seaweedfs/seaweedfs/weed/util/http" +) + +func init() { + Commands = append(Commands, &commandFsDistributeChunks{}) +} + +type commandFsDistributeChunks struct{} + +func (c *commandFsDistributeChunks) Name() string { + return "fs.distributeChunks" +} + +func (c *commandFsDistributeChunks) Help() string { + return `redistribute file chunks evenly across volume server nodes + + Modes: + primary (default) balance chunk count per node using topology-based ownership. + Note: with replication, ownership is derived from volume ID hashing + (vid %% node count), which may not reflect the actual Assign target. + Results can be approximate in replicated environments. + + replica balance both ownership and replica copies across nodes. + Step 1: owner-based balancing (same as primary mode). + Step 2: additional moves to balance total copies (including replicas). + Same ownership limitation as primary mode applies. + + round-robin assign chunks to nodes by file offset order (chunk[0]->A, chunk[1]->B, + chunk[2]->C, ...). Best mode for sequential read performance — ensures + consecutive chunks are on different nodes for I/O pipelining. + Recommended for replication environments as it does not depend on + ownership calculation. + + Files using chunk manifests (very large files) are handled by resolving manifests + to the underlying data chunks, redistributing those, then re-manifestizing before + updating the filer. Old manifest + data chunks are garbage-collected by the filer. + + # analyze current distribution (dry-run, primary mode) + fs.distributeChunks -path=/buckets/my-bucket/large-file.dat + + # apply redistribution + fs.distributeChunks -path=/buckets/my-bucket/large-file.dat -apply + + # distribute across 5 nodes (instead of all) + fs.distributeChunks -path=/buckets/my-bucket/large-file.dat -nodes=5 -apply + + # balance including replica copies + fs.distributeChunks -path=/buckets/my-bucket/large-file.dat -mode=replica -apply + + # round-robin for sequential read performance (recommended with replication) + fs.distributeChunks -path=/buckets/my-bucket/large-file.dat -mode=round-robin -apply +` +} + +func (c *commandFsDistributeChunks) HasTag(CommandTag) bool { + return false +} + +type chunkMove struct { + chunkIndex int + chunk *filer_pb.FileChunk + fromNode string + toNode string +} + + +func (c *commandFsDistributeChunks) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) { + + fsDistributeCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) + filePath := fsDistributeCommand.String("path", "", "file path to redistribute chunks") + apply := fsDistributeCommand.Bool("apply", false, "apply the redistribution") + targetNodeCount := fsDistributeCommand.Int("nodes", 0, "number of nodes to distribute across (0 = all nodes)") + mode := fsDistributeCommand.String("mode", "primary", "distribution mode: primary, replica, round-robin") + if err = fsDistributeCommand.Parse(args); err != nil { + return err + } + if *filePath == "" { + return fmt.Errorf("-path is required") + } + if *mode != "primary" && *mode != "replica" && *mode != "round-robin" { + return fmt.Errorf("-mode must be one of: primary, replica, round-robin") + } + + infoAboutSimulationMode(writer, *apply, "-apply") + + path, err := commandEnv.parseUrl(*filePath) + if err != nil { + return err + } + dir, name := util.FullPath(path).DirAndName() + + entry, err := lookupFileEntry(commandEnv, dir, name) + if err != nil { + return fmt.Errorf("lookup %s: %v", path, err) + } + + chunks, hadManifest, err := resolveEntryDataChunks(commandEnv, entry) + if err != nil { + return err + } + if len(chunks) == 0 { + fmt.Fprintf(writer, "File has no chunks (possibly stored inline)\n") + return nil + } + if hadManifest { + fmt.Fprintf(writer, "Resolved chunk manifests: %d top-level chunk(s) -> %d data chunks.\n", len(entry.GetChunks()), len(chunks)) + } + + volumeInfoMap, volumeNodesList, volumeToOwner, nodeList, err := collectVolumeTopology(commandEnv) + if err != nil { + return err + } + + if len(nodeList) < 2 { + return fmt.Errorf("need at least 2 volume nodes, found %d", len(nodeList)) + } + if *targetNodeCount < 0 { + return fmt.Errorf("-nodes must be >= 0 (0 = all nodes)") + } + if *targetNodeCount > len(nodeList) { + return fmt.Errorf("-nodes=%d exceeds available nodes (%d)", *targetNodeCount, len(nodeList)) + } + + chunkToNode, ownerCount, copiesCount, err := buildDistributionCounts(chunks, volumeToOwner, volumeNodesList) + if err != nil { + return err + } + + activeNodeList, activeSet := selectActiveNodes(nodeList, ownerCount, *targetNodeCount) + + totalChunks := len(chunks) + totalNodes := len(activeNodeList) + totalCopies := 0 + for _, cnt := range copiesCount { + totalCopies += cnt + } + + fmt.Fprintf(writer, "\nFile: %s\n", path) + fmt.Fprintf(writer, "Total chunks: %d, Total nodes: %d, Mode: %s\n", totalChunks, totalNodes, *mode) + printCurrentDistribution(writer, nodeList, ownerCount, copiesCount, activeSet, *targetNodeCount, *mode, totalChunks, totalNodes, totalCopies) + + moves, ownerTarget := planDistribution(*mode, writer, activeNodeList, nodeList, activeSet, ownerCount, copiesCount, chunkToNode, chunks, volumeNodesList, totalChunks, totalNodes, totalCopies) + + if len(moves) == 0 { + fmt.Fprintf(writer, "\nAlready well distributed. No moves needed.\n") + return nil + } + + allRelevantNodes := relevantNodes(nodeList, activeSet, ownerCount) + printRedistributionPlan(writer, allRelevantNodes, ownerCount, ownerTarget, len(moves)) + + if !*apply { + fmt.Fprintf(writer, "\nTo apply, add -apply flag\n") + return nil + } + + fmt.Fprintf(writer, "\nExecuting redistribution...\n") + defer util_http.GetGlobalHttpClient().CloseIdleConnections() + + movedCount, err := executeChunkMoves(commandEnv, writer, moves, volumeInfoMap) + if err != nil { + return err + } + if movedCount == 0 { + fmt.Fprintf(writer, "\nNo chunks were moved successfully.\n") + return nil + } + + finalChunks := chunks + if hadManifest { + remanifested, manErr := filer.MaybeManifestize(newShellSaveAsChunk(commandEnv), chunks) + if manErr != nil { + fmt.Fprintf(writer, "WARNING: re-manifestize failed: %v. Writing flat chunk list.\n", manErr) + } else { + finalChunks = remanifested + fmt.Fprintf(writer, "Re-manifestized %d data chunks into %d top-level chunk(s).\n", len(chunks), len(finalChunks)) + } + } + entry.Chunks = finalChunks + + fmt.Fprintf(writer, "\nUpdating filer metadata (%d chunks moved)...\n", movedCount) + err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return filer_pb.UpdateEntry(context.Background(), client, &filer_pb.UpdateEntryRequest{ + Directory: dir, + Entry: entry, + }) + }) + if err != nil { + fmt.Fprintf(writer, "FAILED to update filer metadata: %v\n", err) + fmt.Fprintf(writer, "WARNING: New chunks were uploaded but metadata was not updated.\n") + fmt.Fprintf(writer, "The original file is still intact. New chunks are orphaned.\n") + return err + } + fmt.Fprintf(writer, "Filer metadata updated successfully.\n") + + fmt.Fprintf(writer, "\nRedistribution complete. %d chunks moved.\n", movedCount) + return nil +} + +// lookupFileEntry fetches the filer entry for the given directory and name. +func lookupFileEntry(commandEnv *CommandEnv, dir, name string) (*filer_pb.Entry, error) { + var entry *filer_pb.Entry + err := commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + resp, err := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{ + Directory: dir, + Name: name, + }) + if err != nil { + return err + } + entry = resp.Entry + return nil + }) + if err != nil { + return nil, err + } + if entry == nil { + return nil, fmt.Errorf("lookup for %s/%s returned no entry", dir, name) + } + if entry.IsDirectory { + return nil, fmt.Errorf("%s/%s is a directory", dir, name) + } + return entry, nil +} + +// resolveEntryDataChunks returns the flattened data chunks of an entry. If the +// entry contains manifest chunks, their contents are resolved. hadManifest +// signals whether any manifest chunks were resolved, so the caller knows to +// re-manifestize after redistribution. +func resolveEntryDataChunks(commandEnv *CommandEnv, entry *filer_pb.Entry) (dataChunks []*filer_pb.FileChunk, hadManifest bool, err error) { + chunks := entry.GetChunks() + if !filer.HasChunkManifest(chunks) { + return chunks, false, nil + } + dataChunks, _, err = filer.ResolveChunkManifest(context.Background(), filer.LookupFn(commandEnv), chunks, 0, math.MaxInt64) + if err != nil { + return nil, true, fmt.Errorf("resolve chunk manifest: %v", err) + } + return dataChunks, true, nil +} + +// collectVolumeTopology queries the master for cluster topology and returns per-volume +// metadata, per-volume node lists, per-volume ownership, and the sorted node list. +func collectVolumeTopology(commandEnv *CommandEnv) ( + volumeInfoMap map[uint32]*master_pb.VolumeInformationMessage, + volumeNodesList map[uint32][]string, + volumeToOwner map[uint32]string, + nodeList []string, + err error, +) { + topoInfo, _, err := collectTopologyInfo(commandEnv, 0) + if err != nil { + return nil, nil, nil, nil, fmt.Errorf("collect topology: %v", err) + } + + volumeInfoMap = make(map[uint32]*master_pb.VolumeInformationMessage) + volumeNodesList = make(map[uint32][]string) + allNodes := make(map[string]bool) + + for _, dc := range topoInfo.GetDataCenterInfos() { + for _, rack := range dc.GetRackInfos() { + for _, dn := range rack.GetDataNodeInfos() { + nodeId := dn.GetId() + allNodes[nodeId] = true + for _, disk := range dn.GetDiskInfos() { + for _, vol := range disk.GetVolumeInfos() { + volumeNodesList[vol.GetId()] = append(volumeNodesList[vol.GetId()], nodeId) + if _, exists := volumeInfoMap[vol.GetId()]; !exists { + volumeInfoMap[vol.GetId()] = vol + } + } + } + } + } + } + + volumeToOwner = make(map[uint32]string) + for vid, nodes := range volumeNodesList { + sort.Strings(nodes) + volumeToOwner[vid] = nodes[int(vid)%len(nodes)] + } + + nodeList = make([]string, 0, len(allNodes)) + for node := range allNodes { + nodeList = append(nodeList, node) + } + sort.Strings(nodeList) + + return volumeInfoMap, volumeNodesList, volumeToOwner, nodeList, nil +} + +// buildDistributionCounts maps each chunk to its owning node and tallies owner/copy counts. +func buildDistributionCounts( + chunks []*filer_pb.FileChunk, + volumeToOwner map[uint32]string, + volumeNodesList map[uint32][]string, +) (chunkToNode map[int]string, ownerCount map[string]int, copiesCount map[string]int, err error) { + chunkToNode = make(map[int]string) + ownerCount = make(map[string]int) + copiesCount = make(map[string]int) + + for i, chunk := range chunks { + var vid uint32 + if chunk.Fid == nil { + fileId, parseErr := needle.ParseFileIdFromString(chunk.GetFileIdString()) + if parseErr != nil { + return nil, nil, nil, fmt.Errorf("failed to parse file id for chunk %d: %v", i, parseErr) + } + vid = uint32(fileId.VolumeId) + } else { + vid = chunk.Fid.GetVolumeId() + } + owner, ok := volumeToOwner[vid] + if !ok { + return nil, nil, nil, fmt.Errorf("volume %d not found in topology", vid) + } + chunkToNode[i] = owner + ownerCount[owner]++ + for _, node := range volumeNodesList[vid] { + copiesCount[node]++ + } + } + return chunkToNode, ownerCount, copiesCount, nil +} + +// selectActiveNodes picks up to targetNodeCount nodes to participate in redistribution. +// When targetNodeCount is 0 all nodes are active. +func selectActiveNodes(nodeList []string, ownerCount map[string]int, targetNodeCount int) (activeNodeList []string, activeSet map[string]bool) { + activeNodeList = nodeList + if targetNodeCount > 0 && targetNodeCount < len(nodeList) { + var withChunks, withoutChunks []string + for _, node := range nodeList { + if ownerCount[node] > 0 { + withChunks = append(withChunks, node) + } else { + withoutChunks = append(withoutChunks, node) + } + } + n := targetNodeCount + if len(withChunks) >= n { + sort.Slice(withChunks, func(i, j int) bool { + return ownerCount[withChunks[i]] > ownerCount[withChunks[j]] + }) + activeNodeList = withChunks[:n] + } else { + activeNodeList = append(withChunks, withoutChunks[:n-len(withChunks)]...) + } + sort.Strings(activeNodeList) + } + + activeSet = make(map[string]bool, len(activeNodeList)) + for _, node := range activeNodeList { + activeSet[node] = true + } + return activeNodeList, activeSet +} + +// printCurrentDistribution writes the per-node chunk/copy table to writer. +func printCurrentDistribution( + writer io.Writer, + nodeList []string, + ownerCount, copiesCount map[string]int, + activeSet map[string]bool, + targetNodeCount int, + mode string, + totalChunks, totalNodes, totalCopies int, +) { + maxOwner := 0 + for _, cnt := range ownerCount { + if cnt > maxOwner { + maxOwner = cnt + } + } + + fmt.Fprintf(writer, "\nCurrent distribution:\n") + for _, node := range nodeList { + oc := ownerCount[node] + cc := copiesCount[node] + bar := strings.Repeat("#", int(math.Ceil(float64(oc)*30/math.Max(float64(maxOwner), 1)))) + marker := " " + if targetNodeCount > 0 && !activeSet[node] { + marker = "-" + } + if mode == "replica" { + fmt.Fprintf(writer, " %s %-20s %4d chunks %4d copies %-30s\n", marker, shortName(node), oc, cc, bar) + } else { + fmt.Fprintf(writer, " %s %-20s %4d chunks %-30s\n", marker, shortName(node), oc, bar) + } + } + fmt.Fprintf(writer, " Ideal: ~%.0f chunks per node", math.Ceil(float64(totalChunks)/float64(totalNodes))) + if mode == "replica" { + fmt.Fprintf(writer, ", ~%.0f copies per node", math.Ceil(float64(totalCopies)/float64(totalNodes))) + } + fmt.Fprintln(writer) +} + +// planDistribution computes the set of chunk moves required for the chosen mode. +func planDistribution( + mode string, + writer io.Writer, + activeNodeList, nodeList []string, + activeSet map[string]bool, + ownerCount, copiesCount map[string]int, + chunkToNode map[int]string, + chunks []*filer_pb.FileChunk, + volumeNodesList map[uint32][]string, + totalChunks, totalNodes, totalCopies int, +) (moves []chunkMove, ownerTarget map[string]int) { + type nodeExcess struct { + node string + excess int + } + + allRelevantNodes := relevantNodes(nodeList, activeSet, ownerCount) + ownerTarget = make(map[string]int) + + switch mode { + case "primary": + ownerTarget = computeOwnerTarget(activeNodeList, nodeList, activeSet, totalChunks, totalNodes) + moves = planOwnerMoves(ownerCount, ownerTarget, chunkToNode, chunks, allRelevantNodes) + + case "replica": + ownerTarget = computeOwnerTarget(activeNodeList, nodeList, activeSet, totalChunks, totalNodes) + moves = planOwnerMoves(ownerCount, ownerTarget, chunkToNode, chunks, allRelevantNodes) + + copiesMin := totalCopies / totalNodes + copiesRemainder := totalCopies % totalNodes + copiesTarget := make(map[string]int) + for i, node := range activeNodeList { + if i < copiesRemainder { + copiesTarget[node] = copiesMin + 1 + } else { + copiesTarget[node] = copiesMin + } + } + + alreadyMoving := make(map[int]bool) + for _, mv := range moves { + alreadyMoving[mv.chunkIndex] = true + } + + // local copy so the planner does not mutate the caller's copiesCount + localCopiesCount := maps.Clone(copiesCount) + + var copiesOver, copiesUnder []nodeExcess + for _, node := range activeNodeList { + diff := localCopiesCount[node] - copiesTarget[node] + if diff > 0 { + copiesOver = append(copiesOver, nodeExcess{node, diff}) + } else if diff < 0 { + copiesUnder = append(copiesUnder, nodeExcess{node, -diff}) + } + } + sort.Slice(copiesOver, func(i, j int) bool { + return copiesOver[i].excess > copiesOver[j].excess + }) + + for oi := range copiesOver { + over := &copiesOver[oi] + if over.excess <= 0 { + continue + } + for idx, chunk := range chunks { + if over.excess <= 0 { + break + } + if alreadyMoving[idx] { + continue + } + vid, vidErr := chunkVolumeId(chunk) + if vidErr != nil { + continue + } + if !slices.Contains(volumeNodesList[vid], over.node) { + continue + } + sort.Slice(copiesUnder, func(i, j int) bool { + return copiesUnder[i].excess > copiesUnder[j].excess + }) + for j := range copiesUnder { + if copiesUnder[j].excess > 0 { + moves = append(moves, chunkMove{ + chunkIndex: idx, + chunk: chunk, + fromNode: chunkToNode[idx], + toNode: copiesUnder[j].node, + }) + alreadyMoving[idx] = true + for _, srcNode := range volumeNodesList[vid] { + localCopiesCount[srcNode]-- + for k := range copiesOver { + if copiesOver[k].node == srcNode { + copiesOver[k].excess-- + } + } + } + copiesUnder[j].excess-- + localCopiesCount[copiesUnder[j].node]++ + over.excess-- + break + } + } + } + } + + case "round-robin": + type chunkWithIndex struct { + index int + offset int64 + } + sortedChunks := make([]chunkWithIndex, totalChunks) + for i, chunk := range chunks { + sortedChunks[i] = chunkWithIndex{index: i, offset: chunk.GetOffset()} + } + sort.Slice(sortedChunks, func(i, j int) bool { + return sortedChunks[i].offset < sortedChunks[j].offset + }) + + for i, sc := range sortedChunks { + targetNode := activeNodeList[i%totalNodes] + currentOwner := chunkToNode[sc.index] + ownerTarget[targetNode]++ + if currentOwner != targetNode { + moves = append(moves, chunkMove{ + chunkIndex: sc.index, + chunk: chunks[sc.index], + fromNode: currentOwner, + toNode: targetNode, + }) + } + } + + fmt.Fprintf(writer, "\nRound-robin assignment (sequential read optimized):\n") + fmt.Fprintf(writer, " First assignments: ") + for pos := range sortedChunks { + if pos >= 12 { + fmt.Fprintf(writer, "... ") + break + } + fmt.Fprintf(writer, "%s ", shortName(activeNodeList[pos%totalNodes])) + } + fmt.Fprintf(writer, "\n Pattern: ") + for i := 0; i < totalNodes && i < 6; i++ { + fmt.Fprintf(writer, "%s ", shortName(activeNodeList[i])) + } + if totalNodes > 6 { + fmt.Fprintf(writer, "...") + } + fmt.Fprintf(writer, "(repeating)\n") + } + + return moves, ownerTarget +} + +// printRedistributionPlan writes the before/after chunk count table for each node. +func printRedistributionPlan(writer io.Writer, allRelevantNodes []string, ownerCount, ownerTarget map[string]int, numMoves int) { + fmt.Fprintf(writer, "\nRedistribution plan: %d chunks to move\n", numMoves) + for _, node := range allRelevantNodes { + current := ownerCount[node] + target := ownerTarget[node] + diff := target - current + sign := "" + if diff > 0 { + sign = "+" + } + fmt.Fprintf(writer, " %-25s %4d -> %4d (%s%d)\n", shortName(node), current, target, sign, diff) + } +} + +// relevantNodes returns nodes that are either active or currently hold chunks. +func relevantNodes(nodeList []string, activeSet map[string]bool, ownerCount map[string]int) []string { + relevant := make([]string, 0, len(nodeList)) + for _, node := range nodeList { + if activeSet[node] || ownerCount[node] > 0 { + relevant = append(relevant, node) + } + } + return relevant +} + +// chunkVolumeId extracts the volume ID from a chunk, falling back to parsing +// the legacy FileId string when Fid is nil. +func chunkVolumeId(chunk *filer_pb.FileChunk) (uint32, error) { + if chunk.Fid != nil { + return chunk.Fid.GetVolumeId(), nil + } + fileId, err := needle.ParseFileIdFromString(chunk.GetFileIdString()) + if err != nil { + return 0, fmt.Errorf("failed to parse file id %q: %v", chunk.GetFileIdString(), err) + } + return uint32(fileId.VolumeId), nil +} + +// shortName returns the hostname portion (before the first ".") of a node ID. +func shortName(nodeID string) string { + return strings.Split(nodeID, ".")[0] +} + +// computeOwnerTarget returns the target chunk count per node for owner-based balancing. +func computeOwnerTarget(activeNodeList, nodeList []string, activeSet map[string]bool, totalChunks, totalNodes int) map[string]int { + target := make(map[string]int) + perNode := totalChunks / totalNodes + remainder := totalChunks % totalNodes + for i, node := range activeNodeList { + if i < remainder { + target[node] = perNode + 1 + } else { + target[node] = perNode + } + } + for _, node := range nodeList { + if !activeSet[node] { + target[node] = 0 + } + } + return target +} + +// executeChunkMoves runs all moves concurrently and returns the number of successfully moved chunks. +// Old chunk cleanup is handled automatically by the filer: filer.UpdateEntry calls +// deleteChunksIfNotNew which removes chunks present in the old entry but not the new one. +func executeChunkMoves( + commandEnv *CommandEnv, + writer io.Writer, + moves []chunkMove, + volumeInfoMap map[uint32]*master_pb.VolumeInformationMessage, +) (int, error) { + uploader, err := operation.NewUploader() + if err != nil { + return 0, fmt.Errorf("create uploader: %v", err) + } + + var mu sync.Mutex + movedCount := 0 + ewg := NewErrorWaitGroup(DefaultMaxParallelization) + + for i, mv := range moves { + ewg.Add(func() error { + chunk := mv.chunk + oldFidStr := chunk.GetFileIdString() + prefix := fmt.Sprintf(" [%d/%d] Moving %s from %s to %s", + i+1, len(moves), oldFidStr, + shortName(mv.fromNode), shortName(mv.toNode), + ) + + fail := func(msg string) { + mu.Lock() + fmt.Fprintf(writer, "%s ... FAILED (%s)\n", prefix, msg) + mu.Unlock() + } + + oldFid, parseErr := needle.ParseFileIdFromString(oldFidStr) + if parseErr != nil { + fail(fmt.Sprintf("parse fid: %v", parseErr)) + return nil + } + + downloadURLs, lookupErr := commandEnv.MasterClient.LookupVolumeServerUrl(oldFid.VolumeId.String()) + if lookupErr != nil || len(downloadURLs) == 0 { + fail(fmt.Sprintf("lookup source: %v", lookupErr)) + return nil + } + + // dlCancel must be called after reader.Close() — the response body is + // streamed into the uploader, so cancelling the context before the + // body is fully consumed causes "context canceled" upload failures. + dlCtx, dlCancel := context.WithTimeout(context.Background(), 5*time.Minute) + var resp *http.Response + var reader io.ReadCloser + var readErr error + for _, serverURL := range downloadURLs { + var dlReq *http.Request + dlReq, readErr = http.NewRequestWithContext(dlCtx, http.MethodGet, fmt.Sprintf("http://%s/%s", serverURL, oldFidStr), nil) + if readErr != nil { + continue + } + dlReq.Header.Add("Accept-Encoding", "gzip") + resp, readErr = util_http.GetGlobalHttpClient().Do(dlReq) + if readErr == nil && resp.StatusCode >= 400 { + util_http.CloseResponse(resp) + readErr = fmt.Errorf("download %s: %s", serverURL, resp.Status) + } + if readErr == nil { + reader = resp.Body + break + } + } + if readErr != nil { + dlCancel() + fail(fmt.Sprintf("download: %v", readErr)) + return nil + } + + contentType := resp.Header.Get("Content-Type") + isCompressed := resp.Header.Get("Content-Encoding") == "gzip" + md5Hash := resp.Header.Get("Content-MD5") + var filename string + if cd := resp.Header.Get("Content-Disposition"); len(cd) > 0 { + if _, after, found := strings.Cut(cd, "filename="); found { + filename = strings.Trim(after, "\"") + } + } + + // Buffer the full response body so the download context can be + // cancelled before the upload starts — otherwise the 5-minute + // dlCtx timeout would also cancel the (potentially slow) upload. + bodyBytes, readBodyErr := io.ReadAll(reader) + reader.Close() + util_http.CloseResponse(resp) + dlCancel() + if readBodyErr != nil { + fail(fmt.Sprintf("read body: %v", readBodyErr)) + return nil + } + + var replication, collection string + if vInfo, ok := volumeInfoMap[uint32(oldFid.VolumeId)]; ok { + replication = fmt.Sprintf("%03d", vInfo.GetReplicaPlacement()) + collection = vInfo.GetCollection() + } + + assignResult, assignErr := operation.Assign(context.Background(), commandEnv.MasterClient.GetMaster, commandEnv.option.GrpcDialOption, + &operation.VolumeAssignRequest{ + Count: 1, + Replication: replication, + Collection: collection, + DataNode: mv.toNode, + }) + if assignErr != nil { + fail(fmt.Sprintf("assign: %v", assignErr)) + return nil + } + + uploadURL := fmt.Sprintf("http://%s/%s", assignResult.Url, assignResult.Fid) + + var jwt security.EncodedJwt + if assignResult.Auth != "" { + jwt = assignResult.Auth + } else { + v := util.GetViper() + signingKey := v.GetString("jwt.signing.key") + if signingKey != "" { + expiresAfterSec := v.GetInt("jwt.signing.expires_after_seconds") + jwt = security.GenJwtForVolumeServer(security.SigningKey(signingKey), expiresAfterSec, assignResult.Fid) + } + } + + _, uploadErr, _ := uploader.Upload(context.Background(), bytes.NewReader(bodyBytes), &operation.UploadOption{ + UploadUrl: uploadURL, + Filename: filename, + IsInputCompressed: isCompressed, + Cipher: false, + MimeType: contentType, + Md5: md5Hash, + Jwt: jwt, + }) + + if uploadErr != nil { + fail(fmt.Sprintf("upload: %v", uploadErr)) + return nil + } + + newFid, fidParseErr := needle.ParseFileIdFromString(assignResult.Fid) + if fidParseErr != nil { + fail(fmt.Sprintf("parse new fid: %v", fidParseErr)) + return nil + } + + mu.Lock() + movedCount++ + chunk.Fid = &filer_pb.FileId{ + VolumeId: uint32(newFid.VolumeId), + FileKey: uint64(newFid.Key), + Cookie: uint32(newFid.Cookie), + } + chunk.FileId = "" + fmt.Fprintf(writer, "%s ... OK -> %s\n", prefix, assignResult.Fid) + mu.Unlock() + + return nil + }) + } + _ = ewg.Wait() + return movedCount, nil +} + +// newShellSaveAsChunk returns a SaveDataAsChunk function for use with +// filer.MaybeManifestize from shell context. The master selects placement — +// manifest chunks are metadata and don't need fan-out distribution. +func newShellSaveAsChunk(commandEnv *CommandEnv) filer.SaveDataAsChunkFunctionType { + return filer.SaveDataAsChunkFunctionType(func(reader io.Reader, name string, offset int64, tsNs int64, _ uint64) (*filer_pb.FileChunk, error) { + uploader, err := operation.NewUploader() + if err != nil { + return nil, fmt.Errorf("create uploader: %v", err) + } + data, err := io.ReadAll(reader) + if err != nil { + return nil, fmt.Errorf("read manifest data: %v", err) + } + assignResult, err := operation.Assign(context.Background(), commandEnv.MasterClient.GetMaster, commandEnv.option.GrpcDialOption, + &operation.VolumeAssignRequest{Count: 1}) + if err != nil { + return nil, fmt.Errorf("assign manifest chunk: %v", err) + } + var jwt security.EncodedJwt + if assignResult.Auth != "" { + jwt = assignResult.Auth + } else { + v := util.GetViper() + signingKey := v.GetString("jwt.signing.key") + if signingKey != "" { + expiresAfterSec := v.GetInt("jwt.signing.expires_after_seconds") + jwt = security.GenJwtForVolumeServer(security.SigningKey(signingKey), expiresAfterSec, assignResult.Fid) + } + } + uploadResult, uploadErr, _ := uploader.Upload(context.Background(), bytes.NewReader(data), &operation.UploadOption{ + UploadUrl: fmt.Sprintf("http://%s/%s", assignResult.Url, assignResult.Fid), + Filename: name, + Jwt: jwt, + }) + if uploadErr != nil { + return nil, fmt.Errorf("upload manifest chunk: %v", uploadErr) + } + if uploadResult.Error != "" { + return nil, fmt.Errorf("upload manifest chunk: %s", uploadResult.Error) + } + return uploadResult.ToPbFileChunk(assignResult.Fid, offset, tsNs), nil + }) +} + +// planOwnerMoves generates moves to balance owner-based chunk distribution. +func planOwnerMoves(ownerCount, ownerTarget map[string]int, chunkToNode map[int]string, chunks []*filer_pb.FileChunk, allRelevantNodes []string) []chunkMove { + type nodeExcess struct { + node string + excess int + } + var overNodes, underNodes []nodeExcess + for _, node := range allRelevantNodes { + diff := ownerCount[node] - ownerTarget[node] + if diff > 0 { + overNodes = append(overNodes, nodeExcess{node, diff}) + } else if diff < 0 { + underNodes = append(underNodes, nodeExcess{node, -diff}) + } + } + + var moves []chunkMove + for _, over := range overNodes { + toMove := over.excess + var nodeChunks []int + for i, node := range chunkToNode { + if node == over.node { + nodeChunks = append(nodeChunks, i) + } + } + moved := 0 + for _, idx := range nodeChunks { + if moved >= toMove { + break + } + for j := range underNodes { + if underNodes[j].excess > 0 { + moves = append(moves, chunkMove{ + chunkIndex: idx, + chunk: chunks[idx], + fromNode: over.node, + toNode: underNodes[j].node, + }) + underNodes[j].excess-- + moved++ + break + } + } + } + } + return moves +} diff --git a/weed/shell/command_fs_distribute_chunks_test.go b/weed/shell/command_fs_distribute_chunks_test.go new file mode 100644 index 000000000..97d94e564 --- /dev/null +++ b/weed/shell/command_fs_distribute_chunks_test.go @@ -0,0 +1,426 @@ +package shell + +import ( + "bytes" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// makeChunk builds a FileChunk with the given volume ID, file key, and byte offset. +func makeChunk(vid uint32, key uint64, offset int64) *filer_pb.FileChunk { + return &filer_pb.FileChunk{ + Fid: &filer_pb.FileId{VolumeId: vid, FileKey: key, Cookie: 1}, + Offset: offset, + Size: 1024 * 1024 * 16, + } +} + +// threeNodeTopology returns a minimal topology for 3 nodes with 3 volumes (one per node). +// +// node A owns vol 1, node B owns vol 2, node C owns vol 3 +// no replication: each volume exists on exactly one node. +func threeNodeTopology() (volumeToOwner map[uint32]string, volumeNodesList map[uint32][]string) { + volumeToOwner = map[uint32]string{ + 1: "nodeA", + 2: "nodeB", + 3: "nodeC", + } + volumeNodesList = map[uint32][]string{ + 1: {"nodeA"}, + 2: {"nodeB"}, + 3: {"nodeC"}, + } + return volumeToOwner, volumeNodesList +} + +// ── shortName ──────────────────────────────────────────────────────────────── + +func TestShortName(t *testing.T) { + tests := []struct { + input string + want string + }{ + {"nodeA.example.com", "nodeA"}, + {"nodeA", "nodeA"}, + {"10.0.0.1", "10"}, + {"", ""}, + } + for _, tt := range tests { + assert.Equal(t, tt.want, shortName(tt.input), "input=%q", tt.input) + } +} + +// ── relevantNodes ──────────────────────────────────────────────────────────── + +func TestRelevantNodes(t *testing.T) { + nodeList := []string{"nodeA", "nodeB", "nodeC", "nodeD"} + activeSet := map[string]bool{"nodeA": true, "nodeB": true} + ownerCount := map[string]int{"nodeA": 3, "nodeC": 1} // nodeC has chunks but isn't active + + got := relevantNodes(nodeList, activeSet, ownerCount) + + // nodeA (active+chunks), nodeB (active), nodeC (inactive but has chunks) + // nodeD (inactive, no chunks) → excluded + assert.ElementsMatch(t, []string{"nodeA", "nodeB", "nodeC"}, got) +} + +// ── computeOwnerTarget ─────────────────────────────────────────────────────── + +func TestComputeOwnerTarget_EvenDivision(t *testing.T) { + nodes := []string{"nodeA", "nodeB", "nodeC"} + activeSet := map[string]bool{"nodeA": true, "nodeB": true, "nodeC": true} + + target := computeOwnerTarget(nodes, nodes, activeSet, 9, 3) + + assert.Equal(t, 3, target["nodeA"]) + assert.Equal(t, 3, target["nodeB"]) + assert.Equal(t, 3, target["nodeC"]) +} + +func TestComputeOwnerTarget_UnevenDivision(t *testing.T) { + nodes := []string{"nodeA", "nodeB", "nodeC"} + activeSet := map[string]bool{"nodeA": true, "nodeB": true, "nodeC": true} + + // 10 chunks / 3 nodes → 3,3,4 (remainder goes to first nodes) + target := computeOwnerTarget(nodes, nodes, activeSet, 10, 3) + + total := target["nodeA"] + target["nodeB"] + target["nodeC"] + assert.Equal(t, 10, total) + for _, node := range nodes { + assert.True(t, target[node] == 3 || target[node] == 4, + "node %s has unexpected target %d", node, target[node]) + } +} + +func TestComputeOwnerTarget_InactiveNodesDrainToZero(t *testing.T) { + activeNodes := []string{"nodeA", "nodeB"} + allNodes := []string{"nodeA", "nodeB", "nodeC"} + activeSet := map[string]bool{"nodeA": true, "nodeB": true} + + target := computeOwnerTarget(activeNodes, allNodes, activeSet, 4, 2) + + assert.Equal(t, 0, target["nodeC"], "inactive node should have target 0") + assert.Equal(t, 2, target["nodeA"]) + assert.Equal(t, 2, target["nodeB"]) +} + +// ── buildDistributionCounts ────────────────────────────────────────────────── + +func TestBuildDistributionCounts(t *testing.T) { + volumeToOwner, volumeNodesList := threeNodeTopology() + + // 4 chunks on vol1 (nodeA), 2 on vol2 (nodeB), 0 on nodeC + chunks := []*filer_pb.FileChunk{ + makeChunk(1, 1, 0), + makeChunk(1, 2, 1), + makeChunk(1, 3, 2), + makeChunk(1, 4, 3), + makeChunk(2, 5, 4), + makeChunk(2, 6, 5), + } + + chunkToNode, ownerCount, copiesCount, err := buildDistributionCounts(chunks, volumeToOwner, volumeNodesList) + + require.NoError(t, err) + assert.Equal(t, 4, ownerCount["nodeA"]) + assert.Equal(t, 2, ownerCount["nodeB"]) + assert.Equal(t, 0, ownerCount["nodeC"]) + + // each chunk maps to its owner + assert.Equal(t, "nodeA", chunkToNode[0]) + assert.Equal(t, "nodeB", chunkToNode[4]) + + // no replication → copies == owner counts + assert.Equal(t, 4, copiesCount["nodeA"]) + assert.Equal(t, 2, copiesCount["nodeB"]) +} + +func TestBuildDistributionCounts_NilFidFallback(t *testing.T) { + volumeToOwner := map[uint32]string{3: "nodeA"} + volumeNodesList := map[uint32][]string{3: {"nodeA"}} + + // chunk with nil Fid but a valid legacy FileId string "3,0123456789abcdef01" + chunks := []*filer_pb.FileChunk{ + {Fid: nil, FileId: "3,0123456789abcdef01"}, + } + + _, ownerCount, _, err := buildDistributionCounts(chunks, volumeToOwner, volumeNodesList) + + require.NoError(t, err) + assert.Equal(t, 1, ownerCount["nodeA"]) +} + +// ── selectActiveNodes ──────────────────────────────────────────────────────── + +func TestSelectActiveNodes_AllNodes(t *testing.T) { + nodeList := []string{"nodeA", "nodeB", "nodeC"} + ownerCount := map[string]int{"nodeA": 5, "nodeB": 3, "nodeC": 2} + + active, activeSet := selectActiveNodes(nodeList, ownerCount, 0) + + assert.Equal(t, nodeList, active) + assert.Len(t, activeSet, 3) +} + +func TestSelectActiveNodes_LimitToTwo(t *testing.T) { + nodeList := []string{"nodeA", "nodeB", "nodeC"} + ownerCount := map[string]int{"nodeA": 5, "nodeB": 3, "nodeC": 0} + + active, activeSet := selectActiveNodes(nodeList, ownerCount, 2) + + assert.Len(t, active, 2) + // nodeA and nodeB have chunks and are the top-2 heaviest + assert.True(t, activeSet["nodeA"]) + assert.True(t, activeSet["nodeB"]) + assert.False(t, activeSet["nodeC"]) +} + +// ── planOwnerMoves ─────────────────────────────────────────────────────────── + +func TestPlanOwnerMoves_BasicBalance(t *testing.T) { + // nodeA has 5 chunks, nodeB/C have 0; target is 2,2,1 (or similar even split of 5) + ownerCount := map[string]int{"nodeA": 5, "nodeB": 0, "nodeC": 0} + ownerTarget := map[string]int{"nodeA": 1, "nodeB": 2, "nodeC": 2} + nodes := []string{"nodeA", "nodeB", "nodeC"} + + chunkToNode := map[int]string{0: "nodeA", 1: "nodeA", 2: "nodeA", 3: "nodeA", 4: "nodeA"} + chunks := []*filer_pb.FileChunk{ + makeChunk(1, 1, 0), + makeChunk(1, 2, 1), + makeChunk(1, 3, 2), + makeChunk(1, 4, 3), + makeChunk(1, 5, 4), + } + + moves := planOwnerMoves(ownerCount, ownerTarget, chunkToNode, chunks, nodes) + + // 4 chunks must leave nodeA + assert.Len(t, moves, 4) + for _, mv := range moves { + assert.Equal(t, "nodeA", mv.fromNode) + assert.NotEqual(t, "nodeA", mv.toNode) + } +} + +func TestPlanOwnerMoves_AlreadyBalanced(t *testing.T) { + ownerCount := map[string]int{"nodeA": 2, "nodeB": 2, "nodeC": 2} + ownerTarget := map[string]int{"nodeA": 2, "nodeB": 2, "nodeC": 2} + nodes := []string{"nodeA", "nodeB", "nodeC"} + chunkToNode := map[int]string{0: "nodeA", 1: "nodeA", 2: "nodeB", 3: "nodeB", 4: "nodeC", 5: "nodeC"} + chunks := make([]*filer_pb.FileChunk, 6) + for i := range chunks { + chunks[i] = makeChunk(1, uint64(i+1), int64(i)) + } + + moves := planOwnerMoves(ownerCount, ownerTarget, chunkToNode, chunks, nodes) + + assert.Empty(t, moves) +} + +// ── planDistribution: primary ───────────────────────────────────────────────── + +func TestPlanDistribution_Primary_BalancesChunks(t *testing.T) { + // 9 chunks, all on nodeA; 3 nodes; primary should balance to 3 each + volumeToOwner, volumeNodesList := threeNodeTopology() + chunks := make([]*filer_pb.FileChunk, 9) + for i := range chunks { + chunks[i] = makeChunk(1, uint64(i+1), int64(i)) // all on vol1 → nodeA + } + + chunkToNode, ownerCount, copiesCount, err := buildDistributionCounts(chunks, volumeToOwner, volumeNodesList) + require.NoError(t, err) + + nodeList := []string{"nodeA", "nodeB", "nodeC"} + activeNodeList, activeSet := selectActiveNodes(nodeList, ownerCount, 0) + totalChunks := len(chunks) + totalNodes := len(activeNodeList) + totalCopies := 0 + for _, cnt := range copiesCount { + totalCopies += cnt + } + + moves, ownerTarget := planDistribution("primary", &bytes.Buffer{}, + activeNodeList, nodeList, activeSet, + ownerCount, copiesCount, chunkToNode, chunks, volumeNodesList, + totalChunks, totalNodes, totalCopies) + + assert.Equal(t, 3, ownerTarget["nodeA"]) + assert.Equal(t, 3, ownerTarget["nodeB"]) + assert.Equal(t, 3, ownerTarget["nodeC"]) + // 6 chunks must be moved off nodeA + assert.Len(t, moves, 6) + for _, mv := range moves { + assert.Equal(t, "nodeA", mv.fromNode) + } +} + +func TestPlanDistribution_Primary_NoMovesWhenBalanced(t *testing.T) { + volumeToOwner, volumeNodesList := threeNodeTopology() + // 3 chunks evenly spread: one per volume/node + chunks := []*filer_pb.FileChunk{ + makeChunk(1, 1, 0), + makeChunk(2, 2, 1), + makeChunk(3, 3, 2), + } + + chunkToNode, ownerCount, copiesCount, err := buildDistributionCounts(chunks, volumeToOwner, volumeNodesList) + require.NoError(t, err) + + nodeList := []string{"nodeA", "nodeB", "nodeC"} + activeNodeList, activeSet := selectActiveNodes(nodeList, ownerCount, 0) + totalCopies := 0 + for _, cnt := range copiesCount { + totalCopies += cnt + } + + moves, _ := planDistribution("primary", &bytes.Buffer{}, + activeNodeList, nodeList, activeSet, + ownerCount, copiesCount, chunkToNode, chunks, volumeNodesList, + len(chunks), len(activeNodeList), totalCopies) + + assert.Empty(t, moves) +} + +// ── planDistribution: round-robin ───────────────────────────────────────────── + +func TestPlanDistribution_RoundRobin_AssignsByOffset(t *testing.T) { + // 6 chunks all on nodeA; round-robin over 3 nodes should spread them A,B,C,A,B,C + volumeToOwner, volumeNodesList := threeNodeTopology() + chunks := []*filer_pb.FileChunk{ + makeChunk(1, 1, 0), + makeChunk(1, 2, 1), + makeChunk(1, 3, 2), + makeChunk(1, 4, 3), + makeChunk(1, 5, 4), + makeChunk(1, 6, 5), + } + + chunkToNode, ownerCount, copiesCount, err := buildDistributionCounts(chunks, volumeToOwner, volumeNodesList) + require.NoError(t, err) + + nodeList := []string{"nodeA", "nodeB", "nodeC"} + activeNodeList, activeSet := selectActiveNodes(nodeList, ownerCount, 0) + totalCopies := 0 + for _, cnt := range copiesCount { + totalCopies += cnt + } + + var buf bytes.Buffer + moves, ownerTarget := planDistribution("round-robin", &buf, + activeNodeList, nodeList, activeSet, + ownerCount, copiesCount, chunkToNode, chunks, volumeNodesList, + len(chunks), len(activeNodeList), totalCopies) + + // 2 chunks stay on nodeA (positions 0,3), 3 moves to nodeB and 3 to nodeC + assert.Equal(t, 2, ownerTarget["nodeA"]) + assert.Equal(t, 2, ownerTarget["nodeB"]) + assert.Equal(t, 2, ownerTarget["nodeC"]) + // 4 chunks must move (nodeB and nodeC positions) + assert.Len(t, moves, 4) + + // output should contain the round-robin preview + assert.Contains(t, buf.String(), "Round-robin assignment") +} + +func TestPlanDistribution_RoundRobin_OffsetOrdering(t *testing.T) { + // chunks deliberately added in reverse offset order to verify sort + volumeToOwner := map[uint32]string{1: "nodeA", 2: "nodeB", 3: "nodeC"} + volumeNodesList := map[uint32][]string{1: {"nodeA"}, 2: {"nodeB"}, 3: {"nodeC"}} + + // 3 chunks already balanced but in non-sequential order + chunks := []*filer_pb.FileChunk{ + makeChunk(3, 3, 200), // highest offset → should be assigned last + makeChunk(1, 1, 0), // lowest offset → assigned to nodeList[0] + makeChunk(2, 2, 100), + } + + chunkToNode, ownerCount, copiesCount, err := buildDistributionCounts(chunks, volumeToOwner, volumeNodesList) + require.NoError(t, err) + + nodeList := []string{"nodeA", "nodeB", "nodeC"} + activeNodeList, activeSet := selectActiveNodes(nodeList, ownerCount, 0) + totalCopies := 0 + for _, cnt := range copiesCount { + totalCopies += cnt + } + + _, ownerTarget := planDistribution("round-robin", &bytes.Buffer{}, + activeNodeList, nodeList, activeSet, + ownerCount, copiesCount, chunkToNode, chunks, volumeNodesList, + len(chunks), len(activeNodeList), totalCopies) + + // every node gets exactly 1 chunk + assert.Equal(t, 1, ownerTarget["nodeA"]) + assert.Equal(t, 1, ownerTarget["nodeB"]) + assert.Equal(t, 1, ownerTarget["nodeC"]) +} + +// ── planDistribution: replica ───────────────────────────────────────────────── + +func TestPlanDistribution_Replica_BalancesOwnerAndCopies(t *testing.T) { + // 2-node replica topology: each volume has copies on both nodeA and nodeB + volumeToOwner := map[uint32]string{ + 1: "nodeA", // vid%2=1 → nodeA + 2: "nodeB", // vid%2=0 → nodeB (sorted: [nodeA,nodeB], idx 2%2=0 → nodeA... let's be explicit) + } + // explicit: vol1 primary=nodeA, vol2 primary=nodeB; both replicated on each other + volumeNodesList := map[uint32][]string{ + 1: {"nodeA", "nodeB"}, + 2: {"nodeA", "nodeB"}, + } + + // 6 chunks: 5 on vol1 (nodeA), 1 on vol2 (nodeB) + chunks := []*filer_pb.FileChunk{ + makeChunk(1, 1, 0), + makeChunk(1, 2, 1), + makeChunk(1, 3, 2), + makeChunk(1, 4, 3), + makeChunk(1, 5, 4), + makeChunk(2, 6, 5), + } + + chunkToNode, ownerCount, copiesCount, err := buildDistributionCounts(chunks, volumeToOwner, volumeNodesList) + require.NoError(t, err) + + nodeList := []string{"nodeA", "nodeB"} + activeNodeList, activeSet := selectActiveNodes(nodeList, ownerCount, 0) + totalCopies := 0 + for _, cnt := range copiesCount { + totalCopies += cnt + } + + moves, ownerTarget := planDistribution("replica", &bytes.Buffer{}, + activeNodeList, nodeList, activeSet, + ownerCount, copiesCount, chunkToNode, chunks, volumeNodesList, + len(chunks), len(activeNodeList), totalCopies) + + // after balancing, nodeA should have 3 and nodeB should have 3 + assert.Equal(t, 3, ownerTarget["nodeA"]) + assert.Equal(t, 3, ownerTarget["nodeB"]) + assert.NotEmpty(t, moves) + for _, mv := range moves { + assert.Equal(t, "nodeA", mv.fromNode) + assert.Equal(t, "nodeB", mv.toNode) + } +} + +// ── printRedistributionPlan ────────────────────────────────────────────────── + +func TestPrintRedistributionPlan_Output(t *testing.T) { + var buf bytes.Buffer + nodes := []string{"nodeA.local", "nodeB.local"} + ownerCount := map[string]int{"nodeA.local": 5, "nodeB.local": 1} + ownerTarget := map[string]int{"nodeA.local": 3, "nodeB.local": 3} + + printRedistributionPlan(&buf, nodes, ownerCount, ownerTarget, 2) + + out := buf.String() + assert.Contains(t, out, "2 chunks to move") + assert.Contains(t, out, "nodeA") + assert.Contains(t, out, "nodeB") + assert.Contains(t, out, "5 ->") + assert.Contains(t, out, "1 ->") +}