diff --git a/spinifex/daemon/heartbeat_test.go b/spinifex/daemon/heartbeat_test.go index 3d8ef7c53..a5e8d271e 100644 --- a/spinifex/daemon/heartbeat_test.go +++ b/spinifex/daemon/heartbeat_test.go @@ -5,6 +5,7 @@ import ( "testing" "github.com/mulgadc/spinifex/spinifex/config" + "github.com/mulgadc/spinifex/spinifex/instancecache" "github.com/mulgadc/spinifex/spinifex/vm" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" @@ -147,3 +148,14 @@ func TestHeartbeatKVContract(t *testing.T) { require.NotNil(t, loaded) assert.Equal(t, h, loaded) } + +// The gateway judges node staleness as a multiple of this interval, and holds +// its own copy because it cannot import this package. Lengthening the publish +// cadence without moving that copy would flap every instance to impaired. +func TestHeartbeatIntervalMatchesLivenessCopy(t *testing.T) { + if heartbeatInterval != instancecache.HeartbeatInterval { + t.Fatalf("heartbeat publish interval %v != instancecache.HeartbeatInterval %v; "+ + "the gateway's staleness threshold is derived from its copy, so the two must move together", + heartbeatInterval, instancecache.HeartbeatInterval) + } +} diff --git a/spinifex/gateway/ec2.go b/spinifex/gateway/ec2.go index 46f38b36f..ada7c0d3c 100644 --- a/spinifex/gateway/ec2.go +++ b/spinifex/gateway/ec2.go @@ -210,7 +210,7 @@ var ec2Actions = map[string]ec2Action{ return gateway_ec2_instance.DescribeInstanceTypes(ctx, input, gw.NATSConn, gw.ExpectedNodes, accountID) }), "DescribeInstanceStatus": ec2Handler(func(ctx context.Context, input *ec2.DescribeInstanceStatusInput, gw *GatewayConfig, accountID string) (any, error) { - return gateway_ec2_instance.DescribeInstanceStatus(ctx, input, gw.NATSConn, gw.DiscoverActiveNodes(ctx), accountID, gw.AZ) + return gateway_ec2_instance.DescribeInstanceStatus(ctx, input, gw.NATSConn, gw.DiscoverActiveNodes(ctx), accountID, gw.AZ, gw.InstanceStatus) }), "GetConsoleOutput": ec2Handler(func(ctx context.Context, input *ec2.GetConsoleOutputInput, gw *GatewayConfig, accountID string) (any, error) { return gateway_ec2_instance.GetConsoleOutput(ctx, input, gw.NATSConn, accountID) diff --git a/spinifex/gateway/ec2/instance/DescribeInstanceStatus.go b/spinifex/gateway/ec2/instance/DescribeInstanceStatus.go index 382e4eb0e..325d6656b 100644 --- a/spinifex/gateway/ec2/instance/DescribeInstanceStatus.go +++ b/spinifex/gateway/ec2/instance/DescribeInstanceStatus.go @@ -24,7 +24,7 @@ const ( // also augments from the stopped-instance KV bucket. The aggregator propagates // the first deterministic 4xx only when no data was collected, matching // DescribeInstances. -func DescribeInstanceStatus(ctx context.Context, input *ec2.DescribeInstanceStatusInput, natsConn *nats.Conn, expectedNodes int, accountID, az string) (*ec2.DescribeInstanceStatusOutput, error) { +func DescribeInstanceStatus(ctx context.Context, input *ec2.DescribeInstanceStatusInput, natsConn *nats.Conn, expectedNodes int, accountID, az string, synthesis StatusSynthesis) (*ec2.DescribeInstanceStatusOutput, error) { if input == nil { input = &ec2.DescribeInstanceStatusInput{} } @@ -36,7 +36,7 @@ func DescribeInstanceStatus(ctx context.Context, input *ec2.DescribeInstanceStat } frames, sum, err := utils.Gather(ctx, natsConn, "ec2.DescribeInstanceStatus", jsonData, - utils.GatherOpts{Timeout: 3 * time.Second, ExpectedNodes: expectedNodes, AccountID: accountID}) + utils.GatherOpts{Timeout: 3 * time.Second, ExpectedResponders: expectedNodes, AccountID: accountID}) if err != nil { return nil, err } @@ -55,6 +55,15 @@ func DescribeInstanceStatus(ctx context.Context, input *ec2.DescribeInstanceStat } } + // A frame an answering node returned must win: it carries that host's own + // health, which no cached record can reconstruct. The covered set is what + // guarantees that; appending last is defence in depth, since dedupStatuses + // is first-writer-wins and would keep the live frame regardless. + responders := fanoutResponders{SuccessResponders: sum.SuccessResponders, Unidentified: sum.Unidentified} + if synth := synthesis.synthesize(ctx, input, accountID, coveredInstanceIDs(allStatuses), responders); len(synth) > 0 { + allStatuses = append(allStatuses, synth...) + } + finalStatuses := dedupStatuses(allStatuses) if sum.FirstClient4xx != "" && len(finalStatuses) == 0 { diff --git a/spinifex/gateway/ec2/instance/DescribeInstanceStatus_test.go b/spinifex/gateway/ec2/instance/DescribeInstanceStatus_test.go index 6145f2afd..721f793cd 100644 --- a/spinifex/gateway/ec2/instance/DescribeInstanceStatus_test.go +++ b/spinifex/gateway/ec2/instance/DescribeInstanceStatus_test.go @@ -45,17 +45,12 @@ func TestDescribeInstanceStatus_SingleNode(t *testing.T) { t.Parallel() _, nc := startTestNATSServer(t) - _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { - respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{ - InstanceStatuses: []*ec2.InstanceStatus{ - runningStatus("i-001", "us-east-1a"), - runningStatus("i-002", "us-east-1a"), - }, - }) - }) - require.NoError(t, err) + answerWith(t, nc, "node-a", + runningStatus("i-001", "us-east-1a"), + runningStatus("i-002", "us-east-1a"), + ) - out, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 1, "123456789012", "us-east-1a") + out, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 1, "123456789012", "us-east-1a", StatusSynthesis{}) require.NoError(t, err) require.Len(t, out.InstanceStatuses, 2) } @@ -65,29 +60,19 @@ func TestDescribeInstanceStatus_TwoNodesDedup(t *testing.T) { _, nc := startTestNATSServer(t) // First node returns i-001 (running) - _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { - respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{ - InstanceStatuses: []*ec2.InstanceStatus{runningStatus("i-001", "us-east-1a")}, - }) - }) - require.NoError(t, err) + answerWith(t, nc, "node-a", runningStatus("i-001", "us-east-1a")) // Second node also reports i-001 — should dedup nc2, err := nats.Connect(nc.ConnectedUrl()) require.NoError(t, err) defer nc2.Close() - _, err = nc2.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { - respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{ - InstanceStatuses: []*ec2.InstanceStatus{runningStatus("i-001", "us-east-1a")}, - }) - }) - require.NoError(t, err) + answerWith(t, nc2, "node-b", runningStatus("i-001", "us-east-1a")) require.NoError(t, nc.Flush()) require.NoError(t, nc2.Flush()) - out, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 2, "123456789012", "us-east-1a") + out, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 2, "123456789012", "us-east-1a", StatusSynthesis{}) require.NoError(t, err) require.Len(t, out.InstanceStatuses, 1) assert.Equal(t, "i-001", *out.InstanceStatuses[0].InstanceId) @@ -98,26 +83,18 @@ func TestDescribeInstanceStatus_OneNodeErrorOthersData(t *testing.T) { _, nc := startTestNATSServer(t) // Node 1: data - _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { - respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{ - InstanceStatuses: []*ec2.InstanceStatus{runningStatus("i-good", "az-a")}, - }) - }) - require.NoError(t, err) + answerWith(t, nc, "node-a", runningStatus("i-good", "az-a")) // Node 2: error nc2, err := nats.Connect(nc.ConnectedUrl()) require.NoError(t, err) defer nc2.Close() - _, err = nc2.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { - require.NoError(t, msg.Respond(utils.GenerateErrorPayload("InvalidParameterValue"))) - }) - require.NoError(t, err) + subscribeAsNode(t, nc2, "ec2.DescribeInstanceStatus", "node-b", utils.GenerateErrorPayload("InvalidParameterValue")) require.NoError(t, nc.Flush()) require.NoError(t, nc2.Flush()) - out, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 2, "123456789012", "az-a") + out, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 2, "123456789012", "az-a", StatusSynthesis{}) require.NoError(t, err) require.Len(t, out.InstanceStatuses, 1) assert.Equal(t, "i-good", *out.InstanceStatuses[0].InstanceId) @@ -127,12 +104,9 @@ func TestDescribeInstanceStatus_AllNodesError(t *testing.T) { t.Parallel() _, nc := startTestNATSServer(t) - _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { - require.NoError(t, msg.Respond(utils.GenerateErrorPayload("InvalidParameterValue"))) - }) - require.NoError(t, err) + subscribeAsNode(t, nc, "ec2.DescribeInstanceStatus", "node-a", utils.GenerateErrorPayload("InvalidParameterValue")) - _, err = DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 1, "123456789012", "az-a") + _, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 1, "123456789012", "az-a", StatusSynthesis{}) require.Error(t, err) assert.Equal(t, "InvalidParameterValue", err.Error()) } @@ -141,7 +115,7 @@ func TestDescribeInstanceStatus_AllNodesTimeoutReturnsEmpty(t *testing.T) { t.Parallel() _, nc := startTestNATSServer(t) - out, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 0, "123456789012", "az-a") + out, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 0, "123456789012", "az-a", StatusSynthesis{}) require.NoError(t, err) require.NotNil(t, out) assert.Empty(t, out.InstanceStatuses) @@ -152,15 +126,10 @@ func TestDescribeInstanceStatus_IncludeAllAddsStoppedAsNotApplicable(t *testing. _, nc := startTestNATSServer(t) // Fan-out responder: a running instance - _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { - respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{ - InstanceStatuses: []*ec2.InstanceStatus{runningStatus("i-running", "az-a")}, - }) - }) - require.NoError(t, err) + answerWith(t, nc, "node-a", runningStatus("i-running", "az-a")) // Stopped KV responder - _, err = nc.QueueSubscribe("ec2.DescribeStoppedInstances", "spinifex-workers", func(msg *nats.Msg) { + _, err := nc.QueueSubscribe("ec2.DescribeStoppedInstances", "spinifex-workers", func(msg *nats.Msg) { respondJSON(t, msg, &ec2.DescribeInstancesOutput{ Reservations: []*ec2.Reservation{{ ReservationId: aws.String("r-stop"), @@ -174,7 +143,7 @@ func TestDescribeInstanceStatus_IncludeAllAddsStoppedAsNotApplicable(t *testing. require.NoError(t, err) input := &ec2.DescribeInstanceStatusInput{IncludeAllInstances: aws.Bool(true)} - out, err := DescribeInstanceStatus(context.Background(), input, nc, 1, "123456789012", "az-a") + out, err := DescribeInstanceStatus(context.Background(), input, nc, 1, "123456789012", "az-a", StatusSynthesis{}) require.NoError(t, err) require.Len(t, out.InstanceStatuses, 2) @@ -202,15 +171,10 @@ func TestDescribeInstanceStatus_RunningWinsOverStoppedDuringRace(t *testing.T) { _, nc := startTestNATSServer(t) // Fan-out: i-001 is running - _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { - respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{ - InstanceStatuses: []*ec2.InstanceStatus{runningStatus("i-001", "az-a")}, - }) - }) - require.NoError(t, err) + answerWith(t, nc, "node-a", runningStatus("i-001", "az-a")) // Stopped KV also has i-001 (race: stop transition hasn't fully cleared the KV) - _, err = nc.QueueSubscribe("ec2.DescribeStoppedInstances", "spinifex-workers", func(msg *nats.Msg) { + _, err := nc.QueueSubscribe("ec2.DescribeStoppedInstances", "spinifex-workers", func(msg *nats.Msg) { respondJSON(t, msg, &ec2.DescribeInstancesOutput{ Reservations: []*ec2.Reservation{{ ReservationId: aws.String("r-1"), @@ -224,7 +188,7 @@ func TestDescribeInstanceStatus_RunningWinsOverStoppedDuringRace(t *testing.T) { require.NoError(t, err) input := &ec2.DescribeInstanceStatusInput{IncludeAllInstances: aws.Bool(true)} - out, err := DescribeInstanceStatus(context.Background(), input, nc, 1, "123456789012", "az-a") + out, err := DescribeInstanceStatus(context.Background(), input, nc, 1, "123456789012", "az-a", StatusSynthesis{}) require.NoError(t, err) require.Len(t, out.InstanceStatuses, 1) assert.Equal(t, "running", *out.InstanceStatuses[0].InstanceState.Name) @@ -235,12 +199,9 @@ func TestDescribeInstanceStatus_NilInput(t *testing.T) { t.Parallel() _, nc := startTestNATSServer(t) - _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { - respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{}) - }) - require.NoError(t, err) + answerWith(t, nc, "node-a") - out, err := DescribeInstanceStatus(context.Background(), nil, nc, 1, "123456789012", "az-a") + out, err := DescribeInstanceStatus(context.Background(), nil, nc, 1, "123456789012", "az-a", StatusSynthesis{}) require.NoError(t, err) require.NotNil(t, out) assert.Empty(t, out.InstanceStatuses) @@ -254,7 +215,7 @@ func TestDescribeInstanceStatus_ClosedConnection(t *testing.T) { require.NoError(t, err) closed.Close() - _, err = DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, closed, 1, "123456789012", "az-a") + _, err = DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, closed, 1, "123456789012", "az-a", StatusSynthesis{}) require.Error(t, err) } @@ -289,14 +250,11 @@ func TestDescribeInstanceStatus_IncludeAllProjectsInputForStoppedHandler(t *test t.Parallel() _, nc := startTestNATSServer(t) - _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { - respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{}) - }) - require.NoError(t, err) + answerWith(t, nc, "node-a") var capturedReq ec2.DescribeInstancesInput var unmarshalErr error - _, err = nc.QueueSubscribe("ec2.DescribeStoppedInstances", "spinifex-workers", func(msg *nats.Msg) { + _, err := nc.QueueSubscribe("ec2.DescribeStoppedInstances", "spinifex-workers", func(msg *nats.Msg) { dec := json.NewDecoder(bytes.NewReader(msg.Data)) dec.DisallowUnknownFields() unmarshalErr = dec.Decode(&capturedReq) @@ -315,7 +273,7 @@ func TestDescribeInstanceStatus_IncludeAllProjectsInputForStoppedHandler(t *test IncludeAllInstances: aws.Bool(true), InstanceIds: []*string{aws.String("i-stop")}, } - out, err := DescribeInstanceStatus(context.Background(), input, nc, 1, "123456789012", "az-a") + out, err := DescribeInstanceStatus(context.Background(), input, nc, 1, "123456789012", "az-a", StatusSynthesis{}) require.NoError(t, err) require.NoError(t, unmarshalErr, "projected input must decode cleanly under DisallowUnknownFields") require.Len(t, out.InstanceStatuses, 1) @@ -422,12 +380,9 @@ func TestDescribeInstanceStatus_IncludeAllWithAZFilterMatches(t *testing.T) { t.Parallel() _, nc := startTestNATSServer(t) - _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { - respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{}) - }) - require.NoError(t, err) + answerWith(t, nc, "node-a") - _, err = nc.QueueSubscribe("ec2.DescribeStoppedInstances", "spinifex-workers", func(msg *nats.Msg) { + _, err := nc.QueueSubscribe("ec2.DescribeStoppedInstances", "spinifex-workers", func(msg *nats.Msg) { respondJSON(t, msg, &ec2.DescribeInstancesOutput{ Reservations: []*ec2.Reservation{{Instances: []*ec2.Instance{{ InstanceId: aws.String("i-stop"), @@ -443,7 +398,7 @@ func TestDescribeInstanceStatus_IncludeAllWithAZFilterMatches(t *testing.T) { {Name: aws.String("availability-zone"), Values: []*string{aws.String("az-a")}}, }, } - out, err := DescribeInstanceStatus(context.Background(), input, nc, 1, "123456789012", "az-a") + out, err := DescribeInstanceStatus(context.Background(), input, nc, 1, "123456789012", "az-a", StatusSynthesis{}) require.NoError(t, err) require.Len(t, out.InstanceStatuses, 1) assert.Equal(t, "i-stop", *out.InstanceStatuses[0].InstanceId) @@ -453,12 +408,9 @@ func TestDescribeInstanceStatus_IncludeAllWithAZFilterMisses(t *testing.T) { t.Parallel() _, nc := startTestNATSServer(t) - _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { - respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{}) - }) - require.NoError(t, err) + answerWith(t, nc, "node-a") - _, err = nc.QueueSubscribe("ec2.DescribeStoppedInstances", "spinifex-workers", func(msg *nats.Msg) { + _, err := nc.QueueSubscribe("ec2.DescribeStoppedInstances", "spinifex-workers", func(msg *nats.Msg) { respondJSON(t, msg, &ec2.DescribeInstancesOutput{ Reservations: []*ec2.Reservation{{Instances: []*ec2.Instance{{ InstanceId: aws.String("i-stop"), @@ -474,7 +426,7 @@ func TestDescribeInstanceStatus_IncludeAllWithAZFilterMisses(t *testing.T) { {Name: aws.String("availability-zone"), Values: []*string{aws.String("az-other")}}, }, } - out, err := DescribeInstanceStatus(context.Background(), input, nc, 1, "123456789012", "az-a") + out, err := DescribeInstanceStatus(context.Background(), input, nc, 1, "123456789012", "az-a", StatusSynthesis{}) require.NoError(t, err) assert.Empty(t, out.InstanceStatuses) } diff --git a/spinifex/gateway/ec2/instance/status_synthesis.go b/spinifex/gateway/ec2/instance/status_synthesis.go new file mode 100644 index 000000000..c19e5a2d3 --- /dev/null +++ b/spinifex/gateway/ec2/instance/status_synthesis.go @@ -0,0 +1,147 @@ +package gateway_ec2_instance + +import ( + "context" + "log/slog" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance" + "github.com/mulgadc/spinifex/spinifex/instancecache" + "github.com/mulgadc/spinifex/spinifex/vm" +) + +const ( + // insufficientData is AWS's answer for a status the control plane cannot + // determine, as distinct from impaired, which asserts something is wrong. + insufficientData = "insufficient-data" + statusImpaired = "impaired" +) + +// recordLister supplies the cached instance records synthesis reads. The +// bool reports cache readiness; a cold cache must not be treated as empty. +type recordLister interface { + List(ctx context.Context, accountID string) ([]*vm.VM, bool) +} + +// nodeLiveness reports whether the node that last ran an instance is still +// heartbeating. +type nodeLiveness interface { + State(ctx context.Context, node string) instancecache.NodeState +} + +// StatusSynthesis backfills status frames for instances no node answered for. +// The zero value disables it, so a gateway wired without a warm cache answers +// exactly as it did before. +type StatusSynthesis struct { + Records recordLister + Liveness nodeLiveness +} + +// fanoutResponders carries the DescribeInstanceStatus fan-out's own responder +// identity, so synthesis can tell a node that answered and omitted an +// instance from a node that never answered at all. +type fanoutResponders struct { + // SuccessResponders lists nodes that answered with a decodable + // non-error frame, i.e. actually ran their own selection logic. + SuccessResponders map[string]bool + // Unidentified is the count of frames received with no node ID header. + // Above zero, responder identity cannot be trusted for this fan-out. + Unidentified int +} + +// synthesize returns status frames for cached instances that no live frame +// covered and whose owning node is not answering. It is a backfill, never a +// second opinion: an instance a node reported on is not touched. +func (s StatusSynthesis) synthesize(ctx context.Context, input *ec2.DescribeInstanceStatusInput, accountID string, covered map[string]bool, responders fanoutResponders) []*ec2.InstanceStatus { + if s.Records == nil { + return nil + } + + // A cache that has not completed its first whole-set sync cannot support a + // claim about what is missing, so it stays out of the answer entirely. + instances, ready := s.Records.List(ctx, accountID) + if !ready { + return nil + } + + // The fan-out already reported a malformed request through its own error + // path. Adding nothing here avoids answering a request the node rejected. + selection, err := handlers_ec2_instance.ParseStatusSelection(input, accountID) + if err != nil { + return nil + } + + var out []*ec2.InstanceStatus + for _, v := range instances { + if v == nil || covered[v.ID] { + continue + } + + state := instancecache.NodeUnknown + if s.Liveness != nil { + state = s.Liveness.State(ctx, v.LastNode) + } + + if responders.Unidentified > 0 { + // Frames could not be attributed to nodes this fan-out, so the + // responder set is not trustworthy: fall back to heartbeat age. + if state == instancecache.NodeLive { + continue + } + } else if responders.SuccessResponders[v.LastNode] { + // The owning node answered this fan-out and chose not to report + // this instance — that exclusion is deliberate, not silence. + continue + } + + // SystemImpaired is node-local memory pressure only an answering node + // can report. A silent node's is unknowable, so it is not fabricated. + status := selection.StatusFor(v, handlers_ec2_instance.StatusOptions{AZ: v.AZ}) + if status == nil { + continue + } + degradeSystemStatus(status, state) + out = append(out, status) + } + + if len(out) > 0 { + slog.InfoContext(ctx, "DescribeInstanceStatus: synthesised frames for unanswered instances", + "count", len(out)) + } + return out +} + +// degradeSystemStatus marks a frame the owning node did not answer for. A +// stale node's instance is impaired; an unreadable heartbeat store leaves it +// insufficient-data, because not knowing a host is healthy is not knowing it +// is not. +func degradeSystemStatus(status *ec2.InstanceStatus, state instancecache.NodeState) { + // A stopped instance stays not-applicable: node liveness says nothing + // about an instance that is not meant to be running. + if status.SystemStatus == nil || aws.StringValue(status.SystemStatus.Status) == notApplicable { + return + } + + label := insufficientData + if state == instancecache.NodeStale { + label = statusImpaired + } + status.SystemStatus.Status = aws.String(label) + status.SystemStatus.Details = []*ec2.InstanceStatusDetails{{ + Name: aws.String(reachabilityDetailName), + Status: aws.String(label), + }} +} + +// coveredInstanceIDs indexes the instance IDs a node actually answered for, so +// synthesis can tell an unanswered instance from a deliberately excluded one. +func coveredInstanceIDs(statuses []*ec2.InstanceStatus) map[string]bool { + covered := make(map[string]bool, len(statuses)) + for _, s := range statuses { + if s != nil && s.InstanceId != nil { + covered[*s.InstanceId] = true + } + } + return covered +} diff --git a/spinifex/gateway/ec2/instance/status_synthesis_test.go b/spinifex/gateway/ec2/instance/status_synthesis_test.go new file mode 100644 index 000000000..362579cbd --- /dev/null +++ b/spinifex/gateway/ec2/instance/status_synthesis_test.go @@ -0,0 +1,385 @@ +//test:in-package — synthesize is unexported, and these tests reuse the NATS +// fan-out harness the package's existing in-package tests already define. + +package gateway_ec2_instance + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/instancecache" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/mulgadc/spinifex/spinifex/vm" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/require" +) + +const synthAccount = "123456789012" + +// fakeRecords is a cache stand-in: a fixed record set plus a readiness flag, +// so a test can assert the cold-cache case as well as the warm one. +type fakeRecords struct { + vms []*vm.VM + ready bool +} + +func (f fakeRecords) List(context.Context, string) ([]*vm.VM, bool) { return f.vms, f.ready } + +// fakeLiveness answers per node, defaulting to Unknown for anything unlisted. +type fakeLiveness struct { + states map[string]instancecache.NodeState +} + +func (f fakeLiveness) State(_ context.Context, node string) instancecache.NodeState { + return f.states[node] +} + +// cachedVM builds a record as the cache would hand it over. +func cachedVM(id, node, az string, state vm.InstanceState) *vm.VM { + launched := time.Now().Add(-time.Hour) + return &vm.VM{ + ID: id, + AccountID: synthAccount, + Status: state, + LastNode: node, + AZ: az, + Instance: &ec2.Instance{InstanceId: aws.String(id), LaunchTime: &launched}, + } +} + +// statusByID indexes an output for assertions. +func statusByID(out *ec2.DescribeInstanceStatusOutput) map[string]*ec2.InstanceStatus { + byID := make(map[string]*ec2.InstanceStatus, len(out.InstanceStatuses)) + for _, s := range out.InstanceStatuses { + byID[aws.StringValue(s.InstanceId)] = s + } + return byID +} + +// answerWith subscribes as nodeID and replies with the given statuses, +// carrying the X-Node-ID header a real daemon reply sets, so the fan-out's +// responder-identity path is what these tests actually exercise. +func answerWith(t *testing.T, nc *nats.Conn, nodeID string, statuses ...*ec2.InstanceStatus) { + t.Helper() + data, err := json.Marshal(&ec2.DescribeInstanceStatusOutput{InstanceStatuses: statuses}) + require.NoError(t, err) + subscribeAsNode(t, nc, "ec2.DescribeInstanceStatus", nodeID, data) +} + +// answerAnonymously subscribes without a node ID header, as an older daemon +// or a frame that could not be attributed would, to exercise the +// identity-loss fallback deliberately rather than by accident. +func answerAnonymously(t *testing.T, nc *nats.Conn, statuses ...*ec2.InstanceStatus) { + t.Helper() + _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { + respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{InstanceStatuses: statuses}) + }) + require.NoError(t, err) +} + +// An instance whose node stopped answering stays in the output and reports +// impaired, rather than vanishing as it does with fan-out alone. +func TestSynthesis_StaleNodeInstanceReportsImpaired(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + answerWith(t, nc, "node-a", runningStatus("i-live", "az-a")) + + synth := StatusSynthesis{ + Records: fakeRecords{ready: true, vms: []*vm.VM{ + cachedVM("i-live", "node-a", "az-a", vm.StateRunning), + cachedVM("i-orphan", "node-dead", "az-b", vm.StateRunning), + }}, + Liveness: fakeLiveness{states: map[string]instancecache.NodeState{ + "node-a": instancecache.NodeLive, + "node-dead": instancecache.NodeStale, + }}, + } + + out, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 1, synthAccount, "az-a", synth) + require.NoError(t, err) + + byID := statusByID(out) + require.Len(t, byID, 2, "the dead node's instance must not disappear") + require.Equal(t, "impaired", aws.StringValue(byID["i-orphan"].SystemStatus.Status)) + require.Equal(t, "az-b", aws.StringValue(byID["i-orphan"].AvailabilityZone), + "a synthesised frame carries the record's AZ, not this gateway's") +} + +// A frame an answering node returned is authoritative and must survive +// unchanged, even when that node is also reported stale. Two guards deliver +// this and either alone suffices, so see TestSynthesis_SkipsCoveredInstances +// for the one that pins the real mechanism. +func TestSynthesis_LiveFrameWinsUnchanged(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + + // The node reports its own memory pressure, which no cached record can + // reconstruct. Synthesis must not overwrite it. + nodeFrame := runningStatus("i-001", "az-a") + nodeFrame.SystemStatus.Status = aws.String("impaired") + answerWith(t, nc, "node-a", nodeFrame) + + synth := StatusSynthesis{ + Records: fakeRecords{ready: true, vms: []*vm.VM{cachedVM("i-001", "node-a", "az-a", vm.StateRunning)}}, + Liveness: fakeLiveness{states: map[string]instancecache.NodeState{"node-a": instancecache.NodeStale}}, + } + + out, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 1, synthAccount, "az-a", synth) + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 1) + require.Equal(t, "impaired", aws.StringValue(out.InstanceStatuses[0].SystemStatus.Status)) + require.Equal(t, "ok", aws.StringValue(out.InstanceStatuses[0].InstanceStatus.Status), + "the node's own frame must be used unchanged") +} + +// node-a answered this fan-out successfully and chose not to report i-gone. +// That is this request's own answer: the success-responder guard skips it +// regardless of what its heartbeat says. See +// TestSynthesis_SuccessResponderOmissionNotBackfilledEvenWhenStale for the +// same guard exercised with a stale heartbeat instead of a live one. +func TestSynthesis_LiveNodeSilenceIsNotBackfilled(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + answerWith(t, nc, "node-a", runningStatus("i-001", "az-a")) + + synth := StatusSynthesis{ + Records: fakeRecords{ready: true, vms: []*vm.VM{ + cachedVM("i-001", "node-a", "az-a", vm.StateRunning), + cachedVM("i-gone", "node-a", "az-a", vm.StateRunning), + }}, + Liveness: fakeLiveness{states: map[string]instancecache.NodeState{"node-a": instancecache.NodeLive}}, + } + + out, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 1, synthAccount, "az-a", synth) + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 1, "the answering node's own omission is its own answer") +} + +// An unreadable heartbeat store degrades only the synthesised frames to +// insufficient-data. Blanking every frame would hide the genuine health of +// every answering node behind a shrug. +func TestSynthesis_UnknownLivenessDegradesOnlySynthesised(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + + nodeFrame := runningStatus("i-live", "az-a") + nodeFrame.SystemStatus.Status = aws.String("impaired") + answerWith(t, nc, "node-a", nodeFrame) + + synth := StatusSynthesis{ + Records: fakeRecords{ready: true, vms: []*vm.VM{ + cachedVM("i-live", "node-a", "az-a", vm.StateRunning), + cachedVM("i-orphan", "node-b", "az-a", vm.StateRunning), + }}, + // Empty map: every lookup answers Unknown, as an unreadable store does. + Liveness: fakeLiveness{states: map[string]instancecache.NodeState{}}, + } + + out, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 1, synthAccount, "az-a", synth) + require.NoError(t, err) + + byID := statusByID(out) + require.Equal(t, "impaired", aws.StringValue(byID["i-live"].SystemStatus.Status), + "a live node's frame must survive an unreadable heartbeat store") + require.Equal(t, "insufficient-data", aws.StringValue(byID["i-orphan"].SystemStatus.Status), + "not knowing a host is healthy is not knowing it is not") +} + +// A stopped instance on a stale node stays not-applicable. Node liveness says +// nothing about an instance that is not meant to be running, and it must not +// appear in a default request at all. +func TestSynthesis_StoppedOnStaleNodeStaysOutAndNotApplicable(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + answerWith(t, nc, "node-active") + + records := fakeRecords{ready: true, vms: []*vm.VM{ + cachedVM("i-stopped", "node-dead", "az-a", vm.StateStopped), + }} + liveness := fakeLiveness{states: map[string]instancecache.NodeState{"node-dead": instancecache.NodeStale}} + synth := StatusSynthesis{Records: records, Liveness: liveness} + + out, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 1, synthAccount, "az-a", synth) + require.NoError(t, err) + require.Empty(t, out.InstanceStatuses, "a default request is running-only") + + all := &ec2.DescribeInstanceStatusInput{IncludeAllInstances: aws.Bool(true)} + out, err = DescribeInstanceStatus(context.Background(), all, nc, 1, synthAccount, "az-a", synth) + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 1) + require.Equal(t, "not-applicable", aws.StringValue(out.InstanceStatuses[0].SystemStatus.Status), + "a stopped instance is not impaired because its host is unreachable") +} + +// Synthesis honours the request rather than dumping the cache: an instance-ID +// filter excludes everything it did not ask for. +func TestSynthesis_HonoursInstanceIDFilter(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + answerWith(t, nc, "node-active") + + synth := StatusSynthesis{ + Records: fakeRecords{ready: true, vms: []*vm.VM{ + cachedVM("i-asked", "node-dead", "az-a", vm.StateRunning), + cachedVM("i-notasked", "node-dead", "az-a", vm.StateRunning), + }}, + Liveness: fakeLiveness{states: map[string]instancecache.NodeState{"node-dead": instancecache.NodeStale}}, + } + + input := &ec2.DescribeInstanceStatusInput{InstanceIds: []*string{aws.String("i-asked")}} + out, err := DescribeInstanceStatus(context.Background(), input, nc, 1, synthAccount, "az-a", synth) + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 1) + require.Equal(t, "i-asked", aws.StringValue(out.InstanceStatuses[0].InstanceId)) +} + +// A cache that has not finished its first whole-set sync cannot support a +// claim about what is missing, so it contributes nothing. +func TestSynthesis_ColdCacheContributesNothing(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + answerWith(t, nc, "node-active") + + synth := StatusSynthesis{ + Records: fakeRecords{ready: false, vms: []*vm.VM{cachedVM("i-orphan", "node-dead", "az-a", vm.StateRunning)}}, + Liveness: fakeLiveness{states: map[string]instancecache.NodeState{"node-dead": instancecache.NodeStale}}, + } + + out, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 1, synthAccount, "az-a", synth) + require.NoError(t, err) + require.Empty(t, out.InstanceStatuses) +} + +// The covered set is what actually keeps a live node's frame authoritative, +// so it is asserted directly. Through the full handler it is redundant with +// dedupStatuses being first-writer-wins, and a test there passes with either +// guard alone — which would let both rot until they broke together. +func TestSynthesis_SkipsCoveredInstances(t *testing.T) { + t.Parallel() + + synth := StatusSynthesis{ + Records: fakeRecords{ready: true, vms: []*vm.VM{ + cachedVM("i-covered", "node-dead", "az-a", vm.StateRunning), + cachedVM("i-uncovered", "node-dead", "az-a", vm.StateRunning), + }}, + Liveness: fakeLiveness{states: map[string]instancecache.NodeState{"node-dead": instancecache.NodeStale}}, + } + + got := synth.synthesize(context.Background(), &ec2.DescribeInstanceStatusInput{}, synthAccount, + map[string]bool{"i-covered": true}, fanoutResponders{}) + + require.Len(t, got, 1, "an instance a node answered for must not be synthesised at all") + require.Equal(t, "i-uncovered", aws.StringValue(got[0].InstanceId)) +} + +// A node that never answers this fan-out at all, but whose heartbeat has not +// yet gone stale, must not make its instances vanish. This is the regression +// the responder-set fix exists for: heartbeat age alone cannot tell "excluded +// on purpose" from "didn't answer this request". +func TestSynthesis_SilentButLiveNodeShowsInsufficientData(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + + live, err := json.Marshal(&ec2.DescribeInstanceStatusOutput{ + InstanceStatuses: []*ec2.InstanceStatus{runningStatus("i-live", "az-a")}, + }) + require.NoError(t, err) + subscribeAsNode(t, nc, "ec2.DescribeInstanceStatus", "node-live", live) + // node-silent never subscribes: it is expected but does not answer. + + synth := StatusSynthesis{ + Records: fakeRecords{ready: true, vms: []*vm.VM{ + cachedVM("i-live", "node-live", "az-a", vm.StateRunning), + cachedVM("i-orphan", "node-silent", "az-a", vm.StateRunning), + }}, + Liveness: fakeLiveness{states: map[string]instancecache.NodeState{ + "node-live": instancecache.NodeLive, + "node-silent": instancecache.NodeLive, + }}, + } + + out, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 2, synthAccount, "az-a", synth) + require.NoError(t, err) + + byID := statusByID(out) + require.Len(t, byID, 2, "a silent-but-live node's instance must not disappear") + require.Equal(t, "insufficient-data", aws.StringValue(byID["i-orphan"].SystemStatus.Status)) +} + +// A node that answered the fan-out successfully and chose not to report an +// instance is respected even when its heartbeat reads stale: the omission is +// this request's own answer, not a symptom for heartbeat age to reinterpret. +func TestSynthesis_SuccessResponderOmissionNotBackfilledEvenWhenStale(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + + data, err := json.Marshal(&ec2.DescribeInstanceStatusOutput{ + InstanceStatuses: []*ec2.InstanceStatus{runningStatus("i-present", "az-a")}, + }) + require.NoError(t, err) + subscribeAsNode(t, nc, "ec2.DescribeInstanceStatus", "node-a", data) + + synth := StatusSynthesis{ + Records: fakeRecords{ready: true, vms: []*vm.VM{ + cachedVM("i-present", "node-a", "az-a", vm.StateRunning), + cachedVM("i-gone", "node-a", "az-a", vm.StateRunning), + }}, + Liveness: fakeLiveness{states: map[string]instancecache.NodeState{"node-a": instancecache.NodeStale}}, + } + + out, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 1, synthAccount, "az-a", synth) + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 1, "the answering node's own omission must survive even a stale heartbeat") + require.Equal(t, "i-present", aws.StringValue(out.InstanceStatuses[0].InstanceId)) +} + +// A node that answered with an error envelope did not run its own selection +// logic, so its omission is not deliberate: its instances are synthesised. +func TestSynthesis_ErrorResponderInstancesAreSynthesised(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + + subscribeAsNode(t, nc, "ec2.DescribeInstanceStatus", "node-a", utils.GenerateErrorPayload("InvalidParameterValue")) + + synth := StatusSynthesis{ + Records: fakeRecords{ready: true, vms: []*vm.VM{ + cachedVM("i-orphan", "node-a", "az-a", vm.StateRunning), + }}, + Liveness: fakeLiveness{states: map[string]instancecache.NodeState{"node-a": instancecache.NodeLive}}, + } + + out, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 1, synthAccount, "az-a", synth) + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 1, "an error envelope is not a deliberate omission") + require.Equal(t, "i-orphan", aws.StringValue(out.InstanceStatuses[0].InstanceId)) + require.Equal(t, "insufficient-data", aws.StringValue(out.InstanceStatuses[0].SystemStatus.Status)) +} + +// When any frame cannot be attributed to a node, the responder set for the +// whole fan-out is untrustworthy, so synthesis falls back to the pre-fix +// heartbeat-only rule: a live node's silence is treated as deliberate again. +func TestSynthesis_UnidentifiedFrameFallsBackToHeartbeatOnly(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + answerAnonymously(t, nc, runningStatus("i-live", "az-a")) + + synth := StatusSynthesis{ + Records: fakeRecords{ready: true, vms: []*vm.VM{ + cachedVM("i-live", "node-live", "az-a", vm.StateRunning), + cachedVM("i-orphan", "node-silent", "az-a", vm.StateRunning), + }}, + Liveness: fakeLiveness{states: map[string]instancecache.NodeState{ + "node-live": instancecache.NodeLive, + "node-silent": instancecache.NodeLive, + }}, + } + + out, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 1, synthAccount, "az-a", synth) + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 1, + "without responder identity, a live node's instance is skipped by the fallback rule") +} diff --git a/spinifex/gateway/gateway.go b/spinifex/gateway/gateway.go index 04632b65f..e11bdfe8b 100644 --- a/spinifex/gateway/gateway.go +++ b/spinifex/gateway/gateway.go @@ -27,6 +27,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/accountteardown" "github.com/mulgadc/spinifex/spinifex/awserrors" gateway_bedrock "github.com/mulgadc/spinifex/spinifex/gateway/bedrock" + gateway_ec2_instance "github.com/mulgadc/spinifex/spinifex/gateway/ec2/instance" gateway_ecr "github.com/mulgadc/spinifex/spinifex/gateway/ecr" gateway_ecrauth "github.com/mulgadc/spinifex/spinifex/gateway/ecrauth" "github.com/mulgadc/spinifex/spinifex/gateway/policy" @@ -131,7 +132,11 @@ type GatewayConfig struct { // falls back to the per-account .dkr.ecr.. name. RegistryHost string AZ string // Availability zone this gateway is running in - IAMService handlers_iam.IAMService + // InstanceStatus backfills DescribeInstanceStatus for instances whose node + // stopped answering, which would otherwise drop out of the answer entirely. + // Zero value keeps the pre-existing fan-out-only behaviour. + InstanceStatus gateway_ec2_instance.StatusSynthesis + IAMService handlers_iam.IAMService // BucketStore reaps a tenant's S3 buckets during account teardown. It // signs with the config service credential, which predastore already // trusts to reach any bucket, so this grants enumeration, not access. diff --git a/spinifex/handlers/ec2/instance/service_impl.go b/spinifex/handlers/ec2/instance/service_impl.go index ce6614e89..e5b21bce3 100644 --- a/spinifex/handlers/ec2/instance/service_impl.go +++ b/spinifex/handlers/ec2/instance/service_impl.go @@ -2881,7 +2881,9 @@ func (s *InstanceServiceImpl) hostUnderMemoryPressure() bool { return false } -func (s *InstanceServiceImpl) buildInstanceStatus(v *vm.VM, systemImpaired bool) *ec2.InstanceStatus { +// az is passed rather than read from config because a cache-backed caller +// projects a record owned by another node, whose AZ is on the record. +func buildInstanceStatus(v *vm.VM, systemImpaired bool, az string) *ec2.InstanceStatus { state := &ec2.InstanceState{} if info, ok := vm.EC2APIState(v.Status); ok { state.SetCode(info.Code) @@ -2914,7 +2916,7 @@ func (s *InstanceServiceImpl) buildInstanceStatus(v *vm.VM, systemImpaired bool) } return &ec2.InstanceStatus{ - AvailabilityZone: aws.String(s.config.AZ), + AvailabilityZone: aws.String(az), InstanceId: aws.String(v.ID), InstanceState: state, InstanceStatus: &ec2.InstanceStatusSummary{ @@ -3025,43 +3027,21 @@ func instanceStatusMatchesFilters(v *vm.VM, is *ec2.InstanceStatus, filters map[ func (s *InstanceServiceImpl) DescribeInstanceStatus(ctx context.Context, input *ec2.DescribeInstanceStatusInput, accountID string) (*ec2.DescribeInstanceStatusOutput, error) { slog.InfoContext(ctx, "Processing DescribeInstanceStatus request from this node", "accountID", accountID) - instanceIDFilter, err := ParseInstanceIDFilter(input.InstanceIds) + selection, err := ParseStatusSelection(input, accountID) if err != nil { return nil, err } - parsedFilters, err := filterutil.ParseFilters(input.Filters, DescribeInstanceStatusValidFilters) - if err != nil { - slog.WarnContext(ctx, "DescribeInstanceStatus: invalid filter", "err", err) - return nil, errors.New(awserrors.ErrorInvalidParameterValue) - } - - includedStates := describeInstanceStatusRunningOnly - if aws.BoolValue(input.IncludeAllInstances) { - includedStates = describeInstanceStatusAllIncluded - } - // SystemStatus is node-wide: evaluate host memory pressure once per request // rather than per instance. - systemImpaired := s.hostUnderMemoryPressure() + opts := StatusOptions{SystemImpaired: s.hostUnderMemoryPressure(), AZ: s.config.AZ} var statuses []*ec2.InstanceStatus s.vmMgr.View(func(vms map[string]*vm.VM) { for _, v := range vms { - if !IsInstanceVisible(accountID, v.AccountID) { - continue + if is := selection.StatusFor(v, opts); is != nil { + statuses = append(statuses, is) } - if len(instanceIDFilter) > 0 && !instanceIDFilter[v.ID] { - continue - } - if !includedStates[v.Status] { - continue - } - is := s.buildInstanceStatus(v, systemImpaired) - if len(parsedFilters) > 0 && !instanceStatusMatchesFilters(v, is, parsedFilters) { - continue - } - statuses = append(statuses, is) } }) diff --git a/spinifex/handlers/ec2/instance/service_impl_test.go b/spinifex/handlers/ec2/instance/service_impl_test.go index 20cf912c8..db3554934 100644 --- a/spinifex/handlers/ec2/instance/service_impl_test.go +++ b/spinifex/handlers/ec2/instance/service_impl_test.go @@ -3983,7 +3983,7 @@ func TestBuildInstanceStatus_ErrorStateProjectsToStopped(t *testing.T) { errored.Status = vm.StateError svc := instanceStatusService(t, "az-a", map[string]*vm.VM{errored.ID: errored}) - is := svc.buildInstanceStatus(errored, false) + is := buildInstanceStatus(errored, false, svc.config.AZ) require.NotNil(t, is.InstanceState) assert.Equal(t, "stopped", *is.InstanceState.Name) assert.Equal(t, int64(80), *is.InstanceState.Code) @@ -4005,7 +4005,7 @@ func TestBuildInstanceStatus_NeverSurfacesNonAWSName(t *testing.T) { v.Status = state svc := instanceStatusService(t, "az-a", map[string]*vm.VM{v.ID: v}) - is := svc.buildInstanceStatus(v, false) + is := buildInstanceStatus(v, false, svc.config.AZ) require.NotNil(t, is.InstanceState) assert.True(t, valid[*is.InstanceState.Name], "buildInstanceStatus(%s) surfaced non-AWS name %q", state, *is.InstanceState.Name) @@ -4023,7 +4023,7 @@ func TestBuildInstanceStatus_IOErrorPauseIsImpaired(t *testing.T) { paused.Health.IOErrorSince = time.Now().Add(-45 * time.Second) svc := instanceStatusService(t, "az-a", map[string]*vm.VM{paused.ID: paused}) - is := svc.buildInstanceStatus(paused, false) + is := buildInstanceStatus(paused, false, svc.config.AZ) assert.Equal(t, "impaired", *is.InstanceStatus.Status) assert.Equal(t, "failed", *is.InstanceStatus.Details[0].Status) assert.Equal(t, "impaired", *is.SystemStatus.Status, @@ -4043,7 +4043,7 @@ func TestBuildInstanceStatus_ResumedGuestClearsImpairment(t *testing.T) { resumed.Health.IOErrorResumes = 0 svc := instanceStatusService(t, "az-a", map[string]*vm.VM{resumed.ID: resumed}) - is := svc.buildInstanceStatus(resumed, false) + is := buildInstanceStatus(resumed, false, svc.config.AZ) assert.Equal(t, "ok", *is.InstanceStatus.Status) assert.Equal(t, "ok", *is.SystemStatus.Status) assert.Nil(t, is.InstanceStatus.Details[0].ImpairedSince) @@ -4060,7 +4060,7 @@ func TestBuildInstanceStatus_IOErrorBeatsLaunchGrace(t *testing.T) { fresh.Health.IOErrorSince = time.Now().Add(-5 * time.Second) svc := instanceStatusService(t, "az-a", map[string]*vm.VM{fresh.ID: fresh}) - is := svc.buildInstanceStatus(fresh, false) + is := buildInstanceStatus(fresh, false, svc.config.AZ) assert.Equal(t, "impaired", *is.InstanceStatus.Status) } diff --git a/spinifex/handlers/ec2/instance/status_selection.go b/spinifex/handlers/ec2/instance/status_selection.go new file mode 100644 index 000000000..ba53142fb --- /dev/null +++ b/spinifex/handlers/ec2/instance/status_selection.go @@ -0,0 +1,85 @@ +package handlers_ec2_instance + +import ( + "errors" + "log/slog" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/mulgadc/spinifex/spinifex/filterutil" + "github.com/mulgadc/spinifex/spinifex/vm" +) + +// StatusSelection is a parsed DescribeInstanceStatus request: which instances +// the caller asked about and may see. Parse once per request, then apply to +// each candidate VM. +type StatusSelection struct { + accountID string + instanceIDs map[string]bool + filters map[string][]string + includedStates map[vm.InstanceState]bool +} + +// StatusOptions carries what the projection cannot read off a VM. SystemImpaired +// is node-local memory pressure that only an answering node can report, and AZ +// differs by source: the node's own on the node path, the record's on a cached one. +type StatusOptions struct { + SystemImpaired bool + AZ string +} + +// ParseStatusSelection validates the request and resolves which states are in +// scope. Shared so the node path and any cache-backed path cannot drift on what +// a request asked for; the default stays running-only. +func ParseStatusSelection(input *ec2.DescribeInstanceStatusInput, accountID string) (StatusSelection, error) { + instanceIDs, err := ParseInstanceIDFilter(input.InstanceIds) + if err != nil { + return StatusSelection{}, err + } + + filters, err := filterutil.ParseFilters(input.Filters, DescribeInstanceStatusValidFilters) + if err != nil { + slog.Warn("DescribeInstanceStatus: invalid filter", "err", err) + return StatusSelection{}, errors.New(awserrors.ErrorInvalidParameterValue) + } + + includedStates := describeInstanceStatusRunningOnly + if aws.BoolValue(input.IncludeAllInstances) { + includedStates = describeInstanceStatusAllIncluded + } + + return StatusSelection{ + accountID: accountID, + instanceIDs: instanceIDs, + filters: filters, + includedStates: includedStates, + }, nil +} + +// Includes reports whether v is part of this request's answer, before any +// status is built. Split out so a caller can test membership without paying +// for the projection. +func (s StatusSelection) Includes(v *vm.VM) bool { + if v == nil || !IsInstanceVisible(s.accountID, v.AccountID) { + return false + } + if len(s.instanceIDs) > 0 && !s.instanceIDs[v.ID] { + return false + } + return s.includedStates[v.Status] +} + +// StatusFor projects v into a status frame, or returns nil when the request +// does not cover it. The status filters run against the built frame, so they +// cannot be applied before projection. +func (s StatusSelection) StatusFor(v *vm.VM, opts StatusOptions) *ec2.InstanceStatus { + if !s.Includes(v) { + return nil + } + is := buildInstanceStatus(v, opts.SystemImpaired, opts.AZ) + if len(s.filters) > 0 && !instanceStatusMatchesFilters(v, is, s.filters) { + return nil + } + return is +} diff --git a/spinifex/instancecache/export_test.go b/spinifex/instancecache/export_test.go new file mode 100644 index 000000000..1152d1d15 --- /dev/null +++ b/spinifex/instancecache/export_test.go @@ -0,0 +1,34 @@ +package instancecache + +import ( + "encoding/json" + "time" +) + +// LivenessMemo is the read-collapsing window, exported so a test can position +// itself either side of it. +const LivenessMemo = livenessMemo + +// HeartbeatRecord returns the key and payload the daemon writes for a node's +// heartbeat, so a test can seed the bucket without importing the daemon. +func HeartbeatRecord(node string, stamped time.Time) (string, []byte, error) { + data, err := json.Marshal(heartbeat{ + Node: node, + Timestamp: stamped.UTC().Format(time.RFC3339), + }) + return heartbeatPrefix + node, data, err +} + +// SetClock replaces the clock staleness is judged against. +func (l *Liveness) SetClock(now func() time.Time) { + l.mu.Lock() + defer l.mu.Unlock() + l.now = now +} + +// SetFetchedAt moves the memo's last-read time; the zero value expires it. +func (l *Liveness) SetFetchedAt(t time.Time) { + l.mu.Lock() + defer l.mu.Unlock() + l.fetchedAt = t +} diff --git a/spinifex/instancecache/liveness.go b/spinifex/instancecache/liveness.go new file mode 100644 index 000000000..497970f4f --- /dev/null +++ b/spinifex/instancecache/liveness.go @@ -0,0 +1,126 @@ +package instancecache + +import ( + "context" + "sync" + "time" + + "github.com/mulgadc/spinifex/spinifex/kvstore" + "github.com/nats-io/nats.go/jetstream" +) + +const ( + // HeartbeatInterval mirrors the cadence the daemon publishes heartbeats + // at. Exported so the daemon can assert the two have not drifted; a + // longer publish interval would make NodeStaleAfter flap. + HeartbeatInterval = 10 * time.Second + + // NodeStaleAfter is three and a half missed beats. Two flaps every + // instance on a node over a GC pause or a NATS reconnect, four is slow to + // tell the truth. + NodeStaleAfter = 7 * HeartbeatInterval / 2 + + // heartbeatPrefix is the cluster-state key prefix the daemon writes each + // node's heartbeat under. + heartbeatPrefix = "heartbeat." + + // livenessMemo collapses a burst of requests into one KV read. Kept far + // below NodeStaleAfter so it cannot change a verdict: the worst it does is + // report a node stale a second late. + livenessMemo = time.Second +) + +// NodeState is what the heartbeat store says about a node. Unknown is a +// distinct answer from Stale: an unreadable store is not evidence that a node +// is gone, and the two must not collapse into one. +type NodeState int + +const ( + NodeUnknown NodeState = iota + NodeLive + NodeStale +) + +// heartbeat is the subset of the daemon's heartbeat record this needs. +// Decoding only these two fields keeps the gateway independent of the capacity +// figures alongside them, which change for unrelated reasons. +type heartbeat struct { + Node string `json:"node"` + Timestamp string `json:"timestamp"` +} + +// Liveness answers whether a node is still heartbeating. It is a read-only +// view over the cluster-state bucket and holds no watch of its own. +type Liveness struct { + store *kvstore.Store[heartbeat] + now func() time.Time + + mu sync.Mutex + seen map[string]time.Time + fetchedAt time.Time + readOK bool +} + +// NewLiveness opens a reader over the cluster-state bucket. The caller +// resolves the bucket name, as it does for the record cache, so this package +// never has to know the daemon's key-space constants. +func NewLiveness(js jetstream.JetStream, bucket kvstore.Config) *Liveness { + return &Liveness{ + store: kvstore.New[heartbeat](js, bucket), + now: time.Now, + seen: make(map[string]time.Time), + } +} + +// State reports whether node is live, stale, or unknown. A node with no +// heartbeat at all is Unknown rather than Stale: it may simply never have +// published one, which is not proof that it died. +func (l *Liveness) State(ctx context.Context, node string) NodeState { + if l == nil || node == "" { + return NodeUnknown + } + + l.mu.Lock() + defer l.mu.Unlock() + l.refreshLocked(ctx) + + if !l.readOK { + return NodeUnknown + } + written, ok := l.seen[node] + if !ok { + return NodeUnknown + } + if l.now().UTC().Sub(written) > NodeStaleAfter { + return NodeStale + } + return NodeLive +} + +// refreshLocked re-reads every heartbeat when the memo has expired. A failed +// read clears readOK rather than keeping the previous answers, so a store the +// gateway cannot reach degrades to Unknown instead of asserting liveness it +// can no longer confirm. +func (l *Liveness) refreshLocked(ctx context.Context) { + if l.now().Sub(l.fetchedAt) < livenessMemo { + return + } + l.fetchedAt = l.now() + + beats, err := l.store.List(ctx, heartbeatPrefix) + if err != nil { + l.readOK = false + return + } + + seen := make(map[string]time.Time, len(beats)) + for _, b := range beats { + written, err := time.Parse(time.RFC3339, b.Timestamp) + if err != nil || b.Node == "" { + continue + } + seen[b.Node] = written.UTC() + } + l.seen = seen + l.readOK = true +} diff --git a/spinifex/instancecache/liveness_test.go b/spinifex/instancecache/liveness_test.go new file mode 100644 index 000000000..105a3b460 --- /dev/null +++ b/spinifex/instancecache/liveness_test.go @@ -0,0 +1,116 @@ +package instancecache_test + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/instancecache" + "github.com/mulgadc/spinifex/spinifex/kvstore" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/nats-io/nats.go/jetstream" + "github.com/stretchr/testify/require" +) + +// newTestLiveness opens a cluster-state-shaped bucket and returns a Liveness +// over it, plus the raw KV handle so a test can write heartbeats directly. +func newTestLiveness(t *testing.T) (*instancecache.Liveness, jetstream.KeyValue) { + t.Helper() + _, _, js := testutil.StartTestJetStream(t) + + bucket := fmt.Sprintf("cluster-%d", time.Now().UnixNano()) + kv, err := js.CreateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: bucket, History: 1}) + require.NoError(t, err) + + l := instancecache.NewLiveness(js, kvstore.Config{Name: bucket, History: 1, Replicas: 1}) + return l, kv +} + +// putHeartbeat writes a heartbeat for node stamped age before the test's now. +func putHeartbeat(t *testing.T, kv jetstream.KeyValue, node string, now time.Time, age time.Duration) { + t.Helper() + key, data, err := instancecache.HeartbeatRecord(node, now.Add(-age)) + require.NoError(t, err) + _, err = kv.Put(context.Background(), key, data) + require.NoError(t, err) +} + +// A node beating inside the threshold is live; one past it is stale. The +// boundary is checked from both sides so the comparison cannot be inverted +// without a failure. +func TestLiveness_StaleOnlyPastThreshold(t *testing.T) { + l, kv := newTestLiveness(t) + now := time.Now() + l.SetClock(func() time.Time { return now }) + + putHeartbeat(t, kv, "fresh", now, time.Second) + putHeartbeat(t, kv, "justInside", now, instancecache.NodeStaleAfter-time.Second) + putHeartbeat(t, kv, "justOutside", now, instancecache.NodeStaleAfter+time.Second) + + ctx := context.Background() + require.Equal(t, instancecache.NodeLive, l.State(ctx, "fresh")) + + l.SetFetchedAt(time.Time{}) + require.Equal(t, instancecache.NodeLive, l.State(ctx, "justInside")) + require.Equal(t, instancecache.NodeStale, l.State(ctx, "justOutside")) +} + +// A node that has never published a heartbeat is Unknown, not Stale. Silence +// from a node nobody has heard from is not evidence that it died. +func TestLiveness_UnheardNodeIsUnknownNotStale(t *testing.T) { + l, kv := newTestLiveness(t) + now := time.Now() + l.SetClock(func() time.Time { return now }) + putHeartbeat(t, kv, "known", now, time.Second) + + require.Equal(t, instancecache.NodeUnknown, l.State(context.Background(), "never-seen")) +} + +// An unreadable store degrades every answer to Unknown rather than keeping the +// last one. Asserting liveness the gateway can no longer confirm is the lie +// this whole path exists to remove. +func TestLiveness_UnreadableStoreIsUnknown(t *testing.T) { + l, kv := newTestLiveness(t) + now := time.Now() + l.SetClock(func() time.Time { return now }) + putHeartbeat(t, kv, "node-a", now, time.Second) + + ctx := context.Background() + require.Equal(t, instancecache.NodeLive, l.State(ctx, "node-a")) + + // Cancelled context stands in for a bucket the gateway cannot reach. + dead, cancel := context.WithCancel(context.Background()) + cancel() + l.SetFetchedAt(time.Time{}) + require.Equal(t, instancecache.NodeUnknown, l.State(dead, "node-a")) +} + +// An empty node name is Unknown, and a nil Liveness answers rather than +// panicking, so a caller wired without one degrades instead of crashing. +func TestLiveness_EmptyNodeAndNilReceiver(t *testing.T) { + l, _ := newTestLiveness(t) + require.Equal(t, instancecache.NodeUnknown, l.State(context.Background(), "")) + + var nilLiveness *instancecache.Liveness + require.Equal(t, instancecache.NodeUnknown, nilLiveness.State(context.Background(), "node-a")) +} + +// The memo collapses repeated reads but must not outlive its window, or a +// node that goes silent would keep reporting live. +func TestLiveness_MemoExpires(t *testing.T) { + l, kv := newTestLiveness(t) + now := time.Now() + l.SetClock(func() time.Time { return now }) + putHeartbeat(t, kv, "node-a", now, time.Second) + + ctx := context.Background() + require.Equal(t, instancecache.NodeLive, l.State(ctx, "node-a")) + + // Same heartbeat, but the clock has moved past the threshold. Within the + // memo the cached answer stands; past it the re-read reports stale. + now = now.Add(instancecache.NodeStaleAfter + time.Second) + l.SetFetchedAt(now.Add(-instancecache.LivenessMemo / 2)) + require.Equal(t, instancecache.NodeStale, l.State(ctx, "node-a"), + "staleness is judged against the current clock, not the read time") +} diff --git a/spinifex/services/awsgw/awsgw.go b/spinifex/services/awsgw/awsgw.go index 53f204fab..4008f07a2 100644 --- a/spinifex/services/awsgw/awsgw.go +++ b/spinifex/services/awsgw/awsgw.go @@ -22,6 +22,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/daemon" "github.com/mulgadc/spinifex/spinifex/gateway" gateway_bedrock "github.com/mulgadc/spinifex/spinifex/gateway/bedrock" + gateway_ec2_instance "github.com/mulgadc/spinifex/spinifex/gateway/ec2/instance" gateway_ecr "github.com/mulgadc/spinifex/spinifex/gateway/ecr" gateway_ecrauth "github.com/mulgadc/spinifex/spinifex/gateway/ecrauth" handlers_bedrock "github.com/mulgadc/spinifex/spinifex/handlers/bedrock" @@ -572,14 +573,23 @@ func launchService(config *config.ClusterConfig) error { go gw.Quota.RunBedrockRPMSync(janitorCtx, js, len(config.Nodes)) // Instance cache: a read-only informer over the live instance record space, - // started now so it is warm well before anything reads it. Nothing consults - // it yet; the describe path still serves from the fan-out and KV. + // started now so it is warm well before anything reads it. The describe + // path still serves from the fan-out and KV; only status synthesis reads it. instanceCache := instancecache.New(js, instancecache.Config{ Bucket: kvstore.Config{Name: daemon.InstanceStateBucket, History: 1, Replicas: len(config.Nodes)}, Prefix: daemon.InstanceRecordPrefix, }) go instanceCache.Run(janitorCtx) + // Without this a node that stops answering takes its instances out of + // DescribeInstanceStatus entirely, so a dead host reads as no host. + gw.InstanceStatus = gateway_ec2_instance.StatusSynthesis{ + Records: instanceCache, + Liveness: instancecache.NewLiveness(js, kvstore.Config{ + Name: daemon.ClusterStateBucket, History: 1, Replicas: len(config.Nodes), + }), + } + handler := gw.SetupRoutes() // Load TLS certificate