mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
Josh Siviour
f973b306c4
Every daemon read-modify-wrote the same zone object in s3://northstar with no mutual exclusion. A NATS queue group balances load, it does not serialise distinct messages, so concurrent daemons lost each other's records and could leave the object unparseable as TOML. Once unparseable it stayed that way: both the writer and the drift reconciler must parse the zone before they can repair it, so DNS changes stopped applying until the object was edited by hand. The exposure did not need user activity. The drift backstop ran on every node, unelected, on a 60s ticker, so an idle three-node cluster raced its own zone once per tick. Five publisher families feed the same object, so ordinary concurrent API calls collided too. - Add a per-zone JetStream KV lock, following the CAS patterns already used by the vpcd reconcile leader and the vCPU quota store. One key per zone, so unrelated zones never block. The wait budget stays under requestTimeout so a producer's ack is not lost to contention, and a write aborts rather than land outside an expired lease. - Rebuild a zone whose stored bytes do not parse instead of failing every write forever. The reconciler reports a corrupt zone as absent, which suppresses pruning while still publishing the full desired set. Detection matches the decoder's error type so it works against the pinned northstar release. - Elect one node per drift cycle, removing the idle-cluster self-race. - Correct three comments claiming the queue group serialised zone writes; they are why the hazard went unnoticed. - Run the LB e2e internet-facing subtests in parallel. Their serialisation workaround exempted internal LBs on the false premise that those use a different DNS path; ELBChanges sets the same base-domain zone for both.
1877 lines
70 KiB
Go
1877 lines
70 KiB
Go
//go:build e2e
|
||
|
||
package lb
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"crypto/x509"
|
||
_ "embed"
|
||
"encoding/base64"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"net"
|
||
"net/http"
|
||
"os"
|
||
"strings"
|
||
"testing"
|
||
"time"
|
||
|
||
"github.com/aws/aws-sdk-go/aws"
|
||
"github.com/aws/aws-sdk-go/aws/awserr"
|
||
"github.com/aws/aws-sdk-go/service/acm"
|
||
"github.com/aws/aws-sdk-go/service/ec2"
|
||
"github.com/aws/aws-sdk-go/service/elbv2"
|
||
"github.com/mulgadc/spinifex/tests/e2e/harness"
|
||
"github.com/stretchr/testify/assert"
|
||
"github.com/stretchr/testify/require"
|
||
)
|
||
|
||
//go:embed testdata/app-userdata.sh
|
||
var appUserData string
|
||
|
||
const (
|
||
lbVPCCIDR = "10.200.0.0/16"
|
||
lbSubnetCIDR = "10.200.1.0/24"
|
||
lbKeyName = "lb-e2e-key"
|
||
httpPort = 80
|
||
tcpPort = 9000
|
||
udpPort = 9001
|
||
triggerPort = 9090
|
||
probesPerRun = 20
|
||
)
|
||
|
||
// LB kind: ALB or NLB. Used to parameterise the per-suite path.
|
||
type lbKind string
|
||
|
||
const (
|
||
kindALB lbKind = "ALB"
|
||
kindNLB lbKind = "NLB"
|
||
)
|
||
|
||
// TestLoadBalancer runs 8 LB scenarios (ALB/NLB × internet-facing/internal, plus
|
||
// HTTPS/UDP/ModifyListener/ListenerRules variants) against a shared VPC and app-instance
|
||
// fixture. Each scenario stands up its own dedicated LB (+ target group) — never shared
|
||
// across scenarios, since ModifyListener and ListenerRules mutate their LB's listener/rules
|
||
// in place and must not race a sibling reading the same LB.
|
||
//
|
||
// On multinode clusters (peer != "", i.e. the cluster has the node2 the internet-facing
|
||
// scenarios drive traffic from — this suite's multinode target is a 3-node cluster) there is
|
||
// enough spare capacity to stand up every scenario's dedicated LB at once, so subtests run
|
||
// concurrently via t.Parallel() instead of serially tearing one LB down before the next is
|
||
// created. Single-node dev boxes lack that headroom and keep the original serial
|
||
// create -> verify -> teardown path, where createActiveLB's retry/backoff absorbs the wait
|
||
// for the previous LB's sys.micro slot to be reclaimed.
|
||
func TestLoadBalancer(t *testing.T) {
|
||
env := harness.LoadEnv(t)
|
||
skipIfDevNetworking(t, env)
|
||
harness.RequireDNSEnabled(t, env)
|
||
|
||
// Resolve peer availability before building the shared fixture so the "skip" message
|
||
// appears immediately rather than after minutes of VPC/VM setup.
|
||
peer := pickPeer(env)
|
||
var ssh *harness.PeerSSH
|
||
if peer != "" {
|
||
ssh = harness.NewPeerSSH()
|
||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||
defer cancel()
|
||
if err := ssh.Ping(ctx, peer); err != nil {
|
||
t.Logf("cannot SSH to peer %s: %v (internet-facing suites will skip)", peer, err)
|
||
peer = ""
|
||
ssh = nil
|
||
}
|
||
}
|
||
if peer == "" {
|
||
t.Logf("no peer node available — internet-facing subtests will skip; building shared fixture for internal subtests")
|
||
}
|
||
|
||
artifacts := harness.ArtifactDir(t, env)
|
||
client := harness.NewAWSClient(t, env)
|
||
fixture := setupSharedFixture(t, client, artifacts)
|
||
|
||
// A live peer implies the 3-node multinode cluster this restructure targets, which has
|
||
// the spare capacity to stand up every scenario's LB concurrently (see doc comment
|
||
// above). Single-node dev boxes (peer == "") fall through to the serial path below.
|
||
parallelizeLBs := peer != ""
|
||
|
||
// Internal and internet-facing LB records share one zone object: ELBChanges
|
||
// sets Zone to the base domain for both, and the names differ only by an
|
||
// "internal-" prefix. Concurrent registrations used to clobber each other
|
||
// there, which is why these ran serially; the writer now takes a per-zone lock,
|
||
// so all of them register in parallel.
|
||
t.Run("InternetFacing_ALB", func(t *testing.T) {
|
||
if peer == "" {
|
||
t.Skip("no peer node available")
|
||
}
|
||
if parallelizeLBs {
|
||
t.Parallel()
|
||
}
|
||
runLBSuite(t, client, fixture, kindALB, "internet-facing", ssh, peer)
|
||
})
|
||
t.Run("InternetFacing_NLB", func(t *testing.T) {
|
||
if peer == "" {
|
||
t.Skip("no peer node available")
|
||
}
|
||
if parallelizeLBs {
|
||
t.Parallel()
|
||
}
|
||
runLBSuite(t, client, fixture, kindNLB, "internet-facing", ssh, peer)
|
||
})
|
||
t.Run("InternetFacing_ALB_HTTPS", func(t *testing.T) {
|
||
if peer == "" {
|
||
// HTTPS handshake is driven from the test runner against the LB
|
||
// public IP; gate on the same peer-available signal the sibling
|
||
// internet-facing subtests use, where driver→LB reachability holds.
|
||
t.Skip("no peer node available")
|
||
}
|
||
if parallelizeLBs {
|
||
t.Parallel()
|
||
}
|
||
runHTTPSCertSuite(t, client, fixture)
|
||
})
|
||
|
||
if parallelizeLBs {
|
||
// Multinode: each internal subtest stands up its own independent LB and they start
|
||
// together with the internet-facing group above, so there is no serial queue for one
|
||
// stuck LB to back up behind.
|
||
// A single LB never reaching active only fails that one subtest — it can no longer
|
||
// cascade into 5x sequential 5-minute timeouts, so the serial path's fail-fast gate
|
||
// (below) is unnecessary here: concurrency itself bounds the worst-case wall time to
|
||
// one timeout window instead of five.
|
||
t.Run("Internal_ALB", func(t *testing.T) {
|
||
t.Parallel()
|
||
runLBSuite(t, client, fixture, kindALB, "internal", nil, "")
|
||
})
|
||
t.Run("Internal_NLB", func(t *testing.T) {
|
||
t.Parallel()
|
||
runLBSuite(t, client, fixture, kindNLB, "internal", nil, "")
|
||
})
|
||
t.Run("Internal_NLB_UDP", func(t *testing.T) {
|
||
t.Parallel()
|
||
runUDPNLBSuite(t, client, fixture)
|
||
})
|
||
t.Run("Internal_ALB_ModifyListener", func(t *testing.T) {
|
||
t.Parallel()
|
||
runModifyListenerSuite(t, client, fixture)
|
||
})
|
||
t.Run("Internal_ALB_ListenerRules", func(t *testing.T) {
|
||
t.Parallel()
|
||
runListenerRulesSuite(t, client, fixture)
|
||
})
|
||
return
|
||
}
|
||
|
||
// Single-node: gate remaining internal subtests on Internal_ALB — if the LB never
|
||
// reaches active, the rest will time out identically, so fail fast instead of burning
|
||
// ~5min each. No fixed inter-subtest sleeps: createActiveLB retries with backoff when
|
||
// the previous LB's sys.micro slot is still being reclaimed.
|
||
albOK := t.Run("Internal_ALB", func(t *testing.T) {
|
||
runLBSuite(t, client, fixture, kindALB, "internal", nil, "")
|
||
})
|
||
skipIfInternalBroken := func(t *testing.T) {
|
||
if !albOK {
|
||
t.Skip("Internal_ALB failed (LB never reached active) — skipping remaining internal subtests to fail fast")
|
||
}
|
||
}
|
||
t.Run("Internal_NLB", func(t *testing.T) {
|
||
skipIfInternalBroken(t)
|
||
runLBSuite(t, client, fixture, kindNLB, "internal", nil, "")
|
||
})
|
||
t.Run("Internal_NLB_UDP", func(t *testing.T) {
|
||
skipIfInternalBroken(t)
|
||
runUDPNLBSuite(t, client, fixture)
|
||
})
|
||
t.Run("Internal_ALB_ModifyListener", func(t *testing.T) {
|
||
skipIfInternalBroken(t)
|
||
runModifyListenerSuite(t, client, fixture)
|
||
})
|
||
t.Run("Internal_ALB_ListenerRules", func(t *testing.T) {
|
||
skipIfInternalBroken(t)
|
||
runListenerRulesSuite(t, client, fixture)
|
||
})
|
||
}
|
||
|
||
// --- Fixture: shared VPC, subnet, IGW, SG, app instances ----------------
|
||
|
||
type sharedFixture struct {
|
||
VPCID string
|
||
SubnetID string
|
||
IGWID string
|
||
MainRouteTableID string
|
||
SecurityGroup string
|
||
AMIID string
|
||
InstanceType string
|
||
AppInstanceIDs []string
|
||
ClientID string
|
||
ClientPublicIP string
|
||
}
|
||
|
||
func setupSharedFixture(t *testing.T, c *harness.AWSClient, artifacts string) *sharedFixture {
|
||
t.Helper()
|
||
f := &sharedFixture{}
|
||
|
||
harness.Phase(t, "Discovering cluster capacity")
|
||
f.InstanceType = discoverNanoInstanceType(t, c)
|
||
f.AMIID = discoverAMI(t, c)
|
||
|
||
harness.Phase(t, "Ensuring SSH key pair %q", lbKeyName)
|
||
ensureKeyPair(t, c)
|
||
t.Cleanup(func() { deleteKeyPair(t, c) })
|
||
|
||
harness.Phase(t, "Creating shared VPC topology (%s)", lbVPCCIDR)
|
||
createVPC(t, c, f)
|
||
t.Cleanup(func() { deleteVPC(t, c, f) })
|
||
createIGW(t, c, f)
|
||
t.Cleanup(func() { deleteIGW(t, c, f) })
|
||
createSubnet(t, c, f)
|
||
t.Cleanup(func() { deleteSubnet(t, c, f) })
|
||
addPublicRoute(t, c, f)
|
||
t.Cleanup(func() { deletePublicRoute(t, c, f) })
|
||
configureDefaultSG(t, c, f)
|
||
|
||
harness.Phase(t, "Launching app instances (2× %s)", f.InstanceType)
|
||
launchAppInstances(t, c, f)
|
||
t.Cleanup(func() { terminateInstances(t, c, f.AppInstanceIDs) })
|
||
|
||
harness.Phase(t, "Launching probe client")
|
||
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
|
||
var clientUserData string
|
||
|
||
// --- Discovery helpers ---------------------------------------------------
|
||
|
||
func skipIfDevNetworking(t *testing.T, env *harness.Env) {
|
||
t.Helper()
|
||
cfg := os.ExpandEnv("$HOME/spinifex/config/spinifex.toml")
|
||
if env.ConfigDir != "" {
|
||
cfg = env.ConfigDir + "/spinifex.toml"
|
||
}
|
||
raw, err := os.ReadFile(cfg)
|
||
if err != nil {
|
||
return
|
||
}
|
||
if bytes.Contains(raw, []byte("dev_networking = true")) {
|
||
t.Skip("dev_networking enabled in spinifex.toml; LB E2E requires pool mode w/ external IPAM")
|
||
}
|
||
}
|
||
|
||
func pickPeer(env *harness.Env) string {
|
||
if len(env.NodeIPs) < 2 {
|
||
return ""
|
||
}
|
||
return env.NodeIPs[1]
|
||
}
|
||
|
||
func discoverNanoInstanceType(t *testing.T, c *harness.AWSClient) string {
|
||
t.Helper()
|
||
out, err := c.EC2.DescribeInstanceTypes(&ec2.DescribeInstanceTypesInput{})
|
||
require.NoError(t, err)
|
||
for _, it := range out.InstanceTypes {
|
||
name := aws.StringValue(it.InstanceType)
|
||
if strings.Contains(name, "nano") {
|
||
t.Logf("instance type: %s", name)
|
||
return name
|
||
}
|
||
}
|
||
t.Fatal("no nano instance type available")
|
||
return ""
|
||
}
|
||
|
||
func discoverAMI(t *testing.T, c *harness.AWSClient) string {
|
||
t.Helper()
|
||
out, err := c.EC2.DescribeImages(&ec2.DescribeImagesInput{})
|
||
require.NoError(t, err)
|
||
var fallback, nonAlpine, ubuntu string
|
||
for _, img := range out.Images {
|
||
id := aws.StringValue(img.ImageId)
|
||
name := aws.StringValue(img.Name)
|
||
if fallback == "" {
|
||
fallback = id
|
||
}
|
||
if !strings.Contains(strings.ToLower(name), "alpine") && nonAlpine == "" {
|
||
nonAlpine = id
|
||
}
|
||
if strings.HasPrefix(name, "ami-ubuntu") {
|
||
ubuntu = id
|
||
break
|
||
}
|
||
}
|
||
for _, candidate := range []string{ubuntu, nonAlpine, fallback} {
|
||
if candidate != "" {
|
||
t.Logf("AMI: %s", candidate)
|
||
return candidate
|
||
}
|
||
}
|
||
t.Fatal("no AMIs available")
|
||
return ""
|
||
}
|
||
|
||
func ensureKeyPair(t *testing.T, c *harness.AWSClient) {
|
||
t.Helper()
|
||
_, _ = c.EC2.DeleteKeyPair(&ec2.DeleteKeyPairInput{KeyName: aws.String(lbKeyName)})
|
||
_, err := c.EC2.CreateKeyPair(&ec2.CreateKeyPairInput{KeyName: aws.String(lbKeyName)})
|
||
require.NoError(t, err, "create key pair %s", lbKeyName)
|
||
}
|
||
|
||
func deleteKeyPair(t *testing.T, c *harness.AWSClient) {
|
||
t.Helper()
|
||
if _, err := c.EC2.DeleteKeyPair(&ec2.DeleteKeyPairInput{KeyName: aws.String(lbKeyName)}); err != nil {
|
||
t.Logf("delete key pair: %v", err)
|
||
}
|
||
}
|
||
|
||
// --- VPC / Subnet / IGW / SG --------------------------------------------
|
||
|
||
func createVPC(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
|
||
t.Helper()
|
||
out, err := c.EC2.CreateVpc(&ec2.CreateVpcInput{CidrBlock: aws.String(lbVPCCIDR)})
|
||
require.NoError(t, err)
|
||
f.VPCID = aws.StringValue(out.Vpc.VpcId)
|
||
t.Logf("VPC: %s", f.VPCID)
|
||
}
|
||
|
||
func deleteVPC(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
|
||
t.Helper()
|
||
if f.VPCID == "" {
|
||
return
|
||
}
|
||
if _, err := c.EC2.DeleteVpc(&ec2.DeleteVpcInput{VpcId: aws.String(f.VPCID)}); err != nil {
|
||
t.Logf("delete VPC %s: %v", f.VPCID, err)
|
||
}
|
||
}
|
||
|
||
func createIGW(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
|
||
t.Helper()
|
||
out, err := c.EC2.CreateInternetGateway(&ec2.CreateInternetGatewayInput{})
|
||
require.NoError(t, err)
|
||
f.IGWID = aws.StringValue(out.InternetGateway.InternetGatewayId)
|
||
_, err = c.EC2.AttachInternetGateway(&ec2.AttachInternetGatewayInput{
|
||
InternetGatewayId: aws.String(f.IGWID),
|
||
VpcId: aws.String(f.VPCID),
|
||
})
|
||
require.NoError(t, err)
|
||
t.Logf("IGW: %s (attached to %s)", f.IGWID, f.VPCID)
|
||
}
|
||
|
||
func deleteIGW(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
|
||
t.Helper()
|
||
if f.IGWID == "" {
|
||
return
|
||
}
|
||
_, _ = c.EC2.DetachInternetGateway(&ec2.DetachInternetGatewayInput{
|
||
InternetGatewayId: aws.String(f.IGWID),
|
||
VpcId: aws.String(f.VPCID),
|
||
})
|
||
if _, err := c.EC2.DeleteInternetGateway(&ec2.DeleteInternetGatewayInput{
|
||
InternetGatewayId: aws.String(f.IGWID),
|
||
}); err != nil {
|
||
t.Logf("delete IGW: %v", err)
|
||
}
|
||
}
|
||
|
||
func createSubnet(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
|
||
t.Helper()
|
||
out, err := c.EC2.CreateSubnet(&ec2.CreateSubnetInput{
|
||
VpcId: aws.String(f.VPCID),
|
||
CidrBlock: aws.String(lbSubnetCIDR),
|
||
})
|
||
require.NoError(t, err)
|
||
f.SubnetID = aws.StringValue(out.Subnet.SubnetId)
|
||
_, err = c.EC2.ModifySubnetAttribute(&ec2.ModifySubnetAttributeInput{
|
||
SubnetId: aws.String(f.SubnetID),
|
||
MapPublicIpOnLaunch: &ec2.AttributeBooleanValue{Value: aws.Bool(true)},
|
||
})
|
||
require.NoError(t, err)
|
||
t.Logf("subnet: %s (MapPublicIpOnLaunch)", f.SubnetID)
|
||
}
|
||
|
||
func deleteSubnet(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
|
||
t.Helper()
|
||
if f.SubnetID == "" {
|
||
return
|
||
}
|
||
if _, err := c.EC2.DeleteSubnet(&ec2.DeleteSubnetInput{SubnetId: aws.String(f.SubnetID)}); err != nil {
|
||
t.Logf("delete subnet: %v", err)
|
||
}
|
||
}
|
||
|
||
// addPublicRoute installs 0.0.0.0/0 → IGW on the VPC's main route table so the daemon
|
||
// classifies the subnet as public and does not install an OVN LR egress DROP.
|
||
func addPublicRoute(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
|
||
t.Helper()
|
||
out, err := c.EC2.DescribeRouteTables(&ec2.DescribeRouteTablesInput{
|
||
Filters: []*ec2.Filter{
|
||
{Name: aws.String("vpc-id"), Values: []*string{aws.String(f.VPCID)}},
|
||
{Name: aws.String("association.main"), Values: []*string{aws.String("true")}},
|
||
},
|
||
})
|
||
require.NoError(t, err, "describe main route table")
|
||
require.NotEmpty(t, out.RouteTables, "VPC %s has no main route table", f.VPCID)
|
||
f.MainRouteTableID = aws.StringValue(out.RouteTables[0].RouteTableId)
|
||
_, err = c.EC2.CreateRoute(&ec2.CreateRouteInput{
|
||
RouteTableId: aws.String(f.MainRouteTableID),
|
||
DestinationCidrBlock: aws.String("0.0.0.0/0"),
|
||
GatewayId: aws.String(f.IGWID),
|
||
})
|
||
require.NoError(t, err, "create-route 0.0.0.0/0 -> IGW")
|
||
t.Logf("public route: %s 0.0.0.0/0 -> %s", f.MainRouteTableID, f.IGWID)
|
||
}
|
||
|
||
func deletePublicRoute(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
|
||
t.Helper()
|
||
if f.MainRouteTableID == "" {
|
||
return
|
||
}
|
||
if _, err := c.EC2.DeleteRoute(&ec2.DeleteRouteInput{
|
||
RouteTableId: aws.String(f.MainRouteTableID),
|
||
DestinationCidrBlock: aws.String("0.0.0.0/0"),
|
||
}); err != nil {
|
||
t.Logf("delete public route: %v", err)
|
||
}
|
||
}
|
||
|
||
func configureDefaultSG(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
|
||
t.Helper()
|
||
out, err := c.EC2.DescribeSecurityGroups(&ec2.DescribeSecurityGroupsInput{
|
||
Filters: []*ec2.Filter{
|
||
{Name: aws.String("vpc-id"), Values: []*string{aws.String(f.VPCID)}},
|
||
{Name: aws.String("group-name"), Values: []*string{aws.String("default")}},
|
||
},
|
||
})
|
||
require.NoError(t, err)
|
||
require.NotEmpty(t, out.SecurityGroups, "VPC default SG missing")
|
||
f.SecurityGroup = aws.StringValue(out.SecurityGroups[0].GroupId)
|
||
t.Logf("default SG: %s", f.SecurityGroup)
|
||
|
||
// Use structured IpPermissions form — vpcd ignores the top-level shorthand fields
|
||
// so the OVN ACL would never be installed otherwise.
|
||
for _, port := range []int64{httpPort, tcpPort, triggerPort} {
|
||
_, err := c.EC2.AuthorizeSecurityGroupIngress(&ec2.AuthorizeSecurityGroupIngressInput{
|
||
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
|
||
if !errors.As(err, &aerr) || aerr.Code() != "InvalidPermission.Duplicate" {
|
||
t.Fatalf("authorize tcp/%d: %v", port, err)
|
||
}
|
||
}
|
||
}
|
||
|
||
// UDP data plane for the NLB UDP suite (health-checked over TCP/9000).
|
||
if _, err := c.EC2.AuthorizeSecurityGroupIngress(&ec2.AuthorizeSecurityGroupIngressInput{
|
||
GroupId: aws.String(f.SecurityGroup),
|
||
IpPermissions: []*ec2.IpPermission{{
|
||
IpProtocol: aws.String("udp"),
|
||
FromPort: aws.Int64(udpPort),
|
||
ToPort: aws.Int64(udpPort),
|
||
IpRanges: []*ec2.IpRange{{CidrIp: aws.String("0.0.0.0/0")}},
|
||
}},
|
||
}); err != nil {
|
||
var aerr awserr.Error
|
||
if !errors.As(err, &aerr) || aerr.Code() != "InvalidPermission.Duplicate" {
|
||
t.Fatalf("authorize udp/%d: %v", udpPort, err)
|
||
}
|
||
}
|
||
}
|
||
|
||
// --- App instances -------------------------------------------------------
|
||
|
||
func launchAppInstances(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
|
||
t.Helper()
|
||
for i := 0; i < 2; i++ {
|
||
out, err := c.EC2.RunInstances(&ec2.RunInstancesInput{
|
||
ImageId: aws.String(f.AMIID),
|
||
InstanceType: aws.String(f.InstanceType),
|
||
KeyName: aws.String(lbKeyName),
|
||
SubnetId: aws.String(f.SubnetID),
|
||
MinCount: aws.Int64(1),
|
||
MaxCount: aws.Int64(1),
|
||
UserData: aws.String(base64Encode(appUserData)),
|
||
})
|
||
require.NoErrorf(t, err, "run-instances app %d", i+1)
|
||
require.NotEmpty(t, out.Instances)
|
||
id := aws.StringValue(out.Instances[0].InstanceId)
|
||
f.AppInstanceIDs = append(f.AppInstanceIDs, id)
|
||
t.Logf("app instance %d: %s", i+1, id)
|
||
}
|
||
|
||
for _, id := range f.AppInstanceIDs {
|
||
harness.WaitForInstanceRunning(t, c, id, 120*time.Second)
|
||
}
|
||
}
|
||
|
||
func terminateInstances(t *testing.T, c *harness.AWSClient, ids []string) {
|
||
t.Helper()
|
||
if len(ids) == 0 {
|
||
return
|
||
}
|
||
awsIDs := make([]*string, len(ids))
|
||
for i, id := range ids {
|
||
awsIDs[i] = aws.String(id)
|
||
}
|
||
if _, err := c.EC2.TerminateInstances(&ec2.TerminateInstancesInput{InstanceIds: awsIDs}); err != nil {
|
||
t.Logf("terminate instances: %v", err)
|
||
return
|
||
}
|
||
harness.WaitForInstanceTerminated(t, c, ids, 60*time.Second)
|
||
}
|
||
|
||
// --- LB suite: one LB (ALB or NLB) × one scheme (internet-facing/internal)
|
||
|
||
// runLBSuite creates one TG + LB + listener, asserts scheme/DNS/ENI invariants, runs
|
||
// traffic tests, then tears down before returning. Self-contained so subtests don't
|
||
// accumulate LB system instances on capacity-constrained dev nodes.
|
||
func runLBSuite(t *testing.T, c *harness.AWSClient, f *sharedFixture, kind lbKind, scheme string, ssh *harness.PeerSSH, peer string) {
|
||
t.Helper()
|
||
suffix := "int"
|
||
if scheme == "internet-facing" {
|
||
suffix = "inet"
|
||
}
|
||
lbName := strings.ToLower(string(kind))
|
||
label := fmt.Sprintf("%s %s", kind, scheme)
|
||
|
||
var proto, hcPath, eniDescPrefix, lbType string
|
||
var port int64
|
||
if kind == kindALB {
|
||
proto, port, hcPath, eniDescPrefix, lbType = "HTTP", httpPort, "/index.html", "app", "application"
|
||
} else {
|
||
proto, port, hcPath, eniDescPrefix, lbType = "TCP", tcpPort, "", "net", "network"
|
||
}
|
||
|
||
tgArn := createTargetGroup(t, c, f, fmt.Sprintf("lb-e2e-%s-%s-tg", lbName, suffix), proto, port, hcPath)
|
||
t.Cleanup(func() { deleteTargetGroup(t, c, tgArn) })
|
||
|
||
registerTargets(t, c, tgArn, f.AppInstanceIDs)
|
||
t.Cleanup(func() { deregisterTargets(t, c, tgArn, f.AppInstanceIDs) })
|
||
|
||
// ALB scenarios get their own LB-facing SG so concurrent scenarios (parallel
|
||
// internal group) never share one SG; NLBs auto-isolate via their managed SG.
|
||
var sgIDs []string
|
||
if kind == kindALB {
|
||
sgIDs = []string{createScenarioSG(t, c, f, fmt.Sprintf("lb-e2e-%s-%s-sg", lbName, suffix), port)}
|
||
}
|
||
|
||
lb, _ := createActiveLB(t, c, f, fmt.Sprintf("lb-e2e-%s-%s", lbName, suffix), lbType, scheme, proto, port, tgArn, label, sgIDs)
|
||
|
||
assert.Equal(t, scheme, lb.Scheme, label+" scheme")
|
||
assert.Equal(t, lbType, lb.Type, label+" type")
|
||
if kind == kindNLB {
|
||
assert.Contains(t, lb.ARN, "/net/", label+" ARN must contain /net/")
|
||
}
|
||
|
||
if kind == kindNLB {
|
||
captureLBConsoleOnFailure(t, c, eniDescPrefix, lb)
|
||
}
|
||
harness.WaitForTargetsHealthy(t, c, tgArn, 2, label, 2*time.Minute)
|
||
|
||
eni := lbENI(t, c, eniDescPrefix, lb)
|
||
|
||
if scheme == "internet-facing" {
|
||
ip := publicIP(eni)
|
||
require.NotEmpty(t, ip, label+" needs public IP")
|
||
assertLBDNSResolves(t, lb.DNSName, ip, label)
|
||
runInternetFacingTrafficSingle(t, kind, ssh, peer, ip)
|
||
if kind == kindNLB {
|
||
runNLBDeregisterDraining(t, c, tgArn, f.AppInstanceIDs[0])
|
||
}
|
||
return
|
||
}
|
||
|
||
// internal
|
||
assert.Empty(t, publicIP(eni), label+" must not have public IP")
|
||
priv := privateIP(eni)
|
||
require.NotEmpty(t, priv, label+" needs private IP")
|
||
assertInternalDNS(t, c, lb.ARN, label)
|
||
assertLBDNSResolves(t, lb.DNSName, priv, label)
|
||
runInternalTrafficViaClient(t, c, f, kind, priv)
|
||
}
|
||
|
||
// runHTTPSCertSuite exercises the ACM → HTTPS listener path end-to-end:
|
||
// ImportCertificate a self-signed leaf → create an internet-facing ALB HTTPS
|
||
// listener referencing the cert ARN → complete a TLS handshake against the LB
|
||
// and assert HAProxy serves exactly the imported cert → drive one HTTPS GET
|
||
// that terminates TLS at the LB and forwards to a healthy backend →
|
||
// DeleteCertificate. The cert is minted for the LB's public IP (known only
|
||
// after the LB is active) so the end-to-end GET can verify the chain by IP SAN.
|
||
func runHTTPSCertSuite(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
|
||
t.Helper()
|
||
const label = "ALB internet-facing HTTPS (ACM)"
|
||
const httpsPort int64 = 443
|
||
|
||
sgIDs := []string{createScenarioSG(t, c, f, "lb-e2e-https-sg", httpsPort)}
|
||
|
||
tgArn := createTargetGroup(t, c, f, "lb-e2e-https-tg", "HTTP", httpPort, "/index.html")
|
||
t.Cleanup(func() { deleteTargetGroup(t, c, tgArn) })
|
||
registerTargets(t, c, tgArn, f.AppInstanceIDs)
|
||
t.Cleanup(func() { deregisterTargets(t, c, tgArn, f.AppInstanceIDs) })
|
||
|
||
lb, _ := createActiveLB(t, c, f, "lb-e2e-https", "application", "internet-facing", "", 0, "", label, sgIDs)
|
||
eni := lbENI(t, c, "app", lb)
|
||
pubIP := publicIP(eni)
|
||
require.NotEmpty(t, pubIP, label+" needs a public IP for the TLS handshake")
|
||
|
||
// Mint a self-signed leaf for the LB's public IP and import it into ACM.
|
||
certPEM, keyPEM, leaf, err := harness.GenerateSelfSignedCertPEM(pubIP)
|
||
require.NoError(t, err, "generate self-signed cert")
|
||
imp, err := c.ACM.ImportCertificate(&acm.ImportCertificateInput{
|
||
Certificate: certPEM,
|
||
PrivateKey: keyPEM,
|
||
})
|
||
require.NoError(t, err, "acm import-certificate")
|
||
certArn := aws.StringValue(imp.CertificateArn)
|
||
require.NotEmpty(t, certArn, "ImportCertificate returned empty CertificateArn")
|
||
require.Contains(t, certArn, ":certificate/", "cert ARN must be an ACM certificate ARN")
|
||
t.Cleanup(func() {
|
||
if _, derr := c.ACM.DeleteCertificate(&acm.DeleteCertificateInput{
|
||
CertificateArn: aws.String(certArn),
|
||
}); derr != nil {
|
||
t.Logf("delete certificate %s: %v", certArn, derr)
|
||
}
|
||
})
|
||
t.Logf("imported cert: %s", certArn)
|
||
|
||
listener := createHTTPSListener(t, c, lb.ARN, httpsPort, tgArn, certArn)
|
||
t.Cleanup(func() { deleteListener(t, c, listener) })
|
||
|
||
harness.WaitForTargetsHealthy(t, c, tgArn, 2, label, 2*time.Minute)
|
||
|
||
// TLS handshake: HAProxy must present exactly the imported leaf. The
|
||
// lb-agent picks up the new listener cert on its reconcile tick, so poll
|
||
// the handshake until the served cert matches before asserting.
|
||
var served *x509.Certificate
|
||
harness.EventuallyErr(t, func() error {
|
||
got, ferr := harness.FetchServedCert(pubIP, int(httpsPort), 5*time.Second)
|
||
if ferr != nil {
|
||
return ferr
|
||
}
|
||
if !harness.FingerprintMatches(got, leaf) {
|
||
return fmt.Errorf("served cert does not match imported cert yet")
|
||
}
|
||
served = got
|
||
return nil
|
||
}, 90*time.Second, 3*time.Second)
|
||
require.NotNil(t, served, label+" never served the imported cert")
|
||
t.Logf("TLS handshake ok — HAProxy served the imported cert (CN=%s)", served.Subject.CommonName)
|
||
|
||
version, _, err := harness.FetchTLSPosture(pubIP, int(httpsPort), 5*time.Second)
|
||
require.NoError(t, err, "fetch TLS posture")
|
||
assert.GreaterOrEqualf(t, version, uint16(0x0303), "negotiated TLS version must be >= 1.2, got 0x%04x", version)
|
||
|
||
// End-to-end: one HTTPS GET that terminates TLS at the LB and forwards to a
|
||
// backend. Verify the chain against the imported cert (IP SAN matches pubIP).
|
||
pool := x509.NewCertPool()
|
||
require.True(t, pool.AppendCertsFromPEM(certPEM), "build CA pool from imported cert")
|
||
url := fmt.Sprintf("https://%s:%d/index.html", pubIP, httpsPort)
|
||
harness.EventuallyErr(t, func() error {
|
||
status, body, gerr := harness.HTTPSGet(url, pool, 10*time.Second)
|
||
if gerr != nil {
|
||
return gerr
|
||
}
|
||
if status != http.StatusOK {
|
||
return fmt.Errorf("status %d (body=%q)", status, string(body))
|
||
}
|
||
return nil
|
||
}, 90*time.Second, 3*time.Second)
|
||
t.Logf("HTTPS GET %s -> 200 (TLS terminated at LB, forwarded to backend)", url)
|
||
}
|
||
|
||
// createHTTPSListener creates an HTTPS listener referencing certArn. Distinct
|
||
// from createListener because the ELBv2 handler requires Certificates for HTTPS
|
||
// and rejects them for HTTP.
|
||
func createHTTPSListener(t *testing.T, c *harness.AWSClient, lbArn string, port int64, tgArn, certArn string) string {
|
||
t.Helper()
|
||
// e2e:allow-create — the HTTPS listener is the subject under test (ACM cert -> termination).
|
||
out, err := c.ELBv2.CreateListener(&elbv2.CreateListenerInput{
|
||
LoadBalancerArn: aws.String(lbArn),
|
||
Protocol: aws.String("HTTPS"),
|
||
Port: aws.Int64(port),
|
||
Certificates: []*elbv2.Certificate{{CertificateArn: aws.String(certArn)}},
|
||
DefaultActions: []*elbv2.Action{{
|
||
Type: aws.String("forward"),
|
||
TargetGroupArn: aws.String(tgArn),
|
||
}},
|
||
})
|
||
require.NoError(t, err, "create-listener HTTPS")
|
||
require.NotEmpty(t, out.Listeners, "create-listener HTTPS returned no listeners")
|
||
arn := aws.StringValue(out.Listeners[0].ListenerArn)
|
||
t.Logf("listener HTTPS:%d -> %s (cert %s)", port, arn, certArn)
|
||
return arn
|
||
}
|
||
|
||
// createScenarioSG creates a dedicated security group for one ALB scenario's
|
||
// frontend/listener ports and authorizes tcp/0.0.0.0/0 on each of ports. Giving
|
||
// each ALB scenario its own SG means concurrent scenarios (see the parallel
|
||
// internal group in TestLoadBalancer) never authorize into, or otherwise
|
||
// mutate, the same SG object.
|
||
//
|
||
// This only isolates the LB-FACING side. The shared app-instance/probe-client
|
||
// SG (f.SecurityGroup, set up by configureDefaultSG) is left untouched and
|
||
// keeps permitting LB->target health-check and data traffic: SG ingress rules
|
||
// gate inbound traffic at the DESTINATION ENI's own SG regardless of which SG
|
||
// the source ENI carries, so targets never needed to share the LB's SG in the
|
||
// first place.
|
||
//
|
||
// NLBs don't need this: CreateLoadBalancer with no explicit SecurityGroups
|
||
// already gives every NLB its own auto-managed SG with listener ports opened
|
||
// automatically, so only ALB scenarios (which otherwise fall back onto the one
|
||
// shared VPC default SG) require an explicit dedicated SG here.
|
||
func createScenarioSG(t *testing.T, c *harness.AWSClient, f *sharedFixture, name string, ports ...int64) string {
|
||
t.Helper()
|
||
out, err := c.EC2.CreateSecurityGroup(&ec2.CreateSecurityGroupInput{ // e2e:allow-create — a dedicated per-scenario LB SG is the isolation under test; a shared Ensure* fixture would defeat it.
|
||
VpcId: aws.String(f.VPCID),
|
||
GroupName: aws.String(name),
|
||
Description: aws.String("lb-e2e dedicated LB-facing security group"),
|
||
})
|
||
require.NoErrorf(t, err, "create-security-group %s", name)
|
||
sgID := aws.StringValue(out.GroupId)
|
||
t.Cleanup(func() { deleteSecurityGroup(t, c, sgID) })
|
||
|
||
for _, port := range ports {
|
||
_, err := c.EC2.AuthorizeSecurityGroupIngress(&ec2.AuthorizeSecurityGroupIngressInput{
|
||
GroupId: aws.String(sgID),
|
||
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")}},
|
||
}},
|
||
})
|
||
require.NoErrorf(t, err, "authorize tcp/%d on %s", port, name)
|
||
}
|
||
t.Logf("scenario SG %s: %s (tcp ports=%v)", name, sgID, ports)
|
||
return sgID
|
||
}
|
||
|
||
// deleteSecurityGroup best-effort deletes a scenario SG created by
|
||
// createScenarioSG. Registered via t.Cleanup after the LB's own teardown
|
||
// (LIFO order means the LB, and the ENI referencing this SG, is gone first).
|
||
func deleteSecurityGroup(t *testing.T, c *harness.AWSClient, sgID string) {
|
||
if sgID == "" {
|
||
return
|
||
}
|
||
if _, err := c.EC2.DeleteSecurityGroup(&ec2.DeleteSecurityGroupInput{GroupId: aws.String(sgID)}); err != nil {
|
||
t.Logf("delete security group %s: %v", sgID, err)
|
||
}
|
||
}
|
||
|
||
// runUDPNLBSuite exercises an internal NLB with a UDP listener. The lb-agent probes
|
||
// targets over TCP/9000 (nginx stream has no active health-check); once healthy, a
|
||
// UDP datagram round-trip through the VIP proves the L4 data plane.
|
||
func runUDPNLBSuite(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
|
||
t.Helper()
|
||
const label = "NLB internal UDP"
|
||
|
||
tgArn := createUDPTargetGroup(t, c, f, "lb-e2e-nlb-udp-tg")
|
||
t.Cleanup(func() { deleteTargetGroup(t, c, tgArn) })
|
||
|
||
registerTargets(t, c, tgArn, f.AppInstanceIDs)
|
||
t.Cleanup(func() { deregisterTargets(t, c, tgArn, f.AppInstanceIDs) })
|
||
|
||
// NLB — no dedicated SG needed; CreateLoadBalancer auto-provisions one.
|
||
lb, _ := createActiveLB(t, c, f, "lb-e2e-nlb-udp", "network", "internal", "UDP", udpPort, tgArn, label, nil)
|
||
|
||
assert.Equal(t, "network", lb.Type, label+" type")
|
||
assert.Contains(t, lb.ARN, "/net/", label+" ARN must contain /net/")
|
||
|
||
captureLBConsoleOnFailure(t, c, "net", lb)
|
||
// The key assertion: NLB targets reach healthy via the agent's active
|
||
// prober. Pre-feature this timed out (nginx reported empty server lists).
|
||
harness.WaitForTargetsHealthy(t, c, tgArn, 2, label, 2*time.Minute)
|
||
// Targets only reach healthy if the lb-agent authenticated to the gateway;
|
||
// with static keys dropped from user-data that auth is IMDS-only, so assert
|
||
// the cred provenance explicitly (no baked keys, assumed-role principal).
|
||
assertLBAgentAuthViaIMDS(t)
|
||
|
||
eni := lbENI(t, c, "net", lb)
|
||
priv := privateIP(eni)
|
||
require.NotEmpty(t, priv, label+" needs private IP")
|
||
|
||
runInternalUDPViaClient(t, f, priv, label)
|
||
}
|
||
|
||
// runInternalUDPViaClient drives the shared client to send UDP datagrams at the
|
||
// NLB VIP and asserts the echoed hostnames round-robin across both backends.
|
||
func runInternalUDPViaClient(t *testing.T, f *sharedFixture, lbIP, label string) {
|
||
t.Helper()
|
||
resultsFile := fmt.Sprintf("nlb-udp-%d.txt", time.Now().UnixNano())
|
||
order := map[string]any{
|
||
"proto": "udp",
|
||
"ip": lbIP,
|
||
"count": probesPerRun,
|
||
"outfile": resultsFile,
|
||
"udp_port": udpPort,
|
||
}
|
||
body, err := json.Marshal(order)
|
||
require.NoError(t, err)
|
||
|
||
triggerURL := fmt.Sprintf("http://%s:%d/", f.ClientPublicIP, triggerPort)
|
||
client := &http.Client{Timeout: 120 * time.Second}
|
||
resp, err := client.Post(triggerURL, "application/json", bytes.NewReader(body))
|
||
require.NoErrorf(t, err, "trigger %s probe", label)
|
||
resp.Body.Close()
|
||
require.Equalf(t, 200, resp.StatusCode, "trigger %s probe HTTP status", label)
|
||
|
||
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, "udp"),
|
||
1, probesPerRun/2, label)
|
||
}
|
||
|
||
// --- LB CRUD helpers -----------------------------------------------------
|
||
|
||
func createTargetGroup(t *testing.T, c *harness.AWSClient, f *sharedFixture, name, proto string, port int64, hcPath string) string {
|
||
t.Helper()
|
||
in := &elbv2.CreateTargetGroupInput{
|
||
Name: aws.String(name),
|
||
Protocol: aws.String(proto),
|
||
Port: aws.Int64(port),
|
||
VpcId: aws.String(f.VPCID),
|
||
HealthCheckIntervalSeconds: aws.Int64(5),
|
||
HealthyThresholdCount: aws.Int64(2),
|
||
UnhealthyThresholdCount: aws.Int64(2),
|
||
}
|
||
if hcPath != "" {
|
||
in.HealthCheckPath = aws.String(hcPath)
|
||
} else {
|
||
in.HealthCheckProtocol = aws.String("TCP")
|
||
in.HealthCheckIntervalSeconds = aws.Int64(10)
|
||
}
|
||
out, err := c.ELBv2.CreateTargetGroup(in)
|
||
require.NoErrorf(t, err, "create-target-group %s", name)
|
||
arn := aws.StringValue(out.TargetGroups[0].TargetGroupArn)
|
||
t.Logf("TG %s: %s", name, arn)
|
||
return arn
|
||
}
|
||
|
||
// createUDPTargetGroup creates a UDP TG with a TCP health check (UDP can't be probed),
|
||
// exercising the nginx agent's active TCP prober that advances NLB targets to healthy.
|
||
func createUDPTargetGroup(t *testing.T, c *harness.AWSClient, f *sharedFixture, name string) string {
|
||
t.Helper()
|
||
out, err := c.ELBv2.CreateTargetGroup(&elbv2.CreateTargetGroupInput{
|
||
Name: aws.String(name),
|
||
Protocol: aws.String("UDP"),
|
||
Port: aws.Int64(udpPort),
|
||
VpcId: aws.String(f.VPCID),
|
||
HealthCheckProtocol: aws.String("TCP"),
|
||
HealthCheckPort: aws.String(fmt.Sprintf("%d", tcpPort)),
|
||
HealthCheckIntervalSeconds: aws.Int64(10),
|
||
HealthyThresholdCount: aws.Int64(2),
|
||
UnhealthyThresholdCount: aws.Int64(2),
|
||
})
|
||
require.NoErrorf(t, err, "create-target-group %s (UDP)", name)
|
||
arn := aws.StringValue(out.TargetGroups[0].TargetGroupArn)
|
||
t.Logf("UDP TG %s: %s (HC TCP:%d)", name, arn, tcpPort)
|
||
return arn
|
||
}
|
||
|
||
func deleteTargetGroup(t *testing.T, c *harness.AWSClient, arn string) {
|
||
if arn == "" {
|
||
return
|
||
}
|
||
if _, err := c.ELBv2.DeleteTargetGroup(&elbv2.DeleteTargetGroupInput{TargetGroupArn: aws.String(arn)}); err != nil {
|
||
t.Logf("delete TG %s: %v", arn, err)
|
||
}
|
||
}
|
||
|
||
func registerTargets(t *testing.T, c *harness.AWSClient, tgArn string, instanceIDs []string) {
|
||
t.Helper()
|
||
targets := make([]*elbv2.TargetDescription, len(instanceIDs))
|
||
for i, id := range instanceIDs {
|
||
targets[i] = &elbv2.TargetDescription{Id: aws.String(id)}
|
||
}
|
||
_, err := c.ELBv2.RegisterTargets(&elbv2.RegisterTargetsInput{
|
||
TargetGroupArn: aws.String(tgArn),
|
||
Targets: targets,
|
||
})
|
||
require.NoError(t, err, "register-targets")
|
||
}
|
||
|
||
func deregisterTargets(t *testing.T, c *harness.AWSClient, tgArn string, instanceIDs []string) {
|
||
if tgArn == "" || len(instanceIDs) == 0 {
|
||
return
|
||
}
|
||
targets := make([]*elbv2.TargetDescription, len(instanceIDs))
|
||
for i, id := range instanceIDs {
|
||
targets[i] = &elbv2.TargetDescription{Id: aws.String(id)}
|
||
}
|
||
if _, err := c.ELBv2.DeregisterTargets(&elbv2.DeregisterTargetsInput{
|
||
TargetGroupArn: aws.String(tgArn),
|
||
Targets: targets,
|
||
}); err != nil {
|
||
t.Logf("deregister-targets: %v", err)
|
||
}
|
||
}
|
||
|
||
type lbInfo struct {
|
||
ARN, ID, Scheme, Type, DNSName string
|
||
}
|
||
|
||
// createLB builds the CreateLoadBalancer request. sgIDs, when non-empty, is
|
||
// passed as the LB's SecurityGroups — for ALB scenarios this is a dedicated
|
||
// per-scenario SG (see createScenarioSG); nil for NLBs, which get their own
|
||
// managed SG automatically. Nil/empty SecurityGroups on an ALB falls back to
|
||
// the VPC's shared default SG, which is the multi-scenario contention this
|
||
// per-scenario SG threading avoids.
|
||
func createLB(t *testing.T, c *harness.AWSClient, f *sharedFixture, name, lbType, scheme string, sgIDs []string) lbInfo {
|
||
t.Helper()
|
||
in := &elbv2.CreateLoadBalancerInput{
|
||
Name: aws.String(name),
|
||
Subnets: []*string{aws.String(f.SubnetID)},
|
||
Scheme: aws.String(scheme),
|
||
}
|
||
if lbType == "network" {
|
||
in.Type = aws.String("network")
|
||
}
|
||
if len(sgIDs) > 0 {
|
||
in.SecurityGroups = aws.StringSlice(sgIDs)
|
||
}
|
||
out, err := c.ELBv2.CreateLoadBalancer(in)
|
||
require.NoErrorf(t, err, "create-load-balancer %s", name)
|
||
require.NotEmpty(t, out.LoadBalancers)
|
||
lb := out.LoadBalancers[0]
|
||
arn := aws.StringValue(lb.LoadBalancerArn)
|
||
parts := strings.Split(arn, "/")
|
||
info := lbInfo{
|
||
ARN: arn,
|
||
ID: parts[len(parts)-1],
|
||
Scheme: aws.StringValue(lb.Scheme),
|
||
Type: aws.StringValue(lb.Type),
|
||
DNSName: aws.StringValue(lb.DNSName),
|
||
}
|
||
t.Logf("LB %s: %s (scheme=%s type=%s dns=%s)", name, info.ARN, info.Scheme, info.Type, info.DNSName)
|
||
return info
|
||
}
|
||
|
||
// lbCreateAttempts bounds retries when an LB lands in terminal state failed.
|
||
// On single-node dev boxes (serial subtests, one LB torn down before the next
|
||
// is created) this is usually the previous suite's sys.micro VM still being
|
||
// deallocated. On multinode clusters (subtests run concurrently, each owning
|
||
// an independent LB — see TestLoadBalancer) there is no such reclaim wait, so
|
||
// a terminal failure here reflects a genuine transient issue; the same retry
|
||
// serves as a defensive fallback either way.
|
||
const lbCreateAttempts = 3
|
||
|
||
// createActiveLB creates an LB (plus a listener when proto is non-empty) and
|
||
// waits for state=active. A terminal state failed is retried with backoff
|
||
// (see lbCreateAttempts) rather than failing the suite outright. Cleanups for
|
||
// the surviving LB/listener are registered on success.
|
||
func createActiveLB(t *testing.T, c *harness.AWSClient, f *sharedFixture, name, lbType, scheme, proto string, port int64, tgArn, label string, sgIDs []string) (lbInfo, string) {
|
||
t.Helper()
|
||
var lastErr error
|
||
for attempt := 1; attempt <= lbCreateAttempts; attempt++ {
|
||
if attempt > 1 {
|
||
wait := time.Duration(attempt-1) * 15 * time.Second
|
||
t.Logf("%s: waiting %s before retry (serial-mode capacity reclaim, or a transient failure)", label, wait)
|
||
time.Sleep(wait)
|
||
}
|
||
lb := createLB(t, c, f, name, lbType, scheme, sgIDs)
|
||
// Register idempotent teardown up-front so no exit path (fatal, timeout,
|
||
// or success) can leak the LB/listener into the shared VPC. Deletes are
|
||
// idempotent server-side, so a retry's explicit teardown below is safe.
|
||
t.Cleanup(func() { deleteLB(t, c, lb) })
|
||
listener := ""
|
||
if proto != "" {
|
||
listener = createListener(t, c, lb.ARN, proto, port, tgArn)
|
||
t.Cleanup(func() { deleteListener(t, c, listener) })
|
||
}
|
||
lastErr = harness.WaitForLBActiveErr(t, c, lb.ARN, label, 5*time.Minute)
|
||
if lastErr == nil {
|
||
return lb, listener
|
||
}
|
||
if !errors.Is(lastErr, harness.ErrLBTerminalFailed) && !errors.Is(lastErr, harness.ErrLBProvisioningTimeout) {
|
||
t.Fatalf("%s: %v", label, lastErr)
|
||
}
|
||
t.Logf("%s: attempt %d/%d: %v — tearing down and retrying", label, attempt, lbCreateAttempts, lastErr)
|
||
deleteListener(t, c, listener)
|
||
deleteLB(t, c, lb)
|
||
}
|
||
t.Fatalf("%s: LB never reached active after %d attempts: %v", label, lbCreateAttempts, lastErr)
|
||
return lbInfo{}, ""
|
||
}
|
||
|
||
func deleteLB(t *testing.T, c *harness.AWSClient, lb lbInfo) {
|
||
if lb.ARN == "" {
|
||
return
|
||
}
|
||
prefix := "app"
|
||
if lb.Type == "network" {
|
||
prefix = "net"
|
||
}
|
||
parts := strings.Split(lb.ARN, "/")
|
||
require.GreaterOrEqualf(t, len(parts), 3, "deleteLB: malformed LB ARN %q", lb.ARN)
|
||
lbName := parts[len(parts)-2]
|
||
filter := fmt.Sprintf("ELB %s/%s/%s", prefix, lbName, 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 until DescribeLoadBalancers returns LoadBalancerNotFound,
|
||
// indicating the underlying system VM has terminated 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 {
|
||
t.Helper()
|
||
out, err := c.ELBv2.CreateListener(&elbv2.CreateListenerInput{
|
||
LoadBalancerArn: aws.String(lbArn),
|
||
Protocol: aws.String(proto),
|
||
Port: aws.Int64(port),
|
||
DefaultActions: []*elbv2.Action{{
|
||
Type: aws.String("forward"),
|
||
TargetGroupArn: aws.String(tgArn),
|
||
}},
|
||
})
|
||
require.NoError(t, err, "create-listener")
|
||
arn := aws.StringValue(out.Listeners[0].ListenerArn)
|
||
t.Logf("listener %s:%d -> %s", proto, port, arn)
|
||
return arn
|
||
}
|
||
|
||
func deleteListener(t *testing.T, c *harness.AWSClient, arn string) {
|
||
if arn == "" {
|
||
return
|
||
}
|
||
if _, err := c.ELBv2.DeleteListener(&elbv2.DeleteListenerInput{ListenerArn: aws.String(arn)}); err != nil {
|
||
t.Logf("delete listener: %v", err)
|
||
}
|
||
}
|
||
|
||
// modifyListenerPort changes the listener's port in place. Verifies that
|
||
// HAProxy is reconciled without the listener being deleted (no traffic outage).
|
||
func modifyListenerPort(t *testing.T, c *harness.AWSClient, listenerArn string, port int64) {
|
||
t.Helper()
|
||
_, err := c.ELBv2.ModifyListener(&elbv2.ModifyListenerInput{
|
||
ListenerArn: aws.String(listenerArn),
|
||
Port: aws.Int64(port),
|
||
})
|
||
require.NoErrorf(t, err, "modify-listener port=%d", port)
|
||
t.Logf("listener %s: port → %d", listenerArn, port)
|
||
}
|
||
|
||
// modifyListenerDefaultTG swaps the listener's default target group.
|
||
func modifyListenerDefaultTG(t *testing.T, c *harness.AWSClient, listenerArn, tgArn string) {
|
||
t.Helper()
|
||
_, err := c.ELBv2.ModifyListener(&elbv2.ModifyListenerInput{
|
||
ListenerArn: aws.String(listenerArn),
|
||
DefaultActions: []*elbv2.Action{{
|
||
Type: aws.String("forward"),
|
||
TargetGroupArn: aws.String(tgArn),
|
||
}},
|
||
})
|
||
require.NoError(t, err, "modify-listener default actions")
|
||
t.Logf("listener %s: default TG → %s", listenerArn, tgArn)
|
||
}
|
||
|
||
// describeListenerPort fetches the current port of a listener — used to
|
||
// assert that ModifyListener actually persisted before probing.
|
||
func describeListenerPort(t *testing.T, c *harness.AWSClient, listenerArn string) int64 {
|
||
t.Helper()
|
||
out, err := c.ELBv2.DescribeListeners(&elbv2.DescribeListenersInput{
|
||
ListenerArns: []*string{aws.String(listenerArn)},
|
||
})
|
||
require.NoError(t, err, "describe-listeners")
|
||
require.NotEmpty(t, out.Listeners)
|
||
return aws.Int64Value(out.Listeners[0].Port)
|
||
}
|
||
|
||
// runModifyListenerSuite verifies in-place listener edits: port change and DefaultActions swap.
|
||
// Regression gate: the listener must stay alive across edits (not delete+recreate).
|
||
func runModifyListenerSuite(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
|
||
t.Helper()
|
||
const label = "ALB internal ModifyListener"
|
||
const altPort int64 = 8090
|
||
|
||
tgA := createTargetGroup(t, c, f, "lb-e2e-mod-tg-a", "HTTP", httpPort, "/index.html")
|
||
t.Cleanup(func() { deleteTargetGroup(t, c, tgA) })
|
||
|
||
tgB := createTargetGroup(t, c, f, "lb-e2e-mod-tg-b", "HTTP", httpPort, "/index.html")
|
||
t.Cleanup(func() { deleteTargetGroup(t, c, tgB) })
|
||
|
||
registerTargets(t, c, tgA, f.AppInstanceIDs)
|
||
t.Cleanup(func() { deregisterTargets(t, c, tgA, f.AppInstanceIDs) })
|
||
registerTargets(t, c, tgB, f.AppInstanceIDs)
|
||
t.Cleanup(func() { deregisterTargets(t, c, tgB, f.AppInstanceIDs) })
|
||
|
||
// Both the original listener port and the port it's modified to (altPort)
|
||
// must be open on this scenario's own dedicated SG up front.
|
||
sgIDs := []string{createScenarioSG(t, c, f, "lb-e2e-mod-sg", httpPort, altPort)}
|
||
|
||
lb, listener := createActiveLB(t, c, f, "lb-e2e-mod", "application", "internal", "HTTP", httpPort, tgA, label, sgIDs)
|
||
harness.WaitForTargetsHealthy(t, c, tgA, 2, label+" tgA", 2*time.Minute)
|
||
|
||
eni := lbENI(t, c, "app", lb)
|
||
priv := privateIP(eni)
|
||
require.NotEmpty(t, priv, label+" needs private IP")
|
||
|
||
// Baseline: traffic on the original port.
|
||
probeAtPort(t, f, priv, httpPort, label+" before modify (port 80)")
|
||
|
||
// In-place port change.
|
||
modifyListenerPort(t, c, listener, altPort)
|
||
assert.Equal(t, altPort, describeListenerPort(t, c, listener), "listener port persisted")
|
||
// lb-agent picks up the new HAProxy config on its next 5s tick, not
|
||
// synchronously with the ModifyListener API. Poll until the listener
|
||
// actually answers on altPort before driving the assertive probe.
|
||
waitForListenerServing(t, f, priv, altPort, label+" port 8090", 60*time.Second)
|
||
probeAtPort(t, f, priv, altPort, label+" after port-modify (port 8090)")
|
||
|
||
// Defaults: in-place TG swap on same port.
|
||
modifyListenerDefaultTG(t, c, listener, tgB)
|
||
harness.WaitForTargetsHealthy(t, c, tgB, 2, label+" tgB (post-swap)", 2*time.Minute)
|
||
probeAtPort(t, f, priv, altPort, label+" after TG-swap (still port 8090, now tgB)")
|
||
}
|
||
|
||
// runListenerRulesSuite exercises ALB listener rules: CreateRule, ModifyRule, DescribeRules,
|
||
// and DeleteRule. Two TGs with disjoint backends so probes confirm which TG served.
|
||
// Also tests host-header condition to verify a second ACL field reaches HAProxy.
|
||
func runListenerRulesSuite(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
|
||
t.Helper()
|
||
const label = "ALB internal ListenerRules"
|
||
|
||
require.Len(t, f.AppInstanceIDs, 2, label+" needs 2 app instances")
|
||
appA, appB := f.AppInstanceIDs[0], f.AppInstanceIDs[1]
|
||
|
||
tgA := createTargetGroup(t, c, f, "lb-e2e-rul-tg-a", "HTTP", httpPort, "/index.html")
|
||
t.Cleanup(func() { deleteTargetGroup(t, c, tgA) })
|
||
tgB := createTargetGroup(t, c, f, "lb-e2e-rul-tg-b", "HTTP", httpPort, "/alpha/index.html")
|
||
t.Cleanup(func() { deleteTargetGroup(t, c, tgB) })
|
||
|
||
registerTargets(t, c, tgA, []string{appA})
|
||
t.Cleanup(func() { deregisterTargets(t, c, tgA, []string{appA}) })
|
||
registerTargets(t, c, tgB, []string{appB})
|
||
t.Cleanup(func() { deregisterTargets(t, c, tgB, []string{appB}) })
|
||
|
||
sgIDs := []string{createScenarioSG(t, c, f, "lb-e2e-rul-sg", httpPort)}
|
||
|
||
lb, listener := createActiveLB(t, c, f, "lb-e2e-rul", "application", "internal", "HTTP", httpPort, tgA, label, sgIDs)
|
||
harness.WaitForTargetsHealthy(t, c, tgA, 1, label+" tgA", 2*time.Minute)
|
||
|
||
eni := lbENI(t, c, "app", lb)
|
||
priv := privateIP(eni)
|
||
require.NotEmpty(t, priv, label+" needs private IP")
|
||
|
||
waitForPathServing(t, f, priv, httpPort, "/", "", label+" baseline", 60*time.Second)
|
||
base := probeAtPath(t, f, priv, httpPort, "/", "", label+" default -> tgA")
|
||
require.Equal(t, 1, base.Unique(), label+" default expects 1 responder (tgA single backend)")
|
||
appAHost := singleResponder(base)
|
||
|
||
// CreateRule wires tgB to the listener; only then does spinifex begin
|
||
// health-checking tgB. Wait for healthy AFTER rule creation.
|
||
ruleArn := createPathRule(t, c, listener, 10, "/alpha*", tgB)
|
||
ruleCleanup := func() {
|
||
if ruleArn == "" {
|
||
return
|
||
}
|
||
deleteRule(t, c, ruleArn)
|
||
}
|
||
t.Cleanup(ruleCleanup)
|
||
|
||
harness.WaitForTargetsHealthy(t, c, tgB, 1, label+" tgB (post-rule)", 2*time.Minute)
|
||
waitForPathRoutedAway(t, f, priv, httpPort, "/alpha/", appAHost, label+" wait rule active", 60*time.Second)
|
||
ruleResp := probeAtPath(t, f, priv, httpPort, "/alpha/", "", label+" path /alpha/ -> tgB")
|
||
require.Equal(t, 1, ruleResp.Unique(), label+" rule probe expects 1 responder")
|
||
appBHost := singleResponder(ruleResp)
|
||
require.NotEqual(t, appAHost, appBHost, label+" /alpha/ must route to tgB, not default")
|
||
|
||
// Default route unaffected by rule.
|
||
stillDefault := probeAtPath(t, f, priv, httpPort, "/", "", label+" default unchanged")
|
||
require.Equal(t, 1, stillDefault.Unique())
|
||
assert.Equal(t, appAHost, singleResponder(stillDefault), label+" / must still hit tgA")
|
||
|
||
// DescribeRules: rule + default present.
|
||
rules := describeRules(t, c, listener)
|
||
assert.GreaterOrEqual(t, len(rules), 2, label+" expect rule + default")
|
||
assert.True(t, hasRule(rules, ruleArn), label+" CreateRule arn must appear in DescribeRules")
|
||
|
||
// ModifyRule: /alpha* -> /beta*. /alpha now falls back to default;
|
||
// /beta* now reaches tgB (appB).
|
||
modifyPathRule(t, c, ruleArn, "/beta*")
|
||
waitForPathRoutedTo(t, f, priv, httpPort, "/beta/", appBHost, label+" wait modify active", 60*time.Second)
|
||
modified := probeAtPath(t, f, priv, httpPort, "/alpha/", "", label+" /alpha/ falls to default after modify")
|
||
require.Equal(t, 1, modified.Unique())
|
||
assert.Equal(t, appAHost, singleResponder(modified), label+" /alpha/ should hit default after pattern change")
|
||
|
||
// /beta/ now routes to tgB.
|
||
betaResp := probeAtPath(t, f, priv, httpPort, "/beta/", "", label+" /beta/ -> tgB")
|
||
require.Equal(t, 1, betaResp.Unique())
|
||
assert.Equal(t, appBHost, singleResponder(betaResp), label+" /beta/ must hit tgB after modify")
|
||
|
||
// DeleteRule: /beta also falls to default.
|
||
deleteRule(t, c, ruleArn)
|
||
ruleArn = ""
|
||
waitForPathRoutedAway(t, f, priv, httpPort, "/beta/", appBHost, label+" wait delete active", 60*time.Second)
|
||
afterDelete := probeAtPath(t, f, priv, httpPort, "/beta/", "", label+" /beta/ after delete")
|
||
require.Equal(t, 1, afterDelete.Unique())
|
||
assert.Equal(t, appAHost, singleResponder(afterDelete), label+" /beta/ must hit default after delete")
|
||
}
|
||
|
||
// createPathRule adds a path-pattern listener rule. Returns the rule ARN.
|
||
func createPathRule(t *testing.T, c *harness.AWSClient, listenerArn string, priority int64, pattern, tgArn string) string {
|
||
t.Helper()
|
||
out, err := c.ELBv2.CreateRule(&elbv2.CreateRuleInput{
|
||
ListenerArn: aws.String(listenerArn),
|
||
Priority: aws.Int64(priority),
|
||
Conditions: []*elbv2.RuleCondition{{
|
||
Field: aws.String("path-pattern"),
|
||
PathPatternConfig: &elbv2.PathPatternConditionConfig{
|
||
Values: []*string{aws.String(pattern)},
|
||
},
|
||
}},
|
||
Actions: []*elbv2.Action{{
|
||
Type: aws.String("forward"),
|
||
TargetGroupArn: aws.String(tgArn),
|
||
}},
|
||
})
|
||
require.NoErrorf(t, err, "create-rule priority=%d pattern=%s", priority, pattern)
|
||
require.NotEmpty(t, out.Rules)
|
||
arn := aws.StringValue(out.Rules[0].RuleArn)
|
||
t.Logf("rule %s: priority=%d path=%s -> %s", arn, priority, pattern, tgArn)
|
||
return arn
|
||
}
|
||
|
||
func modifyPathRule(t *testing.T, c *harness.AWSClient, ruleArn, pattern string) {
|
||
t.Helper()
|
||
_, err := c.ELBv2.ModifyRule(&elbv2.ModifyRuleInput{
|
||
RuleArn: aws.String(ruleArn),
|
||
Conditions: []*elbv2.RuleCondition{{
|
||
Field: aws.String("path-pattern"),
|
||
PathPatternConfig: &elbv2.PathPatternConditionConfig{
|
||
Values: []*string{aws.String(pattern)},
|
||
},
|
||
}},
|
||
})
|
||
require.NoErrorf(t, err, "modify-rule pattern=%s", pattern)
|
||
t.Logf("rule %s: path -> %s", ruleArn, pattern)
|
||
}
|
||
|
||
func deleteRule(t *testing.T, c *harness.AWSClient, ruleArn string) {
|
||
if ruleArn == "" {
|
||
return
|
||
}
|
||
if _, err := c.ELBv2.DeleteRule(&elbv2.DeleteRuleInput{RuleArn: aws.String(ruleArn)}); err != nil {
|
||
t.Logf("delete rule %s: %v", ruleArn, err)
|
||
}
|
||
}
|
||
|
||
func describeRules(t *testing.T, c *harness.AWSClient, listenerArn string) []*elbv2.Rule {
|
||
t.Helper()
|
||
out, err := c.ELBv2.DescribeRules(&elbv2.DescribeRulesInput{
|
||
ListenerArn: aws.String(listenerArn),
|
||
})
|
||
require.NoError(t, err, "describe-rules")
|
||
return out.Rules
|
||
}
|
||
|
||
func hasRule(rules []*elbv2.Rule, arn string) bool {
|
||
for _, r := range rules {
|
||
if aws.StringValue(r.RuleArn) == arn {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
func singleResponder(r harness.TrafficResult) string {
|
||
for k := range r.Distribution {
|
||
return k
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// probeAtPath drives the shared client to issue probesPerRun probes at lbIP:port with the
|
||
// given path and optional Host header. Asserts at least half succeed and returns the
|
||
// distribution so callers can assert which backend(s) served.
|
||
func probeAtPath(t *testing.T, f *sharedFixture, lbIP string, port int64, path, host, label string) harness.TrafficResult {
|
||
t.Helper()
|
||
r, err := probeOnceAt(f, lbIP, port, path, host, probesPerRun)
|
||
require.NoErrorf(t, err, "trigger probe %s", label)
|
||
for inst, count := range r.Distribution {
|
||
t.Logf(" %s: %s -> %d", label, inst, count)
|
||
}
|
||
t.Logf(" %s: %d/%d successful, %d unique", label, r.Successful, r.Total, r.Unique())
|
||
require.GreaterOrEqualf(t, r.Successful, probesPerRun/2, "%s probes succeeded", label)
|
||
return r
|
||
}
|
||
|
||
// waitForPathServing polls until the path/host combination returns any
|
||
// successful response. Use after CreateRule/ModifyRule/DeleteRule to bridge
|
||
// the lb-agent reconcile window.
|
||
func waitForPathServing(t *testing.T, f *sharedFixture, lbIP string, port int64, path, host, label string, timeout time.Duration) {
|
||
t.Helper()
|
||
harness.EventuallyErr(t, func() error {
|
||
r, err := probeOnceAt(f, lbIP, port, path, host, 2)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if r.Successful == 0 {
|
||
return fmt.Errorf("%s: 0/%d successful", label, r.Total)
|
||
}
|
||
return nil
|
||
}, timeout, 2*time.Second)
|
||
}
|
||
|
||
// waitForPathRoutedAway polls a path until responses stop coming from awayFrom.
|
||
// Use after CreateRule/ModifyRule/DeleteRule to wait out lb-agent reconcile.
|
||
// Requires at least one successful probe to avoid passing on a transient outage.
|
||
func waitForPathRoutedAway(t *testing.T, f *sharedFixture, lbIP string, port int64, path, awayFrom, label string, timeout time.Duration) {
|
||
t.Helper()
|
||
harness.EventuallyErr(t, func() error {
|
||
r, err := probeOnceAt(f, lbIP, port, path, "", 2)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if r.Successful == 0 {
|
||
return fmt.Errorf("%s: 0/%d successful", label, r.Total)
|
||
}
|
||
if r.Distribution[awayFrom] > 0 {
|
||
return fmt.Errorf("%s: still seeing %s", label, awayFrom)
|
||
}
|
||
return nil
|
||
}, timeout, 2*time.Second)
|
||
}
|
||
|
||
// waitForPathRoutedTo polls a path until responses come from wantHost.
|
||
// Use when the caller already knows the expected backend hostname.
|
||
func waitForPathRoutedTo(t *testing.T, f *sharedFixture, lbIP string, port int64, path, wantHost, label string, timeout time.Duration) {
|
||
t.Helper()
|
||
harness.EventuallyErr(t, func() error {
|
||
r, err := probeOnceAt(f, lbIP, port, path, "", 2)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if r.Distribution[wantHost] == 0 {
|
||
return fmt.Errorf("%s: %s not yet observed", label, wantHost)
|
||
}
|
||
return nil
|
||
}, timeout, 2*time.Second)
|
||
}
|
||
|
||
// probeOnceAt is probeOnce with explicit path + Host-header support. Returns
|
||
// errors rather than failing so callers can poll.
|
||
func probeOnceAt(f *sharedFixture, lbIP string, port int64, path, host string, count int) (harness.TrafficResult, error) {
|
||
resultsFile := fmt.Sprintf("rul-%d-%d.txt", port, time.Now().UnixNano())
|
||
order := map[string]any{
|
||
"proto": "http",
|
||
"ip": lbIP,
|
||
"count": count,
|
||
"outfile": resultsFile,
|
||
"http_port": port,
|
||
"http_path": path,
|
||
"host": host,
|
||
"tcp_port": tcpPort,
|
||
}
|
||
body, err := json.Marshal(order)
|
||
if err != nil {
|
||
return harness.TrafficResult{}, 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))
|
||
if err != nil {
|
||
return harness.TrafficResult{}, fmt.Errorf("trigger: %w", err)
|
||
}
|
||
resp.Body.Close()
|
||
if resp.StatusCode != 200 {
|
||
return harness.TrafficResult{}, fmt.Errorf("trigger status %d", resp.StatusCode)
|
||
}
|
||
results, err := plainHTTPGet(
|
||
fmt.Sprintf("http://%s:%d/%s", f.ClientPublicIP, httpPort, resultsFile),
|
||
10*time.Second)
|
||
if err != nil {
|
||
return harness.TrafficResult{}, fmt.Errorf("fetch %s: %w", resultsFile, err)
|
||
}
|
||
return harness.VerifyResultsLines(results, "http"), nil
|
||
}
|
||
|
||
// waitForListenerServing polls the shared client until at least one probe returns
|
||
// a successful instance_id payload. Used after ModifyListener to avoid racing
|
||
// the lb-agent's poll-driven HAProxy reload (5s tick + reload latency).
|
||
func waitForListenerServing(t *testing.T, f *sharedFixture, lbIP string, port int64, label string, timeout time.Duration) {
|
||
t.Helper()
|
||
harness.EventuallyErr(t, func() error {
|
||
r, err := probeOnce(f, lbIP, port, 2)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if r.Successful == 0 {
|
||
return fmt.Errorf("%s: listener not serving on :%d (0/%d successful)", label, port, r.Total)
|
||
}
|
||
return nil
|
||
}, timeout, 2*time.Second)
|
||
}
|
||
|
||
// probeOnce drives the shared client to issue `count` HTTP probes at
|
||
// lbIP:port and returns the parsed TrafficResult. Errors are returned rather
|
||
// than failing the test so callers can poll.
|
||
func probeOnce(f *sharedFixture, lbIP string, port int64, count int) (harness.TrafficResult, error) {
|
||
resultsFile := fmt.Sprintf("mod-%d-%d.txt", port, time.Now().UnixNano())
|
||
order := map[string]any{
|
||
"proto": "http",
|
||
"ip": lbIP,
|
||
"count": count,
|
||
"outfile": resultsFile,
|
||
"http_port": port,
|
||
"tcp_port": tcpPort,
|
||
}
|
||
body, err := json.Marshal(order)
|
||
if err != nil {
|
||
return harness.TrafficResult{}, 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))
|
||
if err != nil {
|
||
return harness.TrafficResult{}, fmt.Errorf("trigger: %w", err)
|
||
}
|
||
resp.Body.Close()
|
||
if resp.StatusCode != 200 {
|
||
return harness.TrafficResult{}, fmt.Errorf("trigger status %d", resp.StatusCode)
|
||
}
|
||
results, err := plainHTTPGet(
|
||
fmt.Sprintf("http://%s:%d/%s", f.ClientPublicIP, httpPort, resultsFile),
|
||
10*time.Second)
|
||
if err != nil {
|
||
return harness.TrafficResult{}, fmt.Errorf("fetch %s: %w", resultsFile, err)
|
||
}
|
||
return harness.VerifyResultsLines(results, "http"), nil
|
||
}
|
||
|
||
// probeAtPort runs round-robin probes from the shared client to lbIP:port and
|
||
// asserts both app instances respond. Mirrors runInternalTrafficViaClient but
|
||
// takes an explicit port so ModifyListener edits can be verified.
|
||
func probeAtPort(t *testing.T, f *sharedFixture, lbIP string, port int64, label string) {
|
||
t.Helper()
|
||
resultsFile := fmt.Sprintf("mod-%d-%d.txt", port, time.Now().UnixNano())
|
||
order := map[string]any{
|
||
"proto": "http",
|
||
"ip": lbIP,
|
||
"count": probesPerRun,
|
||
"outfile": resultsFile,
|
||
"http_port": port,
|
||
"tcp_port": tcpPort,
|
||
}
|
||
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 probe %s", label)
|
||
resp.Body.Close()
|
||
require.Equalf(t, 200, resp.StatusCode, "trigger probe %s status", label)
|
||
|
||
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, "http"),
|
||
1, probesPerRun/2, label)
|
||
}
|
||
|
||
func lbENI(t *testing.T, c *harness.AWSClient, prefix string, lb lbInfo) *ec2.NetworkInterface {
|
||
t.Helper()
|
||
parts := strings.Split(lb.ARN, "/")
|
||
require.GreaterOrEqualf(t, len(parts), 3, "lbENI: malformed LB ARN %q", lb.ARN)
|
||
lbName := parts[len(parts)-2]
|
||
desc := fmt.Sprintf("ELB %s/%s/%s", prefix, lbName, lb.ID)
|
||
var eni *ec2.NetworkInterface
|
||
harness.EventuallyErr(t, func() error {
|
||
out, err := c.EC2.DescribeNetworkInterfaces(&ec2.DescribeNetworkInterfacesInput{
|
||
Filters: []*ec2.Filter{{
|
||
Name: aws.String("description"),
|
||
Values: []*string{aws.String(desc)},
|
||
}},
|
||
})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if len(out.NetworkInterfaces) == 0 {
|
||
return fmt.Errorf("no ENI for %s", desc)
|
||
}
|
||
eni = out.NetworkInterfaces[0]
|
||
return nil
|
||
}, 30*time.Second, 2*time.Second)
|
||
return eni
|
||
}
|
||
|
||
// captureLBConsoleOnFailure registers an on-failure dump of the NLB VM's serial console.
|
||
// nginx activation errors land on the guest console, not the host journal — so an
|
||
// "0/N healthy" timeout is otherwise undiagnosable from CI artifacts.
|
||
func captureLBConsoleOnFailure(t *testing.T, c *harness.AWSClient, eniDescPrefix string, lb lbInfo) {
|
||
t.Helper()
|
||
eni := lbENI(t, c, eniDescPrefix, lb)
|
||
if eni.Attachment == nil {
|
||
return
|
||
}
|
||
instanceID := aws.StringValue(eni.Attachment.InstanceId)
|
||
if instanceID == "" {
|
||
return
|
||
}
|
||
dir := harness.ArtifactDir(t, harness.LoadEnv(t))
|
||
harness.OnFailure(t, func() {
|
||
harness.DumpInstanceConsole(t, c, instanceID, dir, "lb-vm-console.log")
|
||
})
|
||
}
|
||
|
||
func publicIP(eni *ec2.NetworkInterface) string {
|
||
if eni == nil || eni.Association == nil {
|
||
return ""
|
||
}
|
||
return aws.StringValue(eni.Association.PublicIp)
|
||
}
|
||
|
||
func privateIP(eni *ec2.NetworkInterface) string {
|
||
if eni == nil {
|
||
return ""
|
||
}
|
||
return aws.StringValue(eni.PrivateIpAddress)
|
||
}
|
||
|
||
func assertInternalDNS(t *testing.T, c *harness.AWSClient, lbArn, label string) {
|
||
t.Helper()
|
||
out, err := c.ELBv2.DescribeLoadBalancers(&elbv2.DescribeLoadBalancersInput{
|
||
LoadBalancerArns: []*string{aws.String(lbArn)},
|
||
})
|
||
require.NoError(t, err)
|
||
require.NotEmpty(t, out.LoadBalancers)
|
||
dns := aws.StringValue(out.LoadBalancers[0].DNSName)
|
||
assert.True(t, strings.HasPrefix(dns, "internal-"), "%s internal DNS missing internal- prefix: %s", label, dns)
|
||
}
|
||
|
||
// assertLBDNSResolves confirms the SDK-returned DNSName resolves to the LB's
|
||
// frontend IP through the host resolver (the same path an AWS SDK/CLI client
|
||
// uses). The suite requires Northstar, so an empty or legacy name is a failure.
|
||
// Retries because the control-plane writer publishes the record asynchronously
|
||
// (best-effort + reconcile).
|
||
func assertLBDNSResolves(t *testing.T, dnsName, wantIP, label string) {
|
||
t.Helper()
|
||
require.NotEmptyf(t, dnsName, "%s: load balancer returned no DNS name despite required Northstar DNS", label)
|
||
require.Falsef(t, strings.HasSuffix(dnsName, ".spinifex.local"),
|
||
"%s: load balancer returned legacy DNS name %q despite required Northstar DNS", label, dnsName)
|
||
harness.Step(t, "resolve LB DNS %s → %s (northstar path)", dnsName, wantIP)
|
||
deadline := time.Now().Add(90 * time.Second)
|
||
var last []string
|
||
for time.Now().Before(deadline) {
|
||
addrs, err := net.LookupHost(dnsName)
|
||
if err == nil {
|
||
last = addrs
|
||
for _, a := range addrs {
|
||
if a == wantIP {
|
||
return
|
||
}
|
||
}
|
||
}
|
||
time.Sleep(3 * time.Second)
|
||
}
|
||
t.Fatalf("%s: LB DNS %q never resolved to frontend IP %s within 90s (last=%v) — northstar did not serve the A record",
|
||
label, dnsName, wantIP, last)
|
||
}
|
||
|
||
// --- Internet-facing traffic ---------------------------------------------
|
||
|
||
// runInternetFacingTrafficSingle drives one LB (ALB or NLB) over its public
|
||
// IP. Runs probes locally from the test driver, and if a peer node is wired
|
||
// up, also runs the same probes from there to exercise inter-node routing.
|
||
func runInternetFacingTrafficSingle(t *testing.T, kind lbKind, ssh *harness.PeerSSH, peer, ip string) {
|
||
t.Helper()
|
||
if kind == kindALB {
|
||
url := fmt.Sprintf("http://%s:%d", ip, httpPort)
|
||
harness.AssertRoundRobin(t,
|
||
harness.HTTPRoundRobin(url, probesPerRun, 5*time.Second),
|
||
2, probesPerRun/2, "ALB inet (local)")
|
||
if ssh != nil && peer != "" {
|
||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||
defer cancel()
|
||
harness.AssertRoundRobin(t,
|
||
remoteHTTPRoundRobin(t, ctx, ssh, peer, url, probesPerRun),
|
||
2, probesPerRun/2, "ALB inet (remote)")
|
||
}
|
||
return
|
||
}
|
||
harness.AssertRoundRobin(t,
|
||
harness.TCPRoundRobin(ip, tcpPort, probesPerRun, 5*time.Second),
|
||
1, probesPerRun/2, "NLB inet (local)")
|
||
if ssh != nil && peer != "" {
|
||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||
defer cancel()
|
||
harness.AssertRoundRobin(t,
|
||
remoteTCPRoundRobin(t, ctx, ssh, peer, ip, tcpPort, probesPerRun),
|
||
1, probesPerRun/2, "NLB inet (remote)")
|
||
}
|
||
}
|
||
|
||
func remoteHTTPRoundRobin(t *testing.T, ctx context.Context, ssh *harness.PeerSSH, peer, url string, n int) harness.TrafficResult {
|
||
t.Helper()
|
||
var lines []string
|
||
for i := 0; i < n; i++ {
|
||
out, err := ssh.Run(ctx, peer, fmt.Sprintf("curl -s --max-time 5 '%s/'", url))
|
||
if err != nil {
|
||
continue
|
||
}
|
||
lines = append(lines, strings.TrimSpace(string(out)))
|
||
}
|
||
return harness.VerifyResultsLines(strings.Join(lines, "\n"), "http")
|
||
}
|
||
|
||
func remoteTCPRoundRobin(t *testing.T, ctx context.Context, ssh *harness.PeerSSH, peer, host string, port, n int) harness.TrafficResult {
|
||
t.Helper()
|
||
var lines []string
|
||
for i := 0; i < n; i++ {
|
||
out, err := ssh.Run(ctx, peer, fmt.Sprintf("echo '' | nc -w5 '%s' %d", host, port))
|
||
if err != nil {
|
||
continue
|
||
}
|
||
lines = append(lines, strings.TrimSpace(string(out)))
|
||
}
|
||
return harness.VerifyResultsLines(strings.Join(lines, "\n"), "tcp")
|
||
}
|
||
|
||
func runNLBDeregisterDraining(t *testing.T, c *harness.AWSClient, tgArn, targetID string) {
|
||
t.Helper()
|
||
_, err := c.ELBv2.DeregisterTargets(&elbv2.DeregisterTargetsInput{
|
||
TargetGroupArn: aws.String(tgArn),
|
||
Targets: []*elbv2.TargetDescription{{Id: aws.String(targetID)}},
|
||
})
|
||
require.NoError(t, err)
|
||
|
||
time.Sleep(3 * time.Second)
|
||
out, err := c.ELBv2.DescribeTargetHealth(&elbv2.DescribeTargetHealthInput{TargetGroupArn: aws.String(tgArn)})
|
||
require.NoError(t, err)
|
||
|
||
remaining := len(out.TargetHealthDescriptions)
|
||
draining := 0
|
||
for _, th := range out.TargetHealthDescriptions {
|
||
if aws.StringValue(th.TargetHealth.State) == "draining" {
|
||
draining++
|
||
}
|
||
}
|
||
t.Logf("NLB deregister: %d remaining, %d draining", remaining, draining)
|
||
assert.True(t, remaining == 1 || draining >= 1, "NLB deregister: expected 1 remaining or >=1 draining")
|
||
}
|
||
|
||
// --- Shared probe client + per-suite trigger ----------------------------
|
||
|
||
// launchSharedProbeClient launches one client VM whose user-data exposes http.server on :80
|
||
// and a JSON trigger endpoint on :9090. Both internal suites share the same client
|
||
// to avoid the ~60s cold-boot cost of a per-suite client.
|
||
func launchSharedProbeClient(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
|
||
t.Helper()
|
||
out, err := c.EC2.RunInstances(&ec2.RunInstancesInput{
|
||
ImageId: aws.String(f.AMIID),
|
||
InstanceType: aws.String(f.InstanceType),
|
||
KeyName: aws.String(lbKeyName),
|
||
SubnetId: aws.String(f.SubnetID),
|
||
MinCount: aws.Int64(1),
|
||
MaxCount: aws.Int64(1),
|
||
UserData: aws.String(base64Encode(clientUserData)),
|
||
})
|
||
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, 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)
|
||
|
||
// 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 {
|
||
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")
|
||
}
|
||
|
||
// 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())
|
||
}
|
||
|
||
order := map[string]any{
|
||
"proto": proto,
|
||
"ip": lbIP,
|
||
"count": probesPerRun,
|
||
"outfile": resultsFile,
|
||
"http_port": httpPort,
|
||
"tcp_port": tcpPort,
|
||
}
|
||
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
|
||
// over port 80 with no cert, so the harness HTTPSGet helper would refuse.
|
||
func plainHTTPGet(url string, timeout time.Duration) (string, error) {
|
||
client := &http.Client{Timeout: timeout}
|
||
resp, err := client.Get(url)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
defer resp.Body.Close()
|
||
if resp.StatusCode != http.StatusOK {
|
||
return "", fmt.Errorf("GET %s: status %d", url, resp.StatusCode)
|
||
}
|
||
buf := new(bytes.Buffer)
|
||
if _, err := buf.ReadFrom(resp.Body); err != nil {
|
||
return "", err
|
||
}
|
||
return buf.String(), nil
|
||
}
|
||
|
||
func dumpDaemonLogs(t *testing.T, dir, label string) {
|
||
t.Helper()
|
||
harness.DumpCmd(t, dir, fmt.Sprintf("daemon-%s.log", label),
|
||
"journalctl", "-u", "spinifex-daemon", "--no-pager", "-n", "200")
|
||
}
|
||
|
||
// assertLBAgentAuthViaIMDS proves the LB VM authenticated to the gateway with
|
||
// IMDS-served instance-role credentials rather than baked static keys. The LB VM
|
||
// is a system-account instance plugged into a customer-account ENI, so it is
|
||
// invisible to the customer AWS API; the gateway journal on the cluster nodes is
|
||
// the only surface that observes its principal. The awsgw hosts the in-process
|
||
// IMDS STS responder, which logs "AssumeRoleForInstance success" at INFO when the
|
||
// LB VM's IMDS request mints assumed-role (ASIA) creds for the spinifex-lb-agent
|
||
// role — a regression to the three boot-path defects strands the agent on an
|
||
// empty role list, so no recent mint appears and this fails. Keying on that INFO
|
||
// line, not the gateway's per-request SigV4 debug log, keeps the check
|
||
// independent of the awsgw debug level. Skips when no node is SSH-reachable.
|
||
//
|
||
// The window is time-bounded on purpose: console logs accumulate across every VM
|
||
// the node has ever run (including pre-fix failures), so a flat-file scan for the
|
||
// old error would false-fail; the gateway's recent successes do not have that
|
||
// problem.
|
||
func assertLBAgentAuthViaIMDS(t *testing.T) {
|
||
t.Helper()
|
||
env := harness.LoadEnv(t)
|
||
if len(env.NodeIPs) == 0 {
|
||
t.Skip("no node IPs in env; cannot observe the LB VM gateway principal")
|
||
}
|
||
ssh := harness.NewPeerSSH()
|
||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||
defer cancel()
|
||
|
||
sawAssumedRole, anyReachable := false, false
|
||
for _, ip := range env.NodeIPs {
|
||
gw, err := ssh.Run(ctx, ip,
|
||
"sudo journalctl -u spinifex-awsgw --no-pager --since '5 minutes ago' 2>/dev/null | "+
|
||
`grep -F 'AssumeRoleForInstance success' | grep -F 'role/spinifex-lb-agent' | grep -F '"akid":"ASIA' | tail -1`)
|
||
if err != nil {
|
||
continue // node unreachable; another node carries the heartbeat
|
||
}
|
||
anyReachable = true
|
||
if strings.TrimSpace(string(gw)) != "" {
|
||
sawAssumedRole = true
|
||
}
|
||
}
|
||
if !anyReachable {
|
||
t.Skip("no node SSH-reachable; cannot observe the LB VM gateway principal")
|
||
}
|
||
require.True(t, sawAssumedRole,
|
||
"no assumed-role (ASIA) gateway auth on any node — LB agent did not authenticate via IMDS creds")
|
||
}
|
||
|
||
func base64Encode(s string) string {
|
||
return base64.StdEncoding.EncodeToString([]byte(s))
|
||
}
|