Github-Mirror/seaweedfs
1
0 Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-10-30 17:00:48 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into codex-rust-volume-server-bootstrap

Browse Source 
# Conflicts:
#	.github/workflows/volume-server-integration-tests.yml
#	test/volume_server/framework/cluster.go
#	test/volume_server/framework/cluster_dual.go
#	test/volume_server/framework/volume_fixture.go
#	test/volume_server/grpc/data_stream_success_test.go
#	test/volume_server/http/public_cors_methods_test.go
#	test/volume_server/matrix/config_profiles.go
This commit is contained in:
user.email committed 2026-10-24 22:23:33 -07:00
2 parents 73e4173789 ec8f979350
commit 9436099d55
3609 files changed
+709333 -121315
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

test/volume_server/framework/cluster.go
+144 -191
View File
@@ -18,26 +18,44 @@ import (
"testing"
"time"
"github.com/seaweedfs/seaweedfs/test/testutil"
"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
"github.com/seaweedfs/seaweedfs/weed/storage/types"
)
// goBuildTags names the build tags a server the harness compiles must carry to
// hold the same offsets as the test binary asking for it. A 4-byte server and a
// 5-byte one disagree about every .idx row and reject each other's .vif, and
// the mixed Go/Rust suites run both at once.
func goBuildTags() []string {
if types.OffsetSize == 5 {
return []string{"-tags", "5BytesOffset"}
}
return nil
}
const (
defaultWaitTimeout = 30 * time.Second
defaultWaitTick = 200 * time.Millisecond
testVolumeSizeLimitMB = 32
)
var (
weedBinaryOnce sync.Once
weedBinaryPath string
weedBinaryErr error
)
// Cluster is a lightweight SeaweedFS master + one volume server test harness.
type Cluster struct {
testingTB testing.TB
profile matrix.Profile
weedBinary string
volumeBinary string
baseDir string
configDir string
logsDir string
keepLogs bool
weedBinary string
baseDir string
configDir string
logsDir string
keepLogs bool
masterPort int
masterGrpcPort int
@@ -48,16 +66,29 @@ type Cluster struct {
masterCmd *exec.Cmd
volumeCmd *exec.Cmd
volumeDataDirs []string
cleanupOnce sync.Once
}
// StartSingleVolumeCluster boots one master and one volume server.
func StartSingleVolumeCluster(t testing.TB, profile matrix.Profile) *Cluster {
t.Helper()
return StartSingleVolumeClusterWithDataDirs(t, profile, 1)
}
weedBinary, volumeBinary, err := FindOrBuildServerBinaries()
// StartSingleVolumeClusterWithDataDirs boots one master and one volume server
// with dataDirCount separate data directories (passed to -dir as a comma list).
// Each directory becomes its own DiskLocation on the volume server, letting
// tests exercise multi-disk EC placement paths.
func StartSingleVolumeClusterWithDataDirs(t testing.TB, profile matrix.Profile, dataDirCount int) *Cluster {
t.Helper()
if dataDirCount < 1 {
t.Fatalf("dataDirCount must be >= 1, got %d", dataDirCount)
}
weedBinary, err := FindOrBuildWeedBinary()
if err != nil {
t.Fatalf("resolve server binaries: %v", err)
t.Fatalf("resolve weed binary: %v", err)
}
baseDir, keepLogs, err := newWorkDir()
@@ -68,8 +99,19 @@ func StartSingleVolumeCluster(t testing.TB, profile matrix.Profile) *Cluster {
configDir := filepath.Join(baseDir, "config")
logsDir := filepath.Join(baseDir, "logs")
masterDataDir := filepath.Join(baseDir, "master")
volumeDataDir := filepath.Join(baseDir, "volume")
for _, dir := range []string{configDir, logsDir, masterDataDir, volumeDataDir} {
volumeDataDirs := make([]string, dataDirCount)
// Single-dir layout stays at baseDir/volume so existing fixtures
// (CorruptDatFile etc.) that hardcode that path keep working. Only
// multi-dir clusters get the "volumeN" layout.
for i := 0; i < dataDirCount; i++ {
if dataDirCount == 1 {
volumeDataDirs[i] = filepath.Join(baseDir, "volume")
} else {
volumeDataDirs[i] = filepath.Join(baseDir, fmt.Sprintf("volume%d", i))
}
}
setupDirs := append([]string{configDir, logsDir, masterDataDir}, volumeDataDirs...)
for _, dir := range setupDirs {
if mkErr := os.MkdirAll(dir, 0o755); mkErr != nil {
t.Fatalf("create %s: %v", dir, mkErr)
}
@@ -79,21 +121,17 @@ func StartSingleVolumeCluster(t testing.TB, profile matrix.Profile) *Cluster {
t.Fatalf("write security config: %v", err)
}
masterPort, masterGrpcPort, err := allocateMasterPortPair()
if err != nil {
t.Fatalf("allocate master port pair: %v", err)
}
ports, err := allocatePorts(3)
miniPorts, ports, err := testutil.AllocatePortSet(1, 3)
if err != nil {
t.Fatalf("allocate ports: %v", err)
}
masterPort := miniPorts[0]
masterGrpcPort := masterPort + testutil.GrpcPortOffset
c := &Cluster{
testingTB: t,
profile: profile,
weedBinary: weedBinary,
volumeBinary: volumeBinary,
baseDir: baseDir,
configDir: configDir,
logsDir: logsDir,
@@ -118,11 +156,12 @@ func StartSingleVolumeCluster(t testing.TB, profile matrix.Profile) *Cluster {
t.Fatalf("wait for master readiness: %v\nmaster log tail:\n%s", err, masterLog)
}
if err = c.startVolume(volumeDataDir); err != nil {
if err = c.startVolume(volumeDataDirs); err != nil {
masterLog := c.tailLog("master.log")
c.Stop()
t.Fatalf("start volume: %v\nmaster log tail:\n%s", err, masterLog)
}
c.volumeDataDirs = volumeDataDirs
if err = c.waitForHTTP(c.VolumeAdminURL() + "/status"); err != nil {
volumeLog := c.tailLog("volume.log")
c.Stop()
@@ -157,6 +196,38 @@ func (c *Cluster) Stop() {
})
}
// RestartVolumeServer kills the volume server and starts a new one against
// the same data dirs and ports. The master keeps running. The previous run's
// volume.log is moved to volume.log.previous so the first run's logs survive
// a second-run startup failure.
func (c *Cluster) RestartVolumeServer() {
c.testingTB.Helper()
stopProcess(c.volumeCmd)
c.volumeCmd = nil
// Rotate the log; absent on the first call after a clean start, which
// is fine — startVolume will create it. Any real filesystem failure
// surfaces immediately on the next os.Create in startVolume.
oldLog := filepath.Join(c.logsDir, "volume.log")
_ = os.Rename(oldLog, filepath.Join(c.logsDir, "volume.log.previous"))
if err := c.startVolume(c.volumeDataDirs); err != nil {
c.testingTB.Fatalf("restart volume server: %v", err)
}
if err := c.waitForHTTP(c.VolumeAdminURL() + "/status"); err != nil {
c.testingTB.Fatalf("wait for volume admin readiness after restart: %v\nvolume log tail:\n%s", err, c.tailLog("volume.log"))
}
if err := c.waitForTCP(c.VolumeGRPCAddress()); err != nil {
c.testingTB.Fatalf("wait for volume grpc readiness after restart: %v\nvolume log tail:\n%s", err, c.tailLog("volume.log"))
}
}
// StopVolumeServer kills the volume server but leaves the master and data
// dirs alone. Pair with RestartVolumeServer or Stop.
func (c *Cluster) StopVolumeServer() {
c.testingTB.Helper()
stopProcess(c.volumeCmd)
c.volumeCmd = nil
}
func (c *Cluster) startMaster(dataDir string) error {
logFile, err := os.Create(filepath.Join(c.logsDir, "master.log"))
if err != nil {
@@ -182,12 +253,16 @@ func (c *Cluster) startMaster(dataDir string) error {
return c.masterCmd.Start()
}
func (c *Cluster) startVolume(dataDir string) error {
func (c *Cluster) startVolume(dataDirs []string) error {
logFile, err := os.Create(filepath.Join(c.logsDir, "volume.log"))
if err != nil {
return err
}
maxPerDir := make([]string, len(dataDirs))
for i := range dataDirs {
maxPerDir[i] = "16"
}
args := []string{
"-config_dir=" + c.configDir,
"volume",
@@ -195,12 +270,16 @@ func (c *Cluster) startVolume(dataDir string) error {
"-port=" + strconv.Itoa(c.volumePort),
"-port.grpc=" + strconv.Itoa(c.volumeGrpcPort),
"-port.public=" + strconv.Itoa(c.volumePubPort),
"-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),
"-concurrentDownloadLimitMB=" + strconv.Itoa(c.profile.ConcurrentDownloadLimitMB),
// Integration tests deliberately exercise loopback S3 endpoints
// (the test rig boots weed-mini next to the volume server); allow
// the SSRF guard to be bypassed for them.
"-volume.allowUntrustedRemoteEndpoints",
}
if c.profile.InflightUploadTimeout > 0 {
args = append(args, "-inflightUploadDataTimeout="+c.profile.InflightUploadTimeout.String())
@@ -212,7 +291,7 @@ func (c *Cluster) startVolume(dataDir string) error {
args = append(args, "-fileSizeLimitMB="+strconv.Itoa(c.profile.FileSizeLimitMB))
}
c.volumeCmd = exec.Command(c.volumeBinary, args...)
c.volumeCmd = exec.Command(c.weedBinary, args...)
c.volumeCmd.Dir = c.baseDir
c.volumeCmd.Stdout = logFile
c.volumeCmd.Stderr = logFile
@@ -268,73 +347,6 @@ func stopProcess(cmd *exec.Cmd) {
}
}
func allocatePorts(count int) ([]int, error) {
const minPort = 10000
const maxPort = 55535
const host = "127.0.0.1"
rangeSize := maxPort - minPort + 1
listeners := make([]net.Listener, 0, count)
ports := make([]int, 0, count)
seen := make(map[int]struct{}, count)
closeAll := func() {
for _, ll := range listeners {
_ = ll.Close()
}
}
startOffset := int(time.Now().UnixNano() % int64(rangeSize))
for i := 0; i < count; i++ {
found := false
for offset := 0; offset < rangeSize; offset++ {
port := minPort + (startOffset+offset)%rangeSize
if _, exists := seen[port]; exists {
continue
}
l, err := net.Listen("tcp", net.JoinHostPort(host, strconv.Itoa(port)))
if err != nil {
continue
}
seen[port] = struct{}{}
listeners = append(listeners, l)
ports = append(ports, port)
found = true
startOffset = (startOffset + offset + 1) % rangeSize
break
}
if !found {
closeAll()
return nil, fmt.Errorf("unable to allocate %d ports within range [%d,%d]", count, minPort, maxPort)
}
}
for _, l := range listeners {
_ = l.Close()
}
return ports, nil
}
func allocateMasterPortPair() (int, int, error) {
for masterPort := 10000; masterPort <= 55535; masterPort++ {
masterGrpcPort := masterPort + 10000
l1, err := net.Listen("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(masterPort)))
if err != nil {
continue
}
l2, err := net.Listen("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(masterGrpcPort)))
if err != nil {
_ = l1.Close()
continue
}
_ = l2.Close()
_ = l1.Close()
return masterPort, masterGrpcPort, nil
}
return 0, 0, errors.New("unable to find available master port pair")
}
func newWorkDir() (dir string, keepLogs bool, err error) {
keepLogs = os.Getenv("VOLUME_SERVER_IT_KEEP_LOGS") == "1"
dir, err = os.MkdirTemp("", "seaweedfs_volume_server_it_")
@@ -359,7 +371,7 @@ func writeSecurityConfig(configDir string, profile matrix.Profile) error {
b.WriteString("\"\n")
b.WriteString("expires_after_seconds = 60\n")
}
if profile.AccessUI {
if profile.EnableUIAccess {
if b.Len() > 0 {
b.WriteString("\n")
}
@@ -381,111 +393,45 @@ func FindOrBuildWeedBinary() (string, error) {
return "", fmt.Errorf("WEED_BINARY is set but not executable: %s", fromEnv)
}
repoRoot, err := detectRepoRoot()
if err != nil {
return "", err
}
if candidate := filepath.Join(repoRoot, "weed", "weed"); isExecutableFile(candidate) {
return candidate, nil
}
binDir := filepath.Join(os.TempDir(), "seaweedfs_volume_server_it_bin")
if err := os.MkdirAll(binDir, 0o755); err != nil {
return "", fmt.Errorf("create binary directory %s: %w", binDir, err)
}
binPath := filepath.Join(binDir, "weed")
if isExecutableFile(binPath) {
return binPath, nil
}
cmd := exec.Command("go", "build", "-o", binPath, ".")
cmd.Dir = filepath.Join(repoRoot, "weed")
var out bytes.Buffer
cmd.Stdout = &out
cmd.Stderr = &out
if err := cmd.Run(); err != nil {
return "", fmt.Errorf("build weed binary: %w\n%s", err, out.String())
}
if !isExecutableFile(binPath) {
return "", fmt.Errorf("built weed binary is not executable: %s", binPath)
}
return binPath, nil
}
// FindOrBuildServerBinaries returns master and volume executables.
// Master always runs from the Go weed binary; volume can be switched via env.
func FindOrBuildServerBinaries() (masterBinary string, volumeBinary string, err error) {
masterBinary, err = FindOrBuildWeedBinary()
if err != nil {
return "", "", err
}
volumeBinary, err = FindOrBuildVolumeServerBinary(masterBinary)
if err != nil {
return "", "", err
}
return masterBinary, volumeBinary, nil
}
// FindOrBuildVolumeServerBinary resolves the executable used for volume-server processes.
//
// Behavior:
// - `VOLUME_SERVER_BINARY=/path/to/bin`: use explicit executable path.
// - `VOLUME_SERVER_IMPL=rust`: build/use Rust volume server launcher.
// - default: use the same Go `weed` binary.
func FindOrBuildVolumeServerBinary(defaultBinary string) (string, error) {
if fromEnv := os.Getenv("VOLUME_SERVER_BINARY"); fromEnv != "" {
if isExecutableFile(fromEnv) {
return fromEnv, nil
weedBinaryOnce.Do(func() {
repoRoot := ""
if _, file, _, ok := runtime.Caller(0); ok {
repoRoot = filepath.Clean(filepath.Join(filepath.Dir(file), "..", "..", ".."))
}
if repoRoot == "" {
weedBinaryErr = errors.New("unable to detect repository root")
return
}
return "", fmt.Errorf("VOLUME_SERVER_BINARY is set but not executable: %s", fromEnv)
}
impl := strings.ToLower(strings.TrimSpace(os.Getenv("VOLUME_SERVER_IMPL")))
if impl == "" || impl == "go" {
return defaultBinary, nil
}
if impl != "rust" {
return "", fmt.Errorf("unsupported VOLUME_SERVER_IMPL %q (supported: go, rust)", impl)
}
binDir := filepath.Join(os.TempDir(), "seaweedfs_volume_server_it_bin")
if err := os.MkdirAll(binDir, 0o755); err != nil {
weedBinaryErr = fmt.Errorf("create binary directory %s: %w", binDir, err)
return
}
binPath := filepath.Join(binDir, "weed")
repoRoot, err := detectRepoRoot()
if err != nil {
return "", err
}
return FindOrBuildRustVolumeServerBinary(repoRoot)
}
args := append([]string{"build"}, goBuildTags()...)
args = append(args, "-o", binPath, ".")
cmd := exec.Command("go", args...)
cmd.Dir = filepath.Join(repoRoot, "weed")
var out bytes.Buffer
cmd.Stdout = &out
cmd.Stderr = &out
if err := cmd.Run(); err != nil {
weedBinaryErr = fmt.Errorf("build weed binary: %w\n%s", err, out.String())
return
}
if !isExecutableFile(binPath) {
weedBinaryErr = fmt.Errorf("built weed binary is not executable: %s", binPath)
return
}
weedBinaryPath = binPath
})
// FindOrBuildRustVolumeServerBinary builds the Rust volume server launcher when needed.
func FindOrBuildRustVolumeServerBinary(repoRoot string) (string, error) {
manifestPath := filepath.Join(repoRoot, "rust", "volume_server", "Cargo.toml")
if _, err := os.Stat(manifestPath); err != nil {
return "", fmt.Errorf("rust volume server manifest not found at %s: %w", manifestPath, err)
if weedBinaryErr != nil {
return "", weedBinaryErr
}
targetDir := filepath.Join(os.TempDir(), "seaweedfs_volume_server_it_rust_target")
binPath := filepath.Join(targetDir, "release", "weed-volume-rs")
if isExecutableFile(binPath) && os.Getenv("VOLUME_SERVER_RUST_REBUILD") != "1" {
return binPath, nil
}
cmd := exec.Command("cargo", "build", "--release", "--manifest-path", manifestPath, "--target-dir", targetDir)
var out bytes.Buffer
cmd.Stdout = &out
cmd.Stderr = &out
if err := cmd.Run(); err != nil {
return "", fmt.Errorf("build rust volume server binary: %w\n%s", err, out.String())
}
if !isExecutableFile(binPath) {
return "", fmt.Errorf("built rust volume server binary is not executable: %s", binPath)
}
return binPath, nil
}
func detectRepoRoot() (string, error) {
if _, file, _, ok := runtime.Caller(0); ok {
return filepath.Clean(filepath.Join(filepath.Dir(file), "..", "..", "..")), nil
}
return "", errors.New("unable to detect repository root")
return weedBinaryPath, nil
}
func isExecutableFile(path string) bool {
@@ -551,3 +497,10 @@ func (c *Cluster) VolumePublicURL() string {
func (c *Cluster) BaseDir() string {
return c.baseDir
}
// VolumeDataDirs returns the data directories the volume server was started with.
// Index 0 corresponds to DiskLocation 0, index 1 to DiskLocation 1, and so on.
// Tests can scan these directories to verify where files physically landed.
func (c *Cluster) VolumeDataDirs() []string {
return append([]string(nil), c.volumeDataDirs...)
}
test/volume_server/framework/cluster_dual.go
+7 -285
View File
@@ -1,295 +1,17 @@
package framework
import (
"fmt"
"net"
"os"
"os/exec"
"path/filepath"
"strconv"
"sync"
"testing"
"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
)
type DualVolumeCluster struct {
testingTB testing.TB
profile matrix.Profile
// DualVolumeCluster is deprecated. Use MultiVolumeCluster instead.
// For backward compatibility, it's a type alias for MultiVolumeCluster.
type DualVolumeCluster = MultiVolumeCluster
weedBinary string
volumeBinary string
baseDir string
configDir string
logsDir string
keepLogs bool
masterPort int
masterGrpcPort int
volumePort0 int
volumeGrpcPort0 int
volumePubPort0 int
volumePort1 int
volumeGrpcPort1 int
volumePubPort1 int
masterCmd *exec.Cmd
volumeCmd0 *exec.Cmd
volumeCmd1 *exec.Cmd
cleanupOnce sync.Once
}
func StartDualVolumeCluster(t testing.TB, profile matrix.Profile) *DualVolumeCluster {
t.Helper()
weedBinary, volumeBinary, err := FindOrBuildServerBinaries()
if err != nil {
t.Fatalf("resolve server binaries: %v", err)
}
baseDir, keepLogs, err := newWorkDir()
if err != nil {
t.Fatalf("create temp test directory: %v", err)
}
configDir := filepath.Join(baseDir, "config")
logsDir := filepath.Join(baseDir, "logs")
masterDataDir := filepath.Join(baseDir, "master")
volumeDataDir0 := filepath.Join(baseDir, "volume0")
volumeDataDir1 := filepath.Join(baseDir, "volume1")
for _, dir := range []string{configDir, logsDir, masterDataDir, volumeDataDir0, volumeDataDir1} {
if mkErr := os.MkdirAll(dir, 0o755); mkErr != nil {
t.Fatalf("create %s: %v", dir, mkErr)
}
}
if err = writeSecurityConfig(configDir, profile); err != nil {
t.Fatalf("write security config: %v", err)
}
masterPort, masterGrpcPort, err := allocateMasterPortPair()
if err != nil {
t.Fatalf("allocate master port pair: %v", err)
}
ports, err := allocatePorts(6)
if err != nil {
t.Fatalf("allocate volume ports: %v", err)
}
c := &DualVolumeCluster{
testingTB: t,
profile: profile,
weedBinary: weedBinary,
volumeBinary: volumeBinary,
baseDir: baseDir,
configDir: configDir,
logsDir: logsDir,
keepLogs: keepLogs,
masterPort: masterPort,
masterGrpcPort: masterGrpcPort,
volumePort0: ports[0],
volumeGrpcPort0: ports[1],
volumePubPort0: ports[0],
volumePort1: ports[2],
volumeGrpcPort1: ports[3],
volumePubPort1: ports[2],
}
if profile.SplitPublicPort {
c.volumePubPort0 = ports[4]
c.volumePubPort1 = ports[5]
}
if err = c.startMaster(masterDataDir); err != nil {
c.Stop()
t.Fatalf("start master: %v", err)
}
if err = c.waitForHTTP(c.MasterURL() + "/dir/status"); err != nil {
masterLog := c.tailLog("master.log")
c.Stop()
t.Fatalf("wait for master readiness: %v\nmaster log tail:\n%s", err, masterLog)
}
if err = c.startVolume(0, volumeDataDir0); err != nil {
masterLog := c.tailLog("master.log")
c.Stop()
t.Fatalf("start first volume server: %v\nmaster log tail:\n%s", err, masterLog)
}
if err = c.waitForHTTP(c.VolumeAdminURL(0) + "/status"); err != nil {
volumeLog := c.tailLog("volume0.log")
c.Stop()
t.Fatalf("wait for first volume readiness: %v\nvolume log tail:\n%s", err, volumeLog)
}
if err = c.waitForTCP(c.VolumeGRPCAddress(0)); err != nil {
volumeLog := c.tailLog("volume0.log")
c.Stop()
t.Fatalf("wait for first volume grpc readiness: %v\nvolume log tail:\n%s", err, volumeLog)
}
if err = c.startVolume(1, volumeDataDir1); err != nil {
volumeLog := c.tailLog("volume0.log")
c.Stop()
t.Fatalf("start second volume server: %v\nfirst volume log tail:\n%s", err, volumeLog)
}
if err = c.waitForHTTP(c.VolumeAdminURL(1) + "/status"); err != nil {
volumeLog := c.tailLog("volume1.log")
c.Stop()
t.Fatalf("wait for second volume readiness: %v\nvolume log tail:\n%s", err, volumeLog)
}
if err = c.waitForTCP(c.VolumeGRPCAddress(1)); err != nil {
volumeLog := c.tailLog("volume1.log")
c.Stop()
t.Fatalf("wait for second volume grpc readiness: %v\nvolume log tail:\n%s", err, volumeLog)
}
t.Cleanup(func() {
c.Stop()
})
return c
}
func (c *DualVolumeCluster) Stop() {
if c == nil {
return
}
c.cleanupOnce.Do(func() {
stopProcess(c.volumeCmd1)
stopProcess(c.volumeCmd0)
stopProcess(c.masterCmd)
if !c.keepLogs && !c.testingTB.Failed() {
_ = os.RemoveAll(c.baseDir)
} else if c.baseDir != "" {
c.testingTB.Logf("volume server integration logs kept at %s", c.baseDir)
}
})
}
func (c *DualVolumeCluster) startMaster(dataDir string) error {
logFile, err := os.Create(filepath.Join(c.logsDir, "master.log"))
if err != nil {
return err
}
args := []string{
"-config_dir=" + c.configDir,
"master",
"-ip=127.0.0.1",
"-port=" + strconv.Itoa(c.masterPort),
"-port.grpc=" + strconv.Itoa(c.masterGrpcPort),
"-mdir=" + dataDir,
"-peers=none",
"-volumeSizeLimitMB=" + strconv.Itoa(testVolumeSizeLimitMB),
"-defaultReplication=000",
}
c.masterCmd = exec.Command(c.weedBinary, args...)
c.masterCmd.Dir = c.baseDir
c.masterCmd.Stdout = logFile
c.masterCmd.Stderr = logFile
return c.masterCmd.Start()
}
func (c *DualVolumeCluster) startVolume(index int, dataDir string) error {
logName := fmt.Sprintf("volume%d.log", index)
logFile, err := os.Create(filepath.Join(c.logsDir, logName))
if err != nil {
return err
}
volumePort := c.volumePort0
volumeGrpcPort := c.volumeGrpcPort0
volumePubPort := c.volumePubPort0
if index == 1 {
volumePort = c.volumePort1
volumeGrpcPort = c.volumeGrpcPort1
volumePubPort = c.volumePubPort1
}
args := []string{
"-config_dir=" + c.configDir,
"volume",
"-ip=127.0.0.1",
"-port=" + strconv.Itoa(volumePort),
"-port.grpc=" + strconv.Itoa(volumeGrpcPort),
"-port.public=" + strconv.Itoa(volumePubPort),
"-dir=" + dataDir,
"-max=16",
"-master=127.0.0.1:" + strconv.Itoa(c.masterPort),
"-readMode=" + c.profile.ReadMode,
"-concurrentUploadLimitMB=" + strconv.Itoa(c.profile.ConcurrentUploadLimitMB),
"-concurrentDownloadLimitMB=" + strconv.Itoa(c.profile.ConcurrentDownloadLimitMB),
}
if c.profile.InflightUploadTimeout > 0 {
args = append(args, "-inflightUploadDataTimeout="+c.profile.InflightUploadTimeout.String())
}
if c.profile.InflightDownloadTimeout > 0 {
args = append(args, "-inflightDownloadDataTimeout="+c.profile.InflightDownloadTimeout.String())
}
cmd := exec.Command(c.volumeBinary, args...)
cmd.Dir = c.baseDir
cmd.Stdout = logFile
cmd.Stderr = logFile
if err = cmd.Start(); err != nil {
return err
}
if index == 1 {
c.volumeCmd1 = cmd
} else {
c.volumeCmd0 = cmd
}
return nil
}
func (c *DualVolumeCluster) waitForHTTP(url string) error {
return (&Cluster{}).waitForHTTP(url)
}
func (c *DualVolumeCluster) waitForTCP(addr string) error {
return (&Cluster{}).waitForTCP(addr)
}
func (c *DualVolumeCluster) tailLog(logName string) string {
return (&Cluster{logsDir: c.logsDir}).tailLog(logName)
}
func (c *DualVolumeCluster) MasterAddress() string {
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.masterPort))
}
func (c *DualVolumeCluster) MasterURL() string {
return "http://" + c.MasterAddress()
}
func (c *DualVolumeCluster) VolumeAdminAddress(index int) string {
if index == 1 {
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumePort1))
}
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumePort0))
}
func (c *DualVolumeCluster) VolumePublicAddress(index int) string {
if index == 1 {
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumePubPort1))
}
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumePubPort0))
}
func (c *DualVolumeCluster) VolumeGRPCAddress(index int) string {
if index == 1 {
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumeGrpcPort1))
}
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumeGrpcPort0))
}
func (c *DualVolumeCluster) VolumeAdminURL(index int) string {
return "http://" + c.VolumeAdminAddress(index)
}
func (c *DualVolumeCluster) VolumePublicURL(index int) string {
return "http://" + c.VolumePublicAddress(index)
// StartDualVolumeCluster starts a cluster with 2 volume servers.
// Deprecated: Use StartMultiVolumeClusterAuto(t, profile, 2) directly.
func StartDualVolumeCluster(t testing.TB, profile matrix.Profile) MultiCluster {
return StartMultiVolumeClusterAuto(t, profile, 2)
}
test/volume_server/framework/cluster_interface.go
+63
View File
@@ -0,0 +1,63 @@
package framework
import (
"os"
"testing"
"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
)
// TestCluster is the common interface for single-volume cluster harnesses.
// Both *Cluster (Go volume) and *RustCluster (Rust volume) satisfy it.
type TestCluster interface {
MasterAddress() string
VolumeAdminAddress() string
VolumePublicAddress() string
VolumeGRPCAddress() string
VolumeServerAddress() string
MasterURL() string
VolumeAdminURL() string
VolumePublicURL() string
BaseDir() string
Stop()
}
func useRustVolumeServer() bool {
return os.Getenv("VOLUME_SERVER_IMPL") == "rust"
}
// StartVolumeCluster starts a single-volume cluster using either the Go or
// Rust volume server, depending on the VOLUME_SERVER_IMPL environment variable.
// Set VOLUME_SERVER_IMPL=rust to use the Rust volume server.
func StartVolumeCluster(t testing.TB, profile matrix.Profile) TestCluster {
t.Helper()
if useRustVolumeServer() {
return StartRustVolumeCluster(t, profile)
}
return StartSingleVolumeCluster(t, profile)
}
// MultiCluster is the common interface for multi-volume cluster harnesses.
// Both *MultiVolumeCluster (Go) and *RustMultiVolumeCluster (Rust) satisfy it.
type MultiCluster interface {
MasterAddress() string
MasterURL() string
BaseDir() string
VolumeAdminAddress(index int) string
VolumeAdminURL(index int) string
VolumePublicAddress(index int) string
VolumePublicURL(index int) string
VolumeGRPCAddress(index int) string
Stop()
}
// StartMultiVolumeClusterAuto starts a multi-volume cluster using either Go or
// Rust volume servers, depending on the VOLUME_SERVER_IMPL environment variable.
// Set VOLUME_SERVER_IMPL=rust to use Rust volume servers.
func StartMultiVolumeClusterAuto(t testing.TB, profile matrix.Profile, count int) MultiCluster {
t.Helper()
if useRustVolumeServer() {
return StartRustMultiVolumeCluster(t, profile, count)
}
return StartMultiVolumeCluster(t, profile, count)
}
test/volume_server/framework/cluster_interface_test.go
+20
View File
@@ -0,0 +1,20 @@
package framework
import "testing"
func TestUseRustVolumeServer(t *testing.T) {
t.Setenv("VOLUME_SERVER_IMPL", "rust")
if !useRustVolumeServer() {
t.Fatalf("expected rust selection when VOLUME_SERVER_IMPL=rust")
}
t.Setenv("VOLUME_SERVER_IMPL", "go")
if useRustVolumeServer() {
t.Fatalf("expected go selection when VOLUME_SERVER_IMPL=go")
}
t.Setenv("VOLUME_SERVER_IMPL", "")
if useRustVolumeServer() {
t.Fatalf("expected go selection when VOLUME_SERVER_IMPL is unset")
}
}
test/volume_server/framework/cluster_mixed.go
+371
View File
@@ -0,0 +1,371 @@
package framework
import (
"fmt"
"net"
"os"
"os/exec"
"path/filepath"
"strconv"
"sync"
"testing"
"github.com/seaweedfs/seaweedfs/test/testutil"
"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
)
// MixedVolumeCluster wraps a Go master + a mix of Go and Rust volume servers
// for integration testing. Go servers occupy indices [0, goCount) and Rust
// servers occupy indices [goCount, goCount+rustCount).
type MixedVolumeCluster struct {
testingTB testing.TB
profile matrix.Profile
weedBinary string // Go weed binary (master + Go volume servers)
rustVolumeBinary string // Rust volume binary
baseDir string
configDir string
logsDir string
keepLogs bool
masterPort int
masterGrpcPort int
volumePorts []int
volumeGrpcPorts []int
volumePubPorts []int
isRust []bool // which servers are Rust
masterCmd *exec.Cmd
volumeCmds []*exec.Cmd
cleanupOnce sync.Once
}
// StartMixedVolumeCluster starts a cluster with 1 Go master, goCount Go volume
// servers, and rustCount Rust volume servers. Go servers come first in the index.
func StartMixedVolumeCluster(t testing.TB, profile matrix.Profile, goCount, rustCount int) *MixedVolumeCluster {
t.Helper()
if goCount < 0 || rustCount < 0 {
t.Fatalf("goCount and rustCount must be non-negative, got go=%d rust=%d", goCount, rustCount)
}
total := goCount + rustCount
if total < 2 {
t.Fatalf("need at least 2 volume servers, got %d", total)
}
weedBinary, err := FindOrBuildWeedBinary()
if err != nil {
t.Fatalf("resolve weed binary: %v", err)
}
// Only build the Rust binary when Rust servers are requested.
var rustBinary string
if rustCount > 0 {
rustBinary, err = FindOrBuildRustBinary()
if err != nil {
t.Skipf("skipping mixed cluster test: rust binary unavailable: %v", err)
}
}
baseDir, keepLogs, err := newWorkDir()
if err != nil {
t.Fatalf("create temp test directory: %v", err)
}
configDir := filepath.Join(baseDir, "config")
logsDir := filepath.Join(baseDir, "logs")
masterDataDir := filepath.Join(baseDir, "master")
dirs := []string{configDir, logsDir, masterDataDir}
for i := 0; i < total; i++ {
dirs = append(dirs, filepath.Join(baseDir, fmt.Sprintf("volume%d", i)))
}
for _, dir := range dirs {
if mkErr := os.MkdirAll(dir, 0o755); mkErr != nil {
t.Fatalf("create %s: %v", dir, mkErr)
}
}
if err = writeSecurityConfig(configDir, profile); err != nil {
t.Fatalf("write security config: %v", err)
}
// 2 ports per server (admin, grpc); add 1 more when public port is split out.
portsPerServer := 2
if profile.SplitPublicPort {
portsPerServer = 3
}
miniPorts, ports, err := testutil.AllocatePortSet(1, total*portsPerServer)
if err != nil {
t.Fatalf("allocate ports: %v", err)
}
masterPort := miniPorts[0]
masterGrpcPort := masterPort + testutil.GrpcPortOffset
isRust := make([]bool, total)
for i := goCount; i < total; i++ {
isRust[i] = true
}
c := &MixedVolumeCluster{
testingTB: t,
profile: profile,
weedBinary: weedBinary,
rustVolumeBinary: rustBinary,
baseDir: baseDir,
configDir: configDir,
logsDir: logsDir,
keepLogs: keepLogs,
masterPort: masterPort,
masterGrpcPort: masterGrpcPort,
volumePorts: make([]int, total),
volumeGrpcPorts: make([]int, total),
volumePubPorts: make([]int, total),
isRust: isRust,
volumeCmds: make([]*exec.Cmd, total),
}
for i := 0; i < total; i++ {
baseIdx := i * portsPerServer
c.volumePorts[i] = ports[baseIdx]
c.volumeGrpcPorts[i] = ports[baseIdx+1]
if profile.SplitPublicPort {
c.volumePubPorts[i] = ports[baseIdx+2]
} else {
c.volumePubPorts[i] = c.volumePorts[i] // reuse admin port
}
}
// Start master
if err = c.startMaster(masterDataDir); err != nil {
c.Stop()
t.Fatalf("start master: %v", err)
}
helper := &Cluster{logsDir: logsDir}
if err = helper.waitForHTTP(c.MasterURL() + "/dir/status"); err != nil {
masterLog := helper.tailLog("master.log")
c.Stop()
t.Fatalf("wait for master readiness: %v\nmaster log tail:\n%s", err, masterLog)
}
// Start volume servers
for i := 0; i < total; i++ {
volumeDataDir := filepath.Join(baseDir, fmt.Sprintf("volume%d", i))
if isRust[i] {
err = c.startRustVolume(i, volumeDataDir)
} else {
err = c.startGoVolume(i, volumeDataDir)
}
if err != nil {
logTail := helper.tailLog(fmt.Sprintf("volume%d.log", i))
c.Stop()
t.Fatalf("start volume server %d (rust=%v): %v\nlog tail:\n%s", i, isRust[i], err, logTail)
}
// Rust uses /healthz, Go uses /status
healthURL := c.VolumeAdminURL(i) + "/status"
if isRust[i] {
healthURL = c.VolumeAdminURL(i) + "/healthz"
}
if err = helper.waitForHTTP(healthURL); err != nil {
logTail := helper.tailLog(fmt.Sprintf("volume%d.log", i))
c.Stop()
t.Fatalf("wait for volume server %d readiness: %v\nlog tail:\n%s", i, err, logTail)
}
if err = helper.waitForTCP(c.VolumeGRPCAddress(i)); err != nil {
logTail := helper.tailLog(fmt.Sprintf("volume%d.log", i))
c.Stop()
t.Fatalf("wait for volume server %d grpc readiness: %v\nlog tail:\n%s", i, err, logTail)
}
}
t.Cleanup(func() {
c.Stop()
})
return c
}
func (c *MixedVolumeCluster) Stop() {
if c == nil {
return
}
c.cleanupOnce.Do(func() {
for i := len(c.volumeCmds) - 1; i >= 0; i-- {
stopProcess(c.volumeCmds[i])
}
stopProcess(c.masterCmd)
if !c.keepLogs && !c.testingTB.Failed() {
_ = os.RemoveAll(c.baseDir)
} else if c.baseDir != "" {
c.testingTB.Logf("mixed volume server integration logs kept at %s", c.baseDir)
}
})
}
func (c *MixedVolumeCluster) startMaster(dataDir string) error {
logFile, err := os.Create(filepath.Join(c.logsDir, "master.log"))
if err != nil {
return err
}
args := []string{
"-config_dir=" + c.configDir,
"master",
"-ip=127.0.0.1",
"-port=" + strconv.Itoa(c.masterPort),
"-port.grpc=" + strconv.Itoa(c.masterGrpcPort),
"-mdir=" + dataDir,
"-peers=none",
"-volumeSizeLimitMB=" + strconv.Itoa(testVolumeSizeLimitMB),
"-defaultReplication=000",
}
c.masterCmd = exec.Command(c.weedBinary, args...)
c.masterCmd.Dir = c.baseDir
c.masterCmd.Stdout = logFile
c.masterCmd.Stderr = logFile
if err = c.masterCmd.Start(); err != nil {
logFile.Close()
return err
}
logFile.Close() // child inherited the fd
return nil
}
func (c *MixedVolumeCluster) startGoVolume(index int, dataDir string) error {
logName := fmt.Sprintf("volume%d.log", index)
logFile, err := os.Create(filepath.Join(c.logsDir, logName))
if err != nil {
return err
}
args := []string{
"-config_dir=" + c.configDir,
"volume",
"-ip=127.0.0.1",
"-port=" + strconv.Itoa(c.volumePorts[index]),
"-port.grpc=" + strconv.Itoa(c.volumeGrpcPorts[index]),
"-port.public=" + strconv.Itoa(c.volumePubPorts[index]),
"-dir=" + dataDir,
"-max=16",
"-master=127.0.0.1:" + strconv.Itoa(c.masterPort),
"-readMode=" + c.profile.ReadMode,
"-concurrentUploadLimitMB=" + strconv.Itoa(c.profile.ConcurrentUploadLimitMB),
"-concurrentDownloadLimitMB=" + strconv.Itoa(c.profile.ConcurrentDownloadLimitMB),
// Integration tests deliberately exercise loopback S3 endpoints; allow the SSRF guard to be bypassed for them.
"-volume.allowUntrustedRemoteEndpoints",
}
if c.profile.InflightUploadTimeout > 0 {
args = append(args, "-inflightUploadDataTimeout="+c.profile.InflightUploadTimeout.String())
}
if c.profile.InflightDownloadTimeout > 0 {
args = append(args, "-inflightDownloadDataTimeout="+c.profile.InflightDownloadTimeout.String())
}
if c.profile.FileSizeLimitMB > 0 {
args = append(args, "-fileSizeLimitMB="+strconv.Itoa(c.profile.FileSizeLimitMB))
}
cmd := exec.Command(c.weedBinary, args...)
cmd.Dir = c.baseDir
cmd.Stdout = logFile
cmd.Stderr = logFile
if err = cmd.Start(); err != nil {
logFile.Close()
return err
}
logFile.Close() // child inherited the fd
c.volumeCmds[index] = cmd
return nil
}
func (c *MixedVolumeCluster) startRustVolume(index int, dataDir string) error {
logName := fmt.Sprintf("volume%d.log", index)
logFile, err := os.Create(filepath.Join(c.logsDir, logName))
if err != nil {
return err
}
args := rustVolumeArgs(
c.profile,
c.configDir,
c.masterPort,
c.volumePorts[index],
c.volumeGrpcPorts[index],
c.volumePubPorts[index],
dataDir,
)
cmd := exec.Command(c.rustVolumeBinary, args...)
cmd.Dir = c.baseDir
cmd.Stdout = logFile
cmd.Stderr = logFile
if err = cmd.Start(); err != nil {
logFile.Close()
return err
}
logFile.Close() // child inherited the fd
c.volumeCmds[index] = cmd
return nil
}
// --- accessor methods (mirror MultiVolumeCluster) ---
func (c *MixedVolumeCluster) MasterAddress() string {
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.masterPort))
}
func (c *MixedVolumeCluster) MasterURL() string {
return "http://" + c.MasterAddress()
}
func (c *MixedVolumeCluster) VolumeAdminAddress(index int) string {
if index < 0 || index >= len(c.volumePorts) {
return ""
}
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumePorts[index]))
}
func (c *MixedVolumeCluster) VolumePublicAddress(index int) string {
if index < 0 || index >= len(c.volumePubPorts) {
return ""
}
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumePubPorts[index]))
}
func (c *MixedVolumeCluster) VolumeGRPCAddress(index int) string {
if index < 0 || index >= len(c.volumeGrpcPorts) {
return ""
}
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumeGrpcPorts[index]))
}
func (c *MixedVolumeCluster) VolumeAdminURL(index int) string {
return "http://" + c.VolumeAdminAddress(index)
}
func (c *MixedVolumeCluster) VolumePublicURL(index int) string {
return "http://" + c.VolumePublicAddress(index)
}
func (c *MixedVolumeCluster) BaseDir() string {
return c.baseDir
}
// VolumeServerAddress returns SeaweedFS server address format: ip:httpPort.grpcPort
func (c *MixedVolumeCluster) VolumeServerAddress(index int) string {
if index < 0 || index >= len(c.volumePorts) {
return ""
}
return fmt.Sprintf("%s.%d", c.VolumeAdminAddress(index), c.volumeGrpcPorts[index])
}
func (c *MixedVolumeCluster) IsRust(index int) bool {
if index < 0 || index >= len(c.isRust) {
return false
}
return c.isRust[index]
}
test/volume_server/framework/cluster_multi.go
+351
View File
@@ -0,0 +1,351 @@
package framework
import (
"fmt"
"net"
"os"
"os/exec"
"path/filepath"
"strconv"
"strings"
"sync"
"testing"
"github.com/seaweedfs/seaweedfs/test/testutil"
"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
)
type MultiVolumeCluster struct {
testingTB testing.TB
profile matrix.Profile
weedBinary string
baseDir string
configDir string
logsDir string
keepLogs bool
volumeServerCount int
disksPerServer int
masterPort int
masterGrpcPort int
volumePorts []int
volumeGrpcPorts []int
volumePubPorts []int
volumeDiskDirs [][]string // per server: the data directories (one DiskLocation each)
masterCmd *exec.Cmd
volumeCmds []*exec.Cmd
cleanupOnce sync.Once
}
// 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 {
t.Fatalf("resolve weed binary: %v", err)
}
baseDir, keepLogs, err := newWorkDir()
if err != nil {
t.Fatalf("create temp test directory: %v", err)
}
configDir := filepath.Join(baseDir, "config")
logsDir := filepath.Join(baseDir, "logs")
masterDataDir := filepath.Join(baseDir, "master")
// 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++ {
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 {
t.Fatalf("create %s: %v", dir, mkErr)
}
}
if err = writeSecurityConfig(configDir, profile); err != nil {
t.Fatalf("write security config: %v", err)
}
// Allocate ports for all volume servers (3 ports per server: admin, grpc, public)
// If SplitPublicPort is true, we need an additional port per server
portsPerServer := 3
if profile.SplitPublicPort {
portsPerServer = 4
}
totalPorts := serverCount * portsPerServer
miniPorts, ports, err := testutil.AllocatePortSet(1, totalPorts)
if err != nil {
t.Fatalf("allocate ports: %v", err)
}
masterPort := miniPorts[0]
masterGrpcPort := masterPort + testutil.GrpcPortOffset
c := &MultiVolumeCluster{
testingTB: t,
profile: profile,
weedBinary: weedBinary,
baseDir: baseDir,
configDir: configDir,
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),
}
// Assign ports to each volume server
for i := 0; i < serverCount; i++ {
baseIdx := i * portsPerServer
c.volumePorts[i] = ports[baseIdx]
c.volumeGrpcPorts[i] = ports[baseIdx+1]
// Assign public port, using baseIdx+3 if SplitPublicPort, else baseIdx+2
pubPortIdx := baseIdx + 2
if profile.SplitPublicPort {
pubPortIdx = baseIdx + 3
}
c.volumePubPorts[i] = ports[pubPortIdx]
}
// Start master
if err = c.startMaster(masterDataDir); err != nil {
c.Stop()
t.Fatalf("start master: %v", err)
}
if err = c.waitForHTTP(c.MasterURL() + "/dir/status"); err != nil {
masterLog := c.tailLog("master.log")
c.Stop()
t.Fatalf("wait for master readiness: %v\nmaster log tail:\n%s", err, masterLog)
}
// Start all volume servers
for i := 0; i < serverCount; i++ {
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()
t.Fatalf("start volume server %d: %v\nvolume log tail:\n%s", i, err, c.tailLog(volumeLog))
}
if err = c.waitForHTTP(c.VolumeAdminURL(i) + "/status"); err != nil {
volumeLog := fmt.Sprintf("volume%d.log", i)
c.Stop()
t.Fatalf("wait for volume server %d readiness: %v\nvolume log tail:\n%s", i, err, c.tailLog(volumeLog))
}
if err = c.waitForTCP(c.VolumeGRPCAddress(i)); err != nil {
volumeLog := fmt.Sprintf("volume%d.log", i)
c.Stop()
t.Fatalf("wait for volume server %d grpc readiness: %v\nvolume log tail:\n%s", i, err, c.tailLog(volumeLog))
}
}
t.Cleanup(func() {
c.Stop()
})
return c
}
// StartTripleVolumeCluster is a convenience wrapper that starts a cluster with 3 volume servers
func StartTripleVolumeCluster(t testing.TB, profile matrix.Profile) *MultiVolumeCluster {
return StartMultiVolumeCluster(t, profile, 3)
}
func (c *MultiVolumeCluster) Stop() {
if c == nil {
return
}
c.cleanupOnce.Do(func() {
// Stop volume servers in reverse order
for i := len(c.volumeCmds) - 1; i >= 0; i-- {
stopProcess(c.volumeCmds[i])
}
stopProcess(c.masterCmd)
if !c.keepLogs && !c.testingTB.Failed() {
_ = os.RemoveAll(c.baseDir)
} else if c.baseDir != "" {
c.testingTB.Logf("volume server integration logs kept at %s", c.baseDir)
}
})
}
func (c *MultiVolumeCluster) startMaster(dataDir string) error {
logFile, err := os.Create(filepath.Join(c.logsDir, "master.log"))
if err != nil {
return err
}
args := []string{
"-config_dir=" + c.configDir,
"master",
"-ip=127.0.0.1",
"-port=" + strconv.Itoa(c.masterPort),
"-port.grpc=" + strconv.Itoa(c.masterGrpcPort),
"-mdir=" + dataDir,
"-peers=none",
"-volumeSizeLimitMB=" + strconv.Itoa(testVolumeSizeLimitMB),
"-defaultReplication=000",
}
c.masterCmd = exec.Command(c.weedBinary, args...)
c.masterCmd.Dir = c.baseDir
c.masterCmd.Stdout = logFile
c.masterCmd.Stderr = logFile
return c.masterCmd.Start()
}
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",
"-ip=127.0.0.1",
"-port=" + strconv.Itoa(c.volumePorts[index]),
"-port.grpc=" + strconv.Itoa(c.volumeGrpcPorts[index]),
"-port.public=" + strconv.Itoa(c.volumePubPorts[index]),
"-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),
"-concurrentDownloadLimitMB=" + strconv.Itoa(c.profile.ConcurrentDownloadLimitMB),
// Integration tests deliberately exercise loopback S3 endpoints; allow the SSRF guard to be bypassed for them.
"-volume.allowUntrustedRemoteEndpoints",
}
if c.profile.InflightUploadTimeout > 0 {
args = append(args, "-inflightUploadDataTimeout="+c.profile.InflightUploadTimeout.String())
}
if c.profile.InflightDownloadTimeout > 0 {
args = append(args, "-inflightDownloadDataTimeout="+c.profile.InflightDownloadTimeout.String())
}
if c.profile.FileSizeLimitMB > 0 {
args = append(args, "-fileSizeLimitMB="+strconv.Itoa(c.profile.FileSizeLimitMB))
}
cmd := exec.Command(c.weedBinary, args...)
cmd.Dir = c.baseDir
cmd.Stdout = logFile
cmd.Stderr = logFile
if err = cmd.Start(); err != nil {
return err
}
c.volumeCmds[index] = cmd
return nil
}
func (c *MultiVolumeCluster) waitForHTTP(url string) error {
return (&Cluster{}).waitForHTTP(url)
}
func (c *MultiVolumeCluster) waitForTCP(addr string) error {
return (&Cluster{}).waitForTCP(addr)
}
func (c *MultiVolumeCluster) tailLog(logName string) string {
return (&Cluster{logsDir: c.logsDir}).tailLog(logName)
}
func (c *MultiVolumeCluster) MasterAddress() string {
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.masterPort))
}
func (c *MultiVolumeCluster) MasterURL() string {
return "http://" + c.MasterAddress()
}
func (c *MultiVolumeCluster) VolumeAdminAddress(index int) string {
if index < 0 || index >= len(c.volumePorts) {
return ""
}
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumePorts[index]))
}
func (c *MultiVolumeCluster) VolumePublicAddress(index int) string {
if index < 0 || index >= len(c.volumePubPorts) {
return ""
}
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumePubPorts[index]))
}
func (c *MultiVolumeCluster) VolumeGRPCAddress(index int) string {
if index < 0 || index >= len(c.volumeGrpcPorts) {
return ""
}
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumeGrpcPorts[index]))
}
func (c *MultiVolumeCluster) VolumeAdminURL(index int) string {
return "http://" + c.VolumeAdminAddress(index)
}
func (c *MultiVolumeCluster) VolumePublicURL(index int) string {
return "http://" + c.VolumePublicAddress(index)
}
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]
}
test/volume_server/framework/cluster_multi_rust.go
+287
View File
@@ -0,0 +1,287 @@
package framework
import (
"fmt"
"net"
"os"
"os/exec"
"path/filepath"
"strconv"
"sync"
"testing"
"github.com/seaweedfs/seaweedfs/test/testutil"
"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
)
// RustMultiVolumeCluster wraps a Go master + multiple Rust volume servers
// for integration testing. It mirrors MultiVolumeCluster but uses the Rust
// volume binary instead of the Go weed binary for volume servers.
type RustMultiVolumeCluster struct {
testingTB testing.TB
profile matrix.Profile
weedBinary string // Go weed binary (for the master)
rustVolumeBinary string // Rust volume binary
baseDir string
configDir string
logsDir string
keepLogs bool
volumeServerCount int
masterPort int
masterGrpcPort int
volumePorts []int
volumeGrpcPorts []int
volumePubPorts []int
masterCmd *exec.Cmd
volumeCmds []*exec.Cmd
cleanupOnce sync.Once
}
// StartRustMultiVolumeCluster starts a cluster with a Go master and the
// specified number of Rust volume servers.
func StartRustMultiVolumeCluster(t testing.TB, profile matrix.Profile, serverCount int) *RustMultiVolumeCluster {
t.Helper()
if serverCount < 1 {
t.Fatalf("serverCount must be at least 1, got %d", serverCount)
}
weedBinary, err := FindOrBuildWeedBinary()
if err != nil {
t.Fatalf("resolve weed binary: %v", err)
}
rustBinary, err := FindOrBuildRustBinary()
if err != nil {
t.Fatalf("resolve rust volume binary: %v", err)
}
baseDir, keepLogs, err := newWorkDir()
if err != nil {
t.Fatalf("create temp test directory: %v", err)
}
configDir := filepath.Join(baseDir, "config")
logsDir := filepath.Join(baseDir, "logs")
masterDataDir := filepath.Join(baseDir, "master")
// Create directories for master and all volume servers
dirs := []string{configDir, logsDir, masterDataDir}
for i := 0; i < serverCount; i++ {
dirs = append(dirs, filepath.Join(baseDir, fmt.Sprintf("volume%d", i)))
}
for _, dir := range dirs {
if mkErr := os.MkdirAll(dir, 0o755); mkErr != nil {
t.Fatalf("create %s: %v", dir, mkErr)
}
}
if err = writeSecurityConfig(configDir, profile); err != nil {
t.Fatalf("write security config: %v", err)
}
// Allocate ports for all volume servers (3 ports per server: admin, grpc, public)
// If SplitPublicPort is true, we need an additional port per server
portsPerServer := 3
if profile.SplitPublicPort {
portsPerServer = 4
}
totalPorts := serverCount * portsPerServer
miniPorts, ports, err := testutil.AllocatePortSet(1, totalPorts)
if err != nil {
t.Fatalf("allocate ports: %v", err)
}
masterPort := miniPorts[0]
masterGrpcPort := masterPort + testutil.GrpcPortOffset
c := &RustMultiVolumeCluster{
testingTB: t,
profile: profile,
weedBinary: weedBinary,
rustVolumeBinary: rustBinary,
baseDir: baseDir,
configDir: configDir,
logsDir: logsDir,
keepLogs: keepLogs,
volumeServerCount: serverCount,
masterPort: masterPort,
masterGrpcPort: masterGrpcPort,
volumePorts: make([]int, serverCount),
volumeGrpcPorts: make([]int, serverCount),
volumePubPorts: make([]int, serverCount),
volumeCmds: make([]*exec.Cmd, serverCount),
}
// Assign ports to each volume server
for i := 0; i < serverCount; i++ {
baseIdx := i * portsPerServer
c.volumePorts[i] = ports[baseIdx]
c.volumeGrpcPorts[i] = ports[baseIdx+1]
// Assign public port, using baseIdx+3 if SplitPublicPort, else baseIdx+2
pubPortIdx := baseIdx + 2
if profile.SplitPublicPort {
pubPortIdx = baseIdx + 3
}
c.volumePubPorts[i] = ports[pubPortIdx]
}
// Start master (Go)
if err = c.startMaster(masterDataDir); err != nil {
c.Stop()
t.Fatalf("start master: %v", err)
}
helper := &Cluster{logsDir: logsDir}
if err = helper.waitForHTTP(c.MasterURL() + "/dir/status"); err != nil {
masterLog := helper.tailLog("master.log")
c.Stop()
t.Fatalf("wait for master readiness: %v\nmaster log tail:\n%s", err, masterLog)
}
// Start all Rust volume servers
for i := 0; i < serverCount; i++ {
volumeDataDir := filepath.Join(baseDir, fmt.Sprintf("volume%d", i))
if err = c.startRustVolume(i, volumeDataDir); err != nil {
volumeLog := fmt.Sprintf("volume%d.log", i)
c.Stop()
t.Fatalf("start rust volume server %d: %v\nvolume log tail:\n%s", i, err, helper.tailLog(volumeLog))
}
if err = helper.waitForHTTP(c.VolumeAdminURL(i) + "/healthz"); err != nil {
volumeLog := fmt.Sprintf("volume%d.log", i)
c.Stop()
t.Fatalf("wait for rust volume server %d readiness: %v\nvolume log tail:\n%s", i, err, helper.tailLog(volumeLog))
}
if err = helper.waitForTCP(c.VolumeGRPCAddress(i)); err != nil {
volumeLog := fmt.Sprintf("volume%d.log", i)
c.Stop()
t.Fatalf("wait for rust volume server %d grpc readiness: %v\nvolume log tail:\n%s", i, err, helper.tailLog(volumeLog))
}
}
t.Cleanup(func() {
c.Stop()
})
return c
}
func (c *RustMultiVolumeCluster) Stop() {
if c == nil {
return
}
c.cleanupOnce.Do(func() {
// Stop volume servers in reverse order
for i := len(c.volumeCmds) - 1; i >= 0; i-- {
stopProcess(c.volumeCmds[i])
}
stopProcess(c.masterCmd)
if !c.keepLogs && !c.testingTB.Failed() {
_ = os.RemoveAll(c.baseDir)
} else if c.baseDir != "" {
c.testingTB.Logf("rust multi volume server integration logs kept at %s", c.baseDir)
}
})
}
func (c *RustMultiVolumeCluster) startMaster(dataDir string) error {
logFile, err := os.Create(filepath.Join(c.logsDir, "master.log"))
if err != nil {
return err
}
args := []string{
"-config_dir=" + c.configDir,
"master",
"-ip=127.0.0.1",
"-port=" + strconv.Itoa(c.masterPort),
"-port.grpc=" + strconv.Itoa(c.masterGrpcPort),
"-mdir=" + dataDir,
"-peers=none",
"-volumeSizeLimitMB=" + strconv.Itoa(testVolumeSizeLimitMB),
"-defaultReplication=000",
}
c.masterCmd = exec.Command(c.weedBinary, args...)
c.masterCmd.Dir = c.baseDir
c.masterCmd.Stdout = logFile
c.masterCmd.Stderr = logFile
return c.masterCmd.Start()
}
func (c *RustMultiVolumeCluster) startRustVolume(index int, dataDir string) error {
logName := fmt.Sprintf("volume%d.log", index)
logFile, err := os.Create(filepath.Join(c.logsDir, logName))
if err != nil {
return err
}
args := rustVolumeArgs(
c.profile,
c.configDir,
c.masterPort,
c.volumePorts[index],
c.volumeGrpcPorts[index],
c.volumePubPorts[index],
dataDir,
)
cmd := exec.Command(c.rustVolumeBinary, args...)
cmd.Dir = c.baseDir
cmd.Stdout = logFile
cmd.Stderr = logFile
if err = cmd.Start(); err != nil {
return err
}
c.volumeCmds[index] = cmd
return nil
}
// --- accessor methods (mirror MultiVolumeCluster) ---
func (c *RustMultiVolumeCluster) MasterAddress() string {
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.masterPort))
}
func (c *RustMultiVolumeCluster) MasterURL() string {
return "http://" + c.MasterAddress()
}
func (c *RustMultiVolumeCluster) VolumeAdminAddress(index int) string {
if index < 0 || index >= len(c.volumePorts) {
return ""
}
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumePorts[index]))
}
func (c *RustMultiVolumeCluster) VolumePublicAddress(index int) string {
if index < 0 || index >= len(c.volumePubPorts) {
return ""
}
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumePubPorts[index]))
}
func (c *RustMultiVolumeCluster) VolumeGRPCAddress(index int) string {
if index < 0 || index >= len(c.volumeGrpcPorts) {
return ""
}
return net.JoinHostPort("127.0.0.1", strconv.Itoa(c.volumeGrpcPorts[index]))
}
func (c *RustMultiVolumeCluster) VolumeAdminURL(index int) string {
return "http://" + c.VolumeAdminAddress(index)
}
func (c *RustMultiVolumeCluster) VolumePublicURL(index int) string {
return "http://" + c.VolumePublicAddress(index)
}
func (c *RustMultiVolumeCluster) BaseDir() string {
return c.baseDir
}
test/volume_server/framework/cluster_rust.go
+359
View File
@@ -0,0 +1,359 @@
package framework
import (
"bytes"
"fmt"
"net"
"os"
"os/exec"
"path/filepath"
"runtime"
"strconv"
"sync"
"testing"
"github.com/seaweedfs/seaweedfs/test/testutil"
"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
"github.com/seaweedfs/seaweedfs/weed/storage/types"
)
// RustCluster wraps a Go master + Rust volume server for integration testing.
type RustCluster struct {
testingTB testing.TB
profile matrix.Profile
weedBinary string // Go weed binary (for the master)
rustVolumeBinary string // Rust volume binary
baseDir string
configDir string
logsDir string
keepLogs bool
masterPort int
masterGrpcPort int
volumePort int
volumeGrpcPort int
volumePubPort int
masterCmd *exec.Cmd
volumeCmd *exec.Cmd
cleanupOnce sync.Once
}
var (
rustBinaryOnce sync.Once
rustBinaryPath string
rustBinaryErr error
)
// StartRustVolumeCluster starts a Go master + Rust volume server.
func StartRustVolumeCluster(t testing.TB, profile matrix.Profile) *RustCluster {
t.Helper()
weedBinary, err := FindOrBuildWeedBinary()
if err != nil {
t.Fatalf("resolve weed binary: %v", err)
}
rustBinary, err := FindOrBuildRustBinary()
if err != nil {
t.Fatalf("resolve rust volume binary: %v", err)
}
baseDir, keepLogs, err := newWorkDir()
if err != nil {
t.Fatalf("create temp test directory: %v", err)
}
configDir := filepath.Join(baseDir, "config")
logsDir := filepath.Join(baseDir, "logs")
masterDataDir := filepath.Join(baseDir, "master")
volumeDataDir := filepath.Join(baseDir, "volume")
for _, dir := range []string{configDir, logsDir, masterDataDir, volumeDataDir} {
if mkErr := os.MkdirAll(dir, 0o755); mkErr != nil {
t.Fatalf("create %s: %v", dir, mkErr)
}
}
if err = writeSecurityConfig(configDir, profile); err != nil {
t.Fatalf("write security config: %v", err)
}
miniPorts, ports, err := testutil.AllocatePortSet(1, 3)
if err != nil {
t.Fatalf("allocate ports: %v", err)
}
masterPort := miniPorts[0]
masterGrpcPort := masterPort + testutil.GrpcPortOffset
rc := &RustCluster{
testingTB: t,
profile: profile,
weedBinary: weedBinary,
rustVolumeBinary: rustBinary,
baseDir: baseDir,
configDir: configDir,
logsDir: logsDir,
keepLogs: keepLogs,
masterPort: masterPort,
masterGrpcPort: masterGrpcPort,
volumePort: ports[0],
volumeGrpcPort: ports[1],
volumePubPort: ports[0],
}
if profile.SplitPublicPort {
rc.volumePubPort = ports[2]
}
if err = rc.startMaster(masterDataDir); err != nil {
rc.Stop()
t.Fatalf("start master: %v", err)
}
// Reuse the same HTTP readiness helper via an unexported Cluster shim.
helper := &Cluster{logsDir: logsDir}
if err = helper.waitForHTTP(rc.MasterURL() + "/dir/status"); err != nil {
masterLog := helper.tailLog("master.log")
rc.Stop()
t.Fatalf("wait for master readiness: %v\nmaster log tail:\n%s", err, masterLog)
}
if err = rc.startRustVolume(volumeDataDir); err != nil {
masterLog := helper.tailLog("master.log")
rc.Stop()
t.Fatalf("start rust volume: %v\nmaster log tail:\n%s", err, masterLog)
}
if err = helper.waitForHTTP(rc.VolumeAdminURL() + "/healthz"); err != nil {
volumeLog := helper.tailLog("volume.log")
rc.Stop()
t.Fatalf("wait for rust volume readiness: %v\nvolume log tail:\n%s", err, volumeLog)
}
if err = helper.waitForTCP(rc.VolumeGRPCAddress()); err != nil {
volumeLog := helper.tailLog("volume.log")
rc.Stop()
t.Fatalf("wait for rust volume grpc readiness: %v\nvolume log tail:\n%s", err, volumeLog)
}
t.Cleanup(func() {
rc.Stop()
})
return rc
}
// Stop terminates all processes and cleans temporary files.
func (rc *RustCluster) Stop() {
if rc == nil {
return
}
rc.cleanupOnce.Do(func() {
stopProcess(rc.volumeCmd)
stopProcess(rc.masterCmd)
if !rc.keepLogs && !rc.testingTB.Failed() {
_ = os.RemoveAll(rc.baseDir)
} else if rc.baseDir != "" {
rc.testingTB.Logf("rust volume server integration logs kept at %s", rc.baseDir)
}
})
}
func (rc *RustCluster) startMaster(dataDir string) error {
logFile, err := os.Create(filepath.Join(rc.logsDir, "master.log"))
if err != nil {
return err
}
args := []string{
"-config_dir=" + rc.configDir,
"master",
"-ip=127.0.0.1",
"-port=" + strconv.Itoa(rc.masterPort),
"-port.grpc=" + strconv.Itoa(rc.masterGrpcPort),
"-mdir=" + dataDir,
"-peers=none",
"-volumeSizeLimitMB=" + strconv.Itoa(testVolumeSizeLimitMB),
"-defaultReplication=000",
}
rc.masterCmd = exec.Command(rc.weedBinary, args...)
rc.masterCmd.Dir = rc.baseDir
rc.masterCmd.Stdout = logFile
rc.masterCmd.Stderr = logFile
return rc.masterCmd.Start()
}
func rustVolumeArgs(
profile matrix.Profile,
configDir string,
masterPort int,
volumePort int,
volumeGrpcPort int,
volumePubPort int,
dataDir string,
) []string {
args := []string{
"--port", strconv.Itoa(volumePort),
"--port.grpc", strconv.Itoa(volumeGrpcPort),
"--port.public", strconv.Itoa(volumePubPort),
"--ip", "127.0.0.1",
"--ip.bind", "127.0.0.1",
"--dir", dataDir,
"--max", "16",
"--master", "127.0.0.1:" + strconv.Itoa(masterPort),
"--volume.allowUntrustedRemoteEndpoints",
"--securityFile", filepath.Join(configDir, "security.toml"),
"--readMode", profile.ReadMode,
"--concurrentUploadLimitMB", strconv.Itoa(profile.ConcurrentUploadLimitMB),
"--concurrentDownloadLimitMB", strconv.Itoa(profile.ConcurrentDownloadLimitMB),
"--preStopSeconds", "0",
}
if profile.InflightUploadTimeout > 0 {
args = append(args, "--inflightUploadDataTimeout", profile.InflightUploadTimeout.String())
}
if profile.InflightDownloadTimeout > 0 {
args = append(args, "--inflightDownloadDataTimeout", profile.InflightDownloadTimeout.String())
}
if profile.FileSizeLimitMB > 0 {
args = append(args, "--fileSizeLimitMB", strconv.Itoa(profile.FileSizeLimitMB))
}
return args
}
func (rc *RustCluster) startRustVolume(dataDir string) error {
logFile, err := os.Create(filepath.Join(rc.logsDir, "volume.log"))
if err != nil {
return err
}
args := rustVolumeArgs(
rc.profile,
rc.configDir,
rc.masterPort,
rc.volumePort,
rc.volumeGrpcPort,
rc.volumePubPort,
dataDir,
)
rc.volumeCmd = exec.Command(rc.rustVolumeBinary, args...)
rc.volumeCmd.Dir = rc.baseDir
rc.volumeCmd.Stdout = logFile
rc.volumeCmd.Stderr = logFile
return rc.volumeCmd.Start()
}
// FindOrBuildRustBinary returns an executable Rust volume binary, building one when needed.
func FindOrBuildRustBinary() (string, error) {
if fromEnv := os.Getenv("RUST_VOLUME_BINARY"); fromEnv != "" {
if isExecutableFile(fromEnv) {
return fromEnv, nil
}
return "", fmt.Errorf("RUST_VOLUME_BINARY is set but not executable: %s", fromEnv)
}
rustBinaryOnce.Do(func() {
// Derive the Rust volume crate directory from this source file's location.
rustCrateDir := ""
releaseBin := ""
if _, file, _, ok := runtime.Caller(0); ok {
repoRoot := filepath.Clean(filepath.Join(filepath.Dir(file), "..", "..", ".."))
for _, candidate := range []struct {
dir string
bin string
}{
{"seaweed-volume", "weed-volume"},
{"weed-volume", "weed-volume"},
{filepath.Join("rust", "volume_server"), "weed-volume-rs"},
} {
dir := filepath.Join(repoRoot, candidate.dir)
if isDir(dir) && isFile(filepath.Join(dir, "Cargo.toml")) {
rustCrateDir = dir
releaseBin = filepath.Join(dir, "target", "release", candidate.bin)
break
}
}
}
if rustCrateDir == "" {
rustBinaryErr = fmt.Errorf("unable to detect Rust volume crate directory")
return
}
// Always rebuild once per test process so the harness uses current source
// and features. The crate defaults to 5bytes, so a test binary built
// without 5BytesOffset has to turn it off or the Rust server refuses the
// .vif the Go server just wrote.
args := []string{"build", "--release"}
if types.OffsetSize != 5 {
args = append(args, "--no-default-features")
}
cmd := exec.Command("cargo", args...)
cmd.Dir = rustCrateDir
var out bytes.Buffer
cmd.Stdout = &out
cmd.Stderr = &out
if err := cmd.Run(); err != nil {
rustBinaryErr = fmt.Errorf("build rust volume binary: %w\n%s", err, out.String())
return
}
if !isExecutableFile(releaseBin) {
rustBinaryErr = fmt.Errorf("built rust volume binary is not executable: %s", releaseBin)
return
}
rustBinaryPath = releaseBin
})
if rustBinaryErr != nil {
return "", rustBinaryErr
}
return rustBinaryPath, nil
}
func isDir(path string) bool {
info, err := os.Stat(path)
return err == nil && info.IsDir()
}
func isFile(path string) bool {
info, err := os.Stat(path)
return err == nil && info.Mode().IsRegular()
}
// --- accessor methods (mirror Cluster) ---
func (rc *RustCluster) MasterAddress() string {
return net.JoinHostPort("127.0.0.1", strconv.Itoa(rc.masterPort))
}
func (rc *RustCluster) VolumeAdminAddress() string {
return net.JoinHostPort("127.0.0.1", strconv.Itoa(rc.volumePort))
}
func (rc *RustCluster) VolumePublicAddress() string {
return net.JoinHostPort("127.0.0.1", strconv.Itoa(rc.volumePubPort))
}
func (rc *RustCluster) VolumeGRPCAddress() string {
return net.JoinHostPort("127.0.0.1", strconv.Itoa(rc.volumeGrpcPort))
}
// VolumeServerAddress returns SeaweedFS server address format: ip:httpPort.grpcPort
func (rc *RustCluster) VolumeServerAddress() string {
return fmt.Sprintf("%s.%d", rc.VolumeAdminAddress(), rc.volumeGrpcPort)
}
func (rc *RustCluster) MasterURL() string {
return "http://" + rc.MasterAddress()
}
func (rc *RustCluster) VolumeAdminURL() string {
return "http://" + rc.VolumeAdminAddress()
}
func (rc *RustCluster) VolumePublicURL() string {
return "http://" + rc.VolumePublicAddress()
}
func (rc *RustCluster) BaseDir() string {
return rc.baseDir
}
test/volume_server/framework/cluster_rust_test.go
+38
View File
@@ -0,0 +1,38 @@
package framework
import (
"testing"
"time"
"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
)
func TestRustVolumeArgsIncludeReadMode(t *testing.T) {
profile := matrix.P1()
profile.ReadMode = "redirect"
profile.ConcurrentUploadLimitMB = 7
profile.ConcurrentDownloadLimitMB = 9
profile.InflightUploadTimeout = 3 * time.Second
profile.InflightDownloadTimeout = 4 * time.Second
args := rustVolumeArgs(profile, "/tmp/config", 9333, 18080, 28080, 38080, "/tmp/data")
assertArgPair(t, args, "--readMode", "redirect")
assertArgPair(t, args, "--concurrentUploadLimitMB", "7")
assertArgPair(t, args, "--concurrentDownloadLimitMB", "9")
assertArgPair(t, args, "--inflightUploadDataTimeout", "3s")
assertArgPair(t, args, "--inflightDownloadDataTimeout", "4s")
}
func assertArgPair(t *testing.T, args []string, flag string, want string) {
t.Helper()
for i := 0; i+1 < len(args); i += 2 {
if args[i] == flag {
if args[i+1] != want {
t.Fatalf("%s value mismatch: got %q want %q", flag, args[i+1], want)
}
return
}
}
t.Fatalf("missing %s in args: %v", flag, args)
}
test/volume_server/framework/cluster_with_filer.go
+2 -1
View File
@@ -9,6 +9,7 @@ import (
"strconv"
"testing"
"github.com/seaweedfs/seaweedfs/test/testutil"
"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
)
@@ -25,7 +26,7 @@ func StartSingleVolumeClusterWithFiler(t testing.TB, profile matrix.Profile) *Cl
baseCluster := StartSingleVolumeCluster(t, profile)
ports, err := allocatePorts(2)
ports, err := testutil.AllocatePorts(2)
if err != nil {
t.Fatalf("allocate filer ports: %v", err)
}
test/volume_server/framework/volume_fixture.go
+85 -1
View File
@@ -5,6 +5,8 @@ import (
"context"
"fmt"
"net/http"
"os"
"path/filepath"
"testing"
"time"
@@ -14,11 +16,23 @@ import (
func AllocateVolume(t testing.TB, client volume_server_pb.VolumeServerClient, volumeID uint32, collection string) {
t.Helper()
AllocateVolumeWithReplication(t, client, volumeID, collection, "000")
allocateVolume(t, client, volumeID, collection, "", "")
}
func AllocateVolumeWithReplication(t testing.TB, client volume_server_pb.VolumeServerClient, volumeID uint32, collection, replication string) {
t.Helper()
allocateVolume(t, client, volumeID, collection, replication, "")
}
// AllocateVolumeWithTTL allocates a volume whose needles expire after ttl
// (e.g. "10m").
func AllocateVolumeWithTTL(t testing.TB, client volume_server_pb.VolumeServerClient, volumeID uint32, collection string, ttl string) {
t.Helper()
allocateVolume(t, client, volumeID, collection, "", ttl)
}
func allocateVolume(t testing.TB, client volume_server_pb.VolumeServerClient, volumeID uint32, collection string, replication, ttl string) {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
@@ -30,6 +44,7 @@ func AllocateVolumeWithReplication(t testing.TB, client volume_server_pb.VolumeS
VolumeId: volumeID,
Collection: collection,
Replication: replication,
Ttl: ttl,
Version: uint32(needle.GetCurrentVersion()),
})
if err != nil {
@@ -53,6 +68,75 @@ func UploadBytes(t testing.TB, client *http.Client, volumeURL, fid string, data
return DoRequest(t, client, req)
}
// CorruptIndexFile appends garbage bytes to a volume's .idx file on disk so
// that CheckIndexFile detects a size mismatch during scrub.
func CorruptIndexFile(t testing.TB, baseDir string, volumeID uint32) {
t.Helper()
idxPath := filepath.Join(baseDir, "volume", fmt.Sprintf("%d.idx", volumeID))
f, err := os.OpenFile(idxPath, os.O_WRONLY|os.O_APPEND, 0644)
if err != nil {
t.Fatalf("open idx file for corruption: %v", err)
}
defer f.Close()
if _, err := f.Write([]byte{0xDE, 0xAD}); err != nil {
t.Fatalf("corrupt idx file: %v", err)
}
}
// EnableMaintenanceMode puts the volume server into maintenance mode.
func EnableMaintenanceMode(t testing.TB, ctx context.Context, client volume_server_pb.VolumeServerClient) {
t.Helper()
stateResp, err := client.GetState(ctx, &volume_server_pb.GetStateRequest{})
if err != nil {
t.Fatalf("GetState failed: %v", err)
}
_, err = client.SetState(ctx, &volume_server_pb.SetStateRequest{
State: &volume_server_pb.VolumeServerState{
Maintenance: true,
Version: stateResp.GetState().GetVersion(),
},
})
if err != nil {
t.Fatalf("SetState maintenance=true failed: %v", err)
}
}
// CorruptDatFile truncates a volume's .dat file to just the superblock (8 bytes)
// so that needle reads fail during a full scrub due to data file size mismatch.
func CorruptDatFile(t testing.TB, baseDir string, volumeID uint32) {
t.Helper()
datPath := filepath.Join(baseDir, "volume", fmt.Sprintf("%d.dat", volumeID))
// Truncate to superblock size only, removing all needle data.
if err := os.Truncate(datPath, 8); err != nil {
t.Fatalf("truncate dat file for corruption: %v", err)
}
}
// CorruptEcxFile appends garbage bytes to a volume's .ecx file on disk so
// that CheckIndexFile detects a size mismatch during EC index scrub.
func CorruptEcxFile(t testing.TB, baseDir string, volumeID uint32) {
t.Helper()
ecxPath := filepath.Join(baseDir, "volume", fmt.Sprintf("%d.ecx", volumeID))
f, err := os.OpenFile(ecxPath, os.O_WRONLY|os.O_APPEND, 0644)
if err != nil {
t.Fatalf("open ecx file for corruption: %v", err)
}
defer f.Close()
if _, err := f.Write([]byte{0xDE, 0xAD}); err != nil {
t.Fatalf("corrupt ecx file: %v", err)
}
}
// CorruptEcShardFile truncates an EC shard file to 1 byte so that local shard
// reads fail during an EC scrub.
func CorruptEcShardFile(t testing.TB, baseDir string, volumeID uint32, shardID int) {
t.Helper()
shardPath := filepath.Join(baseDir, "volume", fmt.Sprintf("%d.ec%02d", volumeID, shardID))
if err := os.Truncate(shardPath, 1); err != nil {
t.Fatalf("truncate EC shard file %s: %v", shardPath, err)
}
}
func ReadBytes(t testing.TB, client *http.Client, volumeURL, fid string) *http.Response {
t.Helper()