mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-20 11:21:49 +00:00
test(integration): cover RunInstances placement-group strategy routing
Adds a placement-group daemon-lite wiring a real PlacementGroupServiceImpl over NATS, then proves spread strategy assigns exactly one instance per node while cluster strategy pins the whole launch to the single highest-capacity node.
This commit is contained in:
joshsiv-mulga
@@ -0,0 +1,47 @@
|
||||
//go:build integration
|
||||
|
||||
package integration
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/config"
|
||||
handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// StartPlacementGroupDaemonLite subscribes a real
|
||||
// handlers_ec2_placementgroup.PlacementGroupServiceImpl — the same production
|
||||
// code a live daemon runs — to every ec2.*PlacementGroup*/ec2.Reserve*/
|
||||
// ec2.Finalize*/ec2.Release* subject the gateway's NATSPlacementGroupService
|
||||
// client calls (gateway/ec2/placementgroup/*.go and RunInstances.go's spread/
|
||||
// cluster routing in gateway/ec2/instance/placement.go). Unlike a StubSubject
|
||||
// canned reply, this exercises the real KV-backed CAS reservation logic, so a
|
||||
// test that checks placement-group strategy routing actually proves the
|
||||
// gateway drove a genuine reserve/finalize round trip rather than merely
|
||||
// reaching some daemon-shaped subject.
|
||||
//
|
||||
// Kept separate from StartDaemonLite (matching StartLaunchTemplateDaemonLite's
|
||||
// precedent): only a test that actually exercises placement groups should pay
|
||||
// for wiring this service.
|
||||
func StartPlacementGroupDaemonLite(t *testing.T, gw *Gateway) *handlers_ec2_placementgroup.PlacementGroupServiceImpl {
|
||||
t.Helper()
|
||||
|
||||
cfg := &config.Config{AZ: testAZ}
|
||||
svc, err := handlers_ec2_placementgroup.NewPlacementGroupServiceImplWithNATS(cfg, gw.NATSConn)
|
||||
require.NoError(t, err, "construct placement group service")
|
||||
|
||||
nc := gw.NATSConn
|
||||
sub(t, nc, "ec2.CreatePlacementGroup", func(m *nats.Msg) { dispatch(m, svc.CreatePlacementGroup) })
|
||||
sub(t, nc, "ec2.DeletePlacementGroup", func(m *nats.Msg) { dispatch(m, svc.DeletePlacementGroup) })
|
||||
sub(t, nc, "ec2.DescribePlacementGroups", func(m *nats.Msg) { dispatch(m, svc.DescribePlacementGroups) })
|
||||
sub(t, nc, "ec2.ReserveSpreadNodes", func(m *nats.Msg) { dispatch(m, svc.ReserveSpreadNodes) })
|
||||
sub(t, nc, "ec2.FinalizeSpreadInstances", func(m *nats.Msg) { dispatch(m, svc.FinalizeSpreadInstances) })
|
||||
sub(t, nc, "ec2.ReleaseSpreadNodes", func(m *nats.Msg) { dispatch(m, svc.ReleaseSpreadNodes) })
|
||||
sub(t, nc, "ec2.RemoveInstanceFromPlacementGroup", func(m *nats.Msg) { dispatch(m, svc.RemoveInstance) })
|
||||
sub(t, nc, "ec2.ReserveClusterNode", func(m *nats.Msg) { dispatch(m, svc.ReserveClusterNode) })
|
||||
sub(t, nc, "ec2.FinalizeClusterInstances", func(m *nats.Msg) { dispatch(m, svc.FinalizeClusterInstances) })
|
||||
|
||||
return svc
|
||||
}
|
||||
@@ -0,0 +1,145 @@
|
||||
//go:build integration
|
||||
|
||||
package integration
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/service/ec2"
|
||||
"github.com/mulgadc/spinifex/spinifex/gateway"
|
||||
"github.com/mulgadc/spinifex/spinifex/types"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// stubNodeCapacity registers one of several simultaneous spinifex.node.status
|
||||
// responders (NATS core fan-out: every non-queue subscriber on the subject
|
||||
// independently receives and replies to the same request), so a test can
|
||||
// simulate a multi-node cluster where each node reports distinct capacity.
|
||||
// StubSubject cannot do this: a second call on the same subject would add a
|
||||
// duplicate responder replying with the same single payload, not a second
|
||||
// node's distinct status.
|
||||
func stubNodeCapacity(t *testing.T, gw *Gateway, node, instanceType string, available int) {
|
||||
t.Helper()
|
||||
sub, err := gw.NATSConn.Subscribe("spinifex.node.status", func(msg *nats.Msg) {
|
||||
if msg.Reply == "" {
|
||||
return
|
||||
}
|
||||
_ = msg.Respond(mustMarshal(t, &types.NodeStatusResponse{
|
||||
Node: node,
|
||||
InstanceTypes: []types.InstanceTypeCap{{Name: instanceType, Available: available}},
|
||||
}))
|
||||
})
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = sub.Unsubscribe() })
|
||||
}
|
||||
|
||||
// withExpectedNodes overrides StartGateway's default ExpectedNodes: 1 pin,
|
||||
// which utils.Gather uses to stop fanning out once that many nodes reply
|
||||
// rather than always waiting the full 3s timeout.
|
||||
func withExpectedNodes(n int) Option {
|
||||
return func(cfg *gateway.GatewayConfig) { cfg.ExpectedNodes = n }
|
||||
}
|
||||
|
||||
// TestRunInstances_PlacementGroupSpreadRouting proves a spread-strategy group
|
||||
// (RunInstances.go's placement-group switch -> distributeInstancesSpread in
|
||||
// placement.go) genuinely reserves and dispatches to distinct nodes via the
|
||||
// real PlacementGroupServiceImpl, rather than falling through to the default
|
||||
// distributeInstances path: two equal-capacity nodes, MinCount=MaxCount=2,
|
||||
// must each receive exactly one instance.
|
||||
func TestRunInstances_PlacementGroupSpreadRouting(t *testing.T) {
|
||||
gw := StartGateway(t, withExpectedNodes(2))
|
||||
StartPlacementGroupDaemonLite(t, gw)
|
||||
|
||||
const (
|
||||
instanceType = "t3.micro"
|
||||
nodeA = "spread-node-a"
|
||||
nodeB = "spread-node-b"
|
||||
groupName = "pg-spread-routing"
|
||||
)
|
||||
|
||||
stubNodeCapacity(t, gw, nodeA, instanceType, 5)
|
||||
stubNodeCapacity(t, gw, nodeB, instanceType, 5)
|
||||
chA := captureLaunchTemplateNodeInput(t, gw, instanceType, nodeA)
|
||||
chB := captureLaunchTemplateNodeInput(t, gw, instanceType, nodeB)
|
||||
|
||||
_, err := gw.EC2Client(t).CreatePlacementGroup(&ec2.CreatePlacementGroupInput{
|
||||
GroupName: aws.String(groupName),
|
||||
Strategy: aws.String(ec2.PlacementStrategySpread),
|
||||
})
|
||||
require.NoError(t, err, "create-placement-group (spread)")
|
||||
|
||||
out, err := gw.EC2Client(t).RunInstances(&ec2.RunInstancesInput{
|
||||
ImageId: aws.String("ami-0123456789abcdef0"),
|
||||
InstanceType: aws.String(instanceType),
|
||||
MinCount: aws.Int64(2),
|
||||
MaxCount: aws.Int64(2),
|
||||
Placement: &ec2.Placement{GroupName: aws.String(groupName)},
|
||||
})
|
||||
require.NoError(t, err, "run-instances into spread group")
|
||||
require.Len(t, out.Instances, 2)
|
||||
|
||||
gotA := awaitLaunchTemplateNodeInput(t, chA)
|
||||
gotB := awaitLaunchTemplateNodeInput(t, chB)
|
||||
require.Equal(t, int64(1), aws.Int64Value(gotA.MinCount), "spread must assign exactly 1 instance to node A")
|
||||
require.Equal(t, int64(1), aws.Int64Value(gotA.MaxCount))
|
||||
require.Equal(t, int64(1), aws.Int64Value(gotB.MinCount), "spread must assign exactly 1 instance to node B")
|
||||
require.Equal(t, int64(1), aws.Int64Value(gotB.MaxCount))
|
||||
}
|
||||
|
||||
// TestRunInstances_PlacementGroupClusterRouting proves a cluster-strategy
|
||||
// group (distributeInstancesCluster) pins the entire launch to a single node
|
||||
// rather than spreading it: given two nodes of unequal capacity, all
|
||||
// requested instances must land on the higher-capacity node (ReserveClusterNode
|
||||
// picks EligibleNodes[0], which queryNodeCapacity sorts by capacity desc), and
|
||||
// the low-capacity node must receive zero dispatches.
|
||||
func TestRunInstances_PlacementGroupClusterRouting(t *testing.T) {
|
||||
gw := StartGateway(t, withExpectedNodes(2))
|
||||
StartPlacementGroupDaemonLite(t, gw)
|
||||
|
||||
const (
|
||||
instanceType = "t3.micro"
|
||||
nodeHigh = "cluster-node-high"
|
||||
nodeLow = "cluster-node-low"
|
||||
groupName = "pg-cluster-routing"
|
||||
)
|
||||
|
||||
stubNodeCapacity(t, gw, nodeHigh, instanceType, 5)
|
||||
stubNodeCapacity(t, gw, nodeLow, instanceType, 1)
|
||||
chHigh := captureLaunchTemplateNodeInput(t, gw, instanceType, nodeHigh)
|
||||
|
||||
// No responder is stubbed on nodeLow's own launch subject via
|
||||
// captureLaunchTemplateNodeInput; instead a raw counting subscriber proves
|
||||
// it never fires, rather than merely absence-of-error.
|
||||
var lowDispatchCount int
|
||||
lowSub, err := gw.NATSConn.Subscribe("ec2.RunInstances."+instanceType+"."+nodeLow, func(msg *nats.Msg) {
|
||||
lowDispatchCount++
|
||||
})
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = lowSub.Unsubscribe() })
|
||||
|
||||
_, err = gw.EC2Client(t).CreatePlacementGroup(&ec2.CreatePlacementGroupInput{
|
||||
GroupName: aws.String(groupName),
|
||||
Strategy: aws.String(ec2.PlacementStrategyCluster),
|
||||
})
|
||||
require.NoError(t, err, "create-placement-group (cluster)")
|
||||
|
||||
out, err := gw.EC2Client(t).RunInstances(&ec2.RunInstancesInput{
|
||||
ImageId: aws.String("ami-0123456789abcdef0"),
|
||||
InstanceType: aws.String(instanceType),
|
||||
MinCount: aws.Int64(1),
|
||||
MaxCount: aws.Int64(3),
|
||||
Placement: &ec2.Placement{GroupName: aws.String(groupName)},
|
||||
})
|
||||
require.NoError(t, err, "run-instances into cluster group")
|
||||
// captureLaunchTemplateNodeInput's mock always replies with 1 instance
|
||||
// regardless of the dispatched Assigned count; the count actually assigned
|
||||
// is asserted below via the captured RunInstancesInput's MinCount/MaxCount.
|
||||
require.Len(t, out.Instances, 1)
|
||||
|
||||
gotHigh := awaitLaunchTemplateNodeInput(t, chHigh)
|
||||
require.Equal(t, int64(3), aws.Int64Value(gotHigh.MinCount), "cluster must pin all 3 instances to the high-capacity node")
|
||||
require.Equal(t, int64(3), aws.Int64Value(gotHigh.MaxCount))
|
||||
require.Equal(t, 0, lowDispatchCount, "cluster must never dispatch to the low-capacity node")
|
||||
}
|
||||
Reference in New Issue
Block a user