mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-08 23:37:43 +02:00
test: consolidate port allocation into shared test/testutil package (#8982)
* test: consolidate port allocation into shared test/testutil package Move duplicated port allocation logic from 15+ test files into a single shared package at test/testutil/. This fixes a port collision bug where independently allocated ports could overlap via the gRPC offset (port+10000), causing weed mini to reject the configuration. The shared package provides: - AllocatePorts: atomic allocation of N unique ports - AllocateMiniPorts/MustFreeMiniPorts: gRPC-offset-aware allocation that prevents port A+10000 == port B collisions - WaitForPort, WaitForService, FindBindIP, WriteIAMConfig, HasDocker * test: address review feedback and fix FUSE build - Revert fuse_integration change: it has its own go.mod and cannot import the shared testutil package - AllocateMiniPorts: hold all listeners open until the entire batch is allocated, preventing race conditions where other processes steal ports - HasDocker: add 5s context timeout to avoid hanging on stalled Docker - WaitForService: only treat 2xx HTTP status codes as ready * test: use global rand in AllocateMiniPorts for better seeding Go 1.20+ auto-seeds the global rand generator. Using it avoids identical sequences when multiple tests call at the same nanosecond. * test: revert WaitForService status code check S3 endpoints return non-2xx (e.g. 403) on bare GET requests, so requiring 2xx caused the S3 integration test to time out. Any HTTP response is sufficient proof that the service is running. * test: fix gofmt formatting in s3tables test files
This commit is contained in:
1 parent
ac12a735c7
commit
fbe758efa8
26 files changed
+308
-716
No files matched your search
@@ -5,7 +5,6 @@ import (
|
||||
"crypto/rand"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/exec"
|
||||
@@ -13,6 +12,8 @@ import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/test/testutil"
|
||||
)
|
||||
|
||||
type TestEnvironment struct {
|
||||
@@ -96,29 +97,26 @@ func NewTestEnvironment(t *testing.T) *TestEnvironment {
|
||||
t.Fatalf("Failed to create temp dir: %v", err)
|
||||
}
|
||||
|
||||
bindIP := findBindIP()
|
||||
bindIP := testutil.FindBindIP()
|
||||
|
||||
masterPort, masterGrpcPort := mustFreePortPair(t, "Master")
|
||||
volumePort, volumeGrpcPort := mustFreePortPair(t, "Volume")
|
||||
filerPort, filerGrpcPort := mustFreePortPair(t, "Filer")
|
||||
s3Port, s3GrpcPort := mustFreePortPair(t, "S3")
|
||||
icebergPort := mustFreePort(t, "Iceberg")
|
||||
// 9 ports: master(2), volume(2), filer(2), s3(2), iceberg(1)
|
||||
ports := testutil.MustAllocatePorts(t, 9)
|
||||
|
||||
return &TestEnvironment{
|
||||
seaweedDir: seaweedDir,
|
||||
weedBinary: weedBinary,
|
||||
dataDir: dataDir,
|
||||
bindIP: bindIP,
|
||||
s3Port: s3Port,
|
||||
s3GrpcPort: s3GrpcPort,
|
||||
icebergPort: icebergPort,
|
||||
masterPort: masterPort,
|
||||
masterGrpcPort: masterGrpcPort,
|
||||
filerPort: filerPort,
|
||||
filerGrpcPort: filerGrpcPort,
|
||||
volumePort: volumePort,
|
||||
volumeGrpcPort: volumeGrpcPort,
|
||||
dockerAvailable: hasDocker(),
|
||||
masterPort: ports[0],
|
||||
masterGrpcPort: ports[1],
|
||||
volumePort: ports[2],
|
||||
volumeGrpcPort: ports[3],
|
||||
filerPort: ports[4],
|
||||
filerGrpcPort: ports[5],
|
||||
s3Port: ports[6],
|
||||
s3GrpcPort: ports[7],
|
||||
icebergPort: ports[8],
|
||||
dockerAvailable: testutil.HasDocker(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -333,73 +331,6 @@ func createObjectBucket(t *testing.T, env *TestEnvironment, bucketName string) {
|
||||
}
|
||||
}
|
||||
|
||||
func hasDocker() bool {
|
||||
cmd := exec.Command("docker", "version")
|
||||
return cmd.Run() == nil
|
||||
}
|
||||
|
||||
func mustFreePort(t *testing.T, name string) int {
|
||||
t.Helper()
|
||||
|
||||
port, err := getFreePort()
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to get free port for %s: %v", name, err)
|
||||
}
|
||||
return port
|
||||
}
|
||||
|
||||
func mustFreePortPair(t *testing.T, name string) (int, int) {
|
||||
t.Helper()
|
||||
|
||||
httpPort, grpcPort, err := findAvailablePortPair()
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to get free port pair for %s: %v", name, err)
|
||||
}
|
||||
return httpPort, grpcPort
|
||||
}
|
||||
|
||||
func findAvailablePortPair() (int, int, error) {
|
||||
httpPort, err := getFreePort()
|
||||
if err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
grpcPort, err := getFreePort()
|
||||
if err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
return httpPort, grpcPort, nil
|
||||
}
|
||||
|
||||
func getFreePort() (int, error) {
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer listener.Close()
|
||||
|
||||
addr := listener.Addr().(*net.TCPAddr)
|
||||
return addr.Port, nil
|
||||
}
|
||||
|
||||
func findBindIP() string {
|
||||
addrs, err := net.InterfaceAddrs()
|
||||
if err != nil {
|
||||
return "127.0.0.1"
|
||||
}
|
||||
for _, addr := range addrs {
|
||||
ipNet, ok := addr.(*net.IPNet)
|
||||
if !ok || ipNet.IP == nil {
|
||||
continue
|
||||
}
|
||||
ip := ipNet.IP.To4()
|
||||
if ip == nil || ip.IsLoopback() || ip.IsLinkLocalUnicast() {
|
||||
continue
|
||||
}
|
||||
return ip.String()
|
||||
}
|
||||
return "127.0.0.1"
|
||||
}
|
||||
|
||||
func randomString(length int) string {
|
||||
const charset = "abcdefghijklmnopqrstuvwxyz0123456789"
|
||||
b := make([]byte, length)
|
||||
|
||||
@@ -20,6 +20,7 @@ import (
|
||||
"github.com/aws/aws-sdk-go-v2/config"
|
||||
"github.com/aws/aws-sdk-go-v2/credentials"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/seaweedfs/seaweedfs/test/testutil"
|
||||
"github.com/seaweedfs/seaweedfs/test/volume_server/framework"
|
||||
"github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb"
|
||||
@@ -115,7 +116,7 @@ func startDistributedLockCluster(t *testing.T) *distributedLockCluster {
|
||||
require.NoError(t, os.MkdirAll(dir, 0o755), "create %s", dir)
|
||||
}
|
||||
|
||||
ports, err := allocatePorts(12)
|
||||
ports, err := testutil.AllocatePorts(12)
|
||||
require.NoError(t, err, "allocate ports")
|
||||
cluster.masterPort = ports[0]
|
||||
cluster.masterGrpcPort = ports[1]
|
||||
@@ -591,25 +592,6 @@ func (c *distributedLockCluster) tailLog(name string) string {
|
||||
return strings.Join(lines, "\n")
|
||||
}
|
||||
|
||||
func allocatePorts(count int) ([]int, error) {
|
||||
listeners := make([]net.Listener, 0, count)
|
||||
ports := make([]int, 0, count)
|
||||
for i := 0; i < count; i++ {
|
||||
l, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
for _, openListener := range listeners {
|
||||
_ = openListener.Close()
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
listeners = append(listeners, l)
|
||||
ports = append(ports, l.Addr().(*net.TCPAddr).Port)
|
||||
}
|
||||
for _, l := range listeners {
|
||||
_ = l.Close()
|
||||
}
|
||||
return ports, nil
|
||||
}
|
||||
|
||||
func stopProcess(cmd *exec.Cmd) {
|
||||
if cmd == nil || cmd.Process == nil {
|
||||
|
||||
@@ -7,7 +7,6 @@ import (
|
||||
"encoding/base64"
|
||||
"fmt"
|
||||
"math/rand"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/exec"
|
||||
@@ -25,6 +24,7 @@ import (
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/test/testutil"
|
||||
"github.com/seaweedfs/seaweedfs/test/volume_server/framework"
|
||||
"github.com/seaweedfs/seaweedfs/weed/command"
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
@@ -112,37 +112,16 @@ func TestS3Integration(t *testing.T) {
|
||||
}
|
||||
|
||||
// findAvailablePort finds an available port by binding to port 0
|
||||
func findAvailablePort() (int, error) {
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer listener.Close()
|
||||
|
||||
addr := listener.Addr().(*net.TCPAddr)
|
||||
return addr.Port, nil
|
||||
}
|
||||
|
||||
// startMiniCluster starts a weed mini instance directly without exec.
|
||||
// Extra flags (e.g. "-s3.allowDeleteBucketNotEmpty=false") can be appended via extraArgs.
|
||||
func startMiniCluster(t *testing.T, extraArgs ...string) (*TestCluster, error) {
|
||||
// Find available ports
|
||||
masterPort, err := findAvailablePort()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to find master port: %v", err)
|
||||
}
|
||||
volumePort, err := findAvailablePort()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to find volume port: %v", err)
|
||||
}
|
||||
filerPort, err := findAvailablePort()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to find filer port: %v", err)
|
||||
}
|
||||
s3Port, err := findAvailablePort()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to find s3 port: %v", err)
|
||||
}
|
||||
// Allocate non-colliding ports (including gRPC offsets) for weed mini.
|
||||
ports := testutil.MustFreeMiniPorts(t, []string{"Master", "Volume", "Filer", "S3"})
|
||||
masterPort := ports[0]
|
||||
volumePort := ports[1]
|
||||
filerPort := ports[2]
|
||||
s3Port := ports[3]
|
||||
// Create temporary directory for test data
|
||||
testDir := t.TempDir()
|
||||
|
||||
@@ -167,7 +146,7 @@ func startMiniCluster(t *testing.T, extraArgs ...string) (*TestCluster, error) {
|
||||
|
||||
// Create empty security.toml to disable JWT authentication in tests
|
||||
securityToml := filepath.Join(testDir, "security.toml")
|
||||
err = os.WriteFile(securityToml, []byte("# Empty security config for testing\n"), 0644)
|
||||
err := os.WriteFile(securityToml, []byte("# Empty security config for testing\n"), 0644)
|
||||
if err != nil {
|
||||
cancel()
|
||||
return nil, fmt.Errorf("failed to create security.toml: %v", err)
|
||||
@@ -233,10 +212,9 @@ func startMiniCluster(t *testing.T, extraArgs ...string) (*TestCluster, error) {
|
||||
}()
|
||||
|
||||
// Wait for S3 service to be ready
|
||||
err = waitForS3Ready(cluster.s3Endpoint, 30*time.Second)
|
||||
if err != nil {
|
||||
if !testutil.WaitForService(cluster.s3Endpoint, 30*time.Second) {
|
||||
cancel()
|
||||
return nil, fmt.Errorf("S3 service failed to start: %v", err)
|
||||
return nil, fmt.Errorf("S3 service failed to start at %s", cluster.s3Endpoint)
|
||||
}
|
||||
|
||||
// If VOLUME_SERVER_IMPL=rust, start a Rust volume server alongside weed mini
|
||||
@@ -277,14 +255,12 @@ func (c *TestCluster) startRustVolumeServer(t *testing.T) error {
|
||||
return fmt.Errorf("resolve rust volume binary: %v", err)
|
||||
}
|
||||
|
||||
rustVolumePort, err := findAvailablePort()
|
||||
rustPorts, err := testutil.AllocatePorts(2)
|
||||
if err != nil {
|
||||
return fmt.Errorf("find rust volume port: %v", err)
|
||||
}
|
||||
rustVolumeGrpcPort, err := findAvailablePort()
|
||||
if err != nil {
|
||||
return fmt.Errorf("find rust volume grpc port: %v", err)
|
||||
return fmt.Errorf("find rust volume ports: %v", err)
|
||||
}
|
||||
rustVolumePort := rustPorts[0]
|
||||
rustVolumeGrpcPort := rustPorts[1]
|
||||
|
||||
rustVolumeDir := filepath.Join(c.dataDir, "rust-volume")
|
||||
if err := os.MkdirAll(rustVolumeDir, 0o755); err != nil {
|
||||
@@ -377,25 +353,6 @@ func (c *TestCluster) Stop() {
|
||||
}
|
||||
}
|
||||
|
||||
// waitForS3Ready waits for the S3 service to be ready
|
||||
func waitForS3Ready(endpoint string, timeout time.Duration) error {
|
||||
client := &http.Client{Timeout: 1 * time.Second}
|
||||
deadline := time.Now().Add(timeout)
|
||||
|
||||
for time.Now().Before(deadline) {
|
||||
resp, err := client.Get(endpoint)
|
||||
if err == nil {
|
||||
resp.Body.Close()
|
||||
// Wait a bit more to ensure service is fully ready
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
return nil
|
||||
}
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
}
|
||||
|
||||
return fmt.Errorf("timeout waiting for S3 service at %s", endpoint)
|
||||
}
|
||||
|
||||
// Test functions
|
||||
|
||||
func testCreateBucket(t *testing.T, cluster *TestCluster) {
|
||||
|
||||
@@ -4,7 +4,6 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/exec"
|
||||
@@ -21,6 +20,7 @@ import (
|
||||
"github.com/aws/aws-sdk-go/aws/session"
|
||||
"github.com/aws/aws-sdk-go/service/iam"
|
||||
"github.com/aws/aws-sdk-go/service/s3"
|
||||
"github.com/seaweedfs/seaweedfs/test/testutil"
|
||||
"github.com/seaweedfs/seaweedfs/test/volume_server/framework"
|
||||
"github.com/seaweedfs/seaweedfs/weed/command"
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
@@ -703,43 +703,13 @@ func uniqueName(prefix string) string {
|
||||
|
||||
// --- Test setup helpers ---
|
||||
|
||||
func findAvailablePort() (int, error) {
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer listener.Close()
|
||||
addr := listener.Addr().(*net.TCPAddr)
|
||||
return addr.Port, nil
|
||||
}
|
||||
|
||||
// findAvailablePortPair finds an available http port P such that P and P+10000 (grpc) are both available
|
||||
func findAvailablePortPair() (int, int, error) {
|
||||
httpPort, err := findAvailablePort()
|
||||
if err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
for i := 0; i < 100; i++ {
|
||||
grpcPort, err := findAvailablePort()
|
||||
if err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
if grpcPort != httpPort {
|
||||
return httpPort, grpcPort, nil
|
||||
}
|
||||
}
|
||||
return 0, 0, fmt.Errorf("failed to find available port pair")
|
||||
}
|
||||
|
||||
func startMiniCluster(t *testing.T) (*TestCluster, error) {
|
||||
masterPort, masterGrpcPort, err := findAvailablePortPair()
|
||||
require.NoError(t, err)
|
||||
volumePort, volumeGrpcPort, err := findAvailablePortPair()
|
||||
require.NoError(t, err)
|
||||
filerPort, filerGrpcPort, err := findAvailablePortPair()
|
||||
require.NoError(t, err)
|
||||
s3Port, s3GrpcPort, err := findAvailablePortPair()
|
||||
require.NoError(t, err)
|
||||
ports := testutil.MustAllocatePorts(t, 8)
|
||||
masterPort, masterGrpcPort := ports[0], ports[1]
|
||||
volumePort, volumeGrpcPort := ports[2], ports[3]
|
||||
filerPort, filerGrpcPort := ports[4], ports[5]
|
||||
s3Port, s3GrpcPort := ports[6], ports[7]
|
||||
|
||||
testDir := t.TempDir()
|
||||
|
||||
@@ -760,7 +730,7 @@ func startMiniCluster(t *testing.T) (*TestCluster, error) {
|
||||
|
||||
// Disable authentication for tests
|
||||
securityToml := filepath.Join(testDir, "security.toml")
|
||||
err = os.WriteFile(securityToml, []byte("# Empty security config\n"), 0644)
|
||||
err := os.WriteFile(securityToml, []byte("# Empty security config\n"), 0644)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Configure credential store for IAM tests
|
||||
@@ -819,10 +789,9 @@ enabled = true
|
||||
}()
|
||||
|
||||
// Wait for S3
|
||||
err = waitForS3Ready(cluster.s3Endpoint, 60*time.Second)
|
||||
if err != nil {
|
||||
if !testutil.WaitForService(cluster.s3Endpoint, 60*time.Second) {
|
||||
cancel()
|
||||
return nil, err
|
||||
return nil, fmt.Errorf("timeout waiting for S3 at %s", cluster.s3Endpoint)
|
||||
}
|
||||
|
||||
// If VOLUME_SERVER_IMPL=rust, start a Rust volume server alongside weed mini
|
||||
@@ -837,19 +806,6 @@ enabled = true
|
||||
return cluster, nil
|
||||
}
|
||||
|
||||
func waitForS3Ready(endpoint string, timeout time.Duration) error {
|
||||
client := &http.Client{Timeout: 1 * time.Second}
|
||||
deadline := time.Now().Add(timeout)
|
||||
for time.Now().Before(deadline) {
|
||||
resp, err := client.Get(endpoint)
|
||||
if err == nil {
|
||||
resp.Body.Close()
|
||||
return nil
|
||||
}
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
}
|
||||
return fmt.Errorf("timeout waiting for S3")
|
||||
}
|
||||
|
||||
// startRustVolumeServer starts a Rust volume server that registers with the same master.
|
||||
func (c *TestCluster) startRustVolumeServer(t *testing.T) error {
|
||||
@@ -860,14 +816,12 @@ func (c *TestCluster) startRustVolumeServer(t *testing.T) error {
|
||||
return fmt.Errorf("resolve rust volume binary: %v", err)
|
||||
}
|
||||
|
||||
rustVolumePort, err := findAvailablePort()
|
||||
rustPorts, err := testutil.AllocatePorts(2)
|
||||
if err != nil {
|
||||
return fmt.Errorf("find rust volume port: %v", err)
|
||||
}
|
||||
rustVolumeGrpcPort, err := findAvailablePort()
|
||||
if err != nil {
|
||||
return fmt.Errorf("find rust volume grpc port: %v", err)
|
||||
return fmt.Errorf("find rust volume ports: %v", err)
|
||||
}
|
||||
rustVolumePort := rustPorts[0]
|
||||
rustVolumeGrpcPort := rustPorts[1]
|
||||
|
||||
rustVolumeDir := filepath.Join(c.dataDir, "rust-volume")
|
||||
if err := os.MkdirAll(rustVolumeDir, 0o755); err != nil {
|
||||
|
||||
@@ -14,7 +14,7 @@ import (
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/credentials"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/seaweedfs/seaweedfs/test/s3tables/testutil"
|
||||
"github.com/seaweedfs/seaweedfs/test/testutil"
|
||||
"github.com/testcontainers/testcontainers-go"
|
||||
)
|
||||
|
||||
@@ -94,9 +94,10 @@ func (env *TestEnvironment) StartSeaweedFS(t *testing.T) {
|
||||
t.Fatalf("failed to create temp directory: %v", err)
|
||||
}
|
||||
|
||||
env.masterPort = testutil.MustFreeMiniPort(t, "Master")
|
||||
env.filerPort = testutil.MustFreeMiniPort(t, "Filer")
|
||||
env.s3Port = testutil.MustFreeMiniPort(t, "S3")
|
||||
ports := testutil.MustFreeMiniPorts(t, []string{"Master", "Filer", "S3"})
|
||||
env.masterPort = ports[0]
|
||||
env.filerPort = ports[1]
|
||||
env.s3Port = ports[2]
|
||||
|
||||
bindIP := testutil.FindBindIP()
|
||||
iamConfigPath, err := testutil.WriteIAMConfig(env.seaweedfsDataDir, env.accessKey, env.secretKey)
|
||||
|
||||
Reference in new issue
Block a user