console: hourly infrastructure inventory pass
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
4 files changed,
+278
−0
cmd/console/main.go+4 −0internal/console/infrapass.go+157 −0internal/console/infrapass_test.go+105 −0internal/console/server.go+12 −0
diff --git a/cmd/console/main.go b/cmd/console/main.go| index a70e25246..499b023ba 100644 |
| --- a/cmd/console/main.go |
| +++ b/cmd/console/main.go |
| @@ -317,6 +317,10 @@ func serve() { |
| // Same placement rule as the alert watcher: started here, never in |
| // NewServer, so constructing a Server stays goroutine-free. |
| go srv.StartEconomicsPass(context.Background()) |
| + // The infrastructure inventory pass writes infra/inventory.json for |
| + // /fleet/architecture (internal/console/infrapass.go). Same placement |
| + // rule as the economics pass: started here, never in NewServer. |
| + go srv.StartInfraPass(context.Background()) |
| addr := envOr("CONSOLE_ADDR", "127.0.0.1:8090") |
| log.Printf("console: %s listening on %s", origin, addr) |
| // ReadHeaderTimeout guards against Slowloris-style slow-header connection |
diff --git a/internal/console/infrapass.go b/internal/console/infrapass.go| new file mode 100644 |
| index 000000000..3f3c12b9b |
| --- /dev/null |
| +++ b/internal/console/infrapass.go |
| @@ -0,0 +1,157 @@ |
| +// The infrastructure inventory pass: once an hour, ask the deployment's |
| +// cloud provider what it is carrying and write the answer to |
| +// infraview.InventoryKey in the deployment bucket, where |
| +// /fleet/architecture reads it. |
| +// |
| +// The failure posture is cepull.go's, verbatim and for the same reasons. |
| +// Construction-class failures — no AWS config, no usable credential, an |
| +// AccessDenied-shaped answer to the first calls — latch the pass off for |
| +// the process after a single log line. There is no "the IAM grant might |
| +// appear later" retry worth making, and a deployment without the grant must |
| +// show "not granted" rather than filling its log with an hourly error. |
| +// Everything else — throttling, a 5xx, a network blip — logs and the next |
| +// tick tries again. |
| +package console |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "log" |
| + "os" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/awsinv" |
| + "github.com/carlosframework/platform/internal/infraview" |
| +) |
| + |
| +// infraPassInterval matches economicsPassInterval: an estate changes on the |
| +// timescale of a deploy, not a request, and the refresh button covers the |
| +// moment an operator actually needs it now. |
| +const infraPassInterval = time.Hour |
| + |
| +// infraPassTimeout bounds one whole inventory — every region, every |
| +// service. A pass that hangs must not hold the goroutine until the process |
| +// restarts. |
| +const infraPassTimeout = 2 * time.Minute |
| + |
| +// StartInfraPass runs infraPass immediately and then hourly until ctx is |
| +// done — StartEconomicsPass's shape (economicspass.go:38). |
| +func (s *Server) StartInfraPass(ctx context.Context) { |
| + if s.cfg.Releases == nil { |
| + return // no bucket, nowhere to write, nothing to do |
| + } |
| + t := time.NewTicker(infraPassInterval) |
| + defer t.Stop() |
| + for { |
| + if err := s.infraPass(ctx); err != nil { |
| + log.Printf("console: infra pass: %v", err) |
| + } |
| + select { |
| + case <-ctx.Done(): |
| + return |
| + case <-t.C: |
| + } |
| + } |
| +} |
| + |
| +// resolveInventory is the seam: the test fake when one is set, otherwise a |
| +// lazily constructed awsinv.Provider reused for the life of the process. |
| +func (s *Server) resolveInventory(ctx context.Context) (infraview.Inventory, error) { |
| + if s.inventory != nil { |
| + return s.inventory(ctx) |
| + } |
| + if s.infraDisabled { |
| + return infraview.Inventory{}, errors.New("console: infra inventory disabled for this process") |
| + } |
| + s.infraOnce.Do(func() { |
| + p, err := awsinv.New(ctx) |
| + if err != nil { |
| + // Construction failed. Auth-shaped means no grant: latch. |
| + // Anything else means this process cannot build a provider at |
| + // all, which is equally permanent for this process. |
| + log.Printf("console: infra inventory unavailable: %v", err) |
| + s.infraDisabled = true |
| + return |
| + } |
| + s.infraProvider = p |
| + }) |
| + if s.infraProvider == nil { |
| + s.infraDisabled = true |
| + return infraview.Inventory{}, errors.New("console: no infrastructure provider") |
| + } |
| + return s.infraProvider.Inventory(ctx) |
| +} |
| + |
| +// infraPass takes one inventory and stores it. An auth-shaped failure |
| +// latches; everything else is returned for the caller to log and retry. |
| +func (s *Server) infraPass(ctx context.Context) error { |
| + if s.infraDisabled { |
| + return errors.New("console: infra inventory disabled for this process") |
| + } |
| + ctx, cancel := context.WithTimeout(ctx, infraPassTimeout) |
| + defer cancel() |
| + |
| + inv, err := s.resolveInventory(ctx) |
| + if err != nil { |
| + if awsinv.IsAuthShaped(err) || isAuthShapedError(err) { |
| + s.infraDisabled = true |
| + return fmt.Errorf("console: infra inventory not granted (disabled for this process): %w", err) |
| + } |
| + return fmt.Errorf("console: infra inventory: %w", err) |
| + } |
| + return s.putInventory(ctx, inv) |
| +} |
| + |
| +// putInventory CAS-writes the snapshot — upsertBill's read-etag-then-put |
| +// (economicspass.go:942), with os.ErrNotExist meaning "first ever write". |
| +func (s *Server) putInventory(ctx context.Context, inv infraview.Inventory) error { |
| + if s.cfg.Releases == nil { |
| + return nil |
| + } |
| + _, etag, err := s.cfg.Releases.Get(ctx, infraview.InventoryKey) |
| + switch { |
| + case err == nil: |
| + case errors.Is(err, os.ErrNotExist): |
| + etag = "" |
| + default: |
| + return fmt.Errorf("console: inventory: read: %w", err) |
| + } |
| + body, err := json.Marshal(inv) |
| + if err != nil { |
| + return fmt.Errorf("console: inventory: marshal: %w", err) |
| + } |
| + if _, err := s.cfg.Releases.Put(ctx, infraview.InventoryKey, body, etag); err != nil { |
| + return fmt.Errorf("console: inventory: write: %w", err) |
| + } |
| + return nil |
| +} |
| + |
| +// loadInventory reads the stored snapshot for the page. The second return |
| +// is why there is nothing to show, and it distinguishes the three states |
| +// that must never be rendered as one: "never" (no pass has ever succeeded), |
| +// "ungranted" (this process latched off), and a read error. "no resources" |
| +// is a claim; none of these three earn it. |
| +func (s *Server) loadInventory(ctx context.Context) (*infraview.Inventory, string) { |
| + if s.cfg.Releases == nil { |
| + return nil, "never" |
| + } |
| + body, _, err := s.cfg.Releases.Get(ctx, infraview.InventoryKey) |
| + switch { |
| + case errors.Is(err, os.ErrNotExist): |
| + if s.infraDisabled { |
| + return nil, "ungranted" |
| + } |
| + return nil, "never" |
| + case err != nil: |
| + log.Printf("console: inventory read: %v", err) |
| + return nil, "unreadable" |
| + } |
| + var inv infraview.Inventory |
| + if err := json.Unmarshal(body, &inv); err != nil { |
| + log.Printf("console: inventory parse: %v", err) |
| + return nil, "unreadable" |
| + } |
| + return &inv, "" |
| +} |
diff --git a/internal/console/infrapass_test.go b/internal/console/infrapass_test.go| new file mode 100644 |
| index 000000000..459df2f1f |
| --- /dev/null |
| +++ b/internal/console/infrapass_test.go |
| @@ -0,0 +1,105 @@ |
| +package console |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "testing" |
| + "time" |
| + |
| + "github.com/aws/smithy-go" |
| + |
| + "github.com/carlosframework/platform/internal/infraview" |
| +) |
| + |
| +// fakeAPIError is the same four-method smithy.APIError shape used by |
| +// internal/awsinv's fakeAPIErr (rest_test.go) — the minimum a caller needs |
| +// to satisfy errors.As(err, &apiErr) and drive isAuthShapedError/ |
| +// awsinv.IsAuthShaped off ErrorCode() alone. |
| +type fakeAPIError struct{ code string } |
| + |
| +func (e *fakeAPIError) Error() string { return e.code } |
| +func (e *fakeAPIError) ErrorCode() string { return e.code } |
| +func (e *fakeAPIError) ErrorMessage() string { return e.code } |
| +func (e *fakeAPIError) ErrorFault() smithy.ErrorFault { return smithy.FaultClient } |
| + |
| +// The pass writes the snapshot where the page reads it, and loadInventory |
| +// reads back exactly what was written. |
| +func TestInfraPassWritesSnapshot(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + fx.srv.inventory = func(context.Context) (infraview.Inventory, error) { |
| + return infraview.Inventory{ |
| + Provider: "aws", Account: "1234", TakenAt: time.Now().UTC(), |
| + Resources: []infraview.Resource{{Provider: "aws", Kind: infraview.KindMachine, ID: "i-1"}}, |
| + }, nil |
| + } |
| + if err := fx.srv.infraPass(context.Background()); err != nil { |
| + t.Fatalf("infraPass: %v", err) |
| + } |
| + body, _, err := fx.rel.Get(context.Background(), infraview.InventoryKey) |
| + if err != nil { |
| + t.Fatalf("read back: %v", err) |
| + } |
| + var inv infraview.Inventory |
| + if err := json.Unmarshal(body, &inv); err != nil { |
| + t.Fatalf("parse: %v", err) |
| + } |
| + if len(inv.Resources) != 1 || inv.Resources[0].ID != "i-1" { |
| + t.Fatalf("stored = %+v", inv) |
| + } |
| + got, why := fx.srv.loadInventory(context.Background()) |
| + if got == nil || why != "" { |
| + t.Fatalf("loadInventory = %+v, %q; want the snapshot", got, why) |
| + } |
| +} |
| + |
| +// An auth-shaped failure latches the pass off for the process after ONE |
| +// attempt: "this deployment has no inventory grant" is a permanent answer, |
| +// not something to re-ask AWS every hour forever. |
| +func TestInfraPassLatchesOnAuthShapedFailure(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + calls := 0 |
| + fx.srv.inventory = func(context.Context) (infraview.Inventory, error) { |
| + calls++ |
| + return infraview.Inventory{}, &fakeAPIError{code: "AccessDenied"} |
| + } |
| + if err := fx.srv.infraPass(context.Background()); err == nil { |
| + t.Fatal("want an error from the denied pass") |
| + } |
| + if err := fx.srv.infraPass(context.Background()); err == nil { |
| + t.Fatal("want an error from the latched pass") |
| + } |
| + if calls != 1 { |
| + t.Fatalf("provider called %d times, want 1 — the latch must stop the second attempt", calls) |
| + } |
| + if _, why := fx.srv.loadInventory(context.Background()); why != "ungranted" { |
| + t.Fatalf("loadInventory reason = %q, want %q", why, "ungranted") |
| + } |
| +} |
| + |
| +// A transient failure is NOT the latch: CE throttles under load and a blip |
| +// must not silently kill inventory until the next redeploy. |
| +func TestInfraPassRetriesTransientFailure(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + calls := 0 |
| + fx.srv.inventory = func(context.Context) (infraview.Inventory, error) { |
| + calls++ |
| + return infraview.Inventory{}, errors.New("Throttling: rate exceeded") |
| + } |
| + _ = fx.srv.infraPass(context.Background()) |
| + _ = fx.srv.infraPass(context.Background()) |
| + if calls != 2 { |
| + t.Fatalf("provider called %d times, want 2 — a throttle must not latch", calls) |
| + } |
| +} |
| + |
| +// With no snapshot ever written, the page must be told "never", never |
| +// handed an empty inventory that would render as "this account has no |
| +// resources". |
| +func TestLoadInventoryNeverTaken(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + inv, why := fx.srv.loadInventory(context.Background()) |
| + if inv != nil || why != "never" { |
| + t.Fatalf("loadInventory = %+v, %q; want nil, \"never\"", inv, why) |
| + } |
| +} |
diff --git a/internal/console/server.go b/internal/console/server.go| index 660aef4c2..ea262b028 100644 |
| --- a/internal/console/server.go |
| +++ b/internal/console/server.go |
| @@ -23,6 +23,7 @@ import ( |
| "regexp" |
| "strings" |
| "sync" |
| + "sync/atomic" |
| "time" |
| "github.com/aws/aws-sdk-go-v2/service/costexplorer" |
| @@ -31,6 +32,7 @@ import ( |
| "github.com/carlosframework/platform/internal/dnsrec" |
| "github.com/carlosframework/platform/internal/domains" |
| "github.com/carlosframework/platform/internal/economics" |
| + "github.com/carlosframework/platform/internal/infraview" |
| "github.com/carlosframework/platform/internal/mail" |
| "github.com/carlosframework/platform/internal/objstore" |
| "github.com/carlosframework/platform/internal/provision" |
| @@ -616,6 +618,16 @@ type Server struct { |
| ceClient *costexplorer.Client |
| ceWarnOnce sync.Once |
| ceDisabled bool |
| + // The infrastructure inventory seam (infrapass.go), built exactly like |
| + // the Cost Explorer seam above it: tests set inventory to a fake, the |
| + // real path lazily constructs an awsinv.Provider once. infraDisabled |
| + // latches an auth-shaped failure off for the process — a deployment |
| + // with no inventory IAM grant answers "ungranted" once, not hourly. |
| + inventory func(ctx context.Context) (infraview.Inventory, error) |
| + infraOnce sync.Once |
| + infraProvider infraview.Provider |
| + infraDisabled bool |
| + infraRefreshing atomic.Bool |
| // shipSem bounds how many `carlos ship` uploads (apicli.go, |
| // handleAPIShip) may run AT ONCE, as opposed to s.cli.ship, which |
| // bounds them over TIME. A rate limiter is a counter; it says nothing |