gsadmin/spinifex
1
0
Fork 0
mirror of https://github.com/mulgadc/spinifex.git synced 2026-10-15 09:01:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(test): give each predastore fixture its own S3 gate port (#775)

Browse Source 
The fixture pinned its S3 gate to 127.0.0.1:18443. Three test binaries import
it and go test runs packages as concurrent processes, so two clusters raced
for the port. The loser did not fail cleanly: waitForReady probed the winner's
gate on the same port and reported healthy, so the setup client talked to
another process's cluster and failed with BucketAlreadyOwnedByYou.

Reserve a port per process instead, and reach the gate only through
Fixture.Host so it cannot be hardcoded again. A Start that fails partway now
drains the cluster before returning; its nodes hold process-wide pipe names
that a later attempt would otherwise collide with.
This commit is contained in:
Josh Siviour
2026-08-12 22:37:54 +10:00
committed by GitHub
parent 91d22d9f12
commit 9427cdfc2d
2 changed files with 73 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
spinifex/services/predastore/predastore_integration_test.go
+23 -21
View File
@@ -45,8 +45,10 @@ const (
testRegion = testpredastore.Region
)
// createS3Client creates an AWS S3 client for testing.
func createS3Client(accessKey, secretKey string) *s3.S3 {
// createS3Client creates an AWS S3 client for testing. The endpoint comes
// from the fixture rather than a constant: its port is picked at startup, so
// test binaries running concurrently cannot end up on each other's cluster.
func createS3Client(fixture *testpredastore.Fixture, accessKey, secretKey string) *s3.S3 {
tr := &http.Transport{
TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
}
@@ -54,7 +56,7 @@ func createS3Client(accessKey, secretKey string) *s3.S3 {
sess := session.Must(session.NewSession(&aws.Config{
Region: aws.String(testRegion),
Endpoint: aws.String("https://" + testpredastore.Host),
Endpoint: aws.String("https://" + fixture.Host),
Credentials: credentials.NewStaticCredentials(accessKey, secretKey, ""),
S3ForcePathStyle: aws.Bool(true),
DisableSSL: aws.Bool(false),
@@ -71,10 +73,10 @@ func TestIntegration_BucketListing(t *testing.T) {
}
// Start server (only starts once)
testpredastore.Start(t)
fixture := testpredastore.Start(t)
// Create S3 client
client := createS3Client(validAccessKey, validSecretKey)
client := createS3Client(fixture, validAccessKey, validSecretKey)
// List all buckets
result, err := client.ListBuckets(&s3.ListBucketsInput{})
@@ -99,10 +101,10 @@ func TestIntegration_Authentication_Valid(t *testing.T) {
t.Skip("Skipping integration test in short mode")
}
testpredastore.Start(t)
fixture := testpredastore.Start(t)
// Create client with valid credentials
client := createS3Client(validAccessKey, validSecretKey)
client := createS3Client(fixture, validAccessKey, validSecretKey)
// Try to list buckets
result, err := client.ListBuckets(&s3.ListBucketsInput{})
@@ -119,10 +121,10 @@ func TestIntegration_Authentication_Invalid(t *testing.T) {
t.Skip("Skipping integration test in short mode")
}
testpredastore.Start(t)
fixture := testpredastore.Start(t)
// Create client with invalid credentials
client := createS3Client(invalidAccessKey, invalidSecretKey)
client := createS3Client(fixture, invalidAccessKey, invalidSecretKey)
// Try to list buckets - should fail
_, err := client.ListBuckets(&s3.ListBucketsInput{})
@@ -144,9 +146,9 @@ func TestIntegration_FileUpload_TestBucket(t *testing.T) {
t.Skip("Skipping integration test in short mode")
}
testpredastore.Start(t)
fixture := testpredastore.Start(t)
client := createS3Client(validAccessKey, validSecretKey)
client := createS3Client(fixture, validAccessKey, validSecretKey)
// Upload test file
key := "test1.txt"
@@ -186,9 +188,9 @@ func TestIntegration_FileUpload_PublicBucket(t *testing.T) {
t.Skip("Skipping integration test in short mode")
}
testpredastore.Start(t)
fixture := testpredastore.Start(t)
client := createS3Client(validAccessKey, validSecretKey)
client := createS3Client(fixture, validAccessKey, validSecretKey)
// Upload test file
key := "test1.txt"
@@ -227,9 +229,9 @@ func TestIntegration_FileUpload_PublicBucket2(t *testing.T) {
t.Skip("Skipping integration test in short mode")
}
testpredastore.Start(t)
fixture := testpredastore.Start(t)
client := createS3Client(validAccessKey, validSecretKey)
client := createS3Client(fixture, validAccessKey, validSecretKey)
time.Sleep(1 * time.Second)
@@ -273,9 +275,9 @@ func TestIntegration_FileOperations_Complete(t *testing.T) {
t.Skip("Skipping integration test in short mode")
}
testpredastore.Start(t)
fixture := testpredastore.Start(t)
client := createS3Client(validAccessKey, validSecretKey)
client := createS3Client(fixture, validAccessKey, validSecretKey)
// Test with both buckets
buckets := []string{testBucket, publicBucket}
@@ -365,9 +367,9 @@ func TestIntegration_MultipleFiles(t *testing.T) {
t.Skip("Skipping integration test in short mode")
}
testpredastore.Start(t)
fixture := testpredastore.Start(t)
client := createS3Client(validAccessKey, validSecretKey)
client := createS3Client(fixture, validAccessKey, validSecretKey)
// Upload multiple files to test-bucket
fileCount := 5
@@ -414,9 +416,9 @@ func TestIntegration_LargeFile(t *testing.T) {
t.Skip("Skipping integration test in short mode")
}
testpredastore.Start(t)
fixture := testpredastore.Start(t)
client := createS3Client(validAccessKey, validSecretKey)
client := createS3Client(fixture, validAccessKey, validSecretKey)
// Create 1MB file
size := 1024 * 1024
tests/fixtures/predastore/predastore.go
Vendored
+50 -18
View File
@@ -43,10 +43,9 @@ import (
)
// Fixed connection details for the shared predastore fixture started by
// Start. Every cluster node runs in-process, so these values never need to
// vary per caller or per test run.
// Start. The endpoint is not among them: it carries a port picked at startup
// and is only reachable through the Fixture returned by Start.
const (
Host = "127.0.0.1:18443"
Region = "us-east-1"
// AccessKey/SecretKey are the well-known AWS SDK example credentials
// (docs.aws.amazon.com/IAM/latest/UserGuide), used only to authenticate
@@ -65,7 +64,6 @@ const (
DefaultBucket2 = "public-bucket"
testHost = "127.0.0.1"
testPort = 18443
)
// The topology Start writes. One host runs every node, so they talk over the
@@ -96,6 +94,9 @@ const (
// S3/viperblock clients: a reachable endpoint and its default buckets
// created. Its TLS cert is self-signed, so clients skip verification.
type Fixture struct {
// Host is the cluster's host:port S3 endpoint, whose port is picked when
// the cluster starts. It is the only way to reach this fixture: nothing
// may hardcode a port, or concurrently running test binaries collide.
Host string
Region string
AccessKey string
@@ -190,12 +191,18 @@ func Start(t *testing.T) *Fixture {
t.Fatalf("predastore fixture: write encryption key: %v", err)
}
gatePort, err := freeGatePort()
if err != nil {
t.Fatalf("predastore fixture: reserve gate port: %v", err)
}
host := net.JoinHostPort(testHost, strconv.Itoa(gatePort))
// The config goes through the real file and the real loader rather than a
// hand-built struct, so the fixture trips over the same strict decode and
// topology validation an operator's install would.
dataDir := filepath.Join(testDir, "data")
configPath := filepath.Join(testDir, "predastore.toml")
config := topology(dataDir, certPath, keyPath, encryptionKeyPath)
config := topology(dataDir, certPath, keyPath, encryptionKeyPath, gatePort)
if err := os.WriteFile(configPath, []byte(config), 0600); err != nil {
t.Fatalf("predastore fixture: write config: %v", err)
}
@@ -229,23 +236,23 @@ func Start(t *testing.T) *Fixture {
// The gate holds off serving until the local Raft quorum has a leader, so
// an accepted connection also means writes will commit.
if !waitForReady(30*time.Second, done) {
cancel()
if !waitForReady(30*time.Second, done, host) {
shutdown()
t.Fatal("predastore fixture: S3 gate did not become ready")
}
// Create the default buckets through the S3 API so they land in the meta
// plane; config-defined buckets are not visible to ListBuckets.
setupClient := s3Client(AccessKey, SecretKey)
setupClient := s3Client(host, AccessKey, SecretKey)
for _, bucket := range []string{DefaultBucket, DefaultBucket2} {
if _, err := setupClient.CreateBucket(&s3.CreateBucketInput{Bucket: aws.String(bucket)}); err != nil {
cancel()
shutdown()
t.Fatalf("predastore fixture: create bucket %s: %v", bucket, err)
}
}
fixture = &Fixture{
Host: Host,
Host: host,
Region: Region,
AccessKey: AccessKey,
SecretKey: SecretKey,
@@ -268,25 +275,50 @@ func Stop() {
if !started {
return
}
shutdown()
started, fixture = false, nil
}
// shutdown cancels the running cluster and waits for it to drain, with mu
// held. A Start that fails partway must do this before it returns: its nodes
// hold process-wide pipe names a later attempt would otherwise collide with.
func shutdown() {
stop()
select {
case <-drained:
case <-time.After(30 * time.Second):
slog.Error("predastore fixture: cluster did not drain")
}
started, fixture = false, nil
}
// freeGatePort reserves a port for the S3 gate by binding one and handing it
// straight back. Test binaries for different packages run concurrently, and a
// fixed port would silently point one process's clients at another's cluster.
func freeGatePort() (int, error) {
ln, err := net.Listen("tcp", net.JoinHostPort(testHost, "0"))
if err != nil {
return 0, err
}
defer ln.Close()
addr, ok := ln.Addr().(*net.TCPAddr)
if !ok {
return 0, fmt.Errorf("unexpected listener address %T", ln.Addr())
}
return addr.Port, nil
}
// topology is the fixture's predastore configuration file: one host, the gate
// on the fixed S3 port, and the blob and meta nodes the erasure code needs.
func topology(dataDir, certPath, keyPath, encryptionKeyPath string) string {
// on the reserved S3 port, and the blob and meta nodes the erasure code needs.
func topology(dataDir, certPath, keyPath, encryptionKeyPath string, gatePort int) string {
var b strings.Builder
fmt.Fprintf(&b, "version = 1\nregion = %q\n\n[rs]\ndata = %d\nparity = %d\n\n", Region, rsData, rsParity)
fmt.Fprintf(&b, "[[host]]\nid = %d\nbind_addr = %q\naddr = %q\ndata_dir = %q\ntls_cert = %q\ntls_key = %q\nencryption_key = %q\n\n",
fixtureHostID, testHost, testHost, dataDir, certPath, keyPath, encryptionKeyPath)
fmt.Fprintf(&b, "[[host.node]]\nid = %d\nrole = \"gate\"\nport = %d\n\n", gateNodeID, testPort)
fmt.Fprintf(&b, "[[host.node]]\nid = %d\nrole = \"gate\"\nport = %d\n\n", gateNodeID, gatePort)
for i := range blobNodes {
fmt.Fprintf(&b, "[[host.node]]\nid = %d\nrole = \"blob\"\nport = %d\n\n", firstBlobNode+i, firstBlobPort+i)
}
@@ -369,7 +401,7 @@ func generateCertificate(certPath, keyPath string) error {
// waitForReady polls the S3 endpoint until it answers, the cluster exits or
// timeout elapses. A cluster that failed to start closes done, which is the
// difference between waiting out the timeout and reporting at once.
func waitForReady(timeout time.Duration, done <-chan struct{}) bool {
func waitForReady(timeout time.Duration, done <-chan struct{}, host string) bool {
client := &http.Client{
Transport: &http.Transport{
TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec // self-signed cert generated by this same fixture, localhost-only
@@ -384,7 +416,7 @@ func waitForReady(timeout time.Duration, done <-chan struct{}) bool {
return false
default:
}
resp, err := client.Get("https://" + net.JoinHostPort(testHost, strconv.Itoa(testPort)) + "/")
resp, err := client.Get("https://" + host + "/")
if err == nil {
resp.Body.Close()
return true
@@ -397,7 +429,7 @@ func waitForReady(timeout time.Duration, done <-chan struct{}) bool {
// s3Client creates an AWS S3 client against the fixture for fixture setup
// only (bucket creation); test bodies build their own clients against
// whatever bucket/credentials their scenario needs.
func s3Client(accessKey, secretKey string) *s3.S3 {
func s3Client(host, accessKey, secretKey string) *s3.S3 {
tr := &http.Transport{
TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec // self-signed cert generated by this same fixture, localhost-only
}
@@ -405,7 +437,7 @@ func s3Client(accessKey, secretKey string) *s3.S3 {
sess := session.Must(session.NewSession(&aws.Config{
Region: aws.String(Region),
Endpoint: aws.String("https://" + Host),
Endpoint: aws.String("https://" + host),
Credentials: credentials.NewStaticCredentials(accessKey, secretKey, ""),
S3ForcePathStyle: aws.Bool(true),
DisableSSL: aws.Bool(false),