diff --git a/test/volume_server/framework/cluster_multi.go b/test/volume_server/framework/cluster_multi.go index 5a869999b..9076262d8 100644 --- a/test/volume_server/framework/cluster_multi.go +++ b/test/volume_server/framework/cluster_multi.go @@ -7,6 +7,7 @@ import ( "os/exec" "path/filepath" "strconv" + "strings" "sync" "testing" @@ -24,6 +25,7 @@ type MultiVolumeCluster struct { logsDir string keepLogs bool volumeServerCount int + disksPerServer int masterPort int masterGrpcPort int @@ -31,6 +33,7 @@ type MultiVolumeCluster struct { volumePorts []int volumeGrpcPorts []int volumePubPorts []int + volumeDiskDirs [][]string // per server: the data directories (one DiskLocation each) masterCmd *exec.Cmd volumeCmds []*exec.Cmd @@ -38,13 +41,25 @@ type MultiVolumeCluster struct { cleanupOnce sync.Once } -// StartMultiVolumeCluster starts a cluster with a specified number of volume servers +// StartMultiVolumeCluster starts a cluster with serverCount volume servers, one +// data directory (DiskLocation) each. func StartMultiVolumeCluster(t testing.TB, profile matrix.Profile, serverCount int) *MultiVolumeCluster { + return StartMultiVolumeClusterWithDisks(t, profile, serverCount, 1) +} + +// StartMultiVolumeClusterWithDisks starts serverCount volume servers, each with +// disksPerServer data directories passed to -dir as a comma list so every +// directory becomes its own DiskLocation. Lets tests exercise per-disk EC +// layouts (e.g. shards on one disk, a stub/index on a sibling disk). +func StartMultiVolumeClusterWithDisks(t testing.TB, profile matrix.Profile, serverCount, disksPerServer int) *MultiVolumeCluster { t.Helper() if serverCount < 1 { t.Fatalf("serverCount must be at least 1, got %d", serverCount) } + if disksPerServer < 1 { + t.Fatalf("disksPerServer must be at least 1, got %d", disksPerServer) + } weedBinary, err := FindOrBuildWeedBinary() if err != nil { @@ -60,10 +75,21 @@ func StartMultiVolumeCluster(t testing.TB, profile matrix.Profile, serverCount i logsDir := filepath.Join(baseDir, "logs") masterDataDir := filepath.Join(baseDir, "master") - // Create directories for master and all volume servers + // Create directories for master and all volume servers. With one disk the + // layout stays at baseDir/volumeN; multi-disk servers get baseDir/volumeN/diskD. dirs := []string{configDir, logsDir, masterDataDir} + volumeDiskDirs := make([][]string, serverCount) for i := 0; i < serverCount; i++ { - dirs = append(dirs, filepath.Join(baseDir, fmt.Sprintf("volume%d", i))) + serverDir := filepath.Join(baseDir, fmt.Sprintf("volume%d", i)) + volumeDiskDirs[i] = make([]string, disksPerServer) + for d := 0; d < disksPerServer; d++ { + if disksPerServer == 1 { + volumeDiskDirs[i][d] = serverDir + } else { + volumeDiskDirs[i][d] = filepath.Join(serverDir, fmt.Sprintf("disk%d", d)) + } + dirs = append(dirs, volumeDiskDirs[i][d]) + } } for _, dir := range dirs { if mkErr := os.MkdirAll(dir, 0o755); mkErr != nil { @@ -98,11 +124,13 @@ func StartMultiVolumeCluster(t testing.TB, profile matrix.Profile, serverCount i logsDir: logsDir, keepLogs: keepLogs, volumeServerCount: serverCount, + disksPerServer: disksPerServer, masterPort: masterPort, masterGrpcPort: masterGrpcPort, volumePorts: make([]int, serverCount), volumeGrpcPorts: make([]int, serverCount), volumePubPorts: make([]int, serverCount), + volumeDiskDirs: volumeDiskDirs, volumeCmds: make([]*exec.Cmd, serverCount), } @@ -133,8 +161,7 @@ func StartMultiVolumeCluster(t testing.TB, profile matrix.Profile, serverCount i // Start all volume servers for i := 0; i < serverCount; i++ { - volumeDataDir := filepath.Join(baseDir, fmt.Sprintf("volume%d", i)) - if err = c.startVolume(i, volumeDataDir); err != nil { + if err = c.startVolume(i, c.volumeDiskDirs[i]); err != nil { // Log current server's log for debugging startup failures volumeLog := fmt.Sprintf("volume%d.log", i) c.Stop() @@ -207,13 +234,17 @@ func (c *MultiVolumeCluster) startMaster(dataDir string) error { return c.masterCmd.Start() } -func (c *MultiVolumeCluster) startVolume(index int, dataDir string) error { +func (c *MultiVolumeCluster) startVolume(index int, dataDirs []string) error { logName := fmt.Sprintf("volume%d.log", index) logFile, err := os.Create(filepath.Join(c.logsDir, logName)) if err != nil { return err } + maxPerDir := make([]string, len(dataDirs)) + for i := range dataDirs { + maxPerDir[i] = "16" + } args := []string{ "-config_dir=" + c.configDir, "volume", @@ -221,8 +252,8 @@ func (c *MultiVolumeCluster) startVolume(index int, dataDir string) error { "-port=" + strconv.Itoa(c.volumePorts[index]), "-port.grpc=" + strconv.Itoa(c.volumeGrpcPorts[index]), "-port.public=" + strconv.Itoa(c.volumePubPorts[index]), - "-dir=" + dataDir, - "-max=16", + "-dir=" + strings.Join(dataDirs, ","), + "-max=" + strings.Join(maxPerDir, ","), "-master=127.0.0.1:" + strconv.Itoa(c.masterPort), "-readMode=" + c.profile.ReadMode, "-concurrentUploadLimitMB=" + strconv.Itoa(c.profile.ConcurrentUploadLimitMB), @@ -301,3 +332,17 @@ func (c *MultiVolumeCluster) VolumePublicURL(index int) string { func (c *MultiVolumeCluster) BaseDir() string { return c.baseDir } + +// VolumeDiskDir returns the data directory backing disk diskIndex on volume +// server serverIndex — i.e. the directory for that DiskLocation, for file +// inspection in tests. +func (c *MultiVolumeCluster) VolumeDiskDir(serverIndex, diskIndex int) string { + if serverIndex < 0 || serverIndex >= len(c.volumeDiskDirs) { + return "" + } + disks := c.volumeDiskDirs[serverIndex] + if diskIndex < 0 || diskIndex >= len(disks) { + return "" + } + return disks[diskIndex] +} diff --git a/weed/worker/tasks/erasure_coding/ec_task_julor_layout_e2e_test.go b/weed/worker/tasks/erasure_coding/ec_task_julor_layout_e2e_test.go new file mode 100644 index 000000000..854cb0b4d --- /dev/null +++ b/weed/worker/tasks/erasure_coding/ec_task_julor_layout_e2e_test.go @@ -0,0 +1,173 @@ +package erasure_coding + +import ( + "context" + "fmt" + "net/http" + "os" + "path/filepath" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/test/volume_server/framework" + "github.com/seaweedfs/seaweedfs/test/volume_server/matrix" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +// Reproduces the issue-9478 volume-13 layout end to end on a multi-server, +// multi-disk cluster and asserts the cluster converges to exactly one valid EC +// layout. Before the encode the volume is in the stuck state: a real .dat +// source, a 0-byte stub replica left by an interrupted encode, and partial +// stale EC shards from that attempt. After the worker runs (sweep stubs -> +// clear stale shards -> distribute -> mount -> verify -> delete originals): +// +// - every assigned shard is present on its target disk, with .ecx/.vif +// alongside it (the EC volume's info file survives even when shards land on +// the source's own disk); +// - no regular .dat/.idx remains anywhere — the source is deleted after +// verify, the stub was swept before distribute; +// - the stale shards that aren't part of the new assignment are gone. +func TestEcEncodeJulorLayoutConverges(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test in short mode") + } + + const disksPerServer = 2 + cluster := framework.StartMultiVolumeClusterWithDisks(t, matrix.P1(), 3, disksPerServer) + dialOption := grpc.WithTransportCredentials(insecure.NewCredentials()) + + const ( + volumeID = uint32(9500) + collection = "ec-julor" + ) + // S0: real source. S1: 0-byte stub + EC target. S2: pure EC target. + const ( + srcServer = 0 + stubServer = 1 + ) + addr := func(i int) string { return serverAddress(cluster, i) } + + conns := make([]*grpc.ClientConn, 3) + clients := make([]volume_server_pb.VolumeServerClient, 3) + for i := 0; i < 3; i++ { + conns[i], clients[i] = framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(i)) + defer conns[i].Close() + } + + ctx, cancel := context.WithTimeout(context.Background(), 180*time.Second) + defer cancel() + + // S0: real source replica with data. + framework.AllocateVolume(t, clients[srcServer], volumeID, collection) + httpClient := framework.NewHTTPClient() + for i := 0; i < 8; i++ { + fid := framework.NewFileID(volumeID, uint64(950000+i), uint32(0x9500CA00+i)) + payload := make([]byte, 4096) + for j := range payload { + payload[j] = byte(i + 1) + } + resp := framework.UploadBytes(t, httpClient, cluster.VolumeAdminURL(srcServer), fid, payload) + _ = framework.ReadAllAndClose(t, resp) + require.Equal(t, http.StatusCreated, resp.StatusCode) + } + + // Partial stale EC shards from a previous interrupted encode, mounted on the + // source. cleanupStaleEcShards must clear them before re-distribute. + _, err := clients[srcServer].VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{ + VolumeId: volumeID, Collection: collection, + }) + require.NoError(t, err) + staleShards := []uint32{11, 12, 13} + _, err = clients[srcServer].VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{ + VolumeId: volumeID, Collection: collection, ShardIds: staleShards, + }) + require.NoError(t, err) + + // S1: a 0-byte stub replica of the same volume (interrupted-encode leftover). + framework.AllocateVolume(t, clients[stubServer], volumeID, collection) + + // Plan: 14 shards spread across all three servers and both disks. The source + // server (S0) receives shards too, so its regular volume coexists with EC on + // the same disk before deletion. + targets := []*worker_pb.TaskTarget{ + {Node: addr(0), DiskId: 0, ShardIds: shardRange(0, 3)}, // S0 d0: 0,1,2 + {Node: addr(0), DiskId: 1, ShardIds: shardRange(3, 5)}, // S0 d1: 3,4 + {Node: addr(1), DiskId: 0, ShardIds: shardRange(5, 8)}, // S1 d0: 5,6,7 + {Node: addr(1), DiskId: 1, ShardIds: shardRange(8, 10)}, // S1 d1: 8,9 + {Node: addr(2), DiskId: 0, ShardIds: shardRange(10, 12)}, // S2 d0: 10,11 + {Node: addr(2), DiskId: 1, ShardIds: shardRange(12, 14)}, // S2 d1: 12,13 + } + + task := NewErasureCodingTask("ec-julor", addr(srcServer), volumeID, collection, dialOption) + params := &worker_pb.TaskParams{ + VolumeId: volumeID, + Collection: collection, + Sources: []*worker_pb.TaskSource{ + {Node: addr(srcServer), VolumeId: volumeID}, // real source + {Node: addr(stubServer), VolumeId: volumeID}, // 0-byte stub + {Node: addr(srcServer), VolumeId: volumeID, ShardIds: staleShards}, // stale EC shards to clear + }, + Targets: targets, + TaskParams: &worker_pb.TaskParams_ErasureCodingParams{ + ErasureCodingParams: &worker_pb.ErasureCodingTaskParams{ + DataShards: erasure_coding.DataShardsCount, + ParityShards: erasure_coding.ParityShardsCount, + WorkingDir: t.TempDir(), + }, + }, + } + + require.NoError(t, task.Execute(ctx, params)) + + base := fmt.Sprintf("%s_%d", collection, volumeID) + + // Shards are distributed per-server; the disk within a server is auto-selected + // when DiskId is 0, and on a multi-disk server shards can spread to a sibling + // disk while the .ecx/.vif index lives on one of them (the cross-disk EC + // layout). So assert per server (aggregate across its disks): each server + // holds exactly its assigned shards, its .ecx/.vif survive on at least one + // disk (the info file is not stripped, even on the source's own disk), and no + // regular .dat/.idx remains. Stale shards 11-13 mounted on the source must be + // gone — they are not in S0's assignment. + serverShards := map[int][]uint32{ + 0: shardRange(0, 5), // S0 (source): 0..4 + 1: shardRange(5, 10), // S1 (stub): 5..9 + 2: shardRange(10, 14), // S2 (target): 10..13 + } + allShards := map[uint32]bool{} + for server, want := range serverShards { + found := map[uint32]bool{} + hasEcx, hasVif := false, false + for d := 0; d < disksPerServer; d++ { + dir := cluster.VolumeDiskDir(server, d) + requireAbsent(t, dir, base+".dat") + requireAbsent(t, dir, base+".idx") + for id := 0; id < erasure_coding.TotalShardsCount; id++ { + if ecFileExists(dir, fmt.Sprintf("%s.ec%02d", base, id)) { + found[uint32(id)] = true + allShards[uint32(id)] = true + } + } + hasEcx = hasEcx || ecFileExists(dir, base+".ecx") + hasVif = hasVif || ecFileExists(dir, base+".vif") + } + gotIDs := make([]uint32, 0, len(found)) + for id := range found { + gotIDs = append(gotIDs, id) + } + require.ElementsMatch(t, want, gotIDs, "server %d should hold exactly its assigned shards", server) + require.True(t, hasEcx, "server %d must keep its .ecx", server) + require.True(t, hasVif, "server %d must keep its .vif (not stripped by the original-volume delete)", server) + } + require.Len(t, allShards, erasure_coding.TotalShardsCount, "the full shard set must be present across the cluster") +} + +func ecFileExists(dir, name string) bool { + _, err := os.Stat(filepath.Join(dir, name)) + return err == nil +}