feat(e2e): share probe client across both internal LB suites + SG fixes

- Replaces per-suite probe client with a single shared VM launched in the
  fixture. Saves ~60s by avoiding a cold boot between the ALB and NLB
  internal suites.
- Client cloud-init exposes a JSON trigger endpoint on :9090 in addition
  to the http.server on :80 (results dir). Each suite POSTs
  {proto, ip, count, outfile, http_port, tcp_port} to run a probe loop
  against the LB private IP and writes the result file under /tmp/httpd
  for the driver to fetch back.
- configureDefaultSG now uses the structured IpPermissions form (top-level
  shortcut is silently ignored by vpcd — tracked separately) and opens
  port 9090 in addition to 80 + 9000.
- deleteLB captures the underlying sys.micro VM id via the LB ENI before
  delete, then polls DescribeLoadBalancers until LoadBalancerNotFound so
  the next suite doesn't race the capacity reclaim.
- Internal_NLB sleeps 15s up front to let the ALB sys.micro deallocate
  fully before its own createLB fires. DescribeLoadBalancers can report
  the ALB gone before the vCPU/memory allocation is actually released on
  capacity-tight dev hosts; without this settle, NLB's createLB trips the
  sys.micro reserve and lands in terminal state=failed. Still cheaper
  than cold-booting a per-suite client.
