feat: add read-only EC2 platform inventory for account termination (#1212)

* feat: add read-only EC2 platform inventory for account termination

* fix: drop placement-group member instance IDs from the platform inventory reply
This commit is contained in:
Josh Siviour
2026-10-16 12:05:32 +10:00
committed by GitHub
parent 359eb7a386
commit 71301983bd
22 changed files with 2248 additions and 2 deletions
+24
View File
@@ -36,3 +36,27 @@ contract or an approved compatibility plan; it is not a package refactor.
`EnsureDefaultVpcSubject` (`ec2.EnsureDefaultVpc`) is the request/reply route on which the gateway's account creation asks a daemon in the `spinifex-workers` queue group to build the account's default VPC.
Only the subject is owned here today.
The request (`account_id`) and reply (`vpc_id`, `error`) bodies are still declared as anonymous structs on each side; moving them into this package is a contract change that needs its own characterization first.
## Platform inventory
`PlatformInventorySubject` (`ec2.v1.PlatformInventory`) is a request/reply route on which a daemon in the `spinifex-workers` queue group reports one account's key pairs, placement groups, launch templates and their `ec2-tags` entries from the EC2 owner stores.
EC2 owns it: `domains/ec2/accountinventory` produces the reply and the daemon serves it; the only consumer today is `accountteardown`, which binds an adapter but does not call it.
The request names the account in its body (`account_id`, a 12-digit ID) and may carry an opaque `correlation_ref` that is echoed for logs.
The daemon ignores `X-Account-ID` for this subject.
Neither field grants anything: the subject is read-only, writes to no store, and its reply is never authority to delete.
The reply has one `kinds` entry per kind (`key-pair`, `placement-group`, `launch-template`, `tag-index`) with a status and scan start and end times.
The times are zero when no owner scan ran: a consumer-synthesised `scan-incomplete` report for no responder, a timeout or a transport failure carries zero timestamps and must not be read as a scan.
`observed-clean-at-scan-time` means every store for that kind was read completely and held nothing at scan time; writers are not fenced, so it is a snapshot and never evidence of absence.
`items` carry only a resource ID, kind and disposition (`live`, `dependency-held`, or tag-index `residue`), which are the identities the per-kind Describe subjects already return.
A `dependency-held` placement group does not name its member instances, which may be stale and are not exposed by the existing subjects.
Records no existing subject exposes (public-key objects without metadata, launch-template version bodies and name-index entries without a header, and every unreadable record) appear only as aggregate `counts` per kind, disposition and class.
The reply never carries an object key, raw KV key, key or template name, tag key or value, public-key bytes, launch-template data, user data, record body or store error text; `incomplete_reasons` are fixed categories.
### Compatibility boundary
The subject name carries the contract major; `platform_inventory_test.go` preserves the version-1 JSON shape and wire vocabulary.
Producer and consumer are in the coordinated Spinifex release, so the window is N and N-1.
A daemon that predates the subject has no subscription, so the request gets no responder; the consumer treats no responder, a timeout, an error reply, an undecodable reply, a reply for another account, and a missing, repeated or unknown-status kind as `scan-incomplete`, never clean.
An incompatible change to the subject, fields or vocabulary needs a new major.
+123
View File
@@ -0,0 +1,123 @@
package ec2v1
import "time"
// PlatformInventorySubject is the request/reply subject on which a daemon in
// the spinifex-workers queue group reports one account's key pairs, placement
// groups, launch templates and their tag-index entries. It never writes.
const PlatformInventorySubject = "ec2.v1.PlatformInventory"
// Kinds reported by a platform inventory. The tag index is its own kind
// because it is a separate store with its own completeness.
const (
PlatformKindKeyPair = "key-pair"
PlatformKindPlacementGroup = "placement-group"
PlatformKindLaunchTemplate = "launch-template"
PlatformKindTagIndex = "tag-index"
)
// PlatformKinds lists every kind a complete reply reports, in reply order.
func PlatformKinds() []string {
return []string{PlatformKindKeyPair, PlatformKindPlacementGroup, PlatformKindLaunchTemplate, PlatformKindTagIndex}
}
// Per-kind statuses. Only PlatformStatusObservedClean says a kind's stores
// were read completely and held nothing, and only at scan time: writers are
// not fenced, so it is a snapshot and never evidence of absence.
const (
PlatformStatusObservedClean = "observed-clean-at-scan-time"
PlatformStatusRecordsFound = "records-found"
PlatformStatusScanIncomplete = "scan-incomplete"
)
// Item and count dispositions. A store failure is reported once per kind as
// PlatformStatusScanIncomplete rather than per record.
const (
PlatformDispositionLive = "live"
PlatformDispositionResidue = "residue"
PlatformDispositionUnreadable = "unreadable"
PlatformDispositionDependencyHeld = "dependency-held"
)
// Record classes that are reported only as aggregate counts.
const (
PlatformClassPublicKeyWithoutMetadata = "public-key-object-without-metadata"
PlatformClassPublicKeyUnresolved = "public-key-object-unresolved"
PlatformClassUnreadableRecord = "unreadable-record"
PlatformClassVersionBodyWithoutHeader = "version-body-without-header"
PlatformClassNameIndexWithoutHeader = "name-index-without-header"
PlatformClassUnrecognisedTemplateEntry = "unrecognised-template-entry"
)
// Generic reasons a kind is PlatformStatusScanIncomplete. They are categories,
// never store error text, which can quote stored content.
const (
PlatformReasonStoreListFailed = "store-list-failed"
PlatformReasonRecordReadFailed = "record-read-failed"
PlatformReasonOwnerRecordsUnresolved = "owner-records-unresolved"
PlatformReasonNoResponder = "no-responder"
PlatformReasonTimeout = "timeout"
PlatformReasonTransportFailed = "transport-failed"
PlatformReasonErrorReply = "error-reply"
PlatformReasonInvalidReply = "invalid-reply"
PlatformReasonKindNotReported = "kind-not-reported"
)
// PlatformInventoryRequest asks for one account's platform inventory.
// CorrelationRef is an opaque caller reference echoed for logs; it grants
// nothing and is never a fence or authority to write.
type PlatformInventoryRequest struct {
AccountID string `json:"account_id"`
CorrelationRef string `json:"correlation_ref,omitempty"`
}
// PlatformInventoryReply carries per-kind status, the identities a direct
// caller of the per-kind Describe subjects can already read, and counts for
// every record class those subjects do not expose. It carries no record
// keys, names, tags, key material, template data or record bodies.
type PlatformInventoryReply struct {
AccountID string `json:"account_id"`
CorrelationRef string `json:"correlation_ref,omitempty"`
Kinds []PlatformKindReport `json:"kinds"`
Items []PlatformItem `json:"items,omitempty"`
Counts []PlatformRecordCount `json:"counts,omitempty"`
}
// PlatformKindReport is one kind's scan status and when its scan ran. The
// timestamps are zero when no owner scan ran, as in a consumer-synthesised
// scan-incomplete report for no responder, a timeout or a transport failure.
type PlatformKindReport struct {
Kind string `json:"kind"`
Status string `json:"status"`
IncompleteReasons []string `json:"incomplete_reasons,omitempty"`
ScanStartedAt time.Time `json:"scan_started_at"`
ScanEndedAt time.Time `json:"scan_ended_at"`
}
// PlatformItem is one resource by its AWS resource ID. A dependency-held
// placement group does not name its member instances.
type PlatformItem struct {
ResourceID string `json:"resource_id"`
Kind string `json:"kind"`
Disposition string `json:"disposition"`
}
// PlatformRecordCount is the number of records of one class in one kind.
type PlatformRecordCount struct {
Kind string `json:"kind"`
Disposition string `json:"disposition"`
Class string `json:"class"`
Count int `json:"count"`
}
// IncompletePlatformInventory is the reply a consumer substitutes when no
// usable answer arrived: every kind is scan-incomplete for reason.
func IncompletePlatformInventory(accountID, correlationRef, reason string) PlatformInventoryReply {
reply := PlatformInventoryReply{AccountID: accountID, CorrelationRef: correlationRef}
for _, kind := range PlatformKinds() {
reply.Kinds = append(reply.Kinds, PlatformKindReport{
Kind: kind, Status: PlatformStatusScanIncomplete, IncompleteReasons: []string{reason},
})
}
return reply
}
+102
View File
@@ -0,0 +1,102 @@
package ec2v1
import (
"encoding/json"
"testing"
"time"
"github.com/stretchr/testify/require"
)
func TestPlatformInventorySubject(t *testing.T) {
require.Equal(t, "ec2.v1.PlatformInventory", PlatformInventorySubject)
}
func TestPlatformInventoryRequestJSONContract(t *testing.T) {
data, err := json.Marshal(PlatformInventoryRequest{AccountID: "111122223333", CorrelationRef: "req-1"})
require.NoError(t, err)
require.JSONEq(t, `{"account_id":"111122223333","correlation_ref":"req-1"}`, string(data))
data, err = json.Marshal(PlatformInventoryRequest{AccountID: "111122223333"})
require.NoError(t, err)
require.JSONEq(t, `{"account_id":"111122223333"}`, string(data))
}
func TestPlatformInventoryReplyJSONContract(t *testing.T) {
start := time.Date(2026, 10, 16, 1, 2, 3, 0, time.UTC)
reply := PlatformInventoryReply{
AccountID: "111122223333",
CorrelationRef: "req-1",
Kinds: []PlatformKindReport{
{Kind: PlatformKindKeyPair, Status: PlatformStatusObservedClean, ScanStartedAt: start, ScanEndedAt: start.Add(time.Second)},
{Kind: PlatformKindPlacementGroup, Status: PlatformStatusRecordsFound, ScanStartedAt: start, ScanEndedAt: start},
{Kind: PlatformKindLaunchTemplate, Status: PlatformStatusScanIncomplete, IncompleteReasons: []string{PlatformReasonStoreListFailed}, ScanStartedAt: start, ScanEndedAt: start},
{Kind: PlatformKindTagIndex, Status: PlatformStatusRecordsFound, ScanStartedAt: start, ScanEndedAt: start},
},
Items: []PlatformItem{
{ResourceID: "pg-0held", Kind: PlatformKindPlacementGroup, Disposition: PlatformDispositionDependencyHeld},
{ResourceID: "lt-0gone", Kind: PlatformKindTagIndex, Disposition: PlatformDispositionResidue},
},
Counts: []PlatformRecordCount{
{Kind: PlatformKindKeyPair, Disposition: PlatformDispositionResidue, Class: PlatformClassPublicKeyWithoutMetadata, Count: 2},
},
}
data, err := json.Marshal(reply)
require.NoError(t, err)
require.JSONEq(t, `{
"account_id": "111122223333",
"correlation_ref": "req-1",
"kinds": [
{"kind": "key-pair", "status": "observed-clean-at-scan-time", "scan_started_at": "2026-10-16T01:02:03Z", "scan_ended_at": "2026-10-16T01:02:04Z"},
{"kind": "placement-group", "status": "records-found", "scan_started_at": "2026-10-16T01:02:03Z", "scan_ended_at": "2026-10-16T01:02:03Z"},
{"kind": "launch-template", "status": "scan-incomplete", "incomplete_reasons": ["store-list-failed"], "scan_started_at": "2026-10-16T01:02:03Z", "scan_ended_at": "2026-10-16T01:02:03Z"},
{"kind": "tag-index", "status": "records-found", "scan_started_at": "2026-10-16T01:02:03Z", "scan_ended_at": "2026-10-16T01:02:03Z"}
],
"items": [
{"resource_id": "pg-0held", "kind": "placement-group", "disposition": "dependency-held"},
{"resource_id": "lt-0gone", "kind": "tag-index", "disposition": "residue"}
],
"counts": [
{"kind": "key-pair", "disposition": "residue", "class": "public-key-object-without-metadata", "count": 2}
]
}`, string(data))
}
// The wire values of every status, disposition, class and reason.
func TestPlatformInventoryVocabulary(t *testing.T) {
require.Equal(t, []string{"key-pair", "placement-group", "launch-template", "tag-index"}, PlatformKinds())
require.Equal(t, []string{"live", "residue", "unreadable", "dependency-held"},
[]string{PlatformDispositionLive, PlatformDispositionResidue, PlatformDispositionUnreadable, PlatformDispositionDependencyHeld})
require.Equal(t, []string{
"public-key-object-without-metadata", "public-key-object-unresolved", "unreadable-record",
"version-body-without-header", "name-index-without-header", "unrecognised-template-entry",
}, []string{
PlatformClassPublicKeyWithoutMetadata, PlatformClassPublicKeyUnresolved, PlatformClassUnreadableRecord,
PlatformClassVersionBodyWithoutHeader, PlatformClassNameIndexWithoutHeader, PlatformClassUnrecognisedTemplateEntry,
})
require.Equal(t, []string{
"store-list-failed", "record-read-failed", "owner-records-unresolved", "no-responder",
"timeout", "transport-failed", "error-reply", "invalid-reply", "kind-not-reported",
}, []string{
PlatformReasonStoreListFailed, PlatformReasonRecordReadFailed, PlatformReasonOwnerRecordsUnresolved, PlatformReasonNoResponder,
PlatformReasonTimeout, PlatformReasonTransportFailed, PlatformReasonErrorReply, PlatformReasonInvalidReply, PlatformReasonKindNotReported,
})
}
// A substituted reply never reports a kind clean.
func TestIncompletePlatformInventoryMarksEveryKindIncomplete(t *testing.T) {
reply := IncompletePlatformInventory("111122223333", "req-1", PlatformReasonNoResponder)
require.Equal(t, "111122223333", reply.AccountID)
require.Equal(t, "req-1", reply.CorrelationRef)
require.Len(t, reply.Kinds, len(PlatformKinds()))
for i, kind := range PlatformKinds() {
require.Equal(t, kind, reply.Kinds[i].Kind)
require.Equal(t, PlatformStatusScanIncomplete, reply.Kinds[i].Status)
require.Equal(t, []string{PlatformReasonNoResponder}, reply.Kinds[i].IncompleteReasons)
require.True(t, reply.Kinds[i].ScanStartedAt.IsZero(), "no owner scan ran")
require.True(t, reply.Kinds[i].ScanEndedAt.IsZero(), "no owner scan ran")
}
require.Empty(t, reply.Items)
require.Empty(t, reply.Counts)
}
+1 -1
View File
@@ -446,7 +446,7 @@ These residuals were recorded by individual slices and must be resolved once the
- `providers/ebs/viperblock` → `runtime/compute/nbd` and `runtime/host/process` (dependency, moved unchanged from `services/viperblockd`): the adapter formats NBD endpoints and manages nbdkit processes through runtime packages. ADR-0001 does not list providers → runtime as forbidden and the layering test does not enforce it (same category as `providers/ebs/conformance` → `runtime/compute/nbd`), but a provider should not depend on node runtime. Clears when NBD endpoint naming has a provider-facing home (a candidate for `contracts/ebs/v1`) and process supervision is injected by the role. The adapter also imports `providers/objectstore`, a provider-to-provider edge that is allowed.
- `gateway/ec2.go` (registration): the EC2 dispatch map stays a `GatewayConfig` method in `gateway` and imports every `domains/ec2/awsapi/*` package; `gateway/ec2_idempotency.go` also imports `domains/ec2/awsapi/idem` (`KVBucket`, `TTL`, `TokenAndParams`) for the dispatch-time idempotency wrapper. ADR-0001 makes the domain own its action registration; moving the map waits on the generic ingress registration seam and must not be done as a package move.
- `gateway/requestaudit.go`: the outermost request-audit middleware records AWS fields and is updated from four places in the gateway's authentication flow through the gateway's own context key. Moving it would export about seven names used only by that flow, so it stays in `gateway` until the generic ingress registration seam defines how ingress middleware receives authentication results.
- `accountteardown` (dependency): its EC2 reapers call `domains/ec2/awsapi/*` action adapters directly, as they did under `gateway/ec2`. The ADR's `accountteardown` current-to-target map row replaces this with domain-owned cleanup capabilities before the package moves.
- `accountteardown` (dependency): its EC2 reapers call `domains/ec2/awsapi/*` action adapters directly, as they did under `gateway/ec2`. The ADR's `accountteardown` current-to-target map row replaces this with domain-owned cleanup capabilities before the package moves. It also holds a read-only consumer of the EC2 platform-inventory contract (`contracts/ec2/v1`, `ec2.v1.PlatformInventory`), served by `domains/ec2/accountinventory`; no engine path calls it yet, and the key-pair, placement-group and launch-template reapers still use the `awsapi` adapters.
- `domains/admission/quota` (dependency, moved from `handlers/quota` unchanged): `live.go` imports `gateway/elbv2` and `handlers/rds` for its live ELBv2 and RDS counters; `records.go` imports `runtime/compute/vm` for the instance-record vCPU sweep; `live.go`, `vcpu.go` and `storage.go` call `domains/ec2/awsapi/{vpc,eip,volume}` describe adapters to count VPCs, subnets, addresses and volumes, as they did under `gateway/ec2`. The ADR's `handlers/quota` current-to-target map row replaces these with reader capabilities once their final locations exist.
- `domains/ec2/awsapi/instance` (dependency): imports `handlers/iam` (`IAMService` in `DescribeInstances.go`, `RunInstances.go` and `IamInstanceProfileAssociation.go`; `InstanceProfile` in the latter two), `runtime/compute/cache` (`NodeState`, `NodeUnknown`, `NodeLive`) and `runtime/compute/vm` (`VM`), both in `status_synthesis.go`. The IAM edge needs an IAM capability; the runtime edges are domain-to-runtime and need the instance record and node liveness exposed through an EC2-owned projection.
- `domains/ec2/awsapi/spotinstance` (dependency): imports `handlers/iam.IAMService` and `domains/admission/quota.Service` in `RequestSpotInstances.go`.
@@ -0,0 +1,111 @@
// Package readonlystore wraps an object store or a JetStream KV bucket so that
// any write fails the test. It exists to prove a code path is read-only, which
// an assertion on the final store contents cannot: a write that is undone
// before the check would pass it.
package readonlystore
import (
"context"
"errors"
"sync"
"testing"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/s3"
"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
"github.com/nats-io/nats.go/jetstream"
)
// ErrWrite is returned by every write after the test has been failed.
var ErrWrite = errors.New("write to a read-only store")
// ObjectStore fails the test on any write and records every key it reads.
type ObjectStore struct {
objectstore.ObjectStore
t testing.TB
mu sync.Mutex
gets []string
}
var _ objectstore.ObjectStore = (*ObjectStore)(nil)
// NewObjectStore wraps inner for t.
func NewObjectStore(t testing.TB, inner objectstore.ObjectStore) *ObjectStore {
return &ObjectStore{ObjectStore: inner, t: t}
}
func (s *ObjectStore) GetObject(ctx context.Context, in *s3.GetObjectInput) (*s3.GetObjectOutput, error) {
s.mu.Lock()
s.gets = append(s.gets, aws.StringValue(in.Key))
s.mu.Unlock()
return s.ObjectStore.GetObject(ctx, in)
}
// Gets returns every key read so far.
func (s *ObjectStore) Gets() []string {
s.mu.Lock()
defer s.mu.Unlock()
return append([]string(nil), s.gets...)
}
func (s *ObjectStore) PutObject(_ context.Context, in *s3.PutObjectInput) (*s3.PutObjectOutput, error) {
s.t.Errorf("read-only store: PutObject %s", aws.StringValue(in.Key))
return nil, ErrWrite
}
func (s *ObjectStore) DeleteObject(_ context.Context, in *s3.DeleteObjectInput) (*s3.DeleteObjectOutput, error) {
s.t.Errorf("read-only store: DeleteObject %s", aws.StringValue(in.Key))
return nil, ErrWrite
}
func (s *ObjectStore) EnsureBucket(_ context.Context, bucket string) error {
s.t.Errorf("read-only store: EnsureBucket %s", bucket)
return ErrWrite
}
// KV fails the test on any write to the wrapped bucket.
type KV struct {
jetstream.KeyValue
t testing.TB
}
// NewKV wraps inner for t.
func NewKV(t testing.TB, inner jetstream.KeyValue) *KV {
return &KV{KeyValue: inner, t: t}
}
func (k *KV) write(op, key string) error {
k.t.Errorf("read-only KV %s: %s %s", k.Bucket(), op, key)
return ErrWrite
}
func (k *KV) Put(_ context.Context, key string, _ []byte) (uint64, error) {
return 0, k.write("Put", key)
}
func (k *KV) PutString(_ context.Context, key string, _ string) (uint64, error) {
return 0, k.write("PutString", key)
}
func (k *KV) Create(_ context.Context, key string, _ []byte, _ ...jetstream.KVCreateOpt) (uint64, error) {
return 0, k.write("Create", key)
}
func (k *KV) Update(_ context.Context, key string, _ []byte, _ uint64) (uint64, error) {
return 0, k.write("Update", key)
}
func (k *KV) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error {
return k.write("Delete", key)
}
func (k *KV) Purge(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error {
return k.write("Purge", key)
}
func (k *KV) PurgeDeletes(_ context.Context, _ ...jetstream.KVPurgeOpt) error {
return k.write("PurgeDeletes", "")
}
+3 -1
View File
@@ -72,5 +72,7 @@ func NewClusterEngine(
reapers = append(reapers, IAMReapers(svc)...)
reapers = append(reapers, AccountReapers(svc, names, usage)...)
return NewEngine(NewIAMAccounts(svc), reapers...), nil
engine := NewEngine(NewIAMAccounts(svc), reapers...)
engine.ec2Inventory = newEC2PlatformInventoryClient(nc)
return engine, nil
}
@@ -0,0 +1,92 @@
package accountteardown
import (
"context"
"errors"
"slices"
"time"
ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
"github.com/nats-io/nats.go"
)
// ec2PlatformInventoryTimeout bounds one inventory request. The scan reads
// four stores for one account; a reply later than this is scan-incomplete.
const ec2PlatformInventoryTimeout = 30 * time.Second
// ec2PlatformInventory is EC2's read-only report of an account's key pairs,
// placement groups and launch templates. It never deletes, and a failure to
// obtain the report is returned as scan-incomplete kinds, never as clean.
type ec2PlatformInventory interface {
PlatformInventory(ctx context.Context, accountID, correlationRef string) ec2v1.PlatformInventoryReply
}
// ec2PlatformInventoryClient reaches the EC2 platform-inventory contract over NATS.
type ec2PlatformInventoryClient struct {
nc *nats.Conn
timeout time.Duration
}
var _ ec2PlatformInventory = (*ec2PlatformInventoryClient)(nil)
func newEC2PlatformInventoryClient(nc *nats.Conn) *ec2PlatformInventoryClient {
return &ec2PlatformInventoryClient{nc: nc, timeout: ec2PlatformInventoryTimeout}
}
// PlatformInventory asks EC2 for the account's platform inventory. No
// responder, which is also how a daemon predating the subject answers, a
// timeout, an error reply or a malformed reply each make every kind incomplete.
func (c *ec2PlatformInventoryClient) PlatformInventory(ctx context.Context, accountID, correlationRef string) ec2v1.PlatformInventoryReply {
reply, err := natsmsg.NATSRequest[ec2v1.PlatformInventoryReply](ctx, c.nc, ec2v1.PlatformInventorySubject,
&ec2v1.PlatformInventoryRequest{AccountID: accountID, CorrelationRef: correlationRef}, c.timeout, accountID)
if err != nil {
return ec2v1.IncompletePlatformInventory(accountID, correlationRef, inventoryFailureReason(err))
}
return normalisePlatformInventory(*reply, accountID, correlationRef)
}
func inventoryFailureReason(err error) string {
switch {
case errors.Is(err, nats.ErrNoResponders):
return ec2v1.PlatformReasonNoResponder
case errors.Is(err, nats.ErrTimeout), errors.Is(err, context.DeadlineExceeded):
return ec2v1.PlatformReasonTimeout
}
if _, ok := awserrors.ResolveErrorCode(err); ok {
return ec2v1.PlatformReasonErrorReply
}
return ec2v1.PlatformReasonTransportFailed
}
// normalisePlatformInventory refuses a reply for another account and turns a
// missing or repeated kind, or an unknown status, into scan-incomplete.
func normalisePlatformInventory(reply ec2v1.PlatformInventoryReply, accountID, correlationRef string) ec2v1.PlatformInventoryReply {
if reply.AccountID != accountID {
return ec2v1.IncompletePlatformInventory(accountID, correlationRef, ec2v1.PlatformReasonInvalidReply)
}
known := []string{ec2v1.PlatformStatusObservedClean, ec2v1.PlatformStatusRecordsFound, ec2v1.PlatformStatusScanIncomplete}
byKind := make(map[string]ec2v1.PlatformKindReport, len(reply.Kinds))
for _, report := range reply.Kinds {
if !slices.Contains(known, report.Status) {
report.Status = ec2v1.PlatformStatusScanIncomplete
report.IncompleteReasons = append(report.IncompleteReasons, ec2v1.PlatformReasonInvalidReply)
}
if _, dup := byKind[report.Kind]; dup {
report = ec2v1.PlatformKindReport{Kind: report.Kind, Status: ec2v1.PlatformStatusScanIncomplete,
IncompleteReasons: []string{ec2v1.PlatformReasonInvalidReply}}
}
byKind[report.Kind] = report
}
reply.Kinds = nil
for _, kind := range ec2v1.PlatformKinds() {
report, ok := byKind[kind]
if !ok {
report = ec2v1.PlatformKindReport{Kind: kind, Status: ec2v1.PlatformStatusScanIncomplete,
IncompleteReasons: []string{ec2v1.PlatformReasonKindNotReported}}
}
reply.Kinds = append(reply.Kinds, report)
}
return reply
}
@@ -0,0 +1,192 @@
package accountteardown
//test:in-package — the consumer interface, its adapter and the engine field
// that binds it are unexported.
import (
"context"
"encoding/json"
"testing"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountinventory"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// serveInventory answers the inventory subject with reply's bytes, or not at all when nil.
func serveInventory(t *testing.T, nc *nats.Conn, reply func() []byte) {
t.Helper()
sub, err := nc.QueueSubscribe(ec2v1.PlatformInventorySubject, "spinifex-workers", func(msg *nats.Msg) {
if data := reply(); data != nil {
_ = msg.Respond(data)
}
})
require.NoError(t, err)
t.Cleanup(func() { _ = sub.Unsubscribe() })
require.NoError(t, nc.Flush())
}
func assertAllIncomplete(t *testing.T, reply ec2v1.PlatformInventoryReply, reason string) {
t.Helper()
assert.Equal(t, testAccountID, reply.AccountID)
require.Len(t, reply.Kinds, 4)
for _, report := range reply.Kinds {
assert.Equal(t, ec2v1.PlatformStatusScanIncomplete, report.Status, report.Kind)
assert.Equal(t, []string{reason}, report.IncompleteReasons, report.Kind)
}
assert.Empty(t, reply.Items)
assert.Empty(t, reply.Counts)
}
// The adapter sends the account and correlation reference in the body and
// returns EC2's report of the owner stores, residue counts included.
func TestEC2PlatformInventoryClientReturnsEC2sReport(t *testing.T) {
o := newPlatformOwners(t)
ctx := testCtx(t)
inv := accountinventory.New(o.keys, o.groups, o.templates, o.tags)
var seen ec2v1.PlatformInventoryRequest
sub, err := o.nc.QueueSubscribe(ec2v1.PlatformInventorySubject, "spinifex-workers", func(msg *nats.Msg) {
natsmsg.ServeNATSRequestCtx(msg, func(ctx context.Context, req *ec2v1.PlatformInventoryRequest) (*ec2v1.PlatformInventoryReply, error) {
seen = *req
return inv.PlatformInventory(ctx, req)
})
})
require.NoError(t, err)
t.Cleanup(func() { _ = sub.Unsubscribe() })
_, err = o.keys.ImportKeyPair(ctx, &ec2.ImportKeyPairInput{
KeyName: aws.String("deploy"), PublicKeyMaterial: []byte("ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIOMqqnkVzrm0SdG6UOoqKLsabgH5C9okWi0dh2l9GKJl"),
}, testAccountID)
require.NoError(t, err)
o.store.failDelete = func(key string) bool { return key == "keys/"+testAccountID+"/deploy" }
_, err = o.keys.DeleteKeyPair(ctx, &ec2.DeleteKeyPairInput{KeyName: aws.String("deploy")}, testAccountID)
require.NoError(t, err)
reply := newEC2PlatformInventoryClient(o.nc).PlatformInventory(ctx, testAccountID, "req-1")
assert.Equal(t, ec2v1.PlatformInventoryRequest{AccountID: testAccountID, CorrelationRef: "req-1"}, seen)
assert.Equal(t, "req-1", reply.CorrelationRef)
assert.Equal(t, []ec2v1.PlatformRecordCount{{
Kind: ec2v1.PlatformKindKeyPair, Disposition: ec2v1.PlatformDispositionResidue,
Class: ec2v1.PlatformClassPublicKeyWithoutMetadata, Count: 1,
}}, reply.Counts)
statuses := map[string]string{}
for _, report := range reply.Kinds {
statuses[report.Kind] = report.Status
}
assert.Equal(t, map[string]string{
ec2v1.PlatformKindKeyPair: ec2v1.PlatformStatusRecordsFound,
ec2v1.PlatformKindPlacementGroup: ec2v1.PlatformStatusObservedClean,
ec2v1.PlatformKindLaunchTemplate: ec2v1.PlatformStatusObservedClean,
ec2v1.PlatformKindTagIndex: ec2v1.PlatformStatusObservedClean,
}, statuses)
}
// No daemon serving the subject, which is also what a daemon predating it
// looks like, is scan-incomplete for every kind, never clean.
func TestEC2PlatformInventoryClientReportsNoResponderAsIncomplete(t *testing.T) {
_, nc, _ := testutil.StartTestJetStream(t)
reply := newEC2PlatformInventoryClient(nc).PlatformInventory(testCtx(t), testAccountID, "req-1")
assertAllIncomplete(t, reply, ec2v1.PlatformReasonNoResponder)
assert.Equal(t, "req-1", reply.CorrelationRef)
}
func TestEC2PlatformInventoryClientReportsTransportFailuresAsIncomplete(t *testing.T) {
t.Run("timeout", func(t *testing.T) {
_, nc, _ := testutil.StartTestJetStream(t)
serveInventory(t, nc, func() []byte { return nil })
client := newEC2PlatformInventoryClient(nc)
client.timeout = 50 * time.Millisecond
assertAllIncomplete(t, client.PlatformInventory(testCtx(t), testAccountID, ""), ec2v1.PlatformReasonTimeout)
})
t.Run("error reply", func(t *testing.T) {
_, nc, _ := testutil.StartTestJetStream(t)
serveInventory(t, nc, func() []byte {
return awserrors.GenerateErrorPayload(awserrors.ErrorServerInternal)
})
assertAllIncomplete(t, newEC2PlatformInventoryClient(nc).PlatformInventory(testCtx(t), testAccountID, ""), ec2v1.PlatformReasonErrorReply)
})
t.Run("undecodable reply", func(t *testing.T) {
_, nc, _ := testutil.StartTestJetStream(t)
serveInventory(t, nc, func() []byte { return []byte(`{"kinds":"nope"}`) })
assertAllIncomplete(t, newEC2PlatformInventoryClient(nc).PlatformInventory(testCtx(t), testAccountID, ""), ec2v1.PlatformReasonTransportFailed)
})
t.Run("closed connection", func(t *testing.T) {
_, nc, _ := testutil.StartTestJetStream(t)
nc.Close()
assertAllIncomplete(t, newEC2PlatformInventoryClient(nc).PlatformInventory(testCtx(t), testAccountID, ""), ec2v1.PlatformReasonTransportFailed)
})
}
// A reply for another account is discarded; a kind the reply omits or
// repeats, or a status this consumer does not know, is scan-incomplete.
func TestEC2PlatformInventoryClientDistrustsAMalformedReply(t *testing.T) {
clean := ec2v1.PlatformKindReport{Kind: ec2v1.PlatformKindKeyPair, Status: ec2v1.PlatformStatusObservedClean}
_, nc, _ := testutil.StartTestJetStream(t)
serveInventory(t, nc, func() []byte {
data, _ := json.Marshal(ec2v1.PlatformInventoryReply{AccountID: "999999999999", Kinds: []ec2v1.PlatformKindReport{clean}})
return data
})
assertAllIncomplete(t, newEC2PlatformInventoryClient(nc).PlatformInventory(testCtx(t), testAccountID, ""), ec2v1.PlatformReasonInvalidReply)
reply := normalisePlatformInventory(ec2v1.PlatformInventoryReply{AccountID: testAccountID, Kinds: []ec2v1.PlatformKindReport{
clean,
clean,
{Kind: ec2v1.PlatformKindPlacementGroup, Status: "drained"},
{Kind: ec2v1.PlatformKindTagIndex, Status: ec2v1.PlatformStatusRecordsFound},
}}, testAccountID, "")
assert.Equal(t, []ec2v1.PlatformKindReport{
{Kind: ec2v1.PlatformKindKeyPair, Status: ec2v1.PlatformStatusScanIncomplete, IncompleteReasons: []string{ec2v1.PlatformReasonInvalidReply}},
{Kind: ec2v1.PlatformKindPlacementGroup, Status: ec2v1.PlatformStatusScanIncomplete, IncompleteReasons: []string{ec2v1.PlatformReasonInvalidReply}},
{Kind: ec2v1.PlatformKindLaunchTemplate, Status: ec2v1.PlatformStatusScanIncomplete, IncompleteReasons: []string{ec2v1.PlatformReasonKindNotReported}},
{Kind: ec2v1.PlatformKindTagIndex, Status: ec2v1.PlatformStatusRecordsFound},
}, reply.Kinds)
}
// failingInventory fails the test if anything consults it.
type failingInventory struct {
t *testing.T
}
func (f failingInventory) PlatformInventory(context.Context, string, string) ec2v1.PlatformInventoryReply {
f.t.Error("an engine path called the EC2 platform inventory")
return ec2v1.PlatformInventoryReply{}
}
// The cluster wiring binds the NATS adapter, and no engine path consults it:
// precheck, inventory, dry run and a full teardown all leave it uncalled.
func TestNoEnginePathCallsTheEC2PlatformInventory(t *testing.T) {
_, nc, _ := testutil.StartTestJetStream(t)
masterKey, err := handlers_iam.GenerateMasterKey()
require.NoError(t, err)
svc, err := handlers_iam.NewIAMServiceImpl(t.Context(), nc, masterKey)
require.NoError(t, err)
cluster, err := NewClusterEngine(t.Context(), nc, 1, []string{"node-1"}, svc, newFakeBuckets())
require.NoError(t, err)
assert.IsType(t, &ec2PlatformInventoryClient{}, cluster.ec2Inventory)
reaper := &fakeReaper{kind: "key-pair", stage: StagePlatform, resources: []Resource{{Kind: "key-pair", ID: "deploy"}}}
engine := quietEngine(newFakeAccounts(), reaper)
engine.ec2Inventory = failingInventory{t: t}
require.NoError(t, engine.Precheck(testAccountID, testAccountName))
_, err = engine.Inventory(testCtx(t), testAccountID)
require.NoError(t, err)
dryRun := request()
dryRun.DryRun = true
_, err = engine.Teardown(testCtx(t), dryRun)
require.NoError(t, err)
result, err := engine.Teardown(testCtx(t), request())
require.NoError(t, err)
assert.True(t, result.AccountDeleted)
}
+4
View File
@@ -28,6 +28,10 @@ type Engine struct {
// teardown in the background can publish progress. A teardown takes minutes
// and this is the only view into it while it runs.
OnStage func(ctx context.Context, stage StageResult)
// ec2Inventory is EC2's read-only platform inventory. Bound by the cluster
// wiring but consulted by no teardown or inventory path yet.
ec2Inventory ec2PlatformInventory
}
// NewEngine builds an engine with the default timeouts.
+1
View File
@@ -950,6 +950,7 @@ func (d *Daemon) subscribeAll() error {
{"ec2.ModifyLaunchTemplate", handleNATSRequest(d.node, d.launchTemplateService.ModifyLaunchTemplate), "spinifex-workers"},
{"ec2.DescribeLaunchTemplates", handleNATSRequest(d.node, d.launchTemplateService.DescribeLaunchTemplates), "spinifex-workers"},
{"ec2.DescribeLaunchTemplateVersions", handleNATSRequest(d.node, d.launchTemplateService.DescribeLaunchTemplateVersions), "spinifex-workers"},
{ec2v1.PlatformInventorySubject, handleNATSRequest(d.node, d.platformInventory), "spinifex-workers"},
{"ec2.PutSpotInstanceRequests", handleNATSRequest(d.node, d.spotInstanceService.PutSpotInstanceRequests), "spinifex-workers"},
{"ec2.DescribeSpotInstanceRequests", handleNATSRequest(d.node, d.spotInstanceService.DescribeSpotInstanceRequests), "spinifex-workers"},
{"ec2.CancelSpotInstanceRequests", handleNATSRequest(d.node, d.spotInstanceService.CancelSpotInstanceRequests), "spinifex-workers"},
@@ -0,0 +1,16 @@
package daemon
import (
"context"
ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountinventory"
)
// platformInventory serves the read-only EC2 platform inventory. The account
// comes from the request body, not X-Account-ID, so the scanned account is
// part of the versioned contract; the header is ignored.
func (d *Daemon) platformInventory(ctx context.Context, req *ec2v1.PlatformInventoryRequest, _ string) (*ec2v1.PlatformInventoryReply, error) {
inventory := accountinventory.New(d.keyService, d.placementGroupService, d.launchTemplateService, d.tagsService)
return inventory.PlatformInventory(ctx, req)
}
@@ -0,0 +1,287 @@
package daemon
//test:in-package — the subject is served from the daemon's own subscription
// table, registered by the unexported subscribeAll.
import (
"bytes"
"encoding/base64"
"encoding/hex"
"encoding/json"
"testing"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/aws/aws-sdk-go/service/s3"
ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/internal/testkit/readonlystore"
"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
ec2key "github.com/mulgadc/spinifex/spinifex/domains/ec2/key"
ec2launchtemplate "github.com/mulgadc/spinifex/spinifex/domains/ec2/launchtemplate"
ec2placementgroup "github.com/mulgadc/spinifex/spinifex/domains/ec2/placementgroup"
ec2tags "github.com/mulgadc/spinifex/spinifex/domains/ec2/tags"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
const (
inventoryAccount = "111122223333"
inventoryOther = "444455556666"
inventoryEmpty = "777788889999"
inventoryBucket = "test-bucket"
inventoryPubKey = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIOMqqnkVzrm0SdG6UOoqKLsabgH5C9okWi0dh2l9GKJl"
inventoryUser = "#!/bin/sh\necho tenant-user-data-secret\n"
orphanTemplateID = "lt-0orphan000000000"
brokenTemplateID = "lt-0badbadbadbadbad0"
brokenKeyPairID = "key-0badbadbadbadbad0"
deletedGroupID = "pg-0deleted00000000"
)
// inventoryCluster is a daemon serving its real subscription table over owner
// stores seeded with live records and every residue class, and a separate
// client holding no IAM credential.
type inventoryCluster struct {
caller *nats.Conn
js jetstream.JetStream
live map[string]string
heldIDs []string
secrets []string
}
func newInventoryCluster(t *testing.T) *inventoryCluster {
t.Helper()
ns, nc, js := testutil.StartTestJetStream(t)
ctx := t.Context()
store := objectstore.NewMemoryObjectStore()
cfg := &config.Config{Predastore: config.PredastoreConfig{Bucket: inventoryBucket}}
tagsKV, err := ec2tags.GetOrCreateTagsBucket(ctx, js)
require.NoError(t, err)
tags := ec2tags.NewTagsServiceImplWithStore(cfg, store, tagsKV)
keys := ec2key.NewKeyServiceImplWithStore(store, inventoryBucket)
keys.SetCentralTagStore(tags)
groups, err := ec2placementgroup.NewPlacementGroupServiceImplWithNATS(ctx, nil, nc)
require.NoError(t, err)
templates, err := ec2launchtemplate.NewLaunchTemplateServiceImplWithNATS(ctx, nil, nc)
require.NoError(t, err)
c := &inventoryCluster{js: js, live: map[string]string{}}
tagSpec := func(resourceType string) []*ec2.TagSpecification {
return []*ec2.TagSpecification{{ResourceType: aws.String(resourceType),
Tags: []*ec2.Tag{{Key: aws.String("secret-tag-key"), Value: aws.String("secret-tag-value")}}}}
}
secretTags := map[string]string{"secret-tag-key": "secret-tag-value"}
for _, account := range []string{inventoryAccount, inventoryOther} {
key, err := keys.ImportKeyPair(ctx, &ec2.ImportKeyPairInput{
KeyName: aws.String("live-key-name"), PublicKeyMaterial: []byte(inventoryPubKey), TagSpecifications: tagSpec("key-pair"),
}, account)
require.NoError(t, err)
group, err := groups.CreatePlacementGroup(ctx, &ec2.CreatePlacementGroupInput{
GroupName: aws.String("live-group-name"), Strategy: aws.String(ec2.PlacementStrategySpread),
}, account)
require.NoError(t, err)
template, err := templates.CreateLaunchTemplate(ctx, &ec2.CreateLaunchTemplateInput{
LaunchTemplateName: aws.String("live-template-name"),
LaunchTemplateData: &ec2.RequestLaunchTemplateData{UserData: aws.String(base64.StdEncoding.EncodeToString([]byte(inventoryUser)))},
TagSpecifications: tagSpec("launch-template"),
}, account)
require.NoError(t, err)
if account == inventoryAccount {
c.live[aws.StringValue(key.KeyPairId)] = ec2v1.PlatformKindKeyPair
c.live[aws.StringValue(group.PlacementGroup.GroupId)] = ec2v1.PlatformKindPlacementGroup
c.live[aws.StringValue(template.LaunchTemplate.LaunchTemplateId)] = ec2v1.PlatformKindLaunchTemplate
}
}
// A group still holding a launched instance.
held, err := groups.CreatePlacementGroup(ctx, &ec2.CreatePlacementGroupInput{
GroupName: aws.String("held-group-name"), Strategy: aws.String(ec2.PlacementStrategyCluster),
}, inventoryAccount)
require.NoError(t, err)
reserved, err := groups.ReserveClusterNode(ctx, &ec2placementgroup.ReserveClusterNodeInput{
GroupName: "held-group-name", EligibleNodes: []string{"node-1"},
}, inventoryAccount)
require.NoError(t, err)
_, err = groups.FinalizeClusterInstances(ctx, &ec2placementgroup.FinalizeClusterInstancesInput{
GroupName: "held-group-name", NodeInstances: map[string][]string{reserved.TargetNode: {"i-0held0000000000"}},
}, inventoryAccount)
require.NoError(t, err)
c.heldIDs = []string{aws.StringValue(held.PlacementGroup.GroupId)}
// Owner-store residue no Describe subject reports.
prefix := "keys/" + inventoryAccount + "/"
for objectKey, body := range map[string]string{
prefix + "orphan-secret-key-name": inventoryPubKey + "\n",
prefix + brokenKeyPairID + ".json": `{"KeyName":"broken-secret-key-name","KeyPairId":`,
"keys/" + inventoryOther + "/orphan-secret-key-name": inventoryPubKey + "\n",
} {
_, err := store.PutObject(ctx, &s3.PutObjectInput{Bucket: aws.String(inventoryBucket), Key: aws.String(objectKey), Body: bytes.NewReader([]byte(body))})
require.NoError(t, err)
}
ltKV, err := js.KeyValue(ctx, ec2launchtemplate.KVBucketLaunchTemplates)
require.NoError(t, err)
orphanBody, err := json.Marshal(ec2launchtemplate.LaunchTemplateVersionRec{VersionNumber: 1,
Data: &ec2.ResponseLaunchTemplateData{UserData: aws.String(base64.StdEncoding.EncodeToString([]byte(inventoryUser)))}})
require.NoError(t, err)
orphanName := inventoryAccount + ".name." + hex.EncodeToString([]byte("orphan-secret-template-name"))
for kvKey, body := range map[string][]byte{
inventoryAccount + "." + orphanTemplateID + ".v1": orphanBody,
inventoryAccount + "." + orphanTemplateID + ".v2": orphanBody,
orphanName: []byte(orphanTemplateID),
inventoryAccount + "." + brokenTemplateID: []byte(`{"launch_template_name":"broken-secret-template-name"`),
inventoryAccount + "." + brokenTemplateID + ".v1": orphanBody,
} {
_, err := ltKV.Put(ctx, kvKey, body)
require.NoError(t, err)
}
// Tag-index entries a placement-group and launch-template delete leave.
require.NoError(t, tags.PutResourceTags(ctx, inventoryAccount, deletedGroupID, secretTags))
require.NoError(t, tags.PutResourceTags(ctx, inventoryAccount, orphanTemplateID, secretTags))
require.NoError(t, tags.PutResourceTags(ctx, inventoryAccount, brokenTemplateID, secretTags))
c.secrets = []string{
"i-0held0000000000",
"live-key-name", "live-group-name", "live-template-name", "held-group-name",
"orphan-secret-key-name", "broken-secret-key-name", "orphan-secret-template-name",
hex.EncodeToString([]byte("orphan-secret-template-name")), "broken-secret-template-name",
"secret-tag-key", "secret-tag-value", "AAAAC3NzaC1lZDI1NTE5", "tenant-user-data-secret",
base64.StdEncoding.EncodeToString([]byte(inventoryUser)), brokenKeyPairID, brokenTemplateID,
"keys/", ".json", ".name.", ".v1", ".v2", inventoryAccount + ".", "invalid character", "unexpected end",
}
// The daemon reads the object store and tag index through read-only views.
d := newSubscribeTestDaemon(t, nc, "", "")
ro := readonlystore.NewObjectStore(t, store)
d.keyService = ec2key.NewKeyServiceImplWithStore(ro, inventoryBucket)
d.tagsService = ec2tags.NewTagsServiceImplWithStore(cfg, ro, readonlystore.NewKV(t, tagsKV))
d.placementGroupService = groups
d.launchTemplateService = templates
require.NoError(t, d.subscribeAll())
c.caller, err = nats.Connect(ns.ClientURL())
require.NoError(t, err)
t.Cleanup(c.caller.Close)
return c
}
// request sends body with X-Account-ID set to header and returns the raw reply.
func (c *inventoryCluster) request(t *testing.T, body any, header string) []byte {
t.Helper()
data, err := json.Marshal(body)
require.NoError(t, err)
reply, err := natsRequestAs(c.caller, ec2v1.PlatformInventorySubject, data, header, 5*time.Second)
require.NoError(t, err)
return reply.Data
}
func (c *inventoryCluster) lastSeq(t *testing.T, bucket string) uint64 {
t.Helper()
stream, err := c.js.Stream(t.Context(), "KV_"+bucket)
require.NoError(t, err)
info, err := stream.Info(t.Context())
require.NoError(t, err)
return info.State.LastSeq
}
// The daemon answers for the account in the body, whatever X-Account-ID says,
// with live and tag-index residue IDs, owner-store residue as counts, and no
// write to any store.
func TestPlatformInventorySubjectReportsTheAccountsOwnerStores(t *testing.T) {
c := newInventoryCluster(t)
pgSeq := c.lastSeq(t, ec2placementgroup.KVBucketPlacementGroups)
ltSeq := c.lastSeq(t, ec2launchtemplate.KVBucketLaunchTemplates)
raw := c.request(t, ec2v1.PlatformInventoryRequest{AccountID: inventoryAccount, CorrelationRef: "req-1"}, inventoryOther)
require.Empty(t, replyErrCode(t, raw), string(raw))
var reply ec2v1.PlatformInventoryReply
require.NoError(t, json.Unmarshal(raw, &reply))
assert.Equal(t, inventoryAccount, reply.AccountID)
assert.Equal(t, "req-1", reply.CorrelationRef)
want := []ec2v1.PlatformItem{
{ResourceID: c.heldIDs[0], Kind: ec2v1.PlatformKindPlacementGroup, Disposition: ec2v1.PlatformDispositionDependencyHeld},
{ResourceID: deletedGroupID, Kind: ec2v1.PlatformKindTagIndex, Disposition: ec2v1.PlatformDispositionResidue},
{ResourceID: orphanTemplateID, Kind: ec2v1.PlatformKindTagIndex, Disposition: ec2v1.PlatformDispositionResidue},
}
for id, kind := range c.live {
want = append(want, ec2v1.PlatformItem{ResourceID: id, Kind: kind, Disposition: ec2v1.PlatformDispositionLive})
}
assert.ElementsMatch(t, want, reply.Items)
assert.ElementsMatch(t, []ec2v1.PlatformRecordCount{
{Kind: ec2v1.PlatformKindKeyPair, Disposition: ec2v1.PlatformDispositionUnreadable, Class: ec2v1.PlatformClassUnreadableRecord, Count: 1},
{Kind: ec2v1.PlatformKindKeyPair, Disposition: ec2v1.PlatformDispositionUnreadable, Class: ec2v1.PlatformClassPublicKeyUnresolved, Count: 1},
{Kind: ec2v1.PlatformKindLaunchTemplate, Disposition: ec2v1.PlatformDispositionUnreadable, Class: ec2v1.PlatformClassUnreadableRecord, Count: 1},
{Kind: ec2v1.PlatformKindLaunchTemplate, Disposition: ec2v1.PlatformDispositionResidue, Class: ec2v1.PlatformClassVersionBodyWithoutHeader, Count: 2},
{Kind: ec2v1.PlatformKindLaunchTemplate, Disposition: ec2v1.PlatformDispositionResidue, Class: ec2v1.PlatformClassNameIndexWithoutHeader, Count: 1},
}, reply.Counts)
require.Len(t, reply.Kinds, 4)
for _, report := range reply.Kinds {
assert.Equal(t, ec2v1.PlatformStatusRecordsFound, report.Status, report.Kind)
assert.False(t, report.ScanStartedAt.IsZero(), report.Kind)
assert.False(t, report.ScanEndedAt.Before(report.ScanStartedAt), report.Kind)
}
for _, secret := range c.secrets {
assert.NotContains(t, string(raw), secret, "the reply discloses %q", secret)
}
assert.Equal(t, pgSeq, c.lastSeq(t, ec2placementgroup.KVBucketPlacementGroups), "placement-group bucket written")
assert.Equal(t, ltSeq, c.lastSeq(t, ec2launchtemplate.KVBucketLaunchTemplates), "launch-template bucket written")
}
// An account with no records is observed clean for every kind, and another
// account's records never appear.
func TestPlatformInventorySubjectReportsAnEmptyAccountClean(t *testing.T) {
c := newInventoryCluster(t)
var reply ec2v1.PlatformInventoryReply
require.NoError(t, json.Unmarshal(c.request(t, ec2v1.PlatformInventoryRequest{AccountID: inventoryEmpty}, inventoryAccount), &reply))
assert.Empty(t, reply.Items)
assert.Empty(t, reply.Counts)
require.Len(t, reply.Kinds, 4)
for _, report := range reply.Kinds {
assert.Equal(t, ec2v1.PlatformStatusObservedClean, report.Status, report.Kind)
}
}
// A public-key object with no metadata, where no metadata is unreadable, is
// residue; the other account's records stay out of the reply.
func TestPlatformInventorySubjectCountsAnOrphanPublicKeyAsResidue(t *testing.T) {
c := newInventoryCluster(t)
raw := c.request(t, ec2v1.PlatformInventoryRequest{AccountID: inventoryOther}, inventoryAccount)
var reply ec2v1.PlatformInventoryReply
require.NoError(t, json.Unmarshal(raw, &reply))
assert.Equal(t, []ec2v1.PlatformRecordCount{{
Kind: ec2v1.PlatformKindKeyPair, Disposition: ec2v1.PlatformDispositionResidue,
Class: ec2v1.PlatformClassPublicKeyWithoutMetadata, Count: 1,
}}, reply.Counts)
require.Len(t, reply.Items, 3)
for _, item := range reply.Items {
assert.Equal(t, ec2v1.PlatformDispositionLive, item.Disposition)
assert.NotContains(t, c.live, item.ResourceID)
}
for _, id := range append(c.heldIDs, deletedGroupID, orphanTemplateID) {
assert.NotContains(t, string(raw), id)
}
}
// A request without a valid account in its body is refused, not scanned.
func TestPlatformInventorySubjectRefusesARequestWithoutAnAccount(t *testing.T) {
c := newInventoryCluster(t)
assert.Equal(t, awserrors.ErrorMissingParameter, replyErrCode(t, c.request(t, ec2v1.PlatformInventoryRequest{}, inventoryAccount)))
assert.Equal(t, awserrors.ErrorInvalidParameterValue, replyErrCode(t, c.request(t, ec2v1.PlatformInventoryRequest{AccountID: "1111"}, "")))
}
// A daemon that does not serve the subject leaves the request with no responder.
func TestPlatformInventorySubjectHasNoResponderWithoutTheDaemon(t *testing.T) {
_, nc, _ := testutil.StartTestJetStream(t)
_, err := natsRequestAs(nc, ec2v1.PlatformInventorySubject, []byte(`{"account_id":"111122223333"}`), "", time.Second)
require.ErrorIs(t, err, nats.ErrNoResponders)
}
@@ -0,0 +1,259 @@
// Package accountinventory reports one account's EC2 key pairs, placement
// groups and launch templates from their owners' stores. It composes the
// resource packages' read-only scans and owns the dependency knowledge
// between them; it never writes to any store.
package accountinventory
import (
"context"
"log/slog"
"slices"
"strings"
"time"
ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
ec2key "github.com/mulgadc/spinifex/spinifex/domains/ec2/key"
ec2launchtemplate "github.com/mulgadc/spinifex/spinifex/domains/ec2/launchtemplate"
ec2placementgroup "github.com/mulgadc/spinifex/spinifex/domains/ec2/placementgroup"
ec2tags "github.com/mulgadc/spinifex/spinifex/domains/ec2/tags"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
otelsetup "github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
)
// KeyPairScanner reads one account's key-pair objects.
type KeyPairScanner interface {
ScanAccount(ctx context.Context, accountID string) (*ec2key.AccountScan, error)
}
// PlacementGroupScanner reads one account's placement-group records.
type PlacementGroupScanner interface {
ScanAccount(ctx context.Context, accountID string) (*ec2placementgroup.AccountScan, error)
}
// LaunchTemplateScanner reads one account's launch-template records.
type LaunchTemplateScanner interface {
ScanAccount(ctx context.Context, accountID string) (*ec2launchtemplate.AccountScan, error)
}
// TagIndexScanner lists one account's tagged resource IDs by prefix.
type TagIndexScanner interface {
ScanAccountResourceIDs(ctx context.Context, accountID string, prefixes ...string) ([]string, error)
}
var (
_ KeyPairScanner = (*ec2key.KeyServiceImpl)(nil)
_ PlacementGroupScanner = (*ec2placementgroup.PlacementGroupServiceImpl)(nil)
_ LaunchTemplateScanner = (*ec2launchtemplate.LaunchTemplateServiceImpl)(nil)
_ TagIndexScanner = (*ec2tags.TagsServiceImpl)(nil)
)
// Resource-ID prefixes of the three kinds in the tag index.
const (
keyPairIDPrefix = "key-"
placementGroupIDPrefix = "pg-"
launchTemplateIDPrefix = "lt-"
)
// Inventory answers platform-inventory requests from the owner stores.
type Inventory struct {
keys KeyPairScanner
groups PlacementGroupScanner
templates LaunchTemplateScanner
tags TagIndexScanner
now func() time.Time
}
// New builds an inventory over the four owner scans.
func New(keys KeyPairScanner, groups PlacementGroupScanner, templates LaunchTemplateScanner, tags TagIndexScanner) *Inventory {
return &Inventory{keys: keys, groups: groups, templates: templates, tags: tags, now: time.Now}
}
// owned is the set of IDs whose owner record exists, live or unreadable, for
// one kind; resolved is false when an owner record exists whose ID is unknown.
type owned struct {
ids map[string]bool
resolved bool
}
// kindScan accumulates one kind's report while its stores are read.
type kindScan struct {
report ec2v1.PlatformKindReport
found bool
}
func (i *Inventory) startKind(kind string) *kindScan {
return &kindScan{report: ec2v1.PlatformKindReport{Kind: kind, ScanStartedAt: i.now().UTC()}}
}
func (k *kindScan) incomplete(reason string) {
if !slices.Contains(k.report.IncompleteReasons, reason) {
k.report.IncompleteReasons = append(k.report.IncompleteReasons, reason)
}
}
func (i *Inventory) finishKind(k *kindScan) ec2v1.PlatformKindReport {
k.report.ScanEndedAt = i.now().UTC()
switch {
case len(k.report.IncompleteReasons) > 0:
k.report.Status = ec2v1.PlatformStatusScanIncomplete
case k.found:
k.report.Status = ec2v1.PlatformStatusRecordsFound
default:
k.report.Status = ec2v1.PlatformStatusObservedClean
}
return k.report
}
// PlatformInventory scans the account's tag index first and then each owner
// store, so a resource created mid-scan shows as live rather than as a
// tag-index entry with no owner record.
func (i *Inventory) PlatformInventory(ctx context.Context, req *ec2v1.PlatformInventoryRequest) (*ec2v1.PlatformInventoryReply, error) {
if req == nil || req.AccountID == "" {
return nil, awserrors.Errorf(awserrors.ErrorMissingParameter, "The request must specify an account ID.")
}
if !awsidentifiers.IsAccountID(req.AccountID) {
return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "The account ID must be 12 digits.")
}
start := time.Now()
reply := &ec2v1.PlatformInventoryReply{AccountID: req.AccountID, CorrelationRef: req.CorrelationRef}
tagScan := i.startKind(ec2v1.PlatformKindTagIndex)
tagged, tagErr := i.tags.ScanAccountResourceIDs(ctx, req.AccountID, keyPairIDPrefix, placementGroupIDPrefix, launchTemplateIDPrefix)
if tagErr != nil {
slog.WarnContext(ctx, "Platform inventory: tag index listing failed", "err", tagErr)
tagScan.incomplete(ec2v1.PlatformReasonStoreListFailed)
}
tagScan.found = len(tagged) > 0
keyReport, keyOwned := i.scanKeyPairs(ctx, req.AccountID, reply)
groupReport, groupOwned := i.scanPlacementGroups(ctx, req.AccountID, reply)
templateReport, templateOwned := i.scanLaunchTemplates(ctx, req.AccountID, reply)
byPrefix := map[string]owned{
keyPairIDPrefix: keyOwned,
placementGroupIDPrefix: groupOwned,
launchTemplateIDPrefix: templateOwned,
}
for _, id := range tagged {
parent := byPrefix[idPrefix(id)]
switch {
case parent.ids[id]:
case !parent.resolved:
tagScan.incomplete(ec2v1.PlatformReasonOwnerRecordsUnresolved)
default:
reply.Items = append(reply.Items, ec2v1.PlatformItem{
ResourceID: id, Kind: ec2v1.PlatformKindTagIndex, Disposition: ec2v1.PlatformDispositionResidue,
})
}
}
tagReport := i.finishKind(tagScan)
reply.Kinds = []ec2v1.PlatformKindReport{keyReport, groupReport, templateReport, tagReport}
slog.InfoContext(ctx, "Platform inventory completed", "accountID", req.AccountID,
"correlationRef", req.CorrelationRef, "items", len(reply.Items), "duration_ms", otelsetup.Millis(time.Since(start)))
return reply, nil
}
func idPrefix(id string) string {
for _, prefix := range []string{keyPairIDPrefix, placementGroupIDPrefix, launchTemplateIDPrefix} {
if strings.HasPrefix(id, prefix) {
return prefix
}
}
return ""
}
func (i *Inventory) scanKeyPairs(ctx context.Context, accountID string, reply *ec2v1.PlatformInventoryReply) (ec2v1.PlatformKindReport, owned) {
kind := i.startKind(ec2v1.PlatformKindKeyPair)
scan, err := i.keys.ScanAccount(ctx, accountID)
if err != nil {
slog.WarnContext(ctx, "Platform inventory: key-pair listing failed", "err", err)
kind.incomplete(ec2v1.PlatformReasonStoreListFailed)
return i.finishKind(kind), owned{}
}
ids := make(map[string]bool)
for _, id := range scan.LiveKeyPairIDs {
ids[id] = true
addItem(reply, kind, id, ec2v1.PlatformDispositionLive)
}
for _, id := range scan.UnreadableKeyPairIDs {
ids[id] = true
}
addCount(reply, kind, ec2v1.PlatformDispositionUnreadable, ec2v1.PlatformClassUnreadableRecord, len(scan.UnreadableKeyPairIDs))
addCount(reply, kind, ec2v1.PlatformDispositionResidue, ec2v1.PlatformClassPublicKeyWithoutMetadata, scan.OrphanPublicKeys)
addCount(reply, kind, ec2v1.PlatformDispositionUnreadable, ec2v1.PlatformClassPublicKeyUnresolved, scan.UnresolvedPublicKeys)
if scan.ReadFailures > 0 {
kind.incomplete(ec2v1.PlatformReasonRecordReadFailed)
}
return i.finishKind(kind), owned{ids: ids, resolved: true}
}
// scanPlacementGroups reports a group whose member list still names instances
// as dependency-held: it cannot be deleted until those instances leave it.
func (i *Inventory) scanPlacementGroups(ctx context.Context, accountID string, reply *ec2v1.PlatformInventoryReply) (ec2v1.PlatformKindReport, owned) {
kind := i.startKind(ec2v1.PlatformKindPlacementGroup)
scan, err := i.groups.ScanAccount(ctx, accountID)
if err != nil {
slog.WarnContext(ctx, "Platform inventory: placement-group listing failed", "err", err)
kind.incomplete(ec2v1.PlatformReasonStoreListFailed)
return i.finishKind(kind), owned{}
}
ids := make(map[string]bool)
for _, group := range scan.Groups {
ids[group.GroupID] = true
if group.MemberCount > 0 {
addItem(reply, kind, group.GroupID, ec2v1.PlatformDispositionDependencyHeld)
continue
}
addItem(reply, kind, group.GroupID, ec2v1.PlatformDispositionLive)
}
addCount(reply, kind, ec2v1.PlatformDispositionUnreadable, ec2v1.PlatformClassUnreadableRecord, scan.Unreadable)
if scan.ReadFailures > 0 {
kind.incomplete(ec2v1.PlatformReasonRecordReadFailed)
}
return i.finishKind(kind), owned{ids: ids, resolved: scan.Unreadable == 0}
}
func (i *Inventory) scanLaunchTemplates(ctx context.Context, accountID string, reply *ec2v1.PlatformInventoryReply) (ec2v1.PlatformKindReport, owned) {
kind := i.startKind(ec2v1.PlatformKindLaunchTemplate)
scan, err := i.templates.ScanAccount(ctx, accountID)
if err != nil {
slog.WarnContext(ctx, "Platform inventory: launch-template listing failed", "err", err)
kind.incomplete(ec2v1.PlatformReasonStoreListFailed)
return i.finishKind(kind), owned{}
}
ids := make(map[string]bool)
for _, id := range scan.LiveTemplateIDs {
ids[id] = true
addItem(reply, kind, id, ec2v1.PlatformDispositionLive)
}
for _, id := range scan.UnreadableTemplateIDs {
ids[id] = true
}
addCount(reply, kind, ec2v1.PlatformDispositionUnreadable, ec2v1.PlatformClassUnreadableRecord, len(scan.UnreadableTemplateIDs)+scan.UnreadableNameIndexEntries)
addCount(reply, kind, ec2v1.PlatformDispositionUnreadable, ec2v1.PlatformClassUnrecognisedTemplateEntry, scan.UnrecognisedEntries)
addCount(reply, kind, ec2v1.PlatformDispositionResidue, ec2v1.PlatformClassVersionBodyWithoutHeader, scan.OrphanVersionBodies)
addCount(reply, kind, ec2v1.PlatformDispositionResidue, ec2v1.PlatformClassNameIndexWithoutHeader, scan.OrphanNameIndexEntries)
if scan.ReadFailures > 0 {
kind.incomplete(ec2v1.PlatformReasonRecordReadFailed)
}
return i.finishKind(kind), owned{ids: ids, resolved: true}
}
func addItem(reply *ec2v1.PlatformInventoryReply, kind *kindScan, id, disposition string) {
kind.found = true
reply.Items = append(reply.Items, ec2v1.PlatformItem{
ResourceID: id, Kind: kind.report.Kind, Disposition: disposition,
})
}
func addCount(reply *ec2v1.PlatformInventoryReply, kind *kindScan, disposition, class string, n int) {
if n == 0 {
return
}
kind.found = true
reply.Counts = append(reply.Counts, ec2v1.PlatformRecordCount{
Kind: kind.report.Kind, Disposition: disposition, Class: class, Count: n,
})
}
@@ -0,0 +1,250 @@
package accountinventory
//test:in-package — the scan clock is injected to pin the per-kind timestamps.
import (
"context"
"errors"
"testing"
"time"
ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
ec2key "github.com/mulgadc/spinifex/spinifex/domains/ec2/key"
ec2launchtemplate "github.com/mulgadc/spinifex/spinifex/domains/ec2/launchtemplate"
ec2placementgroup "github.com/mulgadc/spinifex/spinifex/domains/ec2/placementgroup"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
const testAccount = "111122223333"
var errList = errors.New("injected listing failure")
type fakeKeys struct {
scan *ec2key.AccountScan
err error
}
func (f fakeKeys) ScanAccount(context.Context, string) (*ec2key.AccountScan, error) {
return f.scan, f.err
}
type fakeGroups struct {
scan *ec2placementgroup.AccountScan
err error
}
func (f fakeGroups) ScanAccount(context.Context, string) (*ec2placementgroup.AccountScan, error) {
return f.scan, f.err
}
type fakeTemplates struct {
scan *ec2launchtemplate.AccountScan
err error
}
func (f fakeTemplates) ScanAccount(context.Context, string) (*ec2launchtemplate.AccountScan, error) {
return f.scan, f.err
}
type fakeTags struct {
ids []string
err error
prefixes []string
}
func (f *fakeTags) ScanAccountResourceIDs(_ context.Context, _ string, prefixes ...string) ([]string, error) {
f.prefixes = prefixes
return f.ids, f.err
}
// fixture is an account holding every class each scan can report.
type fixture struct {
keys fakeKeys
groups fakeGroups
templates fakeTemplates
tags *fakeTags
}
func emptyFixture() fixture {
return fixture{
keys: fakeKeys{scan: &ec2key.AccountScan{}},
groups: fakeGroups{scan: &ec2placementgroup.AccountScan{}},
templates: fakeTemplates{scan: &ec2launchtemplate.AccountScan{}},
tags: &fakeTags{},
}
}
func (f fixture) run(t *testing.T) *ec2v1.PlatformInventoryReply {
t.Helper()
inv := New(f.keys, f.groups, f.templates, f.tags)
tick := time.Date(2026, 10, 16, 0, 0, 0, 0, time.UTC)
inv.now = func() time.Time { tick = tick.Add(time.Second); return tick }
reply, err := inv.PlatformInventory(context.Background(), &ec2v1.PlatformInventoryRequest{AccountID: testAccount, CorrelationRef: "req-1"})
require.NoError(t, err)
return reply
}
func kindsByName(reply *ec2v1.PlatformInventoryReply) map[string]ec2v1.PlatformKindReport {
out := make(map[string]ec2v1.PlatformKindReport)
for _, k := range reply.Kinds {
out[k.Kind] = k
}
return out
}
// An account with nothing in any store is observed clean for every kind, each
// with its own scan window, and asks the tag index for the three kinds only.
func TestEmptyAccountIsObservedCleanForEveryKind(t *testing.T) {
f := emptyFixture()
reply := f.run(t)
assert.Equal(t, testAccount, reply.AccountID)
assert.Equal(t, "req-1", reply.CorrelationRef)
assert.Empty(t, reply.Items)
assert.Empty(t, reply.Counts)
require.Len(t, reply.Kinds, 4)
for i, kind := range ec2v1.PlatformKinds() {
report := reply.Kinds[i]
assert.Equal(t, kind, report.Kind)
assert.Equal(t, ec2v1.PlatformStatusObservedClean, report.Status, kind)
assert.Empty(t, report.IncompleteReasons, kind)
assert.True(t, report.ScanEndedAt.After(report.ScanStartedAt), kind)
}
assert.ElementsMatch(t, []string{"key-", "pg-", "lt-"}, f.tags.prefixes)
}
// Live and dependency-held items carry IDs; owner-store residue and unreadable
// records are counts; a tag-index entry is residue only when its owner record
// is absent, not when it is live or unreadable.
func TestInventoryClassifiesEveryRecordClass(t *testing.T) {
f := emptyFixture()
f.keys.scan = &ec2key.AccountScan{
LiveKeyPairIDs: []string{"key-0live"}, UnreadableKeyPairIDs: []string{"key-0broken"}, UnresolvedPublicKeys: 2,
}
f.groups.scan = &ec2placementgroup.AccountScan{Groups: []ec2placementgroup.ScannedGroup{
{GroupID: "pg-0live"},
{GroupID: "pg-0held", MemberCount: 2},
}}
f.templates.scan = &ec2launchtemplate.AccountScan{
LiveTemplateIDs: []string{"lt-0live"}, UnreadableTemplateIDs: []string{"lt-0broken"},
OrphanVersionBodies: 3, OrphanNameIndexEntries: 2, UnreadableNameIndexEntries: 1, UnrecognisedEntries: 1,
}
f.tags.ids = []string{"key-0broken", "key-0gone", "key-0live", "lt-0broken", "lt-0gone", "pg-0gone", "pg-0held"}
reply := f.run(t)
assert.ElementsMatch(t, []ec2v1.PlatformItem{
{ResourceID: "key-0live", Kind: ec2v1.PlatformKindKeyPair, Disposition: ec2v1.PlatformDispositionLive},
{ResourceID: "pg-0live", Kind: ec2v1.PlatformKindPlacementGroup, Disposition: ec2v1.PlatformDispositionLive},
{ResourceID: "pg-0held", Kind: ec2v1.PlatformKindPlacementGroup, Disposition: ec2v1.PlatformDispositionDependencyHeld},
{ResourceID: "lt-0live", Kind: ec2v1.PlatformKindLaunchTemplate, Disposition: ec2v1.PlatformDispositionLive},
{ResourceID: "key-0gone", Kind: ec2v1.PlatformKindTagIndex, Disposition: ec2v1.PlatformDispositionResidue},
{ResourceID: "lt-0gone", Kind: ec2v1.PlatformKindTagIndex, Disposition: ec2v1.PlatformDispositionResidue},
{ResourceID: "pg-0gone", Kind: ec2v1.PlatformKindTagIndex, Disposition: ec2v1.PlatformDispositionResidue},
}, reply.Items)
assert.ElementsMatch(t, []ec2v1.PlatformRecordCount{
{Kind: ec2v1.PlatformKindKeyPair, Disposition: ec2v1.PlatformDispositionUnreadable, Class: ec2v1.PlatformClassUnreadableRecord, Count: 1},
{Kind: ec2v1.PlatformKindKeyPair, Disposition: ec2v1.PlatformDispositionUnreadable, Class: ec2v1.PlatformClassPublicKeyUnresolved, Count: 2},
{Kind: ec2v1.PlatformKindLaunchTemplate, Disposition: ec2v1.PlatformDispositionUnreadable, Class: ec2v1.PlatformClassUnreadableRecord, Count: 2},
{Kind: ec2v1.PlatformKindLaunchTemplate, Disposition: ec2v1.PlatformDispositionUnreadable, Class: ec2v1.PlatformClassUnrecognisedTemplateEntry, Count: 1},
{Kind: ec2v1.PlatformKindLaunchTemplate, Disposition: ec2v1.PlatformDispositionResidue, Class: ec2v1.PlatformClassVersionBodyWithoutHeader, Count: 3},
{Kind: ec2v1.PlatformKindLaunchTemplate, Disposition: ec2v1.PlatformDispositionResidue, Class: ec2v1.PlatformClassNameIndexWithoutHeader, Count: 2},
}, reply.Counts)
for _, report := range reply.Kinds {
assert.Equal(t, ec2v1.PlatformStatusRecordsFound, report.Status, report.Kind)
}
}
// Orphan public keys are residue when no metadata is unreadable.
func TestInventoryCountsOrphanPublicKeysAsResidue(t *testing.T) {
f := emptyFixture()
f.keys.scan = &ec2key.AccountScan{OrphanPublicKeys: 2}
reply := f.run(t)
assert.Equal(t, []ec2v1.PlatformRecordCount{{
Kind: ec2v1.PlatformKindKeyPair, Disposition: ec2v1.PlatformDispositionResidue,
Class: ec2v1.PlatformClassPublicKeyWithoutMetadata, Count: 2,
}}, reply.Counts)
assert.Empty(t, reply.Items)
assert.Equal(t, ec2v1.PlatformStatusRecordsFound, kindsByName(reply)[ec2v1.PlatformKindKeyPair].Status)
}
// An unreadable placement group's ID is unknown, so a pg- tag entry without a
// readable owner cannot be called residue and the tag index is incomplete.
func TestUnreadablePlacementGroupLeavesItsTagEntriesUnresolved(t *testing.T) {
f := emptyFixture()
f.groups.scan = &ec2placementgroup.AccountScan{Unreadable: 1}
f.tags.ids = []string{"lt-0gone", "pg-0maybe"}
reply := f.run(t)
assert.Equal(t, []ec2v1.PlatformItem{
{ResourceID: "lt-0gone", Kind: ec2v1.PlatformKindTagIndex, Disposition: ec2v1.PlatformDispositionResidue},
}, reply.Items)
tagIndex := kindsByName(reply)[ec2v1.PlatformKindTagIndex]
assert.Equal(t, ec2v1.PlatformStatusScanIncomplete, tagIndex.Status)
assert.Equal(t, []string{ec2v1.PlatformReasonOwnerRecordsUnresolved}, tagIndex.IncompleteReasons)
assert.Equal(t, ec2v1.PlatformStatusRecordsFound, kindsByName(reply)[ec2v1.PlatformKindPlacementGroup].Status)
}
// A failed listing or a failed record read makes that kind scan-incomplete with
// a generic reason, and a listing failure also leaves its tag entries unresolved.
func TestStoreFailuresAreScanIncomplete(t *testing.T) {
tests := []struct {
name string
fault func(f *fixture)
kind string
reasons []string
}{
{"key listing", func(f *fixture) { f.keys.err = errList }, ec2v1.PlatformKindKeyPair, []string{ec2v1.PlatformReasonStoreListFailed}},
{"group listing", func(f *fixture) { f.groups.err = errList }, ec2v1.PlatformKindPlacementGroup, []string{ec2v1.PlatformReasonStoreListFailed}},
{"template listing", func(f *fixture) { f.templates.err = errList }, ec2v1.PlatformKindLaunchTemplate, []string{ec2v1.PlatformReasonStoreListFailed}},
{"tag listing", func(f *fixture) { f.tags.err = errList }, ec2v1.PlatformKindTagIndex, []string{ec2v1.PlatformReasonStoreListFailed}},
{"key read", func(f *fixture) { f.keys.scan.ReadFailures = 1 }, ec2v1.PlatformKindKeyPair, []string{ec2v1.PlatformReasonRecordReadFailed}},
{"group read", func(f *fixture) { f.groups.scan.ReadFailures = 1 }, ec2v1.PlatformKindPlacementGroup, []string{ec2v1.PlatformReasonRecordReadFailed}},
{"template read", func(f *fixture) { f.templates.scan.ReadFailures = 1 }, ec2v1.PlatformKindLaunchTemplate, []string{ec2v1.PlatformReasonRecordReadFailed}},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
f := emptyFixture()
tc.fault(&f)
reply := f.run(t)
for _, report := range reply.Kinds {
if report.Kind == tc.kind {
assert.Equal(t, ec2v1.PlatformStatusScanIncomplete, report.Status)
assert.Equal(t, tc.reasons, report.IncompleteReasons)
continue
}
assert.Equal(t, ec2v1.PlatformStatusObservedClean, report.Status, report.Kind)
}
})
}
for _, prefix := range []string{"key-0x", "pg-0x", "lt-0x"} {
f := emptyFixture()
f.keys.err, f.groups.err, f.templates.err = errList, errList, errList
f.tags.ids = []string{prefix}
reply := f.run(t)
assert.Empty(t, reply.Items, prefix)
assert.Equal(t, []string{ec2v1.PlatformReasonOwnerRecordsUnresolved}, kindsByName(reply)[ec2v1.PlatformKindTagIndex].IncompleteReasons, prefix)
}
}
func TestInventoryRejectsAMissingOrMalformedAccount(t *testing.T) {
inv := New(emptyFixture().keys, emptyFixture().groups, emptyFixture().templates, &fakeTags{})
for account, code := range map[string]string{
"": awserrors.ErrorMissingParameter,
"1111": awserrors.ErrorInvalidParameterValue,
"1111.lt-": awserrors.ErrorInvalidParameterValue,
"11112222333a": awserrors.ErrorInvalidParameterValue,
"1111222233334": awserrors.ErrorInvalidParameterValue,
} {
_, err := inv.PlatformInventory(context.Background(), &ec2v1.PlatformInventoryRequest{AccountID: account})
got, _ := awserrors.ResolveErrorCode(err)
assert.Equal(t, code, got, account)
}
_, err := inv.PlatformInventory(context.Background(), nil)
got, _ := awserrors.ResolveErrorCode(err)
assert.Equal(t, awserrors.ErrorMissingParameter, got)
}
+126
View File
@@ -0,0 +1,126 @@
package key
import (
"context"
"fmt"
"io"
"log/slog"
"strings"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/s3"
"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
)
// AccountScan is a read-only classification of every object under one
// account's key prefix. It holds key-pair IDs and counts only; object keys,
// key names and object bodies never leave the scan.
type AccountScan struct {
// LiveKeyPairIDs are metadata records DescribeKeyPairs reports.
LiveKeyPairIDs []string
// UnreadableKeyPairIDs are metadata records found but not read or decoded.
UnreadableKeyPairIDs []string
// OrphanPublicKeys are public-key objects with no live metadata naming them.
OrphanPublicKeys int
// UnresolvedPublicKeys are public-key objects that may belong to unreadable
// metadata, so they cannot be called residue.
UnresolvedPublicKeys int
// ReadFailures counts objects the store failed to return.
ReadFailures int
}
// ScanAccount lists and classifies every object under keys/<accountID>/. A
// listing failure is returned as an error; per-object read failures are
// counted. It reads only objects named as metadata (key-*.json).
func (s *KeyServiceImpl) ScanAccount(ctx context.Context, accountID string) (*AccountScan, error) {
prefix := fmt.Sprintf("keys/%s/", accountID)
objects, _, err := objectstore.ListAll(ctx, s.store, &s3.ListObjectsV2Input{
Bucket: aws.String(s.bucketName),
Prefix: aws.String(prefix),
})
if err != nil {
return nil, fmt.Errorf("list key-pair objects: %w", err)
}
scan := &AccountScan{}
liveNames := make(map[string]bool)
var rest []string
unreadable := make(map[string]bool)
for _, obj := range objects {
if obj == nil || obj.Key == nil {
continue
}
name := strings.TrimPrefix(*obj.Key, prefix)
id, isJSON := strings.CutSuffix(name, ".json")
if !isJSON || !strings.HasPrefix(id, "key-") {
rest = append(rest, name)
continue
}
metadata, found, readErr := s.readScanMetadata(ctx, *obj.Key)
switch {
case readErr != nil:
scan.ReadFailures++
unreadable[name] = true
case !found:
// Removed between the listing and the read.
case metadata == nil || aws.StringValue(metadata.KeyPairId) != id || aws.StringValue(metadata.KeyName) == "":
unreadable[name] = true
default:
scan.LiveKeyPairIDs = append(scan.LiveKeyPairIDs, id)
liveNames[*metadata.KeyName] = true
}
if unreadable[name] {
rest = append(rest, name)
}
}
// A .json object that is the public key of a live key pair named *.json is
// not unreadable metadata, so live names are settled before anything else.
for name := range unreadable {
if liveNames[name] {
delete(unreadable, name)
}
}
for _, name := range rest {
switch {
case liveNames[name]:
// The public key of a live key pair.
case unreadable[name]:
scan.UnreadableKeyPairIDs = append(scan.UnreadableKeyPairIDs, strings.TrimSuffix(name, ".json"))
case len(unreadable) > 0:
scan.UnresolvedPublicKeys++
default:
scan.OrphanPublicKeys++
}
}
return scan, nil
}
// readScanMetadata reads and decodes one metadata object. A decode failure is
// reported as a nil record, not an error, and its text is discarded because a
// decoder message can quote the stored bytes.
func (s *KeyServiceImpl) readScanMetadata(ctx context.Context, objectKey string) (*keyPairMetadata, bool, error) {
out, err := s.store.GetObject(ctx, &s3.GetObjectInput{
Bucket: aws.String(s.bucketName),
Key: aws.String(objectKey),
})
if err != nil {
if objectstore.IsNoSuchKeyError(err) {
return nil, false, nil
}
slog.WarnContext(ctx, "Key-pair account scan: metadata read failed", "err", err)
return nil, true, err
}
body, err := io.ReadAll(out.Body)
if closeErr := out.Body.Close(); closeErr != nil {
slog.DebugContext(ctx, "Key-pair account scan: metadata close failed", "err", closeErr)
}
if err != nil {
slog.WarnContext(ctx, "Key-pair account scan: metadata body read failed", "err", err)
return nil, true, err
}
if metadata, decodeErr := decodeKeyPairMetadata(body); decodeErr == nil {
return metadata, true, nil
}
return nil, true, nil
}
@@ -0,0 +1,144 @@
package key_test
import (
"context"
"strings"
"testing"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/aws/aws-sdk-go/service/s3"
"github.com/mulgadc/spinifex/internal/testkit/readonlystore"
ec2key "github.com/mulgadc/spinifex/spinifex/domains/ec2/key"
"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// scanner returns a key service over a read-only view of store, so the scan
// fails the test on any write.
func scanner(t *testing.T, store objectstore.ObjectStore) (*ec2key.KeyServiceImpl, *readonlystore.ObjectStore) {
t.Helper()
ro := readonlystore.NewObjectStore(t, store)
return ec2key.NewKeyServiceImplWithStore(ro, ctBucket), ro
}
// Every residue class the delete and create paths leave is classified, and
// a live key pair whose name ends in .json is not mistaken for metadata.
func TestScanAccountClassifiesEveryObjectUnderTheAccountPrefix(t *testing.T) {
store := newFaultyStore()
svc := ec2key.NewKeyServiceImplWithStore(store, ctBucket)
prefix := "keys/" + ctAccount + "/"
live := importKey(t, svc, ctAccount, "live")
dotted := importKey(t, svc, ctAccount, "web.json")
importKey(t, svc, ctAccount, "deleted")
store.failDelete = func(key string) bool { return key == prefix+"deleted" }
_, err := svc.DeleteKeyPair(context.Background(), &ec2.DeleteKeyPairInput{KeyName: aws.String("deleted")}, ctAccount)
require.NoError(t, err)
store.failDelete = nil
store.failPut = func(key string) bool { return strings.HasSuffix(key, ".json") }
store.failDelete = func(string) bool { return true }
_, err = svc.CreateKeyPair(context.Background(), &ec2.CreateKeyPairInput{KeyName: aws.String("half")}, ctAccount)
require.Error(t, err)
store.failPut, store.failDelete = nil, nil
scanSvc, ro := scanner(t, store)
scan, err := scanSvc.ScanAccount(context.Background(), ctAccount)
require.NoError(t, err)
assert.ElementsMatch(t, []string{aws.StringValue(live.KeyPairId), aws.StringValue(dotted.KeyPairId)}, scan.LiveKeyPairIDs)
assert.Equal(t, 2, scan.OrphanPublicKeys, "the failed-delete and failed-create public keys")
assert.Empty(t, scan.UnreadableKeyPairIDs)
assert.Zero(t, scan.UnresolvedPublicKeys)
assert.Zero(t, scan.ReadFailures)
for _, key := range ro.Gets() {
assert.True(t, strings.HasPrefix(key, prefix+"key-") && strings.HasSuffix(key, ".json"), "the scan read %s, which is not metadata", key)
}
}
// Metadata the store cannot return is unreadable and a read failure; metadata
// that does not decode, or names another ID, is unreadable. While any metadata
// is unreadable, an unmatched public key may be its own, so it is unresolved.
func TestScanAccountReportsUnreadableMetadata(t *testing.T) {
store := newFaultyStore()
svc := ec2key.NewKeyServiceImplWithStore(store, ctBucket)
prefix := "keys/" + ctAccount + "/"
importKey(t, svc, ctAccount, "live")
putObject(t, store, prefix+"key-0badbadbadbadbad0.json", "{not json")
putObject(t, store, prefix+"key-0mismatch00000000.json", `{"KeyName":"other","KeyPairId":"key-0different0000000"}`)
putObject(t, store, prefix+"key-0unreadable000000.json", `{"KeyName":"hidden","KeyPairId":"key-0unreadable000000"}`)
putObject(t, store, prefix+"hidden", ctPubKey)
store.failGet = func(key string) bool { return key == prefix+"key-0unreadable000000.json" }
scanSvc, _ := scanner(t, store)
scan, err := scanSvc.ScanAccount(context.Background(), ctAccount)
require.NoError(t, err)
assert.Len(t, scan.LiveKeyPairIDs, 1)
assert.ElementsMatch(t, []string{"key-0badbadbadbadbad0", "key-0mismatch00000000", "key-0unreadable000000"}, scan.UnreadableKeyPairIDs)
assert.Equal(t, 1, scan.ReadFailures)
assert.Equal(t, 1, scan.UnresolvedPublicKeys)
assert.Zero(t, scan.OrphanPublicKeys)
}
// A metadata object removed between the listing and the read is gone, not unreadable.
func TestScanAccountSkipsMetadataRemovedAfterTheListing(t *testing.T) {
store := &vanishingStore{faultyStore: newFaultyStore()}
putObject(t, store, "keys/"+ctAccount+"/key-0gone000000000000.json", "{}")
store.vanish = "keys/" + ctAccount + "/key-0gone000000000000.json"
scan, err := ec2key.NewKeyServiceImplWithStore(store, ctBucket).ScanAccount(context.Background(), ctAccount)
require.NoError(t, err)
assert.Equal(t, &ec2key.AccountScan{}, scan)
}
// The account prefix ends at its slash, so account 1111 never sees 11112's objects.
func TestScanAccountStaysInsideTheAccountPrefix(t *testing.T) {
store := newFaultyStore()
svc := ec2key.NewKeyServiceImplWithStore(store, ctBucket)
importKey(t, svc, "11112", "neighbour")
putObject(t, store, "keys/11112/orphan", ctPubKey)
scanSvc, _ := scanner(t, store)
scan, err := scanSvc.ScanAccount(context.Background(), "1111")
require.NoError(t, err)
assert.Equal(t, &ec2key.AccountScan{}, scan)
scan, err = scanSvc.ScanAccount(context.Background(), "11112")
require.NoError(t, err)
assert.Len(t, scan.LiveKeyPairIDs, 1)
assert.Equal(t, 1, scan.OrphanPublicKeys)
}
// A listing failure is an error, never an empty scan.
func TestScanAccountReturnsAListingFailure(t *testing.T) {
store := &listFailingStore{MemoryObjectStore: objectstore.NewMemoryObjectStore()}
_, err := ec2key.NewKeyServiceImplWithStore(store, ctBucket).ScanAccount(context.Background(), ctAccount)
require.ErrorIs(t, err, errInjected)
}
type listFailingStore struct {
*objectstore.MemoryObjectStore
}
func (s *listFailingStore) ListObjectsV2(context.Context, *s3.ListObjectsV2Input) (*s3.ListObjectsV2Output, error) {
return nil, errInjected
}
// vanishingStore deletes vanish just before it is read, as a concurrent delete would.
type vanishingStore struct {
*faultyStore
vanish string
}
func (s *vanishingStore) GetObject(ctx context.Context, in *s3.GetObjectInput) (*s3.GetObjectOutput, error) {
if aws.StringValue(in.Key) == s.vanish {
_, _ = s.DeleteObject(ctx, &s3.DeleteObjectInput{Bucket: in.Bucket, Key: in.Key})
}
return s.faultyStore.GetObject(ctx, in)
}
@@ -0,0 +1,119 @@
package launchtemplate
import (
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"strconv"
"strings"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/nats-io/nats.go/jetstream"
)
// AccountScan is a read-only classification of every record under one
// account's prefix: headers, name index and version bodies. It holds template
// IDs and counts only; names, raw keys and record bodies never leave the scan.
type AccountScan struct {
// LiveTemplateIDs are headers DescribeLaunchTemplates reports.
LiveTemplateIDs []string
// UnreadableTemplateIDs are headers found but not read or decoded.
UnreadableTemplateIDs []string
// OrphanVersionBodies are version bodies whose header key is absent.
OrphanVersionBodies int
// OrphanNameIndexEntries name a template whose header key is absent.
OrphanNameIndexEntries int
// UnreadableNameIndexEntries could not be read, so their target is unknown.
UnreadableNameIndexEntries int
// UnrecognisedEntries match none of the three record shapes.
UnrecognisedEntries int
// ReadFailures counts records the store failed to return.
ReadFailures int
}
// ScanAccount reads the <accountID>.* records in the launch-template bucket.
// Headers and name-index entries are read; version bodies are classified from
// their keys alone, so template data and user data are never loaded.
func (s *LaunchTemplateServiceImpl) ScanAccount(ctx context.Context, accountID string) (*AccountScan, error) {
keys, err := kvutil.Keys(ctx, s.kv)
if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, fmt.Errorf("list launch-template keys: %w", err)
}
prefix := accountID + "."
headers := make(map[string]bool)
var bodies, names []string
scan := &AccountScan{}
for _, key := range keys {
rest, ok := strings.CutPrefix(key, prefix)
if !ok {
continue
}
switch {
case isHeaderKey(key, prefix):
headers[rest] = s.scanHeader(ctx, key, rest, scan)
case strings.HasPrefix(rest, "name."):
names = append(names, key)
case versionBodyTemplateID(rest) != "":
bodies = append(bodies, versionBodyTemplateID(rest))
default:
scan.UnrecognisedEntries++
}
}
for _, ltID := range bodies {
if !headers[ltID] {
scan.OrphanVersionBodies++
}
}
for _, key := range names {
entry, err := s.kv.Get(ctx, key)
switch {
case errors.Is(err, jetstream.ErrKeyNotFound):
case err != nil:
slog.WarnContext(ctx, "Launch-template account scan: name index read failed", "err", err)
scan.ReadFailures++
scan.UnreadableNameIndexEntries++
case !headers[string(entry.Value())]:
scan.OrphanNameIndexEntries++
}
}
return scan, nil
}
// scanHeader records one header as live or unreadable and reports whether it
// still exists. A decoder's message is discarded: it can quote stored bytes.
func (s *LaunchTemplateServiceImpl) scanHeader(ctx context.Context, key, ltID string, scan *AccountScan) bool {
entry, err := s.kv.Get(ctx, key)
if err != nil {
if errors.Is(err, jetstream.ErrKeyNotFound) {
return false
}
slog.WarnContext(ctx, "Launch-template account scan: header read failed", "err", err)
scan.ReadFailures++
scan.UnreadableTemplateIDs = append(scan.UnreadableTemplateIDs, ltID)
return true
}
var header LaunchTemplateHeader
if err := json.Unmarshal(entry.Value(), &header); err != nil || header.LaunchTemplateId != ltID {
scan.UnreadableTemplateIDs = append(scan.UnreadableTemplateIDs, ltID)
return true
}
scan.LiveTemplateIDs = append(scan.LiveTemplateIDs, ltID)
return true
}
// versionBodyTemplateID returns the template ID of a version-body key suffix
// (lt-<id>.v<n>), or "" for any other shape.
func versionBodyTemplateID(rest string) string {
ltID, version, ok := strings.Cut(rest, ".v")
if !ok || !strings.HasPrefix(ltID, "lt-") || strings.Contains(ltID, ".") {
return ""
}
if n, err := strconv.ParseInt(version, 10, 64); err != nil || n < 1 {
return ""
}
return ltID
}
@@ -0,0 +1,145 @@
package launchtemplate
//test:in-package — the scan is run over a read-only wrapper of the service's
// unexported KV handle, and records are seeded in their private layout.
import (
"context"
"encoding/base64"
"strings"
"testing"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/internal/testkit/readonlystore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// readRecordingKV records every key read through it.
type readRecordingKV struct {
jetstream.KeyValue
gets []string
}
func (k *readRecordingKV) Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) {
k.gets = append(k.gets, key)
return k.KeyValue.Get(ctx, key)
}
func putRaw(t *testing.T, kv jetstream.KeyValue, key, value string) {
t.Helper()
_, err := kv.Put(t.Context(), key, []byte(value))
require.NoError(t, err)
}
// Each record left by a failed delete or an interrupted create is counted by
// shape, a live template's bodies and name are not, and version bodies are
// classified from their keys without being read.
func TestScanAccountClassifiesLaunchTemplateRecords(t *testing.T) {
svc := setupTestService(t)
live := createWithUserData(t, svc, "live")
deleted := createWithUserData(t, svc, "deleted")
inner := svc.kv
deletedHeader, deletedName := kvutil.AccountKey(testAccountID, deleted), nameKey(testAccountID, "deleted")
svc.kv = &faultyKV{KeyValue: inner, failDelete: func(key string) bool {
return strings.HasPrefix(key, deletedHeader+".v") || key == deletedName
}}
_, err := svc.DeleteLaunchTemplate(context.Background(), &ec2.DeleteLaunchTemplateInput{LaunchTemplateId: aws.String(deleted)}, testAccountID)
require.NoError(t, err)
svc.kv = &faultyKV{KeyValue: inner, failPut: func(key string) bool { return isHeaderKey(key, testAccountID+".") }}
_, err = svc.CreateLaunchTemplate(context.Background(), &ec2.CreateLaunchTemplateInput{
LaunchTemplateName: aws.String("interrupted"),
LaunchTemplateData: &ec2.RequestLaunchTemplateData{UserData: aws.String(base64.StdEncoding.EncodeToString([]byte(ctUserData)))},
}, testAccountID)
require.Error(t, err)
svc.kv = inner
putRaw(t, inner, kvutil.AccountKey(testAccountID, "lt-0badbadbadbadbad0"), "{not json")
putRaw(t, inner, kvutil.AccountKey(testAccountID, "lt-0badbadbadbadbad0.v1"), `{"version_number":1}`)
putRaw(t, inner, kvutil.AccountKey(testAccountID, "lt-0mismatch00000000"), `{"launch_template_id":"lt-0other"}`)
putRaw(t, inner, kvutil.AccountKey(testAccountID, "lt-0unreadable000000"), `{"launch_template_id":"lt-0unreadable000000"}`)
putRaw(t, inner, nameKey(testAccountID, "unreadable"), "lt-0unreadable000000")
putRaw(t, inner, kvutil.AccountKey(testAccountID, "lt-0bad.vx"), "{}")
recorder := &readRecordingKV{KeyValue: readonlystore.NewKV(t, inner)}
svc.kv = &faultyKV{KeyValue: recorder, failGet: func(key string) bool {
return key == kvutil.AccountKey(testAccountID, "lt-0unreadable000000")
}}
scan, err := svc.ScanAccount(context.Background(), testAccountID)
require.NoError(t, err)
assert.Equal(t, []string{live}, scan.LiveTemplateIDs)
assert.ElementsMatch(t, []string{"lt-0badbadbadbadbad0", "lt-0mismatch00000000", "lt-0unreadable000000"}, scan.UnreadableTemplateIDs)
assert.Equal(t, 3, scan.OrphanVersionBodies, "two bodies of the deleted template and v1 of the interrupted create")
assert.Equal(t, 2, scan.OrphanNameIndexEntries, "the deleted and the interrupted template's names")
assert.Equal(t, 1, scan.UnrecognisedEntries)
assert.Zero(t, scan.UnreadableNameIndexEntries)
assert.Equal(t, 1, scan.ReadFailures)
for _, key := range recorder.gets {
assert.NotRegexp(t, `\.v[0-9]+$`, key, "the scan read a version body")
}
}
// A name-index entry the store cannot return is unreadable and a read failure.
func TestScanAccountCountsAnUnreadableNameIndexEntry(t *testing.T) {
svc := setupTestService(t)
createTemplate(t, svc, "live", "t3.micro")
name := nameKey(testAccountID, "live")
svc.kv = &faultyKV{KeyValue: readonlystore.NewKV(t, svc.kv), failGet: func(key string) bool { return key == name }}
scan, err := svc.ScanAccount(context.Background(), testAccountID)
require.NoError(t, err)
assert.Len(t, scan.LiveTemplateIDs, 1)
assert.Equal(t, 1, scan.UnreadableNameIndexEntries)
assert.Equal(t, 1, scan.ReadFailures)
assert.Zero(t, scan.OrphanNameIndexEntries)
}
// The account prefix ends at its dot, so account 1111 never sees 11112's records.
func TestScanAccountStaysInsideTheLaunchTemplateAccountPrefix(t *testing.T) {
svc := setupTestService(t)
_, err := svc.CreateLaunchTemplate(context.Background(), &ec2.CreateLaunchTemplateInput{
LaunchTemplateName: aws.String("neighbour"),
LaunchTemplateData: &ec2.RequestLaunchTemplateData{ImageId: aws.String("ami-123")},
}, "11112")
require.NoError(t, err)
svc.kv = readonlystore.NewKV(t, svc.kv)
scan, err := svc.ScanAccount(context.Background(), "1111")
require.NoError(t, err)
assert.Equal(t, &AccountScan{}, scan)
scan, err = svc.ScanAccount(context.Background(), "11112")
require.NoError(t, err)
assert.Len(t, scan.LiveTemplateIDs, 1)
assert.Zero(t, scan.OrphanVersionBodies+scan.OrphanNameIndexEntries+scan.UnrecognisedEntries)
}
// A listing failure is an error, never an empty scan.
func TestScanAccountReturnsALaunchTemplateListingFailure(t *testing.T) {
svc := setupTestService(t)
svc.kv = &faultyKV{KeyValue: svc.kv, failKeys: func() bool { return true }}
_, err := svc.ScanAccount(context.Background(), testAccountID)
require.ErrorIs(t, err, errInjected)
}
func TestVersionBodyTemplateID(t *testing.T) {
for rest, want := range map[string]string{
"lt-0abc.v1": "lt-0abc",
"lt-0abc.v12": "lt-0abc",
"lt-0abc.v0": "",
"lt-0abc.vx": "",
"lt-0abc": "",
"name.6f72.v1": "",
"lt-0a.b.v1": "",
} {
assert.Equal(t, want, versionBodyTemplateID(rest), rest)
}
}
@@ -0,0 +1,80 @@
package placementgroup
import (
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"strings"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/nats-io/nats.go/jetstream"
)
// ScannedGroup is one readable placement group: its ID and how many distinct
// instances its member map still names.
type ScannedGroup struct {
GroupID string
MemberCount int
}
// AccountScan is a read-only classification of every record under one
// account's prefix. Group names, tags and raw keys never leave the scan.
type AccountScan struct {
Groups []ScannedGroup
// Unreadable records were found but could not be read, decoded or
// identified; their group IDs are unknown.
Unreadable int
// ReadFailures counts records the store failed to return.
ReadFailures int
}
// ScanAccount reads every <accountID>.* record in the placement-group bucket.
// A listing failure is returned as an error; per-record read failures are
// counted. The bucket's schema-version key belongs to no account.
func (s *PlacementGroupServiceImpl) ScanAccount(ctx context.Context, accountID string) (*AccountScan, error) {
keys, err := kvutil.Keys(ctx, s.kv)
if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, fmt.Errorf("list placement-group keys: %w", err)
}
prefix := accountID + "."
scan := &AccountScan{}
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := s.kv.Get(ctx, key)
if err != nil {
if errors.Is(err, jetstream.ErrKeyNotFound) {
continue
}
slog.WarnContext(ctx, "Placement-group account scan: record read failed", "err", err)
scan.ReadFailures++
scan.Unreadable++
continue
}
var record PlacementGroupRecord
if err := json.Unmarshal(entry.Value(), &record); err != nil || record.GroupId == "" {
scan.Unreadable++
continue
}
scan.Groups = append(scan.Groups, ScannedGroup{GroupID: record.GroupId, MemberCount: memberCount(record.NodeInstances)})
}
return scan, nil
}
// memberCount counts distinct instances in a member map, skipping the empty
// placeholders a reservation leaves for a node, as DeletePlacementGroup's in-use check does.
func memberCount(nodeInstances map[string][]string) int {
ids := make(map[string]bool)
for _, onNode := range nodeInstances {
for _, id := range onNode {
if id != "" {
ids[id] = true
}
}
}
return len(ids)
}
@@ -0,0 +1,87 @@
package placementgroup
//test:in-package — the scan is run over a read-only wrapper of the service's
// unexported KV handle, and records are seeded in their private layout.
import (
"context"
"errors"
"testing"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/internal/testkit/readonlystore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// keysFailingKV fails every key listing, standing in for a KV outage.
type keysFailingKV struct {
jetstream.KeyValue
}
func (k *keysFailingKV) Keys(context.Context, ...jetstream.WatchOpt) ([]string, error) {
return nil, errors.New("injected listing failure")
}
// Groups are reported with the instances their member map names; a node
// placeholder left by a reservation is not a member. Undecodable and unreadable
// records are counted, the second also as a read failure.
func TestScanAccountClassifiesPlacementGroupRecords(t *testing.T) {
svc := setupTestService(t)
empty := createTestGroup(t, svc, "empty", ec2.PlacementStrategySpread)
reserved := createTestGroup(t, svc, "reserved", ec2.PlacementStrategySpread)
_, err := svc.ReserveSpreadNodes(context.Background(), &ReserveSpreadNodesInput{
GroupName: "reserved", EligibleNodes: []string{"node-1"}, MinCount: 1, MaxCount: 1,
}, testAccountID)
require.NoError(t, err)
held := createTestGroup(t, svc, "held", ec2.PlacementStrategyCluster)
_, err = svc.FinalizeClusterInstances(context.Background(), &FinalizeClusterInstancesInput{
GroupName: "held", NodeInstances: map[string][]string{"node-2": {"i-0bbb", "i-0aaa"}, "node-3": {"i-0aaa"}},
}, testAccountID)
require.NoError(t, err)
_, err = svc.kv.Put(t.Context(), kvutil.AccountKey(testAccountID, "undecodable"), []byte("{not json"))
require.NoError(t, err)
_, err = svc.kv.Put(t.Context(), kvutil.AccountKey(testAccountID, "anonymous"), []byte(`{"group_name":"anonymous"}`))
require.NoError(t, err)
createTestGroup(t, svc, "unreadable", ec2.PlacementStrategySpread)
require.Contains(t, bucketKeys(t, svc.kv), kvutil.VersionKey)
svc.kv = &hookedKV{KeyValue: readonlystore.NewKV(t, svc.kv), failGet: kvutil.AccountKey(testAccountID, "unreadable")}
scan, err := svc.ScanAccount(context.Background(), testAccountID)
require.NoError(t, err)
assert.ElementsMatch(t, []ScannedGroup{
{GroupID: *empty.GroupId},
{GroupID: *reserved.GroupId},
{GroupID: *held.GroupId, MemberCount: 2},
}, scan.Groups)
assert.Equal(t, 3, scan.Unreadable)
assert.Equal(t, 1, scan.ReadFailures)
}
// The account prefix ends at its dot, so account 1111 never sees 11112's groups.
func TestScanAccountStaysInsideThePlacementGroupAccountPrefix(t *testing.T) {
svc := setupTestService(t)
_, err := svc.CreatePlacementGroup(context.Background(), &ec2.CreatePlacementGroupInput{GroupName: aws.String("neighbour")}, "11112")
require.NoError(t, err)
svc.kv = readonlystore.NewKV(t, svc.kv)
scan, err := svc.ScanAccount(context.Background(), "1111")
require.NoError(t, err)
assert.Equal(t, &AccountScan{}, scan)
scan, err = svc.ScanAccount(context.Background(), "11112")
require.NoError(t, err)
assert.Len(t, scan.Groups, 1)
}
// A listing failure is an error, never an empty scan.
func TestScanAccountReturnsAPlacementGroupListingFailure(t *testing.T) {
svc := setupTestService(t)
svc.kv = &keysFailingKV{KeyValue: svc.kv}
_, err := svc.ScanAccount(context.Background(), testAccountID)
require.Error(t, err)
}
+26
View File
@@ -0,0 +1,26 @@
package tags
import (
"context"
"strings"
)
// ScanAccountResourceIDs returns the sorted IDs of the account's tagged
// resources that start with one of prefixes, from the same KV and legacy
// object listing DescribeTags reads. It reads keys only, never a tag value.
func (s *TagsServiceImpl) ScanAccountResourceIDs(ctx context.Context, accountID string, prefixes ...string) ([]string, error) {
all, err := s.taggedResourceIDs(ctx, accountID)
if err != nil {
return nil, err
}
var ids []string
for _, id := range all {
for _, prefix := range prefixes {
if strings.HasPrefix(id, prefix) {
ids = append(ids, id)
break
}
}
}
return ids, nil
}
@@ -0,0 +1,56 @@
package tags
//test:in-package — entries are seeded in the private KV and legacy object
// layouts the scan has to read.
import (
"context"
"errors"
"testing"
"github.com/aws/aws-sdk-go/service/s3"
"github.com/mulgadc/spinifex/internal/testkit/readonlystore"
"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
type listFailingObjectStore struct {
*objectstore.MemoryObjectStore
}
func (s *listFailingObjectStore) ListObjectsV2(context.Context, *s3.ListObjectsV2Input) (*s3.ListObjectsV2Output, error) {
return nil, errors.New("injected listing failure")
}
// The scan returns the IDs DescribeTags would, from KV and legacy objects,
// limited to the asked prefixes and to the account, without any write.
func TestScanAccountResourceIDsListsPrefixedIDsFromBothStores(t *testing.T) {
kv := testTagsBucket(t)
store := objectstore.NewMemoryObjectStore()
cfg := &config.Config{Predastore: config.PredastoreConfig{Bucket: "test-bucket"}}
writer := NewTagsServiceImplWithStore(cfg, store, kv)
for account, ids := range map[string][]string{
"1111": {"pg-0kv", "lt-0kv", "key-0kv", "i-0ignored", "vpc-0ignored"},
"11112": {"pg-0neighbour"},
} {
for _, id := range ids {
require.NoError(t, writer.PutResourceTags(t.Context(), account, id, map[string]string{"team": "secret"}))
}
}
seedLegacyTags(t, store, "test-bucket", "1111", "pg-0legacy", map[string]string{"team": "secret"})
reader := NewTagsServiceImplWithStore(cfg, readonlystore.NewObjectStore(t, store), readonlystore.NewKV(t, kv))
ids, err := reader.ScanAccountResourceIDs(t.Context(), "1111", "key-", "pg-", "lt-")
require.NoError(t, err)
assert.Equal(t, []string{"key-0kv", "lt-0kv", "pg-0kv", "pg-0legacy"}, ids)
}
// A listing failure is an error, never an empty index.
func TestScanAccountResourceIDsReturnsAListingFailure(t *testing.T) {
cfg := &config.Config{Predastore: config.PredastoreConfig{Bucket: "test-bucket"}}
svc := NewTagsServiceImplWithStore(cfg, &listFailingObjectStore{MemoryObjectStore: objectstore.NewMemoryObjectStore()}, testTagsBucket(t))
_, err := svc.ScanAccountResourceIDs(t.Context(), "1111", "pg-")
require.Error(t, err)
}