mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-20 11:21:49 +00:00
feat: compare the EC2 platform inventory with legacy listings on explicit account-deletion dry runs (#1216)
This commit is contained in:
Josh Siviour
@@ -40,7 +40,7 @@ The request (`account_id`) and reply (`vpc_id`, `error`) bodies are still declar
|
||||
## 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.
|
||||
EC2 owns it: `domains/ec2/accountinventory` produces the reply and the daemon serves it; the only consumer today is `accountteardown`, which calls it only on an explicit operator dry run (`spx admin account delete --dry-run` or the admin API `DryRun`) and only logs a comparison with its legacy listings; no teardown calls 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.
|
||||
|
||||
@@ -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. 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.
|
||||
- `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`; only the explicit dry run (`Engine.DryRun`) calls it, to log a shadow comparison, 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`.
|
||||
|
||||
@@ -12,8 +12,8 @@ import (
|
||||
"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.
|
||||
// ec2PlatformInventoryTimeout bounds one inventory request; an earlier caller
|
||||
// deadline wins. A reply later than either is scan-incomplete.
|
||||
const ec2PlatformInventoryTimeout = 30 * time.Second
|
||||
|
||||
// ec2PlatformInventory is EC2's read-only report of an account's key pairs,
|
||||
@@ -39,8 +39,17 @@ func newEC2PlatformInventoryClient(nc *nats.Conn) *ec2PlatformInventoryClient {
|
||||
// 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 {
|
||||
// The NATS request takes a timeout, not a context, so the caller's deadline
|
||||
// shortens it here; a context already done is a timeout without a request.
|
||||
timeout := c.timeout
|
||||
if deadline, ok := ctx.Deadline(); ok {
|
||||
timeout = min(timeout, time.Until(deadline))
|
||||
}
|
||||
if ctx.Err() != nil || timeout <= 0 {
|
||||
return ec2v1.IncompletePlatformInventory(accountID, correlationRef, ec2v1.PlatformReasonTimeout)
|
||||
}
|
||||
reply, err := natsmsg.NATSRequest[ec2v1.PlatformInventoryReply](ctx, c.nc, ec2v1.PlatformInventorySubject,
|
||||
&ec2v1.PlatformInventoryRequest{AccountID: accountID, CorrelationRef: correlationRef}, c.timeout, accountID)
|
||||
&ec2v1.PlatformInventoryRequest{AccountID: accountID, CorrelationRef: correlationRef}, timeout, accountID)
|
||||
if err != nil {
|
||||
return ec2v1.IncompletePlatformInventory(accountID, correlationRef, inventoryFailureReason(err))
|
||||
}
|
||||
|
||||
@@ -6,6 +6,7 @@ package accountteardown
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -153,19 +154,10 @@ func TestEC2PlatformInventoryClientDistrustsAMalformedReply(t *testing.T) {
|
||||
}, 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) {
|
||||
// The cluster wiring binds the NATS adapter, and only the explicit dry run
|
||||
// consults it: precheck, the inventory the CLI prints before a real teardown,
|
||||
// a Teardown request marked dry run and a full teardown all leave it uncalled.
|
||||
func TestOnlyTheExplicitDryRunCallsTheEC2PlatformInventory(t *testing.T) {
|
||||
_, nc, _ := testutil.StartTestJetStream(t)
|
||||
masterKey, err := handlers_iam.GenerateMasterKey()
|
||||
require.NoError(t, err)
|
||||
@@ -175,9 +167,10 @@ func TestNoEnginePathCallsTheEC2PlatformInventory(t *testing.T) {
|
||||
require.NoError(t, err)
|
||||
assert.IsType(t, &ec2PlatformInventoryClient{}, cluster.ec2Inventory)
|
||||
|
||||
inventory := &fakeInventory{reply: ec2v1.IncompletePlatformInventory(testAccountID, "req-1", ec2v1.PlatformReasonNoResponder)}
|
||||
reaper := &fakeReaper{kind: "key-pair", stage: StagePlatform, resources: []Resource{{Kind: "key-pair", ID: "deploy"}}}
|
||||
engine := quietEngine(newFakeAccounts(), reaper)
|
||||
engine.ec2Inventory = failingInventory{t: t}
|
||||
engine.ec2Inventory = inventory
|
||||
|
||||
require.NoError(t, engine.Precheck(testAccountID, testAccountName))
|
||||
_, err = engine.Inventory(testCtx(t), testAccountID)
|
||||
@@ -186,7 +179,49 @@ func TestNoEnginePathCallsTheEC2PlatformInventory(t *testing.T) {
|
||||
dryRun.DryRun = true
|
||||
_, err = engine.Teardown(testCtx(t), dryRun)
|
||||
require.NoError(t, err)
|
||||
assert.Empty(t, inventory.calls())
|
||||
|
||||
_, err = engine.DryRun(testCtx(t), testAccountID, "req-1")
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, []string{"req-1"}, inventory.calls())
|
||||
|
||||
result, err := engine.Teardown(testCtx(t), request())
|
||||
require.NoError(t, err)
|
||||
assert.True(t, result.AccountDeleted)
|
||||
assert.Equal(t, []string{"req-1"}, inventory.calls(), "a real teardown never consults the inventory")
|
||||
}
|
||||
|
||||
// The shadow matches legacy listings to inventory kinds by the reapers' kind
|
||||
// names, so the registered platform reapers must use the contract's names.
|
||||
func TestPlatformReaperKindsAreTheInventoryKinds(t *testing.T) {
|
||||
var kinds []string
|
||||
for _, reaper := range EC2Reapers(nil, 1, nil) {
|
||||
if reaper.Stage() == StagePlatform {
|
||||
kinds = append(kinds, reaper.Kind())
|
||||
}
|
||||
}
|
||||
for _, kind := range ec2ShadowKinds {
|
||||
assert.Contains(t, kinds, kind.kind)
|
||||
}
|
||||
}
|
||||
|
||||
// A caller deadline shorter than the adapter's own timeout ends the request
|
||||
// at that deadline, and a context already done sends nothing.
|
||||
func TestEC2PlatformInventoryClientHonoursTheCallersDeadline(t *testing.T) {
|
||||
_, nc, _ := testutil.StartTestJetStream(t)
|
||||
var requests atomic.Int32
|
||||
serveInventory(t, nc, func() []byte { requests.Add(1); return nil })
|
||||
client := newEC2PlatformInventoryClient(nc)
|
||||
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
|
||||
defer cancel()
|
||||
started := time.Now()
|
||||
assertAllIncomplete(t, client.PlatformInventory(ctx, testAccountID, ""), ec2v1.PlatformReasonTimeout)
|
||||
assert.Less(t, time.Since(started), 5*time.Second)
|
||||
assert.Equal(t, int32(1), requests.Load())
|
||||
|
||||
done, cancelDone := context.WithCancel(t.Context())
|
||||
cancelDone()
|
||||
assertAllIncomplete(t, client.PlatformInventory(done, testAccountID, ""), ec2v1.PlatformReasonTimeout)
|
||||
assert.Equal(t, int32(1), requests.Load())
|
||||
}
|
||||
|
||||
@@ -0,0 +1,286 @@
|
||||
package accountteardown
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"regexp"
|
||||
"slices"
|
||||
"time"
|
||||
|
||||
ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
|
||||
otelsetup "github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
|
||||
)
|
||||
|
||||
// ec2PlatformShadowBudget caps how long a dry run waits for EC2's platform
|
||||
// inventory after its own listings, never past the caller's deadline. A later
|
||||
// answer is scan-incomplete and the dry run returns exactly as it would without.
|
||||
const ec2PlatformShadowBudget = 10 * time.Second
|
||||
|
||||
// ec2PlatformShadowMessage is the one operator audit record a dry run writes.
|
||||
const ec2PlatformShadowMessage = "Account teardown dry-run EC2 platform shadow comparison"
|
||||
|
||||
// Shadow outcomes, per kind and for the whole record. Only agree means the
|
||||
// legacy listing and the inventory matched and nothing hidden was reported.
|
||||
const (
|
||||
shadowAgree = "agree"
|
||||
shadowDiffer = "differ"
|
||||
shadowScanIncomplete = "scan-incomplete"
|
||||
)
|
||||
|
||||
// Bases a kind's visible resources are compared on.
|
||||
const (
|
||||
shadowBasisResourceID = "resource-id"
|
||||
shadowBasisCount = "count"
|
||||
)
|
||||
|
||||
// shadowKind is one compared kind. The key-pair and placement-group reapers
|
||||
// list by name, which the inventory never carries, so those compare by count.
|
||||
type shadowKind struct {
|
||||
kind string
|
||||
field string
|
||||
byName bool
|
||||
}
|
||||
|
||||
var ec2ShadowKinds = []shadowKind{
|
||||
{kind: ec2v1.PlatformKindKeyPair, field: "keyPair", byName: true},
|
||||
{kind: ec2v1.PlatformKindPlacementGroup, field: "placementGroup", byName: true},
|
||||
{kind: ec2v1.PlatformKindLaunchTemplate, field: "launchTemplate"},
|
||||
}
|
||||
|
||||
// loggableResourceID admits only the AWS ID shape of the three kinds, so a
|
||||
// reply that put stored content where an ID belongs never reaches the log.
|
||||
var loggableResourceID = regexp.MustCompile(`^(key|pg|lt)-[0-9a-f]{1,32}$`)
|
||||
|
||||
var knownShadowReasons = []string{
|
||||
ec2v1.PlatformReasonStoreListFailed, ec2v1.PlatformReasonRecordReadFailed,
|
||||
ec2v1.PlatformReasonOwnerRecordsUnresolved, ec2v1.PlatformReasonNoResponder,
|
||||
ec2v1.PlatformReasonTimeout, ec2v1.PlatformReasonTransportFailed,
|
||||
ec2v1.PlatformReasonErrorReply, ec2v1.PlatformReasonInvalidReply,
|
||||
ec2v1.PlatformReasonKindNotReported,
|
||||
}
|
||||
|
||||
var knownShadowClasses = []string{
|
||||
ec2v1.PlatformClassPublicKeyWithoutMetadata, ec2v1.PlatformClassPublicKeyUnresolved,
|
||||
ec2v1.PlatformClassUnreadableRecord, ec2v1.PlatformClassVersionBodyWithoutHeader,
|
||||
ec2v1.PlatformClassNameIndexWithoutHeader, ec2v1.PlatformClassUnrecognisedTemplateEntry,
|
||||
}
|
||||
|
||||
// shadowUnknownClass stands in for a count class this consumer does not know.
|
||||
const shadowUnknownClass = "unrecognised-class"
|
||||
|
||||
func (e *Engine) ec2ShadowBudget() time.Duration {
|
||||
if e.ec2PlatformShadowBudget > 0 {
|
||||
return e.ec2PlatformShadowBudget
|
||||
}
|
||||
return ec2PlatformShadowBudget
|
||||
}
|
||||
|
||||
// shadowEC2Platform compares EC2's platform inventory with the legacy listings
|
||||
// in result and logs the comparison. It reads result and never changes it.
|
||||
func (e *Engine) shadowEC2Platform(ctx context.Context, result *Result, correlationRef string) {
|
||||
start := time.Now()
|
||||
reply := e.awaitEC2PlatformInventory(ctx, result.AccountID, correlationRef)
|
||||
|
||||
reports := make(map[string]ec2v1.PlatformKindReport, len(reply.Kinds))
|
||||
for _, report := range reply.Kinds {
|
||||
reports[report.Kind] = report
|
||||
}
|
||||
|
||||
outcome := shadowAgree
|
||||
attrs := []slog.Attr{
|
||||
slog.String("accountID", result.AccountID),
|
||||
slog.String("correlationRef", correlationRef),
|
||||
}
|
||||
for _, kind := range ec2ShadowKinds {
|
||||
attr, kindOutcome := compareShadowKind(kind, legacyIDs(result, kind.kind), reports[kind.kind], reply)
|
||||
attrs = append(attrs, attr)
|
||||
outcome = worseShadowOutcome(outcome, kindOutcome)
|
||||
}
|
||||
tagAttr, tagOutcome := shadowTagIndex(reports[ec2v1.PlatformKindTagIndex], reply)
|
||||
attrs = append(attrs, tagAttr)
|
||||
outcome = worseShadowOutcome(outcome, tagOutcome)
|
||||
|
||||
attrs = append(attrs, slog.String("outcome", outcome), slog.Int64("duration_ms", otelsetup.Millis(time.Since(start))))
|
||||
e.logger().LogAttrs(ctx, slog.LevelInfo, ec2PlatformShadowMessage, attrs...)
|
||||
}
|
||||
|
||||
// awaitEC2PlatformInventory bounds the inventory call by the shadow budget even
|
||||
// if the implementation ignores its context, and distrusts the reply's shape.
|
||||
func (e *Engine) awaitEC2PlatformInventory(ctx context.Context, accountID, correlationRef string) ec2v1.PlatformInventoryReply {
|
||||
ctx, cancel := context.WithTimeout(ctx, e.ec2ShadowBudget())
|
||||
defer cancel()
|
||||
|
||||
replies := make(chan ec2v1.PlatformInventoryReply, 1)
|
||||
go func() { replies <- e.ec2Inventory.PlatformInventory(ctx, accountID, correlationRef) }()
|
||||
select {
|
||||
case reply := <-replies:
|
||||
return normalisePlatformInventory(reply, accountID, correlationRef)
|
||||
case <-ctx.Done():
|
||||
return ec2v1.IncompletePlatformInventory(accountID, correlationRef, ec2v1.PlatformReasonTimeout)
|
||||
}
|
||||
}
|
||||
|
||||
func legacyIDs(result *Result, kind string) []string {
|
||||
var ids []string
|
||||
for _, stage := range result.Stages {
|
||||
for _, resource := range stage.Deleted {
|
||||
if resource.Kind == kind {
|
||||
ids = append(ids, resource.ID)
|
||||
}
|
||||
}
|
||||
}
|
||||
return ids
|
||||
}
|
||||
|
||||
// compareShadowKind compares one kind's legacy listing with the inventory's
|
||||
// live and dependency-held items, and reports its hidden records as counts.
|
||||
func compareShadowKind(kind shadowKind, legacy []string, report ec2v1.PlatformKindReport, reply ec2v1.PlatformInventoryReply) (slog.Attr, string) {
|
||||
var visible, held []string
|
||||
hidden := map[string]int{}
|
||||
hiddenTotal := 0
|
||||
for _, item := range reply.Items {
|
||||
if item.Kind != kind.kind {
|
||||
continue
|
||||
}
|
||||
switch item.Disposition {
|
||||
case ec2v1.PlatformDispositionLive:
|
||||
visible = append(visible, item.ResourceID)
|
||||
case ec2v1.PlatformDispositionDependencyHeld:
|
||||
visible = append(visible, item.ResourceID)
|
||||
held = append(held, item.ResourceID)
|
||||
default:
|
||||
hidden[shadowUnknownClass]++
|
||||
hiddenTotal++
|
||||
}
|
||||
}
|
||||
for _, count := range reply.Counts {
|
||||
if count.Kind != kind.kind || count.Count <= 0 {
|
||||
continue
|
||||
}
|
||||
class := count.Class
|
||||
if !slices.Contains(knownShadowClasses, class) {
|
||||
class = shadowUnknownClass
|
||||
}
|
||||
hidden[class] += count.Count
|
||||
hiddenTotal += count.Count
|
||||
}
|
||||
|
||||
basis := shadowBasisResourceID
|
||||
var onlyLegacy, onlyInventory []string
|
||||
agree := len(legacy) == len(visible)
|
||||
if kind.byName {
|
||||
basis = shadowBasisCount
|
||||
} else {
|
||||
onlyLegacy, onlyInventory = setDifference(legacy, visible), setDifference(visible, legacy)
|
||||
agree = len(onlyLegacy) == 0 && len(onlyInventory) == 0
|
||||
}
|
||||
|
||||
comparison := shadowDiffer
|
||||
if agree {
|
||||
comparison = shadowAgree
|
||||
}
|
||||
outcome := comparison
|
||||
if hiddenTotal > 0 {
|
||||
outcome = shadowDiffer
|
||||
}
|
||||
if report.Status != ec2v1.PlatformStatusObservedClean && report.Status != ec2v1.PlatformStatusRecordsFound {
|
||||
comparison, outcome = shadowScanIncomplete, shadowScanIncomplete
|
||||
}
|
||||
|
||||
loggableOnlyLegacy, droppedLegacy := loggableIDs(onlyLegacy)
|
||||
loggableOnlyInventory, droppedInventory := loggableIDs(onlyInventory)
|
||||
loggableHeld, droppedHeld := loggableIDs(held)
|
||||
return slog.Group(kind.field,
|
||||
slog.String("status", report.Status),
|
||||
slog.Any("incompleteReasons", fixedReasons(report.IncompleteReasons)),
|
||||
slog.Time("scanStartedAt", report.ScanStartedAt),
|
||||
slog.Time("scanEndedAt", report.ScanEndedAt),
|
||||
slog.String("comparison", comparison),
|
||||
slog.String("matchBasis", basis),
|
||||
slog.Int("legacyCount", len(legacy)),
|
||||
slog.Int("inventoryVisibleCount", len(visible)),
|
||||
slog.Any("onlyInLegacy", loggableOnlyLegacy),
|
||||
slog.Any("onlyInInventory", loggableOnlyInventory),
|
||||
slog.Any("dependencyHeld", loggableHeld),
|
||||
slog.Int("hiddenRecordCount", hiddenTotal),
|
||||
slog.Any("hiddenRecords", hidden),
|
||||
slog.Int("unloggableIDCount", droppedLegacy+droppedInventory+droppedHeld),
|
||||
), outcome
|
||||
}
|
||||
|
||||
// shadowTagIndex reports tag-index entries whose resource has no owner record,
|
||||
// which no legacy reaper lists.
|
||||
func shadowTagIndex(report ec2v1.PlatformKindReport, reply ec2v1.PlatformInventoryReply) (slog.Attr, string) {
|
||||
var residue []string
|
||||
for _, item := range reply.Items {
|
||||
if item.Kind == ec2v1.PlatformKindTagIndex {
|
||||
residue = append(residue, item.ResourceID)
|
||||
}
|
||||
}
|
||||
outcome := shadowAgree
|
||||
if len(residue) > 0 {
|
||||
outcome = shadowDiffer
|
||||
}
|
||||
if report.Status != ec2v1.PlatformStatusObservedClean && report.Status != ec2v1.PlatformStatusRecordsFound {
|
||||
outcome = shadowScanIncomplete
|
||||
}
|
||||
loggable, dropped := loggableIDs(residue)
|
||||
return slog.Group("tagIndex",
|
||||
slog.String("status", report.Status),
|
||||
slog.Any("incompleteReasons", fixedReasons(report.IncompleteReasons)),
|
||||
slog.Time("scanStartedAt", report.ScanStartedAt),
|
||||
slog.Time("scanEndedAt", report.ScanEndedAt),
|
||||
slog.Int("residueCount", len(residue)),
|
||||
slog.Any("residue", loggable),
|
||||
slog.Int("unloggableIDCount", dropped),
|
||||
), outcome
|
||||
}
|
||||
|
||||
func worseShadowOutcome(a, b string) string {
|
||||
rank := map[string]int{shadowAgree: 0, shadowDiffer: 1, shadowScanIncomplete: 2}
|
||||
if rank[b] > rank[a] {
|
||||
return b
|
||||
}
|
||||
return a
|
||||
}
|
||||
|
||||
// setDifference returns the sorted members of a absent from b.
|
||||
func setDifference(a, b []string) []string {
|
||||
out := []string{}
|
||||
for _, id := range a {
|
||||
if !slices.Contains(b, id) && !slices.Contains(out, id) {
|
||||
out = append(out, id)
|
||||
}
|
||||
}
|
||||
slices.Sort(out)
|
||||
return out
|
||||
}
|
||||
|
||||
func loggableIDs(ids []string) ([]string, int) {
|
||||
out := []string{}
|
||||
dropped := 0
|
||||
for _, id := range ids {
|
||||
if loggableResourceID.MatchString(id) {
|
||||
out = append(out, id)
|
||||
continue
|
||||
}
|
||||
dropped++
|
||||
}
|
||||
slices.Sort(out)
|
||||
return out, dropped
|
||||
}
|
||||
|
||||
// fixedReasons keeps only the contract's reason categories; anything else a
|
||||
// reply carried is reported as an invalid reply rather than logged.
|
||||
func fixedReasons(reasons []string) []string {
|
||||
out := []string{}
|
||||
for _, reason := range reasons {
|
||||
if !slices.Contains(knownShadowReasons, reason) {
|
||||
reason = ec2v1.PlatformReasonInvalidReply
|
||||
}
|
||||
if !slices.Contains(out, reason) {
|
||||
out = append(out, reason)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,503 @@
|
||||
package accountteardown
|
||||
|
||||
//test:in-package — the shadow comparison, the inventory binding it reads and
|
||||
// its budget are unexported.
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"sync"
|
||||
"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"
|
||||
natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
var shadowScanTime = time.Date(2026, 10, 16, 9, 0, 0, 0, time.UTC)
|
||||
|
||||
// fakeInventory answers with reply, or never when blocked, and records the
|
||||
// correlation reference of every call.
|
||||
type fakeInventory struct {
|
||||
reply ec2v1.PlatformInventoryReply
|
||||
blocked chan struct{}
|
||||
|
||||
mu sync.Mutex
|
||||
refs []string
|
||||
}
|
||||
|
||||
func (f *fakeInventory) PlatformInventory(_ context.Context, _, correlationRef string) ec2v1.PlatformInventoryReply {
|
||||
f.mu.Lock()
|
||||
f.refs = append(f.refs, correlationRef)
|
||||
f.mu.Unlock()
|
||||
if f.blocked != nil {
|
||||
<-f.blocked
|
||||
}
|
||||
return f.reply
|
||||
}
|
||||
|
||||
func (f *fakeInventory) calls() []string {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return append([]string(nil), f.refs...)
|
||||
}
|
||||
|
||||
// countingReaper counts listings, so a dry run can be shown to list once.
|
||||
type countingReaper struct {
|
||||
Reaper
|
||||
|
||||
lists int
|
||||
}
|
||||
|
||||
func (r *countingReaper) List(ctx context.Context, accountID string) ([]Resource, error) {
|
||||
r.lists++
|
||||
return r.Reaper.List(ctx, accountID)
|
||||
}
|
||||
|
||||
func platformKinds(status string) []ec2v1.PlatformKindReport {
|
||||
var kinds []ec2v1.PlatformKindReport
|
||||
for _, kind := range ec2v1.PlatformKinds() {
|
||||
kinds = append(kinds, ec2v1.PlatformKindReport{
|
||||
Kind: kind, Status: status, ScanStartedAt: shadowScanTime, ScanEndedAt: shadowScanTime.Add(time.Second),
|
||||
})
|
||||
}
|
||||
return kinds
|
||||
}
|
||||
|
||||
func setKind(kinds []ec2v1.PlatformKindReport, kind, status string, reasons ...string) {
|
||||
for i := range kinds {
|
||||
if kinds[i].Kind == kind {
|
||||
kinds[i].Status, kinds[i].IncompleteReasons = status, reasons
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func item(kind, id, disposition string) ec2v1.PlatformItem {
|
||||
return ec2v1.PlatformItem{ResourceID: id, Kind: kind, Disposition: disposition}
|
||||
}
|
||||
|
||||
// legacyPlatform is a key pair, a placement group and a launch template as the
|
||||
// legacy reapers list them: the first two by name, the third by ID.
|
||||
func legacyPlatform() []Reaper {
|
||||
return []Reaper{
|
||||
&fakeReaper{kind: "instance", stage: StageCompute, resources: []Resource{{Kind: "instance", ID: "i-0123456789abcdef0"}}},
|
||||
&fakeReaper{kind: "key-pair", stage: StagePlatform, resources: []Resource{{Kind: "key-pair", ID: "deploy"}}},
|
||||
&fakeReaper{kind: "placement-group", stage: StagePlatform, resources: []Resource{{Kind: "placement-group", ID: "spread"}}},
|
||||
&fakeReaper{kind: "launch-template", stage: StagePlatform, resources: []Resource{{Kind: "launch-template", ID: "lt-0aaaaaaaaaaaaaaaa"}}},
|
||||
}
|
||||
}
|
||||
|
||||
// matchingInventory reports exactly what legacyPlatform lists.
|
||||
func matchingInventory() ec2v1.PlatformInventoryReply {
|
||||
return ec2v1.PlatformInventoryReply{
|
||||
AccountID: testAccountID,
|
||||
Kinds: platformKinds(ec2v1.PlatformStatusRecordsFound),
|
||||
Items: []ec2v1.PlatformItem{
|
||||
item(ec2v1.PlatformKindKeyPair, "key-0aaaaaaaaaaaaaaaa", ec2v1.PlatformDispositionLive),
|
||||
item(ec2v1.PlatformKindPlacementGroup, "pg-0aaaaaaaaaaaaaaaa", ec2v1.PlatformDispositionLive),
|
||||
item(ec2v1.PlatformKindLaunchTemplate, "lt-0aaaaaaaaaaaaaaaa", ec2v1.PlatformDispositionLive),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// shadowEngine is a quiet engine whose log is captured as JSON in logs.
|
||||
func shadowEngine(inventory ec2PlatformInventory, reapers ...Reaper) (*Engine, *bytes.Buffer) {
|
||||
engine := quietEngine(newFakeAccounts(), reapers...)
|
||||
logs := &bytes.Buffer{}
|
||||
engine.Logger = slog.New(slog.NewJSONHandler(logs, nil))
|
||||
engine.ec2Inventory = inventory
|
||||
return engine, logs
|
||||
}
|
||||
|
||||
// shadowRecords returns every shadow comparison record in logs.
|
||||
func shadowRecords(t *testing.T, logs *bytes.Buffer) []map[string]any {
|
||||
t.Helper()
|
||||
var records []map[string]any
|
||||
for line := range strings.SplitSeq(strings.TrimSpace(logs.String()), "\n") {
|
||||
if line == "" {
|
||||
continue
|
||||
}
|
||||
var record map[string]any
|
||||
require.NoError(t, json.Unmarshal([]byte(line), &record))
|
||||
if record["msg"] == ec2PlatformShadowMessage {
|
||||
records = append(records, record)
|
||||
}
|
||||
}
|
||||
return records
|
||||
}
|
||||
|
||||
func shadowRecord(t *testing.T, logs *bytes.Buffer) map[string]any {
|
||||
t.Helper()
|
||||
records := shadowRecords(t, logs)
|
||||
require.Len(t, records, 1, "one shadow record per dry run")
|
||||
return records[0]
|
||||
}
|
||||
|
||||
func group(t *testing.T, record map[string]any, name string) map[string]any {
|
||||
t.Helper()
|
||||
g, ok := record[name].(map[string]any)
|
||||
require.True(t, ok, "record has a %s group", name)
|
||||
return g
|
||||
}
|
||||
|
||||
func strs(values ...string) []any {
|
||||
out := []any{}
|
||||
for _, v := range values {
|
||||
out = append(out, v)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func resultJSON(t *testing.T, result *Result) string {
|
||||
t.Helper()
|
||||
data, err := json.Marshal(result)
|
||||
require.NoError(t, err)
|
||||
return string(data)
|
||||
}
|
||||
|
||||
// noShadowResult is what the same dry run returns with no inventory bound.
|
||||
func noShadowResult(t *testing.T, reapers ...Reaper) string {
|
||||
t.Helper()
|
||||
result, err := quietEngine(newFakeAccounts(), reapers...).Inventory(testCtx(t), testAccountID)
|
||||
require.NoError(t, err)
|
||||
return resultJSON(t, result)
|
||||
}
|
||||
|
||||
// Visible resources that match on both sides agree. The legacy key-pair and
|
||||
// placement-group reapers list names, so those kinds compare by count.
|
||||
func TestDryRunShadowAgreesWhenBothSidesListTheSameResources(t *testing.T) {
|
||||
reapers := legacyPlatform()
|
||||
counted := make([]Reaper, 0, len(reapers))
|
||||
for _, reaper := range reapers {
|
||||
counted = append(counted, &countingReaper{Reaper: reaper})
|
||||
}
|
||||
inventory := &fakeInventory{reply: matchingInventory()}
|
||||
engine, logs := shadowEngine(inventory, counted...)
|
||||
|
||||
result, err := engine.DryRun(testCtx(t), testAccountID, "req-agree")
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.Equal(t, noShadowResult(t, legacyPlatform()...), resultJSON(t, result))
|
||||
assert.Equal(t, []string{"req-agree"}, inventory.calls())
|
||||
for _, reaper := range counted {
|
||||
assert.Equal(t, 1, reaper.(*countingReaper).lists, "%s listed once: the shadow reuses the dry run's listing", reaper.Kind())
|
||||
}
|
||||
|
||||
record := shadowRecord(t, logs)
|
||||
assert.Equal(t, "INFO", record["level"])
|
||||
assert.Equal(t, testAccountID, record["accountID"])
|
||||
assert.Equal(t, "req-agree", record["correlationRef"])
|
||||
assert.Equal(t, shadowAgree, record["outcome"])
|
||||
assert.Contains(t, record, "duration_ms")
|
||||
|
||||
keyPair := group(t, record, "keyPair")
|
||||
assert.Equal(t, ec2v1.PlatformStatusRecordsFound, keyPair["status"])
|
||||
assert.Equal(t, shadowAgree, keyPair["comparison"])
|
||||
assert.Equal(t, shadowBasisCount, keyPair["matchBasis"])
|
||||
assert.InDelta(t, 1, keyPair["legacyCount"], 0)
|
||||
assert.InDelta(t, 1, keyPair["inventoryVisibleCount"], 0)
|
||||
assert.Equal(t, shadowScanTime.Format(time.RFC3339), keyPair["scanStartedAt"])
|
||||
|
||||
launchTemplate := group(t, record, "launchTemplate")
|
||||
assert.Equal(t, shadowAgree, launchTemplate["comparison"])
|
||||
assert.Equal(t, shadowBasisResourceID, launchTemplate["matchBasis"])
|
||||
assert.Equal(t, strs(), launchTemplate["onlyInLegacy"])
|
||||
assert.Equal(t, strs(), launchTemplate["onlyInInventory"])
|
||||
|
||||
tagIndex := group(t, record, "tagIndex")
|
||||
assert.Equal(t, ec2v1.PlatformStatusRecordsFound, tagIndex["status"])
|
||||
assert.InDelta(t, 0, tagIndex["residueCount"], 0)
|
||||
}
|
||||
|
||||
// Visible resources one side has and the other does not are named by ID where
|
||||
// the legacy reaper lists IDs, and counted where it lists names.
|
||||
func TestDryRunShadowReportsVisibleResourcesOnlyOneSideSees(t *testing.T) {
|
||||
reapers := legacyPlatform()
|
||||
reapers[1].(*fakeReaper).resources = append(reapers[1].(*fakeReaper).resources, Resource{Kind: "key-pair", ID: "ci"})
|
||||
reapers[3].(*fakeReaper).resources = []Resource{
|
||||
{Kind: "launch-template", ID: "lt-0aaaaaaaaaaaaaaaa"}, {Kind: "launch-template", ID: "lt-0bbbbbbbbbbbbbbbb"},
|
||||
}
|
||||
reply := matchingInventory()
|
||||
reply.Items = append(reply.Items, item(ec2v1.PlatformKindLaunchTemplate, "lt-0cccccccccccccccc", ec2v1.PlatformDispositionLive))
|
||||
engine, logs := shadowEngine(&fakeInventory{reply: reply}, reapers...)
|
||||
|
||||
_, err := engine.DryRun(testCtx(t), testAccountID, "req-differ")
|
||||
require.NoError(t, err)
|
||||
|
||||
record := shadowRecord(t, logs)
|
||||
assert.Equal(t, shadowDiffer, record["outcome"])
|
||||
keyPair := group(t, record, "keyPair")
|
||||
assert.Equal(t, shadowDiffer, keyPair["comparison"])
|
||||
assert.InDelta(t, 2, keyPair["legacyCount"], 0)
|
||||
assert.InDelta(t, 1, keyPair["inventoryVisibleCount"], 0)
|
||||
assert.Equal(t, strs(), keyPair["onlyInLegacy"], "legacy key-pair names are never logged")
|
||||
|
||||
launchTemplate := group(t, record, "launchTemplate")
|
||||
assert.Equal(t, shadowDiffer, launchTemplate["comparison"])
|
||||
assert.Equal(t, strs("lt-0bbbbbbbbbbbbbbbb"), launchTemplate["onlyInLegacy"])
|
||||
assert.Equal(t, strs("lt-0cccccccccccccccc"), launchTemplate["onlyInInventory"])
|
||||
assert.Equal(t, shadowAgree, group(t, record, "placementGroup")["comparison"])
|
||||
}
|
||||
|
||||
// Residue the legacy reapers cannot see makes the record differ even when the
|
||||
// visible resources agree: as counts per class, and tag-index entries by ID.
|
||||
func TestDryRunShadowReportsHiddenResidueTheLegacyReapersMiss(t *testing.T) {
|
||||
reply := matchingInventory()
|
||||
reply.Counts = []ec2v1.PlatformRecordCount{
|
||||
{Kind: ec2v1.PlatformKindKeyPair, Disposition: ec2v1.PlatformDispositionResidue, Class: ec2v1.PlatformClassPublicKeyWithoutMetadata, Count: 1},
|
||||
{Kind: ec2v1.PlatformKindLaunchTemplate, Disposition: ec2v1.PlatformDispositionResidue, Class: ec2v1.PlatformClassVersionBodyWithoutHeader, Count: 2},
|
||||
{Kind: ec2v1.PlatformKindLaunchTemplate, Disposition: ec2v1.PlatformDispositionUnreadable, Class: ec2v1.PlatformClassUnreadableRecord, Count: 1},
|
||||
}
|
||||
reply.Items = append(reply.Items, item(ec2v1.PlatformKindTagIndex, "pg-0dddddddddddddddd", ec2v1.PlatformDispositionResidue))
|
||||
engine, logs := shadowEngine(&fakeInventory{reply: reply}, legacyPlatform()...)
|
||||
|
||||
_, err := engine.DryRun(testCtx(t), testAccountID, "req-residue")
|
||||
require.NoError(t, err)
|
||||
|
||||
record := shadowRecord(t, logs)
|
||||
assert.Equal(t, shadowDiffer, record["outcome"])
|
||||
keyPair := group(t, record, "keyPair")
|
||||
assert.Equal(t, shadowAgree, keyPair["comparison"], "the visible key pairs still agree")
|
||||
assert.InDelta(t, 1, keyPair["hiddenRecordCount"], 0)
|
||||
assert.Equal(t, map[string]any{ec2v1.PlatformClassPublicKeyWithoutMetadata: float64(1)}, keyPair["hiddenRecords"])
|
||||
launchTemplate := group(t, record, "launchTemplate")
|
||||
assert.InDelta(t, 3, launchTemplate["hiddenRecordCount"], 0)
|
||||
assert.Equal(t, map[string]any{
|
||||
ec2v1.PlatformClassVersionBodyWithoutHeader: float64(2),
|
||||
ec2v1.PlatformClassUnreadableRecord: float64(1),
|
||||
}, launchTemplate["hiddenRecords"])
|
||||
tagIndex := group(t, record, "tagIndex")
|
||||
assert.InDelta(t, 1, tagIndex["residueCount"], 0)
|
||||
assert.Equal(t, strs("pg-0dddddddddddddddd"), tagIndex["residue"])
|
||||
}
|
||||
|
||||
// A placement group still holding instances is a plain listed resource to the
|
||||
// legacy dry run, which learns otherwise only when its delete fails.
|
||||
func TestDryRunShadowReportsADependencyHeldPlacementGroup(t *testing.T) {
|
||||
reply := matchingInventory()
|
||||
reply.Items[1].Disposition = ec2v1.PlatformDispositionDependencyHeld
|
||||
engine, logs := shadowEngine(&fakeInventory{reply: reply}, legacyPlatform()...)
|
||||
|
||||
_, err := engine.DryRun(testCtx(t), testAccountID, "req-held")
|
||||
require.NoError(t, err)
|
||||
|
||||
placementGroup := group(t, shadowRecord(t, logs), "placementGroup")
|
||||
assert.Equal(t, shadowAgree, placementGroup["comparison"])
|
||||
assert.InDelta(t, 1, placementGroup["inventoryVisibleCount"], 0)
|
||||
assert.Equal(t, strs("pg-0aaaaaaaaaaaaaaaa"), placementGroup["dependencyHeld"])
|
||||
}
|
||||
|
||||
// An owner store that could not be read is scan-incomplete, never agreement,
|
||||
// even where the IDs it did report match.
|
||||
func TestDryRunShadowNeverReportsAnIncompleteScanAsAgreement(t *testing.T) {
|
||||
reply := matchingInventory()
|
||||
setKind(reply.Kinds, ec2v1.PlatformKindLaunchTemplate, ec2v1.PlatformStatusScanIncomplete, ec2v1.PlatformReasonRecordReadFailed)
|
||||
engine, logs := shadowEngine(&fakeInventory{reply: reply}, legacyPlatform()...)
|
||||
|
||||
result, err := engine.DryRun(testCtx(t), testAccountID, "req-store")
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, noShadowResult(t, legacyPlatform()...), resultJSON(t, result))
|
||||
|
||||
record := shadowRecord(t, logs)
|
||||
assert.Equal(t, shadowScanIncomplete, record["outcome"])
|
||||
launchTemplate := group(t, record, "launchTemplate")
|
||||
assert.Equal(t, shadowScanIncomplete, launchTemplate["comparison"])
|
||||
assert.Equal(t, strs(ec2v1.PlatformReasonRecordReadFailed), launchTemplate["incompleteReasons"])
|
||||
assert.Equal(t, shadowAgree, group(t, record, "keyPair")["comparison"])
|
||||
}
|
||||
|
||||
// However the inventory fails, the dry run returns what it returns without it
|
||||
// and the record says scan-incomplete with the fixed reason.
|
||||
func TestDryRunShadowFailureLeavesTheDryRunUnchanged(t *testing.T) {
|
||||
want := noShadowResult(t, legacyPlatform()...)
|
||||
for _, reason := range []string{
|
||||
ec2v1.PlatformReasonNoResponder, ec2v1.PlatformReasonTimeout,
|
||||
ec2v1.PlatformReasonTransportFailed, ec2v1.PlatformReasonErrorReply,
|
||||
} {
|
||||
t.Run(reason, func(t *testing.T) {
|
||||
engine, logs := shadowEngine(&fakeInventory{reply: ec2v1.IncompletePlatformInventory(testAccountID, "req-fail", reason)}, legacyPlatform()...)
|
||||
result, err := engine.DryRun(testCtx(t), testAccountID, "req-fail")
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, want, resultJSON(t, result))
|
||||
|
||||
record := shadowRecord(t, logs)
|
||||
assert.Equal(t, shadowScanIncomplete, record["outcome"])
|
||||
for _, name := range []string{"keyPair", "placementGroup", "launchTemplate", "tagIndex"} {
|
||||
g := group(t, record, name)
|
||||
assert.Equal(t, ec2v1.PlatformStatusScanIncomplete, g["status"], name)
|
||||
assert.Equal(t, strs(reason), g["incompleteReasons"], name)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
t.Run("reply for another account", func(t *testing.T) {
|
||||
reply := matchingInventory()
|
||||
reply.AccountID = "000000000099"
|
||||
engine, logs := shadowEngine(&fakeInventory{reply: reply}, legacyPlatform()...)
|
||||
result, err := engine.DryRun(testCtx(t), testAccountID, "req-fail")
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, want, resultJSON(t, result))
|
||||
record := shadowRecord(t, logs)
|
||||
assert.Equal(t, shadowScanIncomplete, record["outcome"])
|
||||
assert.Equal(t, strs(ec2v1.PlatformReasonInvalidReply), group(t, record, "keyPair")["incompleteReasons"])
|
||||
})
|
||||
|
||||
t.Run("no daemon serves the subject", func(t *testing.T) {
|
||||
_, nc, _ := testutil.StartTestJetStream(t)
|
||||
engine, logs := shadowEngine(newEC2PlatformInventoryClient(nc), legacyPlatform()...)
|
||||
result, err := engine.DryRun(testCtx(t), testAccountID, "req-fail")
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, want, resultJSON(t, result))
|
||||
assert.Equal(t, strs(ec2v1.PlatformReasonNoResponder), group(t, shadowRecord(t, logs), "tagIndex")["incompleteReasons"])
|
||||
})
|
||||
}
|
||||
|
||||
// An inventory that never answers, even one ignoring its context, holds the
|
||||
// dry run for the shadow budget at most and is logged as a timeout.
|
||||
func TestDryRunShadowIsBoundedByItsBudget(t *testing.T) {
|
||||
inventory := &fakeInventory{reply: matchingInventory(), blocked: make(chan struct{})}
|
||||
t.Cleanup(func() { close(inventory.blocked) })
|
||||
engine, logs := shadowEngine(inventory, legacyPlatform()...)
|
||||
engine.ec2PlatformShadowBudget = 20 * time.Millisecond
|
||||
|
||||
started := time.Now()
|
||||
result, err := engine.DryRun(testCtx(t), testAccountID, "req-slow")
|
||||
require.NoError(t, err)
|
||||
assert.Less(t, time.Since(started), 5*time.Second)
|
||||
assert.Equal(t, noShadowResult(t, legacyPlatform()...), resultJSON(t, result))
|
||||
assert.Equal(t, strs(ec2v1.PlatformReasonTimeout), group(t, shadowRecord(t, logs), "launchTemplate")["incompleteReasons"])
|
||||
}
|
||||
|
||||
// The caller's deadline wins over the budget, so a dry run never outlasts the
|
||||
// bound its caller already set.
|
||||
func TestDryRunShadowStopsAtTheCallersDeadline(t *testing.T) {
|
||||
_, nc, _ := testutil.StartTestJetStream(t)
|
||||
serveInventory(t, nc, func() []byte { return nil })
|
||||
engine, logs := shadowEngine(newEC2PlatformInventoryClient(nc), legacyPlatform()...)
|
||||
require.Greater(t, engine.ec2ShadowBudget(), 5*time.Second)
|
||||
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 200*time.Millisecond)
|
||||
defer cancel()
|
||||
started := time.Now()
|
||||
result, err := engine.DryRun(ctx, testAccountID, "req-deadline")
|
||||
require.NoError(t, err)
|
||||
assert.Less(t, time.Since(started), 5*time.Second)
|
||||
assert.Equal(t, noShadowResult(t, legacyPlatform()...), resultJSON(t, result))
|
||||
assert.Equal(t, strs(ec2v1.PlatformReasonTimeout), group(t, shadowRecord(t, logs), "keyPair")["incompleteReasons"])
|
||||
}
|
||||
|
||||
// A dry run whose own listing fails returns that error as before, and nothing
|
||||
// is compared against a listing that does not exist.
|
||||
func TestDryRunWithAFailedListingDoesNotCallTheInventory(t *testing.T) {
|
||||
reapers := legacyPlatform()
|
||||
reapers[2].(*fakeReaper).listErr = errors.New("describe failed")
|
||||
inventory := &fakeInventory{reply: matchingInventory()}
|
||||
engine, logs := shadowEngine(inventory, reapers...)
|
||||
|
||||
_, err := engine.DryRun(testCtx(t), testAccountID, "req-list")
|
||||
require.ErrorContains(t, err, "inventory placement-group")
|
||||
assert.Empty(t, inventory.calls())
|
||||
assert.Empty(t, shadowRecords(t, logs))
|
||||
}
|
||||
|
||||
// A reply that puts anything but the contract's fixed vocabulary where a
|
||||
// reason, class or ID belongs never reaches the log; the IDs are counted.
|
||||
func TestDryRunShadowLogsOnlyTheFixedVocabulary(t *testing.T) {
|
||||
reply := matchingInventory()
|
||||
setKind(reply.Kinds, ec2v1.PlatformKindKeyPair, ec2v1.PlatformStatusScanIncomplete, "read keys/000000000042/secret-key-name: denied")
|
||||
reply.Counts = []ec2v1.PlatformRecordCount{{Kind: ec2v1.PlatformKindLaunchTemplate, Disposition: ec2v1.PlatformDispositionResidue, Class: "secret-template-name", Count: 1}}
|
||||
reply.Items = append(reply.Items,
|
||||
item(ec2v1.PlatformKindLaunchTemplate, "lt-secret-template-name", ec2v1.PlatformDispositionLive),
|
||||
item(ec2v1.PlatformKindTagIndex, "pg-team=secret-tag-value", ec2v1.PlatformDispositionResidue),
|
||||
)
|
||||
reapers := legacyPlatform()
|
||||
reapers[1].(*fakeReaper).resources = []Resource{{Kind: "key-pair", ID: "secret-key-name"}}
|
||||
engine, logs := shadowEngine(&fakeInventory{reply: reply}, reapers...)
|
||||
|
||||
_, err := engine.DryRun(testCtx(t), testAccountID, "req-disclosure")
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.NotContains(t, logs.String(), "secret")
|
||||
record := shadowRecord(t, logs)
|
||||
assert.Equal(t, strs(ec2v1.PlatformReasonInvalidReply), group(t, record, "keyPair")["incompleteReasons"])
|
||||
launchTemplate := group(t, record, "launchTemplate")
|
||||
assert.Equal(t, map[string]any{shadowUnknownClass: float64(1)}, launchTemplate["hiddenRecords"])
|
||||
assert.InDelta(t, 1, launchTemplate["unloggableIDCount"], 0)
|
||||
assert.InDelta(t, 1, group(t, record, "tagIndex")["unloggableIDCount"], 0)
|
||||
}
|
||||
|
||||
// End to end over the real owner stores and the NATS adapter: the record names
|
||||
// the residue each reaper misses and nothing a customer named or stored.
|
||||
func TestDryRunShadowOverTheOwnerStores(t *testing.T) {
|
||||
o := newPlatformOwners(t)
|
||||
ctx := testCtx(t)
|
||||
inv := accountinventory.New(o.keys, o.groups, o.templates, o.tags)
|
||||
sub, err := o.nc.QueueSubscribe(ec2v1.PlatformInventorySubject, "spinifex-workers", func(msg *nats.Msg) {
|
||||
natsmsg.ServeNATSRequestCtx(msg, inv.PlatformInventory)
|
||||
})
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = sub.Unsubscribe() })
|
||||
|
||||
material := []byte("ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIOMqqnkVzrm0SdG6UOoqKLsabgH5C9okWi0dh2l9GKJl")
|
||||
for _, name := range []string{"secret-key-name", "secret-orphan-key"} {
|
||||
_, err = o.keys.ImportKeyPair(ctx, &ec2.ImportKeyPairInput{KeyName: aws.String(name), PublicKeyMaterial: material}, testAccountID)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
o.store.failDelete = func(key string) bool { return key == "keys/"+testAccountID+"/secret-orphan-key" }
|
||||
_, err = o.keys.DeleteKeyPair(ctx, &ec2.DeleteKeyPairInput{KeyName: aws.String("secret-orphan-key")}, testAccountID)
|
||||
require.NoError(t, err)
|
||||
|
||||
_, err = o.groups.CreatePlacementGroup(ctx, &ec2.CreatePlacementGroupInput{
|
||||
GroupName: aws.String("secret-group-name"), Strategy: aws.String(ec2.PlacementStrategySpread),
|
||||
}, testAccountID)
|
||||
require.NoError(t, err)
|
||||
doomed, err := o.groups.CreatePlacementGroup(ctx, &ec2.CreatePlacementGroupInput{
|
||||
GroupName: aws.String("secret-doomed-group"), Strategy: aws.String(ec2.PlacementStrategySpread),
|
||||
}, testAccountID)
|
||||
require.NoError(t, err)
|
||||
lt, err := o.templates.CreateLaunchTemplate(ctx, &ec2.CreateLaunchTemplateInput{
|
||||
LaunchTemplateName: aws.String("secret-template-name"),
|
||||
LaunchTemplateData: &ec2.RequestLaunchTemplateData{UserData: aws.String(base64.StdEncoding.EncodeToString([]byte("secret-user-data")))},
|
||||
}, testAccountID)
|
||||
require.NoError(t, err)
|
||||
doomedID, ltID := aws.StringValue(doomed.PlacementGroup.GroupId), aws.StringValue(lt.LaunchTemplate.LaunchTemplateId)
|
||||
_, err = o.tags.CreateTags(ctx, &ec2.CreateTagsInput{
|
||||
Resources: []*string{aws.String(doomedID), aws.String(ltID)},
|
||||
Tags: []*ec2.Tag{{Key: aws.String("secret-tag-key"), Value: aws.String("secret-tag-value")}},
|
||||
}, testAccountID)
|
||||
require.NoError(t, err)
|
||||
_, err = o.groups.DeletePlacementGroup(ctx, &ec2.DeletePlacementGroupInput{GroupName: aws.String("secret-doomed-group")}, testAccountID)
|
||||
require.NoError(t, err)
|
||||
|
||||
engine, logs := shadowEngine(newEC2PlatformInventoryClient(o.nc), platformReapers(o.nc)...)
|
||||
result, err := engine.DryRun(ctx, testAccountID, "req-owners")
|
||||
require.NoError(t, err)
|
||||
plain, err := quietEngine(newFakeAccounts(), platformReapers(o.nc)...).Inventory(ctx, testAccountID)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, resultJSON(t, plain), resultJSON(t, result))
|
||||
|
||||
assert.NotContains(t, logs.String(), "secret")
|
||||
assert.NotContains(t, logs.String(), base64.StdEncoding.EncodeToString([]byte("secret-user-data")))
|
||||
record := shadowRecord(t, logs)
|
||||
assert.Equal(t, shadowDiffer, record["outcome"])
|
||||
keyPair := group(t, record, "keyPair")
|
||||
assert.Equal(t, shadowAgree, keyPair["comparison"])
|
||||
assert.Equal(t, map[string]any{ec2v1.PlatformClassPublicKeyWithoutMetadata: float64(1)}, keyPair["hiddenRecords"])
|
||||
assert.Equal(t, shadowAgree, group(t, record, "placementGroup")["comparison"])
|
||||
launchTemplate := group(t, record, "launchTemplate")
|
||||
assert.Equal(t, shadowAgree, launchTemplate["comparison"])
|
||||
assert.Equal(t, shadowBasisResourceID, launchTemplate["matchBasis"])
|
||||
assert.Equal(t, strs(doomedID), group(t, record, "tagIndex")["residue"])
|
||||
}
|
||||
@@ -29,9 +29,12 @@ type Engine struct {
|
||||
// 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 is EC2's read-only platform inventory. Only DryRun consults
|
||||
// it, and only to log a comparison; no teardown or Inventory call does.
|
||||
ec2Inventory ec2PlatformInventory
|
||||
|
||||
// ec2PlatformShadowBudget overrides ec2PlatformShadowBudget when positive.
|
||||
ec2PlatformShadowBudget time.Duration
|
||||
}
|
||||
|
||||
// NewEngine builds an engine with the default timeouts.
|
||||
@@ -87,6 +90,20 @@ func (e *Engine) Inventory(ctx context.Context, accountID string) (*Result, erro
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// DryRun is the explicit operator dry run. It returns exactly what Inventory
|
||||
// returns; beside it, EC2's platform inventory is compared with the legacy
|
||||
// listings and only logged, under correlationRef, which authorises nothing.
|
||||
func (e *Engine) DryRun(ctx context.Context, accountID, correlationRef string) (*Result, error) {
|
||||
result, err := e.Inventory(ctx, accountID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if e.ec2Inventory != nil {
|
||||
e.shadowEC2Platform(ctx, result, correlationRef)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// Teardown empties the account and then removes it.
|
||||
//
|
||||
// The account is marked TERMINATING before anything is deleted. Without that
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
@@ -54,6 +55,15 @@ func adminPathMethod(path string) (string, bool) {
|
||||
return rest, true
|
||||
}
|
||||
|
||||
// ctxAdminRequestID carries the /admin request ID to handlers that log under it.
|
||||
// It correlates log records only and authorises nothing.
|
||||
const ctxAdminRequestID contextKey = "admin.requestID"
|
||||
|
||||
func adminRequestIDFrom(ctx context.Context) string {
|
||||
id, _ := ctx.Value(ctxAdminRequestID).(string)
|
||||
return id
|
||||
}
|
||||
|
||||
// adminMaxBodyBytes caps the JSON request body. The largest legitimate request
|
||||
// is a few hundred bytes; the cap exists so an unauthenticated-shaped mistake
|
||||
// cannot make the gateway buffer megabytes.
|
||||
@@ -81,7 +91,7 @@ func (gw *GatewayConfig) Admin_Request(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("X-Amzn-Requestid", requestID)
|
||||
|
||||
method := chi.URLParam(r, "method")
|
||||
ctx := r.Context()
|
||||
ctx := context.WithValue(r.Context(), ctxAdminRequestID, requestID)
|
||||
|
||||
// One handler for every verb rather than a POST route plus a catch-all:
|
||||
// chi resolves the two to the same pattern, and the catch-all wins.
|
||||
|
||||
@@ -121,7 +121,7 @@ func (gw *GatewayConfig) adminDeleteAccount(ctx context.Context, body []byte) (a
|
||||
inventoryCtx, cancel := context.WithTimeout(ctx, accountDeletionInventoryTimeout)
|
||||
defer cancel()
|
||||
|
||||
inventory, err := engine.Inventory(inventoryCtx, req.AccountID)
|
||||
inventory, err := engine.DryRun(inventoryCtx, req.AccountID, adminRequestIDFrom(ctx))
|
||||
if err != nil {
|
||||
return nil, teardownError("DeleteAccount", req.AccountID, err)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,99 @@
|
||||
package gateway
|
||||
|
||||
//test:in-package — the admin dispatcher's request-ID context and the
|
||||
// deletion handler it reaches are unexported.
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/accountteardown"
|
||||
handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
const dryRunShadowMessage = "Account teardown dry-run EC2 platform shadow comparison"
|
||||
|
||||
// deleteAccountGrant is the real IAM service with a policy granting
|
||||
// spinifex:DeleteAccount, so a request clears the admin gates.
|
||||
type deleteAccountGrant struct {
|
||||
handlers_iam.IAMService
|
||||
}
|
||||
|
||||
func (deleteAccountGrant) GetUserPolicies(string, string) ([]handlers_iam.PolicyDocument, error) {
|
||||
return []handlers_iam.PolicyDocument{{
|
||||
Version: "2012-10-17",
|
||||
Statement: []handlers_iam.Statement{{
|
||||
Effect: "Allow", Action: handlers_iam.StringOrArr{"spinifex:DeleteAccount"}, Resource: handlers_iam.StringOrArr{"*"},
|
||||
}},
|
||||
}}, nil
|
||||
}
|
||||
|
||||
// syncBuffer is a log sink safe for the gateway's concurrent writers.
|
||||
type syncBuffer struct {
|
||||
mu sync.Mutex
|
||||
buf bytes.Buffer
|
||||
}
|
||||
|
||||
func (b *syncBuffer) Write(p []byte) (int, error) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
return b.buf.Write(p)
|
||||
}
|
||||
|
||||
func (b *syncBuffer) String() string {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
return b.buf.String()
|
||||
}
|
||||
|
||||
func captureDefaultLog(t *testing.T) *syncBuffer {
|
||||
t.Helper()
|
||||
logs := &syncBuffer{}
|
||||
prev := slog.Default()
|
||||
slog.SetDefault(slog.New(slog.NewJSONHandler(logs, nil)))
|
||||
t.Cleanup(func() { slog.SetDefault(prev) })
|
||||
return logs
|
||||
}
|
||||
|
||||
// The admin API dry run answers exactly as the plain inventory would, and its
|
||||
// shadow comparison is logged under the request ID the response carries.
|
||||
func TestAdminDeleteAccountDryRunLogsTheShadowUnderTheRequestID(t *testing.T) {
|
||||
gw := newDeleteAccountGateway(t)
|
||||
gw.IAMService = deleteAccountGrant{IAMService: gw.IAMService}
|
||||
account := newTenantAccount(t, gw, "tenant@example.com")
|
||||
logs := captureDefaultLog(t)
|
||||
|
||||
rec := authorizedAdminRequestForMethod(gw, "DeleteAccount", `{"accountId":"`+account.AccountID+`","dryRun":true}`)
|
||||
require.Equal(t, http.StatusOK, rec.Code, rec.Body.String())
|
||||
requestID := rec.Header().Get("X-Amzn-Requestid")
|
||||
require.NotEmpty(t, requestID)
|
||||
|
||||
engine, err := accountteardown.NewClusterEngine(t.Context(), gw.NATSConn, gw.ExpectedNodes, gw.NodeIDs, gw.IAMService, gw.BucketStore)
|
||||
require.NoError(t, err)
|
||||
inventory, err := engine.Inventory(t.Context(), account.AccountID)
|
||||
require.NoError(t, err)
|
||||
want, err := json.Marshal(&DeleteAccountResponse{AccountID: account.AccountID, State: "DRY_RUN", DryRun: true, Inventory: inventory})
|
||||
require.NoError(t, err)
|
||||
assert.JSONEq(t, string(want), rec.Body.String())
|
||||
|
||||
var records []map[string]any
|
||||
for line := range strings.SplitSeq(strings.TrimSpace(logs.String()), "\n") {
|
||||
var record map[string]any
|
||||
if json.Unmarshal([]byte(line), &record) == nil && record["msg"] == dryRunShadowMessage {
|
||||
records = append(records, record)
|
||||
}
|
||||
}
|
||||
require.Len(t, records, 1)
|
||||
assert.Equal(t, requestID, records[0]["correlationRef"])
|
||||
assert.Equal(t, account.AccountID, records[0]["accountID"])
|
||||
// The stub cluster answers every ec2 subject with an empty body, which is
|
||||
// no inventory for this account, so the comparison cannot be clean.
|
||||
assert.Equal(t, "scan-incomplete", records[0]["outcome"])
|
||||
}
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"slices"
|
||||
"strings"
|
||||
"time"
|
||||
"uuid"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/accountteardown"
|
||||
"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
|
||||
@@ -101,13 +102,7 @@ func runAccountDelete(cmd *cobra.Command, args []string) error {
|
||||
}
|
||||
|
||||
if dryRun {
|
||||
result, err := engine.Inventory(ctx, accountID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
printTeardownPlan(result)
|
||||
fmt.Printf("\nDry run: nothing was deleted. %d resources would be removed.\n", result.DeletedCount())
|
||||
return nil
|
||||
return runAccountDryRun(ctx, engine, accountID)
|
||||
}
|
||||
|
||||
// Always show the inventory before destroying it. An operator who has not
|
||||
@@ -149,6 +144,25 @@ func runAccountDelete(cmd *cobra.Command, args []string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// accountDryRunner is the engine's explicit dry run.
|
||||
type accountDryRunner interface {
|
||||
DryRun(ctx context.Context, accountID, correlationRef string) (*accountteardown.Result, error)
|
||||
}
|
||||
|
||||
var _ accountDryRunner = (*accountteardown.Engine)(nil)
|
||||
|
||||
// runAccountDryRun prints what a teardown would remove. Each run gets a fresh
|
||||
// correlation reference, which only ties it to its log records.
|
||||
func runAccountDryRun(ctx context.Context, engine accountDryRunner, accountID string) error {
|
||||
result, err := engine.DryRun(ctx, accountID, uuid.NewV4().String())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
printTeardownPlan(result)
|
||||
fmt.Printf("\nDry run: nothing was deleted. %d resources would be removed.\n", result.DeletedCount())
|
||||
return nil
|
||||
}
|
||||
|
||||
// promptAccountName asks for the confirmation name on a terminal.
|
||||
func promptAccountName(accountID string) (string, error) {
|
||||
fmt.Printf("\nThis permanently deletes account %s and everything above.\n", accountID)
|
||||
|
||||
@@ -2,9 +2,11 @@ package cli_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"io"
|
||||
"os"
|
||||
"testing"
|
||||
"uuid"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/accountteardown"
|
||||
"github.com/mulgadc/spinifex/spinifex/operator/cli"
|
||||
@@ -160,3 +162,47 @@ func TestPromptAccountNameTreatsAnEmptyReplyAsCancelled(t *testing.T) {
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "cancelled")
|
||||
}
|
||||
|
||||
// dryRunEngine stands in for the teardown engine's explicit dry run and keeps
|
||||
// the correlation reference each run was given.
|
||||
type dryRunEngine struct {
|
||||
result *accountteardown.Result
|
||||
err error
|
||||
refs []string
|
||||
}
|
||||
|
||||
func (e *dryRunEngine) DryRun(_ context.Context, _, correlationRef string) (*accountteardown.Result, error) {
|
||||
e.refs = append(e.refs, correlationRef)
|
||||
return e.result, e.err
|
||||
}
|
||||
|
||||
// The dry run prints what it always printed, and each run hands the engine a
|
||||
// fresh UUID to log its shadow comparison under.
|
||||
func TestRunAccountDryRunPrintsThePlanUnderAFreshCorrelationRef(t *testing.T) {
|
||||
engine := &dryRunEngine{result: teardownPlan()}
|
||||
|
||||
var err error
|
||||
output := captureOutput(t, func() { err = cli.RunAccountDryRun(t.Context(), engine, "000000000042") })
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "Account 000000000042 (tenant@example.com)\n\n"+
|
||||
" compute\n instance i-1 (running)\n"+
|
||||
" storage\n volume vol-1\n"+
|
||||
"\nDry run: nothing was deleted. 2 resources would be removed.\n", output)
|
||||
|
||||
captureOutput(t, func() { err = cli.RunAccountDryRun(t.Context(), engine, "000000000042") })
|
||||
require.NoError(t, err)
|
||||
require.Len(t, engine.refs, 2)
|
||||
for _, ref := range engine.refs {
|
||||
_, parseErr := uuid.Parse(ref)
|
||||
require.NoError(t, parseErr, ref)
|
||||
}
|
||||
assert.NotEqual(t, engine.refs[0], engine.refs[1])
|
||||
}
|
||||
|
||||
func TestRunAccountDryRunReturnsTheEngineError(t *testing.T) {
|
||||
engine := &dryRunEngine{err: accountteardown.ErrProtectedAccount}
|
||||
var err error
|
||||
output := captureOutput(t, func() { err = cli.RunAccountDryRun(t.Context(), engine, "000000000001") })
|
||||
require.ErrorIs(t, err, accountteardown.ErrProtectedAccount)
|
||||
assert.Empty(t, output)
|
||||
}
|
||||
|
||||
@@ -100,6 +100,9 @@ func SetAccountDeletePollInterval(d time.Duration) func() {
|
||||
// PrintTeardownPlan exposes printTeardownPlan for testing.
|
||||
var PrintTeardownPlan = printTeardownPlan
|
||||
|
||||
// RunAccountDryRun exposes runAccountDryRun for testing.
|
||||
var RunAccountDryRun = runAccountDryRun
|
||||
|
||||
// PrintTeardownResult exposes printTeardownResult for testing.
|
||||
var PrintTeardownResult = printTeardownResult
|
||||
|
||||
|
||||
Reference in New Issue
Block a user