This commit is contained in:
joshsiv-mulga
2026-05-18 12:40:35 +10:00
parent 5591915163
commit afa933ac05
3 changed files with 208 additions and 101 deletions
+149 -61
View File
@@ -7,13 +7,13 @@ import (
"context"
_ "embed"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"net/http"
"os"
"strings"
"testing"
"text/template"
"time"
"github.com/aws/aws-sdk-go/aws"
@@ -34,6 +34,7 @@ const (
lbKeyName = "lb-e2e-key"
httpPort = 80
tcpPort = 9000
triggerPort = 9090
probesPerRun = 20
)
@@ -88,6 +89,12 @@ func TestLoadBalancer(t *testing.T) {
runLBSuite(t, client, fixture, kindALB, "internal", nil, "")
})
t.Run("Internal_NLB", func(t *testing.T) {
// DescribeLoadBalancers reports the ALB gone before the sys.micro VM's
// vCPU/memory allocation is actually reclaimed. Without this settle,
// NLB's createLB races the deallocate on capacity-tight dev hosts and
// trips the sys.micro reserve, ending up in terminal state=failed.
// Cheaper than cold-booting a fresh probe client per suite.
time.Sleep(15 * time.Second)
runLBSuite(t, client, fixture, kindNLB, "internal", nil, "")
})
}
@@ -102,6 +109,8 @@ type sharedFixture struct {
AMIID string
InstanceType string
AppInstanceIDs []string
ClientID string
ClientPublicIP string
}
func setupSharedFixture(t *testing.T, c *harness.AWSClient, artifacts string) *sharedFixture {
@@ -125,14 +134,17 @@ func setupSharedFixture(t *testing.T, c *harness.AWSClient, artifacts string) *s
launchAppInstances(t, c, f)
t.Cleanup(func() { terminateInstances(t, c, f.AppInstanceIDs) })
launchSharedProbeClient(t, c, f)
t.Cleanup(func() { terminateInstances(t, c, []string{f.ClientID}) })
harness.OnFailure(t, func() {
dumpDaemonLogs(t, artifacts, "setup")
})
return f
}
//go:embed testdata/client-userdata.sh.tmpl
var clientUserDataTmpl string
//go:embed testdata/client-userdata.sh
var clientUserData string
// --- Discovery helpers ---------------------------------------------------
@@ -304,13 +316,19 @@ func configureDefaultSG(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
f.SecurityGroup = aws.StringValue(out.SecurityGroups[0].GroupId)
t.Logf("default SG: %s", f.SecurityGroup)
for _, port := range []int64{httpPort, tcpPort} {
// Use structured IpPermissions form — spinifex's vpcd ignores the
// top-level IpProtocol/FromPort/ToPort/CidrIp shortcut and silently
// returns newRules=0, so the OVN ACL is never installed and ingress
// drops at the port group. Filed as a separate bead on the daemon side.
for _, port := range []int64{httpPort, tcpPort, triggerPort} {
_, err := c.EC2.AuthorizeSecurityGroupIngress(&ec2.AuthorizeSecurityGroupIngressInput{
GroupId: aws.String(f.SecurityGroup),
IpProtocol: aws.String("tcp"),
FromPort: aws.Int64(port),
ToPort: aws.Int64(port),
CidrIp: aws.String("0.0.0.0/0"),
GroupId: aws.String(f.SecurityGroup),
IpPermissions: []*ec2.IpPermission{{
IpProtocol: aws.String("tcp"),
FromPort: aws.Int64(port),
ToPort: aws.Int64(port),
IpRanges: []*ec2.IpRange{{CidrIp: aws.String("0.0.0.0/0")}},
}},
})
if err != nil {
var aerr awserr.Error
@@ -526,12 +544,6 @@ func deleteLB(t *testing.T, c *harness.AWSClient, lb lbInfo) {
if lb.ARN == "" {
return
}
if _, err := c.ELBv2.DeleteLoadBalancer(&elbv2.DeleteLoadBalancerInput{
LoadBalancerArn: aws.String(lb.ARN),
}); err != nil {
t.Logf("delete LB %s: %v", lb.ARN, err)
return
}
prefix, lbName := "app", "alb"
if lb.Type == "network" {
prefix, lbName = "net", "nlb"
@@ -541,7 +553,68 @@ func deleteLB(t *testing.T, c *harness.AWSClient, lb lbInfo) {
suffix = "inet"
}
filter := fmt.Sprintf("ELB %s/lb-e2e-%s-%s/%s", prefix, lbName, suffix, lb.ID)
// Capture the underlying sys.micro VM id before deleting so we can wait
// for it to actually terminate. ELBv2.DeleteLoadBalancer returns once
// the LB resource is gone, but the system VM termination is async and
// holds the vCPU/memory allocation until reaped — without this wait,
// the next suite's createLB can race the capacity reclaim and trip the
// reserve, ending up in terminal state=failed.
var sysInstanceID string
if eniOut, err := c.EC2.DescribeNetworkInterfaces(&ec2.DescribeNetworkInterfacesInput{
Filters: []*ec2.Filter{{
Name: aws.String("description"),
Values: []*string{aws.String(filter)},
}},
}); err == nil && len(eniOut.NetworkInterfaces) > 0 && eniOut.NetworkInterfaces[0].Attachment != nil {
sysInstanceID = aws.StringValue(eniOut.NetworkInterfaces[0].Attachment.InstanceId)
}
if _, err := c.ELBv2.DeleteLoadBalancer(&elbv2.DeleteLoadBalancerInput{
LoadBalancerArn: aws.String(lb.ARN),
}); err != nil {
t.Logf("delete LB %s: %v", lb.ARN, err)
return
}
// Block until describe reports LoadBalancerNotFound — the daemon doesn't
// mark the LB gone until the underlying sys.micro VM has been torn down
// and its vCPU/memory deallocated. Polling this avoids the capacity race
// where the next LB createLB fires before deallocation completes (sys
// instances are filtered from DescribeInstances so WaitForInstanceTerminated
// is a no-op for them).
waitForLBGone(t, c, lb.ARN, 60*time.Second)
harness.WaitForENICleanup(t, c, filter, lb.ARN, 30*time.Second)
if sysInstanceID != "" {
harness.WaitForInstanceTerminated(t, c, []string{sysInstanceID}, 60*time.Second)
}
}
// waitForLBGone polls DescribeLoadBalancers until the daemon reports
// LoadBalancerNotFound (or the LB no longer appears in the result), which
// in spinifex corresponds to the underlying system VM having finished
// terminating and released its capacity.
func waitForLBGone(t *testing.T, c *harness.AWSClient, arn string, timeout time.Duration) {
t.Helper()
deadline := time.Now().Add(timeout)
for {
out, err := c.ELBv2.DescribeLoadBalancers(&elbv2.DescribeLoadBalancersInput{
LoadBalancerArns: []*string{aws.String(arn)},
})
if err != nil {
var aerr awserr.Error
if errors.As(err, &aerr) && aerr.Code() == "LoadBalancerNotFound" {
return
}
}
if err == nil && len(out.LoadBalancers) == 0 {
return
}
if time.Now().After(deadline) {
t.Logf("waitForLBGone: %s still visible after %s (continuing)", arn, timeout)
return
}
time.Sleep(2 * time.Second)
}
}
func createListener(t *testing.T, c *harness.AWSClient, lbArn, proto string, port int64, tgArn string) string {
@@ -708,25 +781,14 @@ func runNLBDeregisterDraining(t *testing.T, c *harness.AWSClient, tgArn, targetI
assert.True(t, remaining == 1 || draining >= 1, "NLB deregister: expected 1 remaining or >=1 draining")
}
// --- Internal traffic via per-suite client VM -----------------------------
// --- Shared probe client + per-suite trigger ----------------------------
// runInternalTrafficViaClient launches a fresh client VM whose user-data
// probes the supplied LB private IP from inside the VPC, then fetches the
// results file from the client's public-port-80 server. One client per
// suite (ALB OR NLB) keeps peak instance count at 4 (2 app + 1 LB +
// 1 client) on capacity-constrained dev nodes.
func runInternalTrafficViaClient(t *testing.T, c *harness.AWSClient, f *sharedFixture, kind lbKind, lbIP string) {
// launchSharedProbeClient launches one client VM whose user-data exposes
// http.server on :80 (results dir) and a JSON trigger endpoint on :9090.
// Both internal suites hit the same client — avoids the ~60s second cold
// boot of a per-suite client.
func launchSharedProbeClient(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
t.Helper()
var albIP, nlbIP, resultsFile, proto string
if kind == kindALB {
albIP, resultsFile, proto = lbIP, "alb_results.txt", "http"
} else {
nlbIP, resultsFile, proto = lbIP, "nlb_results.txt", "tcp"
}
userData, err := renderClientUserData(albIP, nlbIP, probesPerRun)
require.NoError(t, err)
out, err := c.EC2.RunInstances(&ec2.RunInstancesInput{
ImageId: aws.String(f.AMIID),
InstanceType: aws.String(f.InstanceType),
@@ -734,43 +796,69 @@ func runInternalTrafficViaClient(t *testing.T, c *harness.AWSClient, f *sharedFi
SubnetId: aws.String(f.SubnetID),
MinCount: aws.Int64(1),
MaxCount: aws.Int64(1),
UserData: aws.String(base64Encode(userData)),
UserData: aws.String(base64Encode(clientUserData)),
})
require.NoErrorf(t, err, "run-instances client (%s)", kind)
clientID := aws.StringValue(out.Instances[0].InstanceId)
t.Cleanup(func() { terminateInstances(t, c, []string{clientID}) })
t.Logf("%s client: %s", kind, clientID)
require.NoError(t, err, "run-instances probe client")
f.ClientID = aws.StringValue(out.Instances[0].InstanceId)
t.Logf("probe client: %s", f.ClientID)
harness.WaitForInstanceRunning(t, c, clientID, 120*time.Second)
eni := harness.InstanceENI(t, c, clientID)
clientIP := publicIP(eni)
require.NotEmpty(t, clientIP, "client VM needs public IP")
harness.WaitForInstanceRunning(t, c, f.ClientID, 120*time.Second)
eni := harness.InstanceENI(t, c, f.ClientID)
f.ClientPublicIP = publicIP(eni)
require.NotEmpty(t, f.ClientPublicIP, "probe client needs public IP")
t.Logf("probe client public IP: %s", f.ClientPublicIP)
statusURL := fmt.Sprintf("http://%s:%d/status.txt", clientIP, httpPort)
// Wait until the trigger server is accepting connections before the
// first suite starts firing probes at it.
triggerURL := fmt.Sprintf("http://%s:%d/", f.ClientPublicIP, triggerPort)
harness.Eventually(t, func() bool {
body, err := plainHTTPGet(statusURL, 5*time.Second)
return err == nil && strings.TrimSpace(body) == "done"
}, 5*time.Minute, 5*time.Second, "client probe did not complete")
results, err := plainHTTPGet(fmt.Sprintf("http://%s:%d/%s", clientIP, httpPort, resultsFile), 10*time.Second)
require.NoErrorf(t, err, "fetch %s", resultsFile)
harness.AssertRoundRobin(t, harness.VerifyResultsLines(results, proto), 1, probesPerRun/2, string(kind)+" internal")
req, _ := http.NewRequest(http.MethodGet, triggerURL, nil)
client := &http.Client{Timeout: 3 * time.Second}
resp, err := client.Do(req)
if err != nil {
return false
}
resp.Body.Close()
return true
}, 5*time.Minute, 5*time.Second, "probe client trigger server not ready")
}
func renderClientUserData(albIP, nlbIP string, n int) (string, error) {
tmpl, err := template.New("client").Parse(clientUserDataTmpl)
if err != nil {
return "", err
// runInternalTrafficViaClient POSTs a probe order to the shared client's
// trigger server, then fetches the results file it wrote and parses it.
func runInternalTrafficViaClient(t *testing.T, _ *harness.AWSClient, f *sharedFixture, kind lbKind, lbIP string) {
t.Helper()
var proto, resultsFile string
if kind == kindALB {
proto, resultsFile = "http", fmt.Sprintf("alb-%d.txt", time.Now().UnixNano())
} else {
proto, resultsFile = "tcp", fmt.Sprintf("nlb-%d.txt", time.Now().UnixNano())
}
var buf bytes.Buffer
if err := tmpl.Execute(&buf, map[string]any{
"ALBPrivateIP": albIP,
"NLBPrivateIP": nlbIP,
"NumRequests": n,
}); err != nil {
return "", err
order := map[string]any{
"proto": proto,
"ip": lbIP,
"count": probesPerRun,
"outfile": resultsFile,
"http_port": httpPort,
"tcp_port": tcpPort,
}
return buf.String(), nil
body, err := json.Marshal(order)
require.NoError(t, err)
triggerURL := fmt.Sprintf("http://%s:%d/", f.ClientPublicIP, triggerPort)
client := &http.Client{Timeout: 90 * time.Second}
resp, err := client.Post(triggerURL, "application/json", bytes.NewReader(body))
require.NoErrorf(t, err, "trigger %s probe", kind)
resp.Body.Close()
require.Equalf(t, 200, resp.StatusCode, "trigger %s probe HTTP status", kind)
results, err := plainHTTPGet(
fmt.Sprintf("http://%s:%d/%s", f.ClientPublicIP, httpPort, resultsFile),
10*time.Second)
require.NoErrorf(t, err, "fetch %s", resultsFile)
harness.AssertRoundRobin(t,
harness.VerifyResultsLines(results, proto),
1, probesPerRun/2, string(kind)+" internal")
}
// plainHTTPGet fetches a plain-HTTP URL (no TLS). The probe client serves
+59
View File
@@ -0,0 +1,59 @@
#!/bin/bash
# Shared probe client. Long-lived across both internal LB suites — saves
# the ~60s second cold boot the per-suite client would otherwise incur.
#
# - Port 80: python http.server, serves /tmp/httpd (status + results files)
# - Port 9090: tiny trigger server. POST JSON {proto, ip, count, outfile}
# runs the probe loop synchronously and writes the result
# file under /tmp/httpd so the driver can curl it back.
set -eu
mkdir -p /tmp/httpd
cd /tmp/httpd
echo "ready" > status.txt
nohup python3 -m http.server 80 --bind 0.0.0.0 > /dev/null 2>&1 &
cat > /tmp/trigger.py <<'PYEOF'
import http.server, json, os, socketserver, subprocess
OUT_DIR = "/tmp/httpd"
class Handler(http.server.BaseHTTPRequestHandler):
def do_POST(self):
try:
n = int(self.headers.get("Content-Length", 0))
body = json.loads(self.rfile.read(n).decode())
proto = body["proto"]
ip = body["ip"]
count = int(body.get("count", 20))
outfile = body["outfile"]
tcp_port = int(body.get("tcp_port", 9000))
http_port = int(body.get("http_port", 80))
except Exception as exc:
self.send_response(400); self.end_headers()
self.wfile.write(f"bad request: {exc}".encode())
return
path = os.path.join(OUT_DIR, outfile)
with open(path, "w") as fh:
for _ in range(count):
if proto == "http":
r = subprocess.run(
["curl", "-s", "--max-time", "5", f"http://{ip}:{http_port}/"],
capture_output=True, text=True)
fh.write(r.stdout.strip() + "\n")
else:
r = subprocess.run(
["bash", "-c", f'echo "" | nc -w2 {ip} {tcp_port} 2>/dev/null || true'],
capture_output=True, text=True)
fh.write(r.stdout.strip() + "\n")
self.send_response(200); self.end_headers()
self.wfile.write(b"ok")
def log_message(self, *a, **kw):
pass
socketserver.TCPServer.allow_reuse_address = True
socketserver.TCPServer(("0.0.0.0", 9090), Handler).serve_forever()
PYEOF
nohup python3 /tmp/trigger.py > /tmp/trigger.log 2>&1 &
-40
View File
@@ -1,40 +0,0 @@
#!/bin/bash
# Client VM user-data template for internal LB suites. Probes ALB (HTTP)
# or NLB (TCP) private IP (one per suite) and publishes results over an
# HTTP server on port 80 for the test driver to fetch.
NUM_REQUESTS={{.NumRequests}}
mkdir -p /tmp/httpd && cd /tmp/httpd
echo "waiting" > status.txt
nohup python3 -m http.server 80 --bind 0.0.0.0 > /dev/null 2>&1 &
{{if .ALBPrivateIP}}
ALB_IP="{{.ALBPrivateIP}}"
for i in $(seq 1 60); do
if curl -s --max-time 3 "http://${ALB_IP}:80/" 2>/dev/null | grep -q instance_id; then break; fi
sleep 5
done
> alb_results.txt
for i in $(seq 1 $NUM_REQUESTS); do
curl -s --max-time 5 "http://${ALB_IP}:80/" >> alb_results.txt 2>/dev/null
echo "" >> alb_results.txt
done
{{end}}
{{if .NLBPrivateIP}}
NLB_IP="{{.NLBPrivateIP}}"
for i in $(seq 1 60); do
PROBE=$(echo "" | nc -w3 ${NLB_IP} 9000 2>/dev/null || true)
if [ -n "$PROBE" ]; then break; fi
sleep 5
done
> nlb_results.txt
for i in $(seq 1 $NUM_REQUESTS); do
RESP=$(echo "" | nc -w2 ${NLB_IP} 9000 2>/dev/null || true)
echo "$RESP" >> nlb_results.txt
done
{{end}}
echo "done" > status.txt