Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8c3ccb55ac | ||
|
|
6422afdcf7 | ||
|
|
e92cf60378 | ||
|
|
8316db1f9b |
+13
-5
@@ -15,8 +15,8 @@ import (
|
||||
//
|
||||
// THE COMPLECTION IT REMOVES. The operator SPA authenticates via /v1/signin (a cloud
|
||||
// PKCE session → the X-User-* principal the middleware mints) but read its IDENTITY
|
||||
// from /v1/get-account, which the embedded IAM (casibase) answers from ITS OWN session
|
||||
// cookie. A PKCE session is not a casibase session, so get-account returned
|
||||
// from the account read, which the embedded IAM (casibase) answers from ITS OWN session
|
||||
// cookie. A PKCE session is not a casibase session, so that read returned
|
||||
// owner:"hanzo" (anonymous) or "Unauthorized operation" — and the SPA's SuperAdmin
|
||||
// gate (owner == "admin" && isAdmin), reading that, bounced the operator UI to login
|
||||
// even though the SAME session got 200 from every /v1/admin/* route. Two session
|
||||
@@ -24,16 +24,24 @@ import (
|
||||
//
|
||||
// THE DECOMPLECTION. Identity is now the principal: when a VALIDATED principal is
|
||||
// present (X-User-Id is set ONLY by IdentityMiddleware from a real credential, never a
|
||||
// raw client header — see middleware_identity.go), /v1/get-account reflects it. owner
|
||||
// raw client header — see middleware_identity.go), /v1/ai/account reflects it. owner
|
||||
// is the HOME org (principal.Owner) so a SuperAdmin org-switched into a tenant stays a
|
||||
// SuperAdmin; isAdmin is the validated bit. With NO principal it falls through
|
||||
// (c.Next()) to the casibase account surface unchanged — the anonymous sign-in page
|
||||
// and any legacy casibase-session caller are untouched. One truth, additive, fail-open
|
||||
// to the old path. MUST be registered AFTER IdentityMiddleware (needs the minted
|
||||
// headers) and BEFORE MountAll (so it precedes the casibase /v1/get-account handler).
|
||||
// headers) and BEFORE MountAll (so it precedes the casibase account handler).
|
||||
// accountPath is the account read this middleware fronts. It is a named constant
|
||||
// because the interception is a PATH MATCH: when the /v1 surface was namespaced
|
||||
// (/v1/get-account → /v1/ai/account) a literal left un-updated here would not
|
||||
// error — the middleware would simply stop firing, fall through to the casibase
|
||||
// account surface, and hand the SPA the anonymous owner again. That is precisely
|
||||
// the bug this file exists to fix, silently restored.
|
||||
const accountPath = "/v1/ai/account"
|
||||
|
||||
func AccountFromPrincipal() zip.Handler {
|
||||
return func(c *zip.Ctx) error {
|
||||
if c.Method() != http.MethodGet || c.Path() != "/v1/get-account" {
|
||||
if c.Method() != http.MethodGet || c.Path() != accountPath {
|
||||
return c.Next()
|
||||
}
|
||||
user := c.User() // X-User-Id — minted only from a validated credential
|
||||
|
||||
+2
-2
@@ -105,8 +105,8 @@ type Config struct {
|
||||
PlatformURL string `json:"platform_url,omitempty"`
|
||||
CloudURL string `json:"cloud_url,omitempty"`
|
||||
ClientID string `json:"client_id,omitempty"`
|
||||
APIKey string `json:"apiKey,omitempty"` // hk-… key; what `hanzo code` hands the agents
|
||||
CodeTool string `json:"code_tool,omitempty"` // default agent for bare `hanzo` / `hanzo code`: dev|claude|codex
|
||||
APIKey string `json:"apiKey,omitempty"` // hk-… key; what `hanzo code` hands the agents
|
||||
CodeTool string `json:"code_tool,omitempty"` // default agent for bare `hanzo` / `hanzo code`: dev|claude|codex
|
||||
CodeModel string `json:"code_model,omitempty"` // default model for `hanzo code` (else defaultCodeModel)
|
||||
}
|
||||
|
||||
|
||||
+12
-12
@@ -144,19 +144,19 @@ func openaiWire(base, token, _ string) map[string]string {
|
||||
}
|
||||
|
||||
type codeAgent struct {
|
||||
bin string // executable to exec
|
||||
wire wire // how it finds the cloud
|
||||
fullAuto []string // flags that bypass approval prompts
|
||||
continueArgs []string // harness-native form of Hanzo -c/--continue
|
||||
modelArg []string // how the model is passed on argv (empty: via env)
|
||||
carrier func(model string) string // maps the resolved model to a client-recognized id (claude: zen→carrier); nil = pass through
|
||||
bin string // executable to exec
|
||||
wire wire // how it finds the cloud
|
||||
fullAuto []string // flags that bypass approval prompts
|
||||
continueArgs []string // harness-native form of Hanzo -c/--continue
|
||||
modelArg []string // how the model is passed on argv (empty: via env)
|
||||
carrier func(model string) string // maps the resolved model to a client-recognized id (claude: zen→carrier); nil = pass through
|
||||
provider func(base, model string) []string // agents that need the endpoint declared, not just env'd
|
||||
clear []string // env that would shadow the wire (a stale key in the shell)
|
||||
configHome string // env var that relocates the agent's config dir to ~/.hanzo ("" = share the user's own install)
|
||||
seed func(dir string) error // one-time defaults for the isolated config dir
|
||||
appendSystem []string // --append-system-prompt + text; ALWAYS applied (identity, not a permission bypass — present in --safe too)
|
||||
mcp bool // auto-wire the Hanzo MCP server (code/vector/web/vision tools) as an stdio server scoped to the cwd
|
||||
install string // hint when the binary is missing
|
||||
clear []string // env that would shadow the wire (a stale key in the shell)
|
||||
configHome string // env var that relocates the agent's config dir to ~/.hanzo ("" = share the user's own install)
|
||||
seed func(dir string) error // one-time defaults for the isolated config dir
|
||||
appendSystem []string // --append-system-prompt + text; ALWAYS applied (identity, not a permission bypass — present in --safe too)
|
||||
mcp bool // auto-wire the Hanzo MCP server (code/vector/web/vision tools) as an stdio server scoped to the cwd
|
||||
install string // hint when the binary is missing
|
||||
}
|
||||
|
||||
// codeContextWindow is the input context (tokens) the served coding model
|
||||
|
||||
+8
-8
@@ -18,7 +18,7 @@ package cli
|
||||
// --serve-engine adds the `engine.serve` capability: the worker probes a local
|
||||
// hanzo-engine (the OpenAI + Anthropic model server on :1234), advertises its model
|
||||
// endpoint in the presence record, and prints (or with --register-provider, POSTs)
|
||||
// the /v1/add-provider call that routes api.hanzo.ai model traffic to this GPU. One
|
||||
// the POST /v1/ai/providers call that routes api.hanzo.ai model traffic to this GPU. One
|
||||
// fleet, two job types: engine.serve (model serving) alongside studio.render.
|
||||
|
||||
import (
|
||||
@@ -935,7 +935,7 @@ type connectOpts struct {
|
||||
serveEngine bool
|
||||
engineURL string // local URL to probe hanzo-engine
|
||||
engineEndpoint string // public URL to advertise (defaults to engineURL)
|
||||
registerProvider bool // auto POST /v1/add-provider for the engine
|
||||
registerProvider bool // auto POST /v1/ai/providers for the engine
|
||||
studioDir string // local Studio checkout to launch + supervise on :8188
|
||||
studioURL string // studio base the render mirror uploads finished images to
|
||||
mirror bool // sweep local renders into the org studio library (default on)
|
||||
@@ -2228,7 +2228,7 @@ func describeEngine(adv *engineAdvertisement) string {
|
||||
return fmt.Sprintf("ready · %d models", n)
|
||||
}
|
||||
|
||||
// providerBody is the POST /v1/add-provider payload registering this node's engine
|
||||
// providerBody is the POST /v1/ai/providers payload registering this node's engine
|
||||
// as an org model provider. hanzo-engine is OpenAI-compatible, so Type=Local: the
|
||||
// gateway speaks the OpenAI wire format to it and auto-appends /v1 to providerUrl.
|
||||
func (w *worker) providerBody() map[string]any {
|
||||
@@ -2260,23 +2260,23 @@ func (w *worker) printEngineHint(out io.Writer) {
|
||||
fmt.Fprintf(out, "serving hanzo-engine (OpenAI + Anthropic) at %s — %s\n", adv.URL, describeEngine(adv))
|
||||
body, _ := json.Marshal(w.providerBody())
|
||||
fmt.Fprintln(out, " route api.hanzo.ai model calls to this GPU by registering it as an org provider:")
|
||||
fmt.Fprintf(out, " curl -sS %s/v1/add-provider -H \"Authorization: Bearer $HANZO_TOKEN\" \\\n", w.baseURL)
|
||||
fmt.Fprintf(out, " curl -sS %s/v1/ai/providers -H \"Authorization: Bearer $HANZO_TOKEN\" \\\n", w.baseURL)
|
||||
fmt.Fprintf(out, " -H 'Content-Type: application/json' -d '%s'\n", body)
|
||||
fmt.Fprintln(out, " (or pass --register-provider. The endpoint must be reachable from api.hanzo.ai —")
|
||||
fmt.Fprintln(out, " a cloud GPU is in-cluster; a BYO node needs a public URL/tunnel. add-provider needs a platform-admin token today.)")
|
||||
fmt.Fprintln(out, " a cloud GPU is in-cluster; a BYO node needs a public URL/tunnel. registering a provider needs a platform-admin token today.)")
|
||||
}
|
||||
|
||||
// registerProvider POSTs /v1/add-provider so the gateway routes model calls to this
|
||||
// registerProvider POSTs /v1/ai/providers so the gateway routes model calls to this
|
||||
// node's engine. Requires the engine to be reachable and (today) a platform-admin
|
||||
// token; both failures are reported clearly rather than swallowed.
|
||||
func (w *worker) registerProvider(ctx context.Context, adv *engineAdvertisement) error {
|
||||
if adv == nil || adv.Status != "ready" {
|
||||
return fmt.Errorf("engine not ready at %s — start hanzo-engine, then retry", w.engineURL)
|
||||
}
|
||||
code, err := w.call(ctx, http.MethodPost, "/v1/add-provider", w.providerBody(), nil)
|
||||
code, err := w.call(ctx, http.MethodPost, "/v1/ai/providers", w.providerBody(), nil)
|
||||
if err != nil {
|
||||
if code == http.StatusForbidden {
|
||||
return fmt.Errorf("add-provider is gated to a platform-admin token today; register from the console or with an admin token: %w", err)
|
||||
return fmt.Errorf("registering a provider is gated to a platform-admin token today; register from the console or with an admin token: %w", err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
+1
-1
@@ -72,7 +72,7 @@ func newLinkCmd(envOf func() *Env, _ *globalFlags) *cobra.Command {
|
||||
f.BoolVar(&opts.serveEngine, "serve-engine", false, "also advertise a hanzo-engine model server (OpenAI + Anthropic) running on this node")
|
||||
f.StringVar(&opts.engineURL, "engine-url", defaultEngineURL, "local URL where hanzo-engine is probed (GET /v1/models)")
|
||||
f.StringVar(&opts.engineEndpoint, "engine-endpoint", "", "public URL to advertise for gateway routing (defaults to --engine-url; a node behind NAT needs a reachable URL/tunnel)")
|
||||
f.BoolVar(&opts.registerProvider, "register-provider", false, "auto-register the engine endpoint as an org model provider (POST /v1/add-provider)")
|
||||
f.BoolVar(&opts.registerProvider, "register-provider", false, "auto-register the engine endpoint as an org model provider (POST /v1/ai/providers)")
|
||||
f.StringVar(&opts.studioDir, "studio-dir", os.Getenv("HANZO_STUDIO_DIR"), "local Hanzo Studio checkout; when set, link launches and supervises the render backend on 127.0.0.1:8188")
|
||||
f.StringVar(&opts.studioURL, "studio-url", firstNonEmpty(os.Getenv("HANZO_STUDIO_UPLOAD_URL"), defaultStudioUploadURL), "studio base URL the render mirror uploads finished images to (POST /v1/library/upload)")
|
||||
f.BoolVar(&opts.mirror, "mirror", true, "sweep local renders into the org studio library; --mirror=false serves jobs only")
|
||||
|
||||
@@ -39,6 +39,7 @@ import (
|
||||
"github.com/hanzoai/cloud/clients/admin/finance"
|
||||
"github.com/hanzoai/cloud/clients/admin/health"
|
||||
"github.com/hanzoai/cloud/clients/admin/iam"
|
||||
"github.com/hanzoai/cloud/clients/admin/infra"
|
||||
"github.com/hanzoai/cloud/clients/admin/invoices"
|
||||
"github.com/hanzoai/cloud/clients/admin/metrics"
|
||||
"github.com/hanzoai/cloud/clients/admin/revenue"
|
||||
@@ -141,6 +142,7 @@ func routes(app *zip.App, s *cloud.Service[core.State]) {
|
||||
revenue.Routes(app, s)
|
||||
finance.Routes(app, s)
|
||||
metrics.Routes(app, s)
|
||||
infra.Routes(app, s)
|
||||
invoices.Routes(app, s)
|
||||
subscriptions.Routes(app, s)
|
||||
}
|
||||
|
||||
@@ -107,6 +107,10 @@ var platformAdminRoutes = []adminRoute{
|
||||
{"GET", "/v1/admin/flags"},
|
||||
{"GET", "/v1/admin/waitlist"},
|
||||
{"POST", "/v1/admin/waitlist/boost"},
|
||||
{"GET", "/v1/admin/infra"},
|
||||
{"POST", "/v1/admin/infra/volumes/v1/snapshot"},
|
||||
{"DELETE", "/v1/admin/infra/volumes/v1"},
|
||||
{"POST", "/v1/admin/infra/nodes/1/cordon"},
|
||||
}
|
||||
|
||||
// adminRoutes is the full surface (both tiers) — the fail-closed gate test denies an
|
||||
|
||||
@@ -1,7 +1,12 @@
|
||||
// Package do reads DigitalOcean's billing API for the finance dashboard's cost
|
||||
// side. DO is our PRIMARY venue (a large promotional credit); this client turns
|
||||
// the customer balance + billing history into money.Cents the finance aggregator
|
||||
// folds into gross margin and runway.
|
||||
// Package digitalocean reads DigitalOcean's billing and infrastructure APIs. DO is
|
||||
// our PRIMARY venue (a large promotional credit); this client turns the customer
|
||||
// balance + billing history into money.Cents the finance aggregator folds into gross
|
||||
// margin and runway, and exposes the account's physical inventory — droplets,
|
||||
// block-storage volumes, DOKS clusters, load balancers — that the /v1/admin/infra
|
||||
// board reads.
|
||||
//
|
||||
// This is the ONE DigitalOcean client the admin plane uses. A new DO read is a
|
||||
// method here calling the shared get/send primitive, never a second client.
|
||||
//
|
||||
// Auth is a single personal-access token, DO_API_TOKEN, sourced from a KMSSecret on
|
||||
// the cloud env — NEVER hard-coded. When the token is unset the client is not Ready
|
||||
@@ -15,12 +20,14 @@
|
||||
package digitalocean
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"math"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -152,6 +159,12 @@ type Volume struct {
|
||||
Region string
|
||||
SizeGiB int
|
||||
DropletIDs []int
|
||||
// Tags carries DO's resource tags. DOKS stamps `k8s:<cluster-uuid>` on the volumes
|
||||
// it provisions, but that tag is ADVISORY ONLY — it survives cluster deletion and
|
||||
// is wrong often enough that it must never decide whether a volume is garbage. The
|
||||
// only sound liveness test is a PV cross-reference (see clients/admin/infra).
|
||||
Tags []string
|
||||
CreatedAt string
|
||||
}
|
||||
|
||||
// volumeWire is the raw DO /v2/volumes row.
|
||||
@@ -162,43 +175,326 @@ type volumeWire struct {
|
||||
Region struct {
|
||||
Slug string `json:"slug"`
|
||||
} `json:"region"`
|
||||
DropletIDs []int `json:"droplet_ids"`
|
||||
DropletIDs []int `json:"droplet_ids"`
|
||||
Tags []string `json:"tags"`
|
||||
CreatedAt string `json:"created_at"`
|
||||
}
|
||||
|
||||
// Volumes lists ALL block-storage volumes across the account, following DO's
|
||||
// page-number pagination (200/page, a hard 25-page cap so a runaway can never loop).
|
||||
// The fleet inventory (count, capacity, monthly cost) is real; per-volume fill is NOT
|
||||
// exposed by DO and stays absent (honest) until a filesystem source reports it.
|
||||
// Volumes lists ALL block-storage volumes across the account. Capacity and attachment
|
||||
// are real; per-volume fill is NOT exposed by DO and stays absent (honest) until a
|
||||
// filesystem source reports it.
|
||||
func (c *Client) Volumes(ctx context.Context) ([]Volume, error) {
|
||||
rows, err := listAll[volumeWire](ctx, c, "/v2/volumes", "volumes")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]Volume, len(rows))
|
||||
for i, v := range rows {
|
||||
out[i] = Volume{
|
||||
ID: v.ID,
|
||||
Name: v.Name,
|
||||
Region: v.Region.Slug,
|
||||
SizeGiB: v.SizeGigabytes,
|
||||
DropletIDs: v.DropletIDs,
|
||||
Tags: v.Tags,
|
||||
CreatedAt: v.CreatedAt,
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// Droplet is one DO droplet. LocalDiskGiB is the droplet's own disk, which is
|
||||
// INCLUDED in MonthlyCents — it is NOT separately billed, and conflating it with
|
||||
// block storage is how a fleet appears to hold terabytes it never pays for.
|
||||
type Droplet struct {
|
||||
ID int
|
||||
Name string
|
||||
Region string
|
||||
Status string
|
||||
SizeSlug string
|
||||
VCPUs int
|
||||
MemoryMiB int
|
||||
LocalDiskGiB int
|
||||
MonthlyCents money.Cents
|
||||
CreatedAt string
|
||||
PrivateIP string
|
||||
PublicIP string
|
||||
Tags []string
|
||||
VolumeIDs []string
|
||||
}
|
||||
|
||||
// dropletWire is the raw DO /v2/droplets row.
|
||||
type dropletWire struct {
|
||||
ID int `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Status string `json:"status"`
|
||||
SizeSlug string `json:"size_slug"`
|
||||
VCPUs int `json:"vcpus"`
|
||||
Memory int `json:"memory"`
|
||||
Disk int `json:"disk"`
|
||||
Size struct {
|
||||
PriceMonthly float64 `json:"price_monthly"`
|
||||
} `json:"size"`
|
||||
Region struct {
|
||||
Slug string `json:"slug"`
|
||||
} `json:"region"`
|
||||
Networks struct {
|
||||
V4 []struct {
|
||||
Type string `json:"type"`
|
||||
IPAddress string `json:"ip_address"`
|
||||
} `json:"v4"`
|
||||
} `json:"networks"`
|
||||
CreatedAt string `json:"created_at"`
|
||||
Tags []string `json:"tags"`
|
||||
VolumeIDs []string `json:"volume_ids"`
|
||||
}
|
||||
|
||||
// Droplets lists ALL droplets across the account.
|
||||
func (c *Client) Droplets(ctx context.Context) ([]Droplet, error) {
|
||||
rows, err := listAll[dropletWire](ctx, c, "/v2/droplets", "droplets")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]Droplet, len(rows))
|
||||
for i, d := range rows {
|
||||
dr := Droplet{
|
||||
ID: d.ID,
|
||||
Name: d.Name,
|
||||
Region: d.Region.Slug,
|
||||
Status: d.Status,
|
||||
SizeSlug: d.SizeSlug,
|
||||
VCPUs: d.VCPUs,
|
||||
MemoryMiB: d.Memory,
|
||||
LocalDiskGiB: d.Disk,
|
||||
MonthlyCents: centsOf(d.Size.PriceMonthly),
|
||||
CreatedAt: d.CreatedAt,
|
||||
Tags: d.Tags,
|
||||
VolumeIDs: d.VolumeIDs,
|
||||
}
|
||||
for _, n := range d.Networks.V4 {
|
||||
switch n.Type {
|
||||
case "private":
|
||||
dr.PrivateIP = n.IPAddress
|
||||
case "public":
|
||||
dr.PublicIP = n.IPAddress
|
||||
}
|
||||
}
|
||||
out[i] = dr
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// Cluster is one DOKS cluster.
|
||||
type Cluster struct {
|
||||
ID string
|
||||
Name string
|
||||
Region string
|
||||
Version string
|
||||
Status string
|
||||
NodePools int
|
||||
CreatedAt string
|
||||
}
|
||||
|
||||
// clusterWire is the raw DO /v2/kubernetes/clusters row.
|
||||
type clusterWire struct {
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Region string `json:"region"`
|
||||
Version string `json:"version"`
|
||||
Status struct {
|
||||
State string `json:"state"`
|
||||
} `json:"status"`
|
||||
NodePools []struct {
|
||||
Name string `json:"name"`
|
||||
} `json:"node_pools"`
|
||||
CreatedAt string `json:"created_at"`
|
||||
}
|
||||
|
||||
// Clusters lists ALL DOKS clusters. This is the authoritative denominator for the
|
||||
// orphan analysis: a volume may only be called unreferenced once EVERY cluster here
|
||||
// has been searched for a PV that claims it.
|
||||
func (c *Client) Clusters(ctx context.Context) ([]Cluster, error) {
|
||||
rows, err := listAll[clusterWire](ctx, c, "/v2/kubernetes/clusters", "kubernetes_clusters")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]Cluster, len(rows))
|
||||
for i, k := range rows {
|
||||
out[i] = Cluster{
|
||||
ID: k.ID,
|
||||
Name: k.Name,
|
||||
Region: k.Region,
|
||||
Version: k.Version,
|
||||
Status: k.Status.State,
|
||||
NodePools: len(k.NodePools),
|
||||
CreatedAt: k.CreatedAt,
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// Kubeconfig fetches a cluster's admin kubeconfig. DO returns a token-based config
|
||||
// against the cluster's public https endpoint (never an exec plugin), which the
|
||||
// caller must still funnel through fleet.SafeRESTConfig before dialing.
|
||||
func (c *Client) Kubeconfig(ctx context.Context, clusterID string) ([]byte, error) {
|
||||
if !c.Ready() {
|
||||
return nil, fmt.Errorf("DO_API_TOKEN not configured")
|
||||
}
|
||||
var out []Volume
|
||||
for page := 1; page <= 25; page++ {
|
||||
body, err := c.get(ctx, "/v2/volumes?per_page=200&page="+strconv.Itoa(page))
|
||||
if strings.TrimSpace(clusterID) == "" {
|
||||
return nil, fmt.Errorf("cluster id required")
|
||||
}
|
||||
return c.get(ctx, "/v2/kubernetes/clusters/"+url.PathEscape(clusterID)+"/kubeconfig")
|
||||
}
|
||||
|
||||
// LoadBalancer is one DO load balancer. DO does not price LBs in the API, so cost is
|
||||
// derived from the billed unit count (see lbUnitCents).
|
||||
type LoadBalancer struct {
|
||||
ID string
|
||||
Name string
|
||||
Region string
|
||||
Status string
|
||||
IP string
|
||||
SizeUnit int
|
||||
MonthlyCents money.Cents
|
||||
DropletIDs []int
|
||||
}
|
||||
|
||||
// lbWire is the raw DO /v2/load_balancers row.
|
||||
type lbWire struct {
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Status string `json:"status"`
|
||||
IP string `json:"ip"`
|
||||
Region struct {
|
||||
Slug string `json:"slug"`
|
||||
} `json:"region"`
|
||||
SizeUnit int `json:"size_unit"`
|
||||
DropletIDs []int `json:"droplet_ids"`
|
||||
}
|
||||
|
||||
// LoadBalancers lists ALL load balancers across the account.
|
||||
func (c *Client) LoadBalancers(ctx context.Context) ([]LoadBalancer, error) {
|
||||
rows, err := listAll[lbWire](ctx, c, "/v2/load_balancers", "load_balancers")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]LoadBalancer, len(rows))
|
||||
for i, l := range rows {
|
||||
units := l.SizeUnit
|
||||
if units <= 0 {
|
||||
units = 1
|
||||
}
|
||||
out[i] = LoadBalancer{
|
||||
ID: l.ID,
|
||||
Name: l.Name,
|
||||
Region: l.Region.Slug,
|
||||
Status: l.Status,
|
||||
IP: l.IP,
|
||||
SizeUnit: units,
|
||||
MonthlyCents: money.Cents(units) * lbUnitCents,
|
||||
DropletIDs: l.DropletIDs,
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// Snapshot is a created block-storage snapshot.
|
||||
type Snapshot struct {
|
||||
ID string
|
||||
Name string
|
||||
SizeGiB int
|
||||
}
|
||||
|
||||
// SnapshotVolume takes a point-in-time snapshot of a volume. This is the "undo" that
|
||||
// makes a delete recoverable, so the delete path takes one FIRST by default.
|
||||
func (c *Client) SnapshotVolume(ctx context.Context, volumeID, name string) (Snapshot, error) {
|
||||
var out Snapshot
|
||||
if !c.Ready() {
|
||||
return out, fmt.Errorf("DO_API_TOKEN not configured")
|
||||
}
|
||||
if strings.TrimSpace(volumeID) == "" {
|
||||
return out, fmt.Errorf("volume id required")
|
||||
}
|
||||
body, err := c.send(ctx, http.MethodPost, "/v2/volumes/"+url.PathEscape(volumeID)+"/snapshots",
|
||||
map[string]string{"name": name})
|
||||
if err != nil {
|
||||
return out, err
|
||||
}
|
||||
var w struct {
|
||||
Snapshot struct {
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
SizeGigabytes int `json:"size_gigabytes"`
|
||||
} `json:"snapshot"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &w); err != nil {
|
||||
return out, fmt.Errorf("do snapshot decode: %w", err)
|
||||
}
|
||||
return Snapshot{ID: w.Snapshot.ID, Name: w.Snapshot.Name, SizeGiB: w.Snapshot.SizeGigabytes}, nil
|
||||
}
|
||||
|
||||
// DeleteVolume destroys a block-storage volume. Irreversible: callers MUST have
|
||||
// proven the volume is referenced by no PV in any cluster first.
|
||||
func (c *Client) DeleteVolume(ctx context.Context, volumeID string) error {
|
||||
if !c.Ready() {
|
||||
return fmt.Errorf("DO_API_TOKEN not configured")
|
||||
}
|
||||
if strings.TrimSpace(volumeID) == "" {
|
||||
return fmt.Errorf("volume id required")
|
||||
}
|
||||
_, err := c.send(ctx, http.MethodDelete, "/v2/volumes/"+url.PathEscape(volumeID), nil)
|
||||
return err
|
||||
}
|
||||
|
||||
// Pagination bounds for every DO collection read: 200 rows a page, a hard 25-page
|
||||
// cap so a runaway can never loop, and the 8 MiB body ceiling a full droplet page
|
||||
// needs (a volume page fits in far less).
|
||||
const (
|
||||
perPage = 200
|
||||
maxPages = 25
|
||||
maxBody = 8 << 20
|
||||
maxRespLen = maxBody
|
||||
)
|
||||
|
||||
// lbUnitCents is DO's published price for one load-balancer node ($12/mo). DO does
|
||||
// not return LB pricing in the API, so this is the one place the rate is written.
|
||||
const lbUnitCents = money.Cents(1200)
|
||||
|
||||
// listAll follows DO's page-number pagination for a collection endpoint, decoding
|
||||
// rows out of the response's named key. It is the ONE pagination loop in this client
|
||||
// — every collection read goes through it, so "stop on the short page or the reported
|
||||
// total" is stated once and cannot drift between endpoints.
|
||||
func listAll[T any](ctx context.Context, c *Client, path, key string) ([]T, error) {
|
||||
if !c.Ready() {
|
||||
return nil, fmt.Errorf("DO_API_TOKEN not configured")
|
||||
}
|
||||
var out []T
|
||||
for page := 1; page <= maxPages; page++ {
|
||||
body, err := c.get(ctx, fmt.Sprintf("%s?per_page=%d&page=%d", path, perPage, page))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var w struct {
|
||||
Volumes []volumeWire `json:"volumes"`
|
||||
Meta struct {
|
||||
Meta struct {
|
||||
Total int `json:"total"`
|
||||
} `json:"meta"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &w); err != nil {
|
||||
return nil, fmt.Errorf("do volumes decode: %w", err)
|
||||
return nil, fmt.Errorf("do %s decode: %w", key, err)
|
||||
}
|
||||
for _, v := range w.Volumes {
|
||||
out = append(out, Volume{
|
||||
ID: v.ID,
|
||||
Name: v.Name,
|
||||
Region: v.Region.Slug,
|
||||
SizeGiB: v.SizeGigabytes,
|
||||
DropletIDs: v.DropletIDs,
|
||||
})
|
||||
var keyed map[string]json.RawMessage
|
||||
if err := json.Unmarshal(body, &keyed); err != nil {
|
||||
return nil, fmt.Errorf("do %s decode: %w", key, err)
|
||||
}
|
||||
var rows []T
|
||||
if raw, ok := keyed[key]; ok && len(raw) > 0 {
|
||||
if err := json.Unmarshal(raw, &rows); err != nil {
|
||||
return nil, fmt.Errorf("do %s decode: %w", key, err)
|
||||
}
|
||||
}
|
||||
out = append(out, rows...)
|
||||
// Stop on the last (short) page, or once we've collected the reported total.
|
||||
if len(w.Volumes) < 200 || (w.Meta.Total > 0 && len(out) >= w.Meta.Total) {
|
||||
if len(rows) < perPage || (w.Meta.Total > 0 && len(out) >= w.Meta.Total) {
|
||||
break
|
||||
}
|
||||
}
|
||||
@@ -207,22 +503,48 @@ func (c *Client) Volumes(ctx context.Context) ([]Volume, error) {
|
||||
|
||||
// get performs one token-authenticated DO GET and returns the raw body.
|
||||
func (c *Client) get(ctx context.Context, path string) ([]byte, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.base+path, nil)
|
||||
return c.send(ctx, http.MethodGet, path, nil)
|
||||
}
|
||||
|
||||
// send performs one token-authenticated DO request and returns the raw body. It is
|
||||
// the single HTTP primitive of this client: every read and every mutation funnels
|
||||
// through it, so auth, timeouts, the body ceiling and status handling exist once.
|
||||
func (c *Client) send(ctx context.Context, method, path string, payload any) ([]byte, error) {
|
||||
var rdr io.Reader
|
||||
if payload != nil {
|
||||
enc, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
rdr = bytes.NewReader(enc)
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, method, c.base+path, rdr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req.Header.Set("Accept", "application/json")
|
||||
req.Header.Set("Authorization", "Bearer "+c.token)
|
||||
if payload != nil {
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
}
|
||||
resp, err := c.http.Do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("digitalocean unreachable: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
||||
body, err := io.ReadAll(io.LimitReader(resp.Body, maxRespLen))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
// DO returns {"id":"...","message":"..."} on error — surface the message so a
|
||||
// failed mutation says WHY, not just a bare status.
|
||||
var e struct {
|
||||
Message string `json:"message"`
|
||||
}
|
||||
if json.Unmarshal(body, &e) == nil && strings.TrimSpace(e.Message) != "" {
|
||||
return nil, fmt.Errorf("digitalocean status %d: %s", resp.StatusCode, e.Message)
|
||||
}
|
||||
return nil, fmt.Errorf("digitalocean status %d", resp.StatusCode)
|
||||
}
|
||||
return body, nil
|
||||
@@ -241,5 +563,8 @@ func dollarsToCents(s string) money.Cents {
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
return money.Cents(math.Round(f * 100))
|
||||
return centsOf(f)
|
||||
}
|
||||
|
||||
// centsOf rounds decimal dollars to integer cents.
|
||||
func centsOf(f float64) money.Cents { return money.Cents(math.Round(f * 100)) }
|
||||
|
||||
@@ -0,0 +1,768 @@
|
||||
// Package infra is the platform's DigitalOcean fleet board: the physical inventory
|
||||
// (DOKS clusters, droplets, block-storage volumes, load balancers) cross-referenced
|
||||
// against what every cluster's Kubernetes actually claims, with the cost of each and
|
||||
// an orphan analysis that is safe BY CONSTRUCTION.
|
||||
//
|
||||
// THE RULE THIS PACKAGE EXISTS TO ENFORCE. A volume being detached, or carrying a
|
||||
// `k8s:<cluster-uuid>` tag for some other cluster, does NOT make it garbage. Those two
|
||||
// signals together would have condemned 4.39 TiB of live data belonging to running
|
||||
// clusters. The ONLY sound liveness test is a cross-reference against the
|
||||
// `spec.csi.volumeHandle` of every PersistentVolume in EVERY cluster — and it is only
|
||||
// a valid test when every cluster answered. So:
|
||||
//
|
||||
// - a volume is deletable only when NO PV in ANY cluster names it, and
|
||||
// - if even one cluster failed to scan, NOTHING is deletable (Complete=false).
|
||||
//
|
||||
// Absence of evidence is not evidence of absence: an unreachable cluster is treated as
|
||||
// a cluster that might be holding the volume. The analysis fails CLOSED.
|
||||
//
|
||||
// "No pod mounts it" is a REVIEW signal, never a delete signal — an idle Bound PVC is
|
||||
// an idle database, not garbage. Idle volumes are surfaced as a queue for a human and
|
||||
// are never counted as reclaimable.
|
||||
//
|
||||
// Analyze is a pure function of (DO inventory, cluster scans) so every rule above is
|
||||
// unit-testable without a network or a cluster.
|
||||
package infra
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/hanzoai/cloud/clients/admin/core"
|
||||
"github.com/hanzoai/cloud/clients/admin/digitalocean"
|
||||
"github.com/hanzoai/cloud/clients/admin/money"
|
||||
)
|
||||
|
||||
// volumeGiBCents is DO's block-storage rate: $0.10 per GiB per month. Droplet LOCAL
|
||||
// disk is NOT billed at this rate (or at all, separately) — it is included in the
|
||||
// droplet's own price. Conflating the two invents terabytes of phantom cost.
|
||||
const volumeGiBCents = money.Cents(10)
|
||||
|
||||
// outlierShareBP flags any single resource costing at least this share of total fleet
|
||||
// spend, in basis points (500bp = 5%). One explicable rule, no magic thresholds.
|
||||
const outlierShareBP = money.Cents(500)
|
||||
|
||||
// Volume states. The state machine is total and ordered: attachment beats reference,
|
||||
// reference beats absence. Only Unreferenced is ever deletable.
|
||||
const (
|
||||
// StateAttached — DO reports the volume attached to a droplet. In use, now.
|
||||
StateAttached = "attached"
|
||||
// StateBound — detached, but a PV in some cluster claims it and that PV is Bound
|
||||
// to a PVC. This is live data between mounts; deleting it destroys a database.
|
||||
StateBound = "bound"
|
||||
// StateReleased — a PV claims it but is no longer Bound (Released/Available/Failed).
|
||||
// A genuine cleanup candidate, but the PV still exists, so a human retires the PV.
|
||||
StateReleased = "released"
|
||||
// StateUnreferenced — no PV in ANY scanned cluster names it. The ONLY deletable state.
|
||||
StateUnreferenced = "unreferenced"
|
||||
)
|
||||
|
||||
// Finding severities.
|
||||
const (
|
||||
SevCritical = "critical"
|
||||
SevWarn = "warn"
|
||||
SevInfo = "info"
|
||||
)
|
||||
|
||||
// firstParty are our own registries: anything here is ours by construction.
|
||||
var firstParty = []string{
|
||||
"ghcr.io/hanzoai/", "ghcr.io/luxfi/", "ghcr.io/zooai/",
|
||||
"registry.hanzo.ai/", "registry.lux.network/", "registry.zoo.network/",
|
||||
"registry.digitalocean.com/hanzo/",
|
||||
}
|
||||
|
||||
// knownVendors is the REVIEWED third-party set — upstream images we deliberately run.
|
||||
// Kept deliberately short: an image outside both lists is reported for a human to
|
||||
// judge, which is the point. Growing this list is a review decision, not a reflex.
|
||||
var knownVendors = []string{
|
||||
"docker.io/library/", "library/", "registry.k8s.io/", "k8s.gcr.io/", "quay.io/",
|
||||
"grafana/", "prom/", "prometheus/", "bitnami/", "minio/", "moby/",
|
||||
"digitalocean/", "docker.digitalocean.com/", "acmglobaltech/", "hanzozt/",
|
||||
}
|
||||
|
||||
// Inventory is the DigitalOcean account read — the half of the analysis input that
|
||||
// needs no cluster.
|
||||
type Inventory struct {
|
||||
Clusters []digitalocean.Cluster
|
||||
Droplets []digitalocean.Droplet
|
||||
Volumes []digitalocean.Volume
|
||||
LoadBalancers []digitalocean.LoadBalancer
|
||||
}
|
||||
|
||||
// PVRef is one PersistentVolume's identity: which DO volume it claims, and whether
|
||||
// that claim is still live.
|
||||
type PVRef struct {
|
||||
Name string
|
||||
Phase string
|
||||
VolumeHandle string
|
||||
ClaimNS string
|
||||
ClaimName string
|
||||
}
|
||||
|
||||
// PVCRef is one PersistentVolumeClaim.
|
||||
type PVCRef struct {
|
||||
Namespace string
|
||||
Name string
|
||||
Phase string
|
||||
Volume string
|
||||
}
|
||||
|
||||
// PodRef is one pod, reduced to what the board needs: where it runs, whether it is
|
||||
// healthy, which PVCs it mounts, and what images it runs.
|
||||
type PodRef struct {
|
||||
Namespace string
|
||||
Name string
|
||||
Phase string
|
||||
Reason string
|
||||
Node string
|
||||
Claims []string
|
||||
Images []string
|
||||
}
|
||||
|
||||
// NodeState is one Kubernetes node's own view of itself.
|
||||
type NodeState struct {
|
||||
Name string
|
||||
Ready bool
|
||||
Schedulable bool
|
||||
}
|
||||
|
||||
// ClusterScan is ONE cluster's Kubernetes truth. Err non-nil means the cluster did
|
||||
// not answer — which forces the whole analysis incomplete.
|
||||
type ClusterScan struct {
|
||||
ClusterID string
|
||||
Err error
|
||||
Nodes []NodeState
|
||||
PVs []PVRef
|
||||
PVCs []PVCRef
|
||||
Pods []PodRef
|
||||
}
|
||||
|
||||
// Snapshot is the whole board in one value.
|
||||
type Snapshot struct {
|
||||
At string `json:"at"`
|
||||
Complete bool `json:"complete"`
|
||||
IncompleteReason string `json:"incompleteReason"`
|
||||
Sources []core.SourceStatus `json:"sources"`
|
||||
Totals Totals `json:"totals"`
|
||||
Cost Cost `json:"cost"`
|
||||
Clusters []Cluster `json:"clusters"`
|
||||
Nodes []Node `json:"nodes"`
|
||||
Volumes []Volume `json:"volumes"`
|
||||
LoadBalancers []LoadBalancer `json:"loadBalancers"`
|
||||
Findings []Finding `json:"findings"`
|
||||
}
|
||||
|
||||
// Totals are fleet counts. LocalDiskGiB is broken out precisely so it can be shown
|
||||
// as NOT separately billed.
|
||||
type Totals struct {
|
||||
Clusters int `json:"clusters"`
|
||||
Nodes int `json:"nodes"`
|
||||
Volumes int `json:"volumes"`
|
||||
LoadBalancers int `json:"loadBalancers"`
|
||||
VolumeGiB int `json:"volumeGiB"`
|
||||
AttachedVolumes int `json:"attachedVolumes"`
|
||||
AttachedGiB int `json:"attachedGiB"`
|
||||
DetachedVolumes int `json:"detachedVolumes"`
|
||||
DetachedGiB int `json:"detachedGiB"`
|
||||
UnreferencedVolumes int `json:"unreferencedVolumes"`
|
||||
UnreferencedGiB int `json:"unreferencedGiB"`
|
||||
IdlePVCs int `json:"idlePVCs"`
|
||||
LocalDiskGiB int `json:"localDiskGiB"`
|
||||
}
|
||||
|
||||
// Cost is monthly spend in cents. Reclaimable counts ONLY unreferenced volumes —
|
||||
// never idle ones, which are live data awaiting a human verdict.
|
||||
type Cost struct {
|
||||
DropletsMonthly money.Cents `json:"dropletsMonthly"`
|
||||
VolumesMonthly money.Cents `json:"volumesMonthly"`
|
||||
LoadBalancersMonthly money.Cents `json:"loadBalancersMonthly"`
|
||||
TotalMonthly money.Cents `json:"totalMonthly"`
|
||||
ReclaimableMonthly money.Cents `json:"reclaimableMonthly"`
|
||||
}
|
||||
|
||||
// Cluster is one DOKS cluster with its scanned Kubernetes rollup.
|
||||
type Cluster struct {
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Region string `json:"region"`
|
||||
Version string `json:"version"`
|
||||
Status string `json:"status"`
|
||||
NodePools int `json:"nodePools"`
|
||||
Nodes int `json:"nodes"`
|
||||
Pods int `json:"pods"`
|
||||
PVs int `json:"pvs"`
|
||||
PVCs int `json:"pvcs"`
|
||||
IdlePVCs int `json:"idlePVCs"`
|
||||
Scanned bool `json:"scanned"`
|
||||
ScanError string `json:"scanError"`
|
||||
MonthlyCents money.Cents `json:"monthlyCents"`
|
||||
}
|
||||
|
||||
// Node is one droplet, joined to the Kubernetes node of the same name.
|
||||
type Node struct {
|
||||
ID int `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Cluster string `json:"cluster"`
|
||||
ClusterID string `json:"clusterId"`
|
||||
Region string `json:"region"`
|
||||
Status string `json:"status"`
|
||||
SizeSlug string `json:"sizeSlug"`
|
||||
VCPUs int `json:"vcpus"`
|
||||
MemoryMiB int `json:"memoryMiB"`
|
||||
LocalDiskGiB int `json:"localDiskGiB"`
|
||||
MonthlyCents money.Cents `json:"monthlyCents"`
|
||||
CreatedAt string `json:"createdAt"`
|
||||
PrivateIP string `json:"privateIp"`
|
||||
PublicIP string `json:"publicIp"`
|
||||
Tags []string `json:"tags"`
|
||||
Ready bool `json:"ready"`
|
||||
Schedulable bool `json:"schedulable"`
|
||||
Pods int `json:"pods"`
|
||||
Volumes int `json:"volumes"`
|
||||
}
|
||||
|
||||
// Volume is one block-storage volume with its PROVEN cluster ownership.
|
||||
type Volume struct {
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Region string `json:"region"`
|
||||
SizeGiB int `json:"sizeGiB"`
|
||||
MonthlyCents money.Cents `json:"monthlyCents"`
|
||||
State string `json:"state"`
|
||||
DropletIDs []int `json:"dropletIds"`
|
||||
NodeName string `json:"nodeName"`
|
||||
// Cluster/ClusterID are the PROVEN owner — resolved through a PV that names this
|
||||
// volume, never through the tag.
|
||||
Cluster string `json:"cluster"`
|
||||
ClusterID string `json:"clusterId"`
|
||||
// TagCluster is the `k8s:<uuid>` tag. ADVISORY ONLY: it outlives the cluster that
|
||||
// set it. Shown so the operator can see tag-vs-truth disagree, never acted on.
|
||||
TagCluster string `json:"tagCluster"`
|
||||
PV string `json:"pv"`
|
||||
PVPhase string `json:"pvPhase"`
|
||||
PVCNamespace string `json:"pvcNamespace"`
|
||||
PVCName string `json:"pvcName"`
|
||||
MountedBy []string `json:"mountedBy"`
|
||||
Idle bool `json:"idle"`
|
||||
CreatedAt string `json:"createdAt"`
|
||||
Deletable bool `json:"deletable"`
|
||||
BlockedReason string `json:"blockedReason"`
|
||||
}
|
||||
|
||||
// LoadBalancer is one DO load balancer, attributed to a cluster via its members.
|
||||
type LoadBalancer struct {
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Region string `json:"region"`
|
||||
Status string `json:"status"`
|
||||
IP string `json:"ip"`
|
||||
SizeUnit int `json:"sizeUnit"`
|
||||
MonthlyCents money.Cents `json:"monthlyCents"`
|
||||
Droplets int `json:"droplets"`
|
||||
Cluster string `json:"cluster"`
|
||||
}
|
||||
|
||||
// Finding is one audit result — the "is anything bad" surface.
|
||||
type Finding struct {
|
||||
ID string `json:"id"`
|
||||
Severity string `json:"severity"`
|
||||
Kind string `json:"kind"`
|
||||
Title string `json:"title"`
|
||||
Detail string `json:"detail"`
|
||||
Resource string `json:"resource"`
|
||||
Cluster string `json:"cluster"`
|
||||
MonthlyCents money.Cents `json:"monthlyCents"`
|
||||
}
|
||||
|
||||
// pvHit is a PV that claims a given DO volume, plus the cluster it lives in.
|
||||
type pvHit struct {
|
||||
cluster string
|
||||
clusterID string
|
||||
pv PVRef
|
||||
}
|
||||
|
||||
// Analyze folds the DO inventory and the per-cluster Kubernetes scans into the board.
|
||||
// PURE: no clock, no network, no cluster — `at` is passed in so the result is
|
||||
// byte-reproducible in tests.
|
||||
func Analyze(inv Inventory, scans []ClusterScan, sources []core.SourceStatus, at time.Time) Snapshot {
|
||||
snap := Snapshot{
|
||||
At: at.UTC().Format(time.RFC3339),
|
||||
Sources: sources,
|
||||
}
|
||||
if snap.Sources == nil {
|
||||
snap.Sources = []core.SourceStatus{}
|
||||
}
|
||||
|
||||
scanByID := make(map[string]ClusterScan, len(scans))
|
||||
for _, s := range scans {
|
||||
scanByID[s.ClusterID] = s
|
||||
}
|
||||
nameByID := make(map[string]string, len(inv.Clusters))
|
||||
for _, c := range inv.Clusters {
|
||||
nameByID[c.ID] = c.Name
|
||||
}
|
||||
|
||||
// ---- completeness gate -------------------------------------------------------
|
||||
// Every cluster must have answered. One silent gap and no volume may be condemned.
|
||||
var unreachable []string
|
||||
for _, c := range inv.Clusters {
|
||||
s, ok := scanByID[c.ID]
|
||||
if !ok || s.Err != nil {
|
||||
unreachable = append(unreachable, c.Name)
|
||||
}
|
||||
}
|
||||
switch {
|
||||
case len(inv.Clusters) == 0:
|
||||
snap.IncompleteReason = "DigitalOcean returned no clusters — the set of places a volume could be in use is unknown."
|
||||
case len(unreachable) > 0:
|
||||
snap.IncompleteReason = fmt.Sprintf(
|
||||
"%d of %d clusters did not answer (%s) — a volume they hold would look unreferenced, so nothing is classified as deletable.",
|
||||
len(unreachable), len(inv.Clusters), strings.Join(unreachable, ", "))
|
||||
default:
|
||||
snap.Complete = true
|
||||
}
|
||||
|
||||
// ---- cross-cluster PV index --------------------------------------------------
|
||||
// THE safety index: every volume handle claimed by any PV in any cluster.
|
||||
byHandle := make(map[string]pvHit)
|
||||
// mounted[clusterID/ns/pvc] -> pods currently mounting it.
|
||||
mounted := make(map[string][]string)
|
||||
for _, s := range scans {
|
||||
if s.Err != nil {
|
||||
continue
|
||||
}
|
||||
cname := nameByID[s.ClusterID]
|
||||
for _, pv := range s.PVs {
|
||||
if h := strings.TrimSpace(pv.VolumeHandle); h != "" {
|
||||
byHandle[h] = pvHit{cluster: cname, clusterID: s.ClusterID, pv: pv}
|
||||
}
|
||||
}
|
||||
for _, p := range s.Pods {
|
||||
for _, claim := range p.Claims {
|
||||
k := claimKey(s.ClusterID, p.Namespace, claim)
|
||||
mounted[k] = append(mounted[k], p.Namespace+"/"+p.Name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---- nodes -------------------------------------------------------------------
|
||||
nodeByName := make(map[string]NodeState)
|
||||
podsPerNode := make(map[string]int)
|
||||
for _, s := range scans {
|
||||
if s.Err != nil {
|
||||
continue
|
||||
}
|
||||
for _, n := range s.Nodes {
|
||||
nodeByName[n.Name] = n
|
||||
}
|
||||
for _, p := range s.Pods {
|
||||
if p.Node != "" {
|
||||
podsPerNode[p.Node]++
|
||||
}
|
||||
}
|
||||
}
|
||||
volsPerDroplet := make(map[int]int)
|
||||
for _, v := range inv.Volumes {
|
||||
for _, id := range v.DropletIDs {
|
||||
volsPerDroplet[id]++
|
||||
}
|
||||
}
|
||||
clusterByDroplet := make(map[int]string, len(inv.Droplets))
|
||||
snap.Nodes = make([]Node, 0, len(inv.Droplets))
|
||||
for _, d := range inv.Droplets {
|
||||
cid := clusterIDFromTags(d.Tags)
|
||||
clusterByDroplet[d.ID] = cid
|
||||
ks := nodeByName[d.Name]
|
||||
snap.Nodes = append(snap.Nodes, Node{
|
||||
ID: d.ID, Name: d.Name, Cluster: nameByID[cid], ClusterID: cid,
|
||||
Region: d.Region, Status: d.Status, SizeSlug: d.SizeSlug,
|
||||
VCPUs: d.VCPUs, MemoryMiB: d.MemoryMiB, LocalDiskGiB: d.LocalDiskGiB,
|
||||
MonthlyCents: d.MonthlyCents, CreatedAt: d.CreatedAt,
|
||||
PrivateIP: d.PrivateIP, PublicIP: d.PublicIP, Tags: nonNilStrings(d.Tags),
|
||||
Ready: ks.Ready, Schedulable: ks.Schedulable,
|
||||
Pods: podsPerNode[d.Name], Volumes: volsPerDroplet[d.ID],
|
||||
})
|
||||
snap.Cost.DropletsMonthly += d.MonthlyCents
|
||||
snap.Totals.LocalDiskGiB += d.LocalDiskGiB
|
||||
}
|
||||
dropletName := make(map[int]string, len(inv.Droplets))
|
||||
for _, d := range inv.Droplets {
|
||||
dropletName[d.ID] = d.Name
|
||||
}
|
||||
|
||||
// ---- volumes: the state machine ----------------------------------------------
|
||||
snap.Volumes = make([]Volume, 0, len(inv.Volumes))
|
||||
for _, v := range inv.Volumes {
|
||||
hit, referenced := byHandle[v.ID]
|
||||
vol := Volume{
|
||||
ID: v.ID, Name: v.Name, Region: v.Region, SizeGiB: v.SizeGiB,
|
||||
MonthlyCents: money.Cents(v.SizeGiB) * volumeGiBCents,
|
||||
DropletIDs: nonNilInts(v.DropletIDs),
|
||||
TagCluster: nameByID[clusterIDFromTags(v.Tags)],
|
||||
CreatedAt: v.CreatedAt,
|
||||
MountedBy: []string{},
|
||||
}
|
||||
if len(v.DropletIDs) > 0 {
|
||||
vol.NodeName = dropletName[v.DropletIDs[0]]
|
||||
}
|
||||
if referenced {
|
||||
vol.Cluster, vol.ClusterID = hit.cluster, hit.clusterID
|
||||
vol.PV, vol.PVPhase = hit.pv.Name, hit.pv.Phase
|
||||
vol.PVCNamespace, vol.PVCName = hit.pv.ClaimNS, hit.pv.ClaimName
|
||||
if hit.pv.ClaimName != "" {
|
||||
vol.MountedBy = nonNilStrings(mounted[claimKey(hit.clusterID, hit.pv.ClaimNS, hit.pv.ClaimName)])
|
||||
}
|
||||
} else if len(v.DropletIDs) > 0 {
|
||||
// No PV names it, but it is physically mounted on a node — that node's
|
||||
// cluster owns it. Attachment is hard evidence, unlike the tag: it is the
|
||||
// live kernel state, so the cost rolls up to the right cluster.
|
||||
vol.ClusterID = clusterByDroplet[v.DropletIDs[0]]
|
||||
vol.Cluster = nameByID[vol.ClusterID]
|
||||
}
|
||||
|
||||
// Attachment beats reference; reference beats absence.
|
||||
switch {
|
||||
case len(v.DropletIDs) > 0:
|
||||
vol.State = StateAttached
|
||||
case referenced && strings.EqualFold(hit.pv.Phase, "Bound"):
|
||||
vol.State = StateBound
|
||||
case referenced:
|
||||
vol.State = StateReleased
|
||||
default:
|
||||
vol.State = StateUnreferenced
|
||||
}
|
||||
|
||||
// Idle is a REVIEW signal on live data, never a delete signal.
|
||||
vol.Idle = vol.State == StateBound && len(vol.MountedBy) == 0
|
||||
|
||||
vol.Deletable = snap.Complete && vol.State == StateUnreferenced
|
||||
vol.BlockedReason = blockedReason(vol, snap.Complete, snap.IncompleteReason)
|
||||
|
||||
snap.Cost.VolumesMonthly += vol.MonthlyCents
|
||||
snap.Totals.VolumeGiB += vol.SizeGiB
|
||||
switch vol.State {
|
||||
case StateAttached:
|
||||
snap.Totals.AttachedVolumes++
|
||||
snap.Totals.AttachedGiB += vol.SizeGiB
|
||||
default:
|
||||
snap.Totals.DetachedVolumes++
|
||||
snap.Totals.DetachedGiB += vol.SizeGiB
|
||||
}
|
||||
if vol.State == StateUnreferenced {
|
||||
snap.Totals.UnreferencedVolumes++
|
||||
snap.Totals.UnreferencedGiB += vol.SizeGiB
|
||||
// Reclaimable is exactly the unreferenced set — and only when the scan was
|
||||
// complete enough to have earned that verdict.
|
||||
if snap.Complete {
|
||||
snap.Cost.ReclaimableMonthly += vol.MonthlyCents
|
||||
}
|
||||
}
|
||||
if vol.Idle {
|
||||
snap.Totals.IdlePVCs++
|
||||
}
|
||||
snap.Volumes = append(snap.Volumes, vol)
|
||||
}
|
||||
|
||||
// ---- load balancers ----------------------------------------------------------
|
||||
snap.LoadBalancers = make([]LoadBalancer, 0, len(inv.LoadBalancers))
|
||||
for _, l := range inv.LoadBalancers {
|
||||
lb := LoadBalancer{
|
||||
ID: l.ID, Name: l.Name, Region: l.Region, Status: l.Status, IP: l.IP,
|
||||
SizeUnit: l.SizeUnit, MonthlyCents: l.MonthlyCents, Droplets: len(l.DropletIDs),
|
||||
}
|
||||
for _, id := range l.DropletIDs {
|
||||
if cid := clusterByDroplet[id]; cid != "" {
|
||||
lb.Cluster = nameByID[cid]
|
||||
break
|
||||
}
|
||||
}
|
||||
snap.Cost.LoadBalancersMonthly += lb.MonthlyCents
|
||||
snap.LoadBalancers = append(snap.LoadBalancers, lb)
|
||||
}
|
||||
|
||||
// ---- cluster rollup ----------------------------------------------------------
|
||||
idleByCluster := make(map[string]int)
|
||||
costByCluster := make(map[string]money.Cents)
|
||||
for _, v := range snap.Volumes {
|
||||
if v.ClusterID != "" {
|
||||
costByCluster[v.ClusterID] += v.MonthlyCents
|
||||
if v.Idle {
|
||||
idleByCluster[v.ClusterID]++
|
||||
}
|
||||
}
|
||||
}
|
||||
nodesByCluster := make(map[string]int)
|
||||
for _, n := range snap.Nodes {
|
||||
if n.ClusterID != "" {
|
||||
nodesByCluster[n.ClusterID]++
|
||||
costByCluster[n.ClusterID] += n.MonthlyCents
|
||||
}
|
||||
}
|
||||
snap.Clusters = make([]Cluster, 0, len(inv.Clusters))
|
||||
for _, c := range inv.Clusters {
|
||||
row := Cluster{
|
||||
ID: c.ID, Name: c.Name, Region: c.Region, Version: c.Version,
|
||||
Status: c.Status, NodePools: c.NodePools, Nodes: nodesByCluster[c.ID],
|
||||
IdlePVCs: idleByCluster[c.ID], MonthlyCents: costByCluster[c.ID],
|
||||
}
|
||||
if s, ok := scanByID[c.ID]; ok {
|
||||
if s.Err != nil {
|
||||
row.ScanError = s.Err.Error()
|
||||
} else {
|
||||
row.Scanned = true
|
||||
row.Pods, row.PVs, row.PVCs = len(s.Pods), len(s.PVs), len(s.PVCs)
|
||||
}
|
||||
} else {
|
||||
row.ScanError = "not scanned"
|
||||
}
|
||||
snap.Clusters = append(snap.Clusters, row)
|
||||
}
|
||||
|
||||
snap.Totals.Clusters = len(snap.Clusters)
|
||||
snap.Totals.Nodes = len(snap.Nodes)
|
||||
snap.Totals.Volumes = len(snap.Volumes)
|
||||
snap.Totals.LoadBalancers = len(snap.LoadBalancers)
|
||||
snap.Cost.TotalMonthly = snap.Cost.DropletsMonthly + snap.Cost.VolumesMonthly + snap.Cost.LoadBalancersMonthly
|
||||
|
||||
snap.Findings = findings(snap, scans, nameByID)
|
||||
return snap
|
||||
}
|
||||
|
||||
// blockedReason states, in the operator's language, exactly why a volume may not be
|
||||
// deleted. An empty string means it may.
|
||||
func blockedReason(v Volume, complete bool, incomplete string) string {
|
||||
if v.Deletable {
|
||||
return ""
|
||||
}
|
||||
switch {
|
||||
case !complete:
|
||||
return incomplete
|
||||
case v.State == StateAttached:
|
||||
if v.NodeName != "" {
|
||||
return "Attached to " + v.NodeName + " and in use."
|
||||
}
|
||||
return "Attached to a droplet and in use."
|
||||
case v.State == StateBound:
|
||||
return fmt.Sprintf("Live data: PV %s is Bound to %s/%s in %s.", v.PV, v.PVCNamespace, v.PVCName, v.Cluster)
|
||||
case v.State == StateReleased:
|
||||
return fmt.Sprintf("PV %s in %s still references it (%s) — retire the PV first.", v.PV, v.Cluster, v.PVPhase)
|
||||
}
|
||||
return "Not eligible for deletion."
|
||||
}
|
||||
|
||||
// findings is the audit pass: what a human should look at, worst first.
|
||||
func findings(s Snapshot, scans []ClusterScan, nameByID map[string]string) []Finding {
|
||||
out := []Finding{}
|
||||
|
||||
if !s.Complete {
|
||||
out = append(out, Finding{
|
||||
ID: "scan-incomplete", Severity: SevCritical, Kind: "scan-incomplete",
|
||||
Title: "Cluster scan incomplete — deletion disabled",
|
||||
Detail: s.IncompleteReason,
|
||||
})
|
||||
}
|
||||
|
||||
for _, v := range s.Volumes {
|
||||
switch {
|
||||
case v.State == StateUnreferenced && s.Complete:
|
||||
out = append(out, Finding{
|
||||
ID: "unref/" + v.ID, Severity: SevWarn, Kind: "unreferenced-volume",
|
||||
Title: fmt.Sprintf("Unreferenced volume %s (%d GiB)", v.Name, v.SizeGiB),
|
||||
Detail: "No PersistentVolume in any cluster references this volume. " +
|
||||
"Verified against every cluster, so it is safe to snapshot and delete.",
|
||||
Resource: v.ID, MonthlyCents: v.MonthlyCents,
|
||||
})
|
||||
case v.State == StateReleased:
|
||||
out = append(out, Finding{
|
||||
ID: "released/" + v.ID, Severity: SevWarn, Kind: "released-pv",
|
||||
Title: fmt.Sprintf("Released PV holding %s (%d GiB)", v.Name, v.SizeGiB),
|
||||
Detail: fmt.Sprintf("PV %s is %s. Retire the PV to release the volume.", v.PV, v.PVPhase),
|
||||
Resource: v.ID, Cluster: v.Cluster, MonthlyCents: v.MonthlyCents,
|
||||
})
|
||||
case v.Idle:
|
||||
out = append(out, Finding{
|
||||
ID: "idle/" + v.ID, Severity: SevInfo, Kind: "idle-pvc",
|
||||
Title: fmt.Sprintf("Idle volume %s (%d GiB) — no pod mounts it", v.Name, v.SizeGiB),
|
||||
Detail: fmt.Sprintf("PVC %s/%s is Bound but no running pod mounts it. "+
|
||||
"REVIEW ONLY: this is live data (typically a stopped database), not garbage.",
|
||||
v.PVCNamespace, v.PVCName),
|
||||
Resource: v.ID, Cluster: v.Cluster, MonthlyCents: v.MonthlyCents,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Unhealthy pods + unknown images, per cluster.
|
||||
type imgSeen struct {
|
||||
pods int
|
||||
cluster string
|
||||
}
|
||||
unknown := map[string]*imgSeen{}
|
||||
for _, sc := range scans {
|
||||
if sc.Err != nil {
|
||||
continue
|
||||
}
|
||||
cname := nameByID[sc.ClusterID]
|
||||
for _, p := range sc.Pods {
|
||||
if bad := podProblem(p); bad != "" {
|
||||
out = append(out, Finding{
|
||||
ID: "pod/" + sc.ClusterID + "/" + p.Namespace + "/" + p.Name, Severity: SevWarn,
|
||||
Kind: "pod-unhealthy", Title: fmt.Sprintf("Pod %s/%s is %s", p.Namespace, p.Name, bad),
|
||||
Detail: fmt.Sprintf("Phase %s%s on node %s.", p.Phase, reasonSuffix(p.Reason), p.Node),
|
||||
Resource: p.Namespace + "/" + p.Name, Cluster: cname,
|
||||
})
|
||||
}
|
||||
for _, img := range p.Images {
|
||||
if knownImage(img) {
|
||||
continue
|
||||
}
|
||||
repo := imageRepo(img)
|
||||
if e, ok := unknown[repo]; ok {
|
||||
e.pods++
|
||||
} else {
|
||||
unknown[repo] = &imgSeen{pods: 1, cluster: cname}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
for repo, e := range unknown {
|
||||
out = append(out, Finding{
|
||||
ID: "image/" + repo, Severity: SevWarn, Kind: "unknown-image",
|
||||
Title: "Unrecognised container image: " + repo,
|
||||
Detail: fmt.Sprintf("Run by %d pod(s), from neither our registries nor the reviewed vendor set.", e.pods),
|
||||
Resource: repo, Cluster: e.cluster,
|
||||
})
|
||||
}
|
||||
|
||||
// Cost outliers: any single resource at or above outlierShareBP of total spend.
|
||||
if s.Cost.TotalMonthly > 0 {
|
||||
threshold := s.Cost.TotalMonthly * outlierShareBP / 10000
|
||||
for _, n := range s.Nodes {
|
||||
if n.MonthlyCents >= threshold {
|
||||
out = append(out, Finding{
|
||||
ID: "cost/node/" + n.Name, Severity: SevInfo, Kind: "cost-outlier",
|
||||
Title: fmt.Sprintf("Node %s is %s of fleet spend", n.Name, shareLabel(n.MonthlyCents, s.Cost.TotalMonthly)),
|
||||
Detail: fmt.Sprintf("%s, %d vCPU / %d MiB.", n.SizeSlug, n.VCPUs, n.MemoryMiB),
|
||||
Resource: n.Name, Cluster: n.Cluster, MonthlyCents: n.MonthlyCents,
|
||||
})
|
||||
}
|
||||
}
|
||||
for _, v := range s.Volumes {
|
||||
if v.MonthlyCents >= threshold {
|
||||
out = append(out, Finding{
|
||||
ID: "cost/volume/" + v.ID, Severity: SevInfo, Kind: "cost-outlier",
|
||||
Title: fmt.Sprintf("Volume %s is %s of fleet spend", v.Name, shareLabel(v.MonthlyCents, s.Cost.TotalMonthly)),
|
||||
Detail: fmt.Sprintf("%d GiB, %s.", v.SizeGiB, v.State),
|
||||
Resource: v.ID, Cluster: v.Cluster, MonthlyCents: v.MonthlyCents,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
rank := map[string]int{SevCritical: 0, SevWarn: 1, SevInfo: 2}
|
||||
sort.SliceStable(out, func(i, j int) bool {
|
||||
if rank[out[i].Severity] != rank[out[j].Severity] {
|
||||
return rank[out[i].Severity] < rank[out[j].Severity]
|
||||
}
|
||||
if out[i].MonthlyCents != out[j].MonthlyCents {
|
||||
return out[i].MonthlyCents > out[j].MonthlyCents
|
||||
}
|
||||
return out[i].ID < out[j].ID
|
||||
})
|
||||
return out
|
||||
}
|
||||
|
||||
// podProblem names the failure a pod is in, or "" when it is fine.
|
||||
func podProblem(p PodRef) string {
|
||||
switch {
|
||||
case strings.EqualFold(p.Reason, "Evicted"):
|
||||
return "Evicted"
|
||||
case strings.EqualFold(p.Phase, "Failed"):
|
||||
return "Failed"
|
||||
case strings.Contains(p.Reason, "CrashLoopBackOff"):
|
||||
return "CrashLoopBackOff"
|
||||
case strings.Contains(p.Reason, "ImagePullBackOff"), strings.Contains(p.Reason, "ErrImagePull"):
|
||||
return "ImagePullBackOff"
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func reasonSuffix(r string) string {
|
||||
if strings.TrimSpace(r) == "" {
|
||||
return ""
|
||||
}
|
||||
return " (" + r + ")"
|
||||
}
|
||||
|
||||
// knownImage reports whether an image comes from our registries or the reviewed
|
||||
// vendor set.
|
||||
func knownImage(img string) bool {
|
||||
l := strings.ToLower(strings.TrimSpace(img))
|
||||
l = strings.TrimPrefix(l, "docker.io/")
|
||||
for _, p := range firstParty {
|
||||
if strings.HasPrefix(l, strings.TrimPrefix(p, "docker.io/")) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
for _, p := range knownVendors {
|
||||
if strings.HasPrefix(l, strings.TrimPrefix(p, "docker.io/")) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
// A bare `name:tag` with no slash is an official Docker Hub library image.
|
||||
return !strings.Contains(strings.SplitN(l, ":", 2)[0], "/")
|
||||
}
|
||||
|
||||
// imageRepo strips the tag/digest so findings group by repository, not by build.
|
||||
func imageRepo(img string) string {
|
||||
s := strings.TrimSpace(img)
|
||||
if i := strings.Index(s, "@"); i > 0 {
|
||||
s = s[:i]
|
||||
}
|
||||
if i := strings.LastIndex(s, ":"); i > strings.LastIndex(s, "/") {
|
||||
s = s[:i]
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// shareLabel renders a cents-of-total share as a percentage.
|
||||
func shareLabel(part, total money.Cents) string {
|
||||
if total <= 0 {
|
||||
return "0%"
|
||||
}
|
||||
return fmt.Sprintf("%.1f%%", float64(part)*100/float64(total))
|
||||
}
|
||||
|
||||
// clusterIDFromTags extracts the DOKS cluster UUID from a `k8s:<uuid>` resource tag.
|
||||
// On droplets this is authoritative (DOKS owns the droplet); on VOLUMES it is
|
||||
// advisory only — see the Volume.TagCluster doc.
|
||||
func clusterIDFromTags(tags []string) string {
|
||||
for _, t := range tags {
|
||||
v := strings.TrimPrefix(t, "k8s:")
|
||||
if v == t || v == "" {
|
||||
continue
|
||||
}
|
||||
// Cluster tags are UUIDs; DOKS also stamps role tags like `k8s:worker`.
|
||||
if len(v) == 36 && strings.Count(v, "-") == 4 {
|
||||
return v
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func claimKey(clusterID, ns, name string) string { return clusterID + "/" + ns + "/" + name }
|
||||
|
||||
func nonNilStrings(s []string) []string {
|
||||
if s == nil {
|
||||
return []string{}
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
func nonNilInts(s []int) []int {
|
||||
if s == nil {
|
||||
return []int{}
|
||||
}
|
||||
return s
|
||||
}
|
||||
@@ -0,0 +1,390 @@
|
||||
package infra
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/hanzoai/cloud/clients/admin/digitalocean"
|
||||
)
|
||||
|
||||
// Two live clusters. The near-miss that motivated this package involved volumes tagged
|
||||
// for one cluster whose PVs actually live in another, so every fixture here has two.
|
||||
const (
|
||||
cidA = "aaaaaaaa-1111-2222-3333-444444444444"
|
||||
cidB = "bbbbbbbb-1111-2222-3333-444444444444"
|
||||
)
|
||||
|
||||
func baseInventory() Inventory {
|
||||
return Inventory{
|
||||
Clusters: []digitalocean.Cluster{
|
||||
{ID: cidA, Name: "hanzo-k8s", Region: "sfo3", Status: "running"},
|
||||
{ID: cidB, Name: "lux-k8s", Region: "sfo3", Status: "running"},
|
||||
},
|
||||
Droplets: []digitalocean.Droplet{{
|
||||
ID: 101, Name: "node-a1", Region: "sfo3", Status: "active",
|
||||
SizeSlug: "s-8vcpu-16gb-amd", VCPUs: 8, MemoryMiB: 16384,
|
||||
LocalDiskGiB: 320, MonthlyCents: 11200,
|
||||
Tags: []string{"k8s", "k8s:" + cidA, "k8s:worker"},
|
||||
}},
|
||||
}
|
||||
}
|
||||
|
||||
func scansOK() []ClusterScan {
|
||||
return []ClusterScan{{ClusterID: cidA}, {ClusterID: cidB}}
|
||||
}
|
||||
|
||||
func volByID(t *testing.T, s Snapshot, id string) Volume {
|
||||
t.Helper()
|
||||
v, ok := findVolume(s, id)
|
||||
if !ok {
|
||||
t.Fatalf("volume %s missing from snapshot", id)
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
func analyze(inv Inventory, scans []ClusterScan) Snapshot {
|
||||
return Analyze(inv, scans, nil, time.Unix(0, 0).UTC())
|
||||
}
|
||||
|
||||
// TestCrossClusterPVProtectsMisTaggedVolume is THE regression test. A detached volume
|
||||
// tagged `k8s:<cluster A>` whose PV actually lives in cluster B must be classified from
|
||||
// the PV, not the tag. Trusting the tag here is what nearly destroyed 4.39 TiB.
|
||||
func TestCrossClusterPVProtectsMisTaggedVolume(t *testing.T) {
|
||||
inv := baseInventory()
|
||||
inv.Volumes = []digitalocean.Volume{{
|
||||
ID: "vol-mistagged", Name: "pvc-neon-pageserver", Region: "sfo3", SizeGiB: 50,
|
||||
DropletIDs: nil, // detached
|
||||
Tags: []string{"k8s:" + cidA}, // tag says cluster A …
|
||||
}}
|
||||
scans := scansOK()
|
||||
// … but the PV that owns it lives in cluster B, and is Bound to a live PVC.
|
||||
scans[1].PVs = []PVRef{{
|
||||
Name: "pv-neon", Phase: "Bound", VolumeHandle: "vol-mistagged",
|
||||
ClaimNS: "neon", ClaimName: "pageserver-data",
|
||||
}}
|
||||
|
||||
got := analyze(inv, scans)
|
||||
v := volByID(t, got, "vol-mistagged")
|
||||
|
||||
if v.State != StateBound {
|
||||
t.Fatalf("state = %q, want %q — a detached, mis-tagged volume with a Bound PV is LIVE DATA", v.State, StateBound)
|
||||
}
|
||||
if v.Deletable {
|
||||
t.Fatal("volume marked deletable: this is the 4.39 TiB data-loss bug")
|
||||
}
|
||||
if v.Cluster != "lux-k8s" {
|
||||
t.Errorf("proven cluster = %q, want lux-k8s (from the PV, not the tag)", v.Cluster)
|
||||
}
|
||||
if v.TagCluster != "hanzo-k8s" {
|
||||
t.Errorf("tagCluster = %q, want hanzo-k8s (advisory, surfaced so tag-vs-truth is visible)", v.TagCluster)
|
||||
}
|
||||
if got.Cost.ReclaimableMonthly != 0 {
|
||||
t.Errorf("reclaimable = %d, want 0", got.Cost.ReclaimableMonthly)
|
||||
}
|
||||
}
|
||||
|
||||
// TestIncompleteScanBlocksEveryDeletion: one unreachable cluster and NOTHING is
|
||||
// deletable, even a volume no reachable cluster references. Absence of evidence is not
|
||||
// evidence of absence.
|
||||
func TestIncompleteScanBlocksEveryDeletion(t *testing.T) {
|
||||
inv := baseInventory()
|
||||
inv.Volumes = []digitalocean.Volume{{ID: "vol-orphan", Name: "stray", SizeGiB: 100}}
|
||||
scans := scansOK()
|
||||
scans[1].Err = errors.New("dial tcp: i/o timeout")
|
||||
|
||||
got := analyze(inv, scans)
|
||||
if got.Complete {
|
||||
t.Fatal("Complete = true with an unreachable cluster")
|
||||
}
|
||||
v := volByID(t, got, "vol-orphan")
|
||||
if v.State != StateUnreferenced {
|
||||
t.Errorf("state = %q, want %q (state is observable; the VERDICT is what is withheld)", v.State, StateUnreferenced)
|
||||
}
|
||||
if v.Deletable {
|
||||
t.Fatal("deletable with an incomplete scan — fail-closed violated")
|
||||
}
|
||||
if got.Cost.ReclaimableMonthly != 0 {
|
||||
t.Errorf("reclaimable = %d, want 0 when the scan is incomplete", got.Cost.ReclaimableMonthly)
|
||||
}
|
||||
if v.BlockedReason == "" || !strings.Contains(v.BlockedReason, "lux-k8s") {
|
||||
t.Errorf("blockedReason = %q, want it to name the unreachable cluster", v.BlockedReason)
|
||||
}
|
||||
if got.Findings[0].Kind != "scan-incomplete" || got.Findings[0].Severity != SevCritical {
|
||||
t.Errorf("first finding = %+v, want a critical scan-incomplete", got.Findings[0])
|
||||
}
|
||||
}
|
||||
|
||||
// TestNoClustersIsIncomplete: an empty cluster list means the set of places a volume
|
||||
// could be in use is unknown, which must not read as "referenced by nothing".
|
||||
func TestNoClustersIsIncomplete(t *testing.T) {
|
||||
inv := Inventory{Volumes: []digitalocean.Volume{{ID: "v1", Name: "x", SizeGiB: 10}}}
|
||||
got := analyze(inv, nil)
|
||||
if got.Complete {
|
||||
t.Fatal("Complete = true with zero clusters")
|
||||
}
|
||||
if volByID(t, got, "v1").Deletable {
|
||||
t.Fatal("deletable with zero clusters known")
|
||||
}
|
||||
}
|
||||
|
||||
// TestVolumeStateMachine covers the full ordering: attachment beats reference,
|
||||
// reference beats absence, and only the unreferenced volume is ever deletable.
|
||||
func TestVolumeStateMachine(t *testing.T) {
|
||||
inv := baseInventory()
|
||||
inv.Volumes = []digitalocean.Volume{
|
||||
{ID: "v-attached", Name: "attached", SizeGiB: 10, DropletIDs: []int{101}, Tags: []string{"k8s:" + cidA}},
|
||||
{ID: "v-bound", Name: "bound", SizeGiB: 20},
|
||||
{ID: "v-released", Name: "released", SizeGiB: 30},
|
||||
{ID: "v-unref", Name: "unref", SizeGiB: 40},
|
||||
// Attached AND referenced by a Bound PV: attachment wins.
|
||||
{ID: "v-both", Name: "both", SizeGiB: 50, DropletIDs: []int{101}},
|
||||
}
|
||||
scans := scansOK()
|
||||
scans[0].PVs = []PVRef{
|
||||
{Name: "pv-bound", Phase: "Bound", VolumeHandle: "v-bound", ClaimNS: "ns", ClaimName: "c1"},
|
||||
{Name: "pv-rel", Phase: "Released", VolumeHandle: "v-released", ClaimNS: "ns", ClaimName: "c2"},
|
||||
{Name: "pv-both", Phase: "Bound", VolumeHandle: "v-both", ClaimNS: "ns", ClaimName: "c3"},
|
||||
}
|
||||
scans[0].Pods = []PodRef{{Namespace: "ns", Name: "p1", Node: "node-a1", Claims: []string{"c1", "c3"}}}
|
||||
|
||||
got := analyze(inv, scans)
|
||||
if !got.Complete {
|
||||
t.Fatalf("Complete = false, want true: %s", got.IncompleteReason)
|
||||
}
|
||||
for _, tc := range []struct {
|
||||
id, want string
|
||||
deletable bool
|
||||
}{
|
||||
{"v-attached", StateAttached, false},
|
||||
{"v-bound", StateBound, false},
|
||||
{"v-released", StateReleased, false},
|
||||
{"v-unref", StateUnreferenced, true},
|
||||
{"v-both", StateAttached, false},
|
||||
} {
|
||||
v := volByID(t, got, tc.id)
|
||||
if v.State != tc.want {
|
||||
t.Errorf("%s: state = %q, want %q", tc.id, v.State, tc.want)
|
||||
}
|
||||
if v.Deletable != tc.deletable {
|
||||
t.Errorf("%s: deletable = %v, want %v (reason %q)", tc.id, v.Deletable, tc.deletable, v.BlockedReason)
|
||||
}
|
||||
if !v.Deletable && v.BlockedReason == "" {
|
||||
t.Errorf("%s: not deletable but no reason given", tc.id)
|
||||
}
|
||||
if v.Deletable && v.BlockedReason != "" {
|
||||
t.Errorf("%s: deletable but carries reason %q", tc.id, v.BlockedReason)
|
||||
}
|
||||
}
|
||||
// Exactly one volume is reclaimable: 40 GiB × $0.10 = $4.00.
|
||||
if got.Cost.ReclaimableMonthly != 400 {
|
||||
t.Errorf("reclaimable = %d cents, want 400", got.Cost.ReclaimableMonthly)
|
||||
}
|
||||
// v-bound is mounted by p1; v-both is mounted by p1 too — neither is idle.
|
||||
if got.Totals.IdlePVCs != 0 {
|
||||
t.Errorf("idlePVCs = %d, want 0", got.Totals.IdlePVCs)
|
||||
}
|
||||
}
|
||||
|
||||
// TestIdleIsReviewNotReclaimable: a Bound volume no pod mounts is flagged for review
|
||||
// but is never deletable and never counted as money we can get back.
|
||||
func TestIdleIsReviewNotReclaimable(t *testing.T) {
|
||||
inv := baseInventory()
|
||||
inv.Volumes = []digitalocean.Volume{{ID: "v-idle", Name: "registry-data", SizeGiB: 50}}
|
||||
scans := scansOK()
|
||||
scans[1].PVs = []PVRef{{Name: "pv-idle", Phase: "Bound", VolumeHandle: "v-idle", ClaimNS: "registry", ClaimName: "registry-data"}}
|
||||
// No pod mounts registry-data.
|
||||
scans[1].Pods = []PodRef{{Namespace: "other", Name: "unrelated", Claims: []string{"something-else"}}}
|
||||
|
||||
got := analyze(inv, scans)
|
||||
v := volByID(t, got, "v-idle")
|
||||
if !v.Idle {
|
||||
t.Fatal("idle = false, want true (Bound but unmounted)")
|
||||
}
|
||||
if v.Deletable {
|
||||
t.Fatal("an idle volume must never be deletable — it is a stopped database, not garbage")
|
||||
}
|
||||
if got.Cost.ReclaimableMonthly != 0 {
|
||||
t.Errorf("reclaimable = %d, want 0: idle capacity is not reclaimable", got.Cost.ReclaimableMonthly)
|
||||
}
|
||||
if got.Totals.IdlePVCs != 1 {
|
||||
t.Errorf("idlePVCs = %d, want 1", got.Totals.IdlePVCs)
|
||||
}
|
||||
var f *Finding
|
||||
for i := range got.Findings {
|
||||
if got.Findings[i].Kind == "idle-pvc" {
|
||||
f = &got.Findings[i]
|
||||
}
|
||||
}
|
||||
if f == nil {
|
||||
t.Fatal("no idle-pvc finding")
|
||||
}
|
||||
if f.Severity != SevInfo {
|
||||
t.Errorf("idle-pvc severity = %q, want info (a review queue, not an alarm)", f.Severity)
|
||||
}
|
||||
if !strings.Contains(f.Detail, "REVIEW ONLY") {
|
||||
t.Errorf("idle-pvc detail must say it is review-only, got %q", f.Detail)
|
||||
}
|
||||
}
|
||||
|
||||
// TestLegacyFlexVolumePVProtects: a pre-CSI PV still shields its volume.
|
||||
func TestLegacyFlexVolumePVProtects(t *testing.T) {
|
||||
inv := baseInventory()
|
||||
inv.Volumes = []digitalocean.Volume{{ID: "v-flex", Name: "legacy", SizeGiB: 10}}
|
||||
scans := scansOK()
|
||||
scans[0].PVs = []PVRef{{Name: "pv-flex", Phase: "Bound", VolumeHandle: "v-flex", ClaimNS: "old", ClaimName: "data"}}
|
||||
if volByID(t, analyze(inv, scans), "v-flex").Deletable {
|
||||
t.Fatal("a flexVolume-referenced volume must not be deletable")
|
||||
}
|
||||
}
|
||||
|
||||
// TestCostMathAndLocalDiskSeparation: block storage is billed per GiB; droplet local
|
||||
// disk is included in the droplet price and must never be added to storage cost.
|
||||
func TestCostMathAndLocalDiskSeparation(t *testing.T) {
|
||||
inv := baseInventory()
|
||||
inv.Volumes = []digitalocean.Volume{
|
||||
{ID: "a", Name: "a", SizeGiB: 200},
|
||||
{ID: "b", Name: "b", SizeGiB: 300},
|
||||
}
|
||||
inv.LoadBalancers = []digitalocean.LoadBalancer{
|
||||
{ID: "lb1", Name: "ingress", SizeUnit: 1, MonthlyCents: 1200, DropletIDs: []int{101}},
|
||||
}
|
||||
got := analyze(inv, scansOK())
|
||||
|
||||
if got.Cost.VolumesMonthly != 5000 {
|
||||
t.Errorf("volumes = %d cents, want 5000 (500 GiB × $0.10)", got.Cost.VolumesMonthly)
|
||||
}
|
||||
if got.Cost.DropletsMonthly != 11200 {
|
||||
t.Errorf("droplets = %d cents, want 11200", got.Cost.DropletsMonthly)
|
||||
}
|
||||
if got.Cost.LoadBalancersMonthly != 1200 {
|
||||
t.Errorf("load balancers = %d cents, want 1200", got.Cost.LoadBalancersMonthly)
|
||||
}
|
||||
if got.Cost.TotalMonthly != 5000+11200+1200 {
|
||||
t.Errorf("total = %d cents, want %d", got.Cost.TotalMonthly, 5000+11200+1200)
|
||||
}
|
||||
// The 320 GiB of local disk is reported, but is NOT in any cost line.
|
||||
if got.Totals.LocalDiskGiB != 320 {
|
||||
t.Errorf("localDiskGiB = %d, want 320", got.Totals.LocalDiskGiB)
|
||||
}
|
||||
if got.Totals.VolumeGiB != 500 {
|
||||
t.Errorf("volumeGiB = %d, want 500 — local disk must never be folded into block storage", got.Totals.VolumeGiB)
|
||||
}
|
||||
if got.LoadBalancers[0].Cluster != "hanzo-k8s" {
|
||||
t.Errorf("lb cluster = %q, want hanzo-k8s (attributed via its member droplet)", got.LoadBalancers[0].Cluster)
|
||||
}
|
||||
}
|
||||
|
||||
// TestNodeJoinAndClusterRollup: droplets join their Kubernetes node by name and roll
|
||||
// up into the owning cluster.
|
||||
func TestNodeJoinAndClusterRollup(t *testing.T) {
|
||||
inv := baseInventory()
|
||||
inv.Volumes = []digitalocean.Volume{{ID: "v1", Name: "v1", SizeGiB: 10, DropletIDs: []int{101}}}
|
||||
scans := scansOK()
|
||||
scans[0].Nodes = []NodeState{{Name: "node-a1", Ready: true, Schedulable: false}}
|
||||
scans[0].Pods = []PodRef{
|
||||
{Namespace: "hanzo", Name: "p1", Node: "node-a1"},
|
||||
{Namespace: "hanzo", Name: "p2", Node: "node-a1"},
|
||||
}
|
||||
got := analyze(inv, scans)
|
||||
|
||||
n := got.Nodes[0]
|
||||
if n.Cluster != "hanzo-k8s" || n.ClusterID != cidA {
|
||||
t.Errorf("node cluster = %q/%q, want hanzo-k8s", n.Cluster, n.ClusterID)
|
||||
}
|
||||
if !n.Ready || n.Schedulable {
|
||||
t.Errorf("node ready=%v schedulable=%v, want ready + cordoned", n.Ready, n.Schedulable)
|
||||
}
|
||||
if n.Pods != 2 || n.Volumes != 1 {
|
||||
t.Errorf("node pods=%d volumes=%d, want 2/1", n.Pods, n.Volumes)
|
||||
}
|
||||
var ca Cluster
|
||||
for _, c := range got.Clusters {
|
||||
if c.ID == cidA {
|
||||
ca = c
|
||||
}
|
||||
}
|
||||
if ca.Nodes != 1 || ca.Pods != 2 || !ca.Scanned {
|
||||
t.Errorf("cluster A rollup = %+v, want 1 node / 2 pods / scanned", ca)
|
||||
}
|
||||
// Node $112.00 + its 10 GiB volume $1.00.
|
||||
if ca.MonthlyCents != 11200+100 {
|
||||
t.Errorf("cluster monthly = %d, want 11300", ca.MonthlyCents)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUnknownImageDetection: our registries and the reviewed vendor set stay quiet;
|
||||
// anything else is reported once per repository.
|
||||
func TestUnknownImageDetection(t *testing.T) {
|
||||
inv := baseInventory()
|
||||
scans := scansOK()
|
||||
scans[0].Pods = []PodRef{{Namespace: "hanzo", Name: "p1", Images: []string{
|
||||
"ghcr.io/hanzoai/cloud:v1.801.218",
|
||||
"ghcr.io/luxfi/node:v1.2.3",
|
||||
"registry.k8s.io/pause:3.9",
|
||||
"grafana/grafana:11.0.0",
|
||||
"acmglobaltech/thing:1",
|
||||
"redis:7", // bare official library image
|
||||
"evil.example.com/miner:latest", // ← the only one that should be reported
|
||||
}}}
|
||||
got := analyze(inv, scans)
|
||||
|
||||
var unknown []string
|
||||
for _, f := range got.Findings {
|
||||
if f.Kind == "unknown-image" {
|
||||
unknown = append(unknown, f.Resource)
|
||||
}
|
||||
}
|
||||
if len(unknown) != 1 || unknown[0] != "evil.example.com/miner" {
|
||||
t.Fatalf("unknown images = %v, want exactly [evil.example.com/miner]", unknown)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUnhealthyPodFindings covers the pod states the audit surfaces.
|
||||
func TestUnhealthyPodFindings(t *testing.T) {
|
||||
inv := baseInventory()
|
||||
scans := scansOK()
|
||||
scans[0].Pods = []PodRef{
|
||||
{Namespace: "a", Name: "ok", Phase: "Running"},
|
||||
{Namespace: "a", Name: "gone", Phase: "Failed", Reason: "Evicted"},
|
||||
{Namespace: "a", Name: "loop", Phase: "Pending", Reason: "CrashLoopBackOff"},
|
||||
{Namespace: "a", Name: "pull", Phase: "Pending", Reason: "ImagePullBackOff"},
|
||||
}
|
||||
got := analyze(inv, scans)
|
||||
seen := map[string]bool{}
|
||||
for _, f := range got.Findings {
|
||||
if f.Kind == "pod-unhealthy" {
|
||||
seen[f.Resource] = true
|
||||
}
|
||||
}
|
||||
if len(seen) != 3 || !seen["a/gone"] || !seen["a/loop"] || !seen["a/pull"] {
|
||||
t.Fatalf("unhealthy pods = %v, want gone/loop/pull and not ok", seen)
|
||||
}
|
||||
}
|
||||
|
||||
// TestClusterTagParsing: only a UUID-shaped k8s: tag is a cluster id; role tags are not.
|
||||
func TestClusterTagParsing(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
tags []string
|
||||
want string
|
||||
}{
|
||||
{[]string{"k8s", "k8s:" + cidA, "k8s:worker"}, cidA},
|
||||
{[]string{"k8s", "k8s:worker"}, ""},
|
||||
{[]string{"unrelated"}, ""},
|
||||
{nil, ""},
|
||||
} {
|
||||
if got := clusterIDFromTags(tc.tags); got != tc.want {
|
||||
t.Errorf("clusterIDFromTags(%v) = %q, want %q", tc.tags, got, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestJSONArraysNeverNull: the console renders these directly; a null array is a crash.
|
||||
func TestJSONArraysNeverNull(t *testing.T) {
|
||||
got := analyze(Inventory{}, nil)
|
||||
if got.Volumes == nil || got.Nodes == nil || got.Clusters == nil ||
|
||||
got.LoadBalancers == nil || got.Findings == nil || got.Sources == nil {
|
||||
t.Fatalf("empty snapshot has nil slices: %+v", got)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,284 @@
|
||||
package infra
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/zap-proto/zip"
|
||||
|
||||
"github.com/hanzoai/cloud"
|
||||
"github.com/hanzoai/cloud/audit"
|
||||
"github.com/hanzoai/cloud/clients/admin/core"
|
||||
"github.com/hanzoai/cloud/clients/admin/digitalocean"
|
||||
)
|
||||
|
||||
// cacheTTL bounds how stale a READ may be. It exists because one board is a fan-out
|
||||
// over the DO API plus every cluster's full pod/PV listing — not because staleness is
|
||||
// acceptable when it matters: every MUTATION re-scans from scratch, ignoring this.
|
||||
const cacheTTL = 60 * time.Second
|
||||
|
||||
// board holds the one cached snapshot behind /v1/admin/infra.
|
||||
type board struct {
|
||||
mu sync.Mutex
|
||||
snap Snapshot
|
||||
at time.Time
|
||||
}
|
||||
|
||||
// Routes registers the DigitalOcean infrastructure board. SuperAdmin only: this is
|
||||
// the whole account's physical inventory and the controls that destroy parts of it.
|
||||
//
|
||||
// NOTE ON THE NOUN: this is INFRASTRUCTURE — droplets, volumes, DOKS clusters, load
|
||||
// balancers. The pre-existing /v1/fleet surface is compute workers and jobs. Different
|
||||
// nouns, deliberately not merged.
|
||||
func Routes(app *zip.App, s *cloud.Service[core.State]) {
|
||||
b := &board{}
|
||||
g := app.Group("/v1/admin")
|
||||
g.Get("/infra", core.Guard(s, b.read))
|
||||
g.Post("/infra/volumes/:id/snapshot", core.Guard(s, b.snapshotVolume))
|
||||
g.Delete("/infra/volumes/:id", core.Guard(s, b.deleteVolume))
|
||||
g.Post("/infra/nodes/:id/cordon", core.Guard(s, b.cordonNode))
|
||||
}
|
||||
|
||||
// read serves the board, from cache unless ?refresh=1.
|
||||
func (b *board) read(s *cloud.Service[core.State], c *zip.Ctx) error {
|
||||
snap, err := b.load(c.Context(), s.State.DO, c.Query("refresh") != "")
|
||||
if err != nil {
|
||||
return core.Fail(c, err.Error())
|
||||
}
|
||||
return core.OK(c, snap)
|
||||
}
|
||||
|
||||
// load returns the snapshot, recomputing when forced or stale. A forced load is the
|
||||
// authority every mutation checks itself against.
|
||||
func (b *board) load(ctx context.Context, do *digitalocean.Client, force bool) (Snapshot, error) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
if !force && !b.at.IsZero() && time.Since(b.at) < cacheTTL {
|
||||
return b.snap, nil
|
||||
}
|
||||
snap, err := collect(ctx, do)
|
||||
if err != nil {
|
||||
return Snapshot{}, err
|
||||
}
|
||||
b.snap, b.at = snap, time.Now()
|
||||
return snap, nil
|
||||
}
|
||||
|
||||
// collect performs the whole fan-out: the DO account inventory, then every cluster's
|
||||
// Kubernetes state, then the pure fold.
|
||||
//
|
||||
// Only an unusable DO account is a hard error. A partial DO read (say load balancers
|
||||
// fail) still produces a board, with the failure named in Sources — EXCEPT for the
|
||||
// two reads the safety verdict depends on. Clusters and Volumes are load-bearing: if
|
||||
// either is missing, the analysis cannot honestly classify anything, so it degrades
|
||||
// via the completeness gate rather than pretending.
|
||||
func collect(ctx context.Context, do *digitalocean.Client) (Snapshot, error) {
|
||||
if do == nil || !do.Ready() {
|
||||
return Snapshot{}, fmt.Errorf("DO_API_TOKEN not configured — DigitalOcean inventory unavailable")
|
||||
}
|
||||
at := time.Now().UTC()
|
||||
stamp := at.Format(time.RFC3339)
|
||||
|
||||
var (
|
||||
inv Inventory
|
||||
sources []core.SourceStatus
|
||||
mu sync.Mutex
|
||||
wg sync.WaitGroup
|
||||
)
|
||||
run := func(name string, fn func() (int, error)) {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
n, err := fn()
|
||||
mu.Lock()
|
||||
sources = append(sources, core.SrcOf(name, err, n, stamp))
|
||||
mu.Unlock()
|
||||
}()
|
||||
}
|
||||
run("do.clusters", func() (int, error) {
|
||||
v, err := do.Clusters(ctx)
|
||||
inv.Clusters = v
|
||||
return len(v), err
|
||||
})
|
||||
run("do.droplets", func() (int, error) {
|
||||
v, err := do.Droplets(ctx)
|
||||
inv.Droplets = v
|
||||
return len(v), err
|
||||
})
|
||||
run("do.volumes", func() (int, error) {
|
||||
v, err := do.Volumes(ctx)
|
||||
inv.Volumes = v
|
||||
return len(v), err
|
||||
})
|
||||
run("do.loadBalancers", func() (int, error) {
|
||||
v, err := do.LoadBalancers(ctx)
|
||||
inv.LoadBalancers = v
|
||||
return len(v), err
|
||||
})
|
||||
wg.Wait()
|
||||
|
||||
scans := Scan(ctx, do, inv.Clusters)
|
||||
for i, sc := range scans {
|
||||
name := "k8s." + inv.Clusters[i].Name
|
||||
rows := len(sc.PVs) + len(sc.PVCs) + len(sc.Pods) + len(sc.Nodes)
|
||||
sources = append(sources, core.SrcOf(name, sc.Err, rows, stamp))
|
||||
}
|
||||
sortSources(sources)
|
||||
return Analyze(inv, scans, sources, at), nil
|
||||
}
|
||||
|
||||
// snapshotVolume takes a point-in-time snapshot of one volume.
|
||||
func (b *board) snapshotVolume(s *cloud.Service[core.State], c *zip.Ctx) error {
|
||||
id := strings.TrimSpace(c.Param("id"))
|
||||
snap, err := b.load(c.Context(), s.State.DO, true)
|
||||
if err != nil {
|
||||
return core.Fail(c, err.Error())
|
||||
}
|
||||
v, ok := findVolume(snap, id)
|
||||
if !ok {
|
||||
return core.Fail(c, "volume not found")
|
||||
}
|
||||
var body struct {
|
||||
Name string `json:"name"`
|
||||
}
|
||||
_ = c.Bind(&body)
|
||||
out, err := takeSnapshot(c.Context(), s.State.DO, v, body.Name)
|
||||
if err != nil {
|
||||
core.EmitAudit(s, c, "infra.volume.snapshot", "do_volume", id, v, nil,
|
||||
audit.Outcome{Result: "failure", Status: 200, Reason: err.Error()})
|
||||
return core.Fail(c, err.Error())
|
||||
}
|
||||
core.EmitAudit(s, c, "infra.volume.snapshot", "do_volume", id, v, out,
|
||||
audit.Outcome{Result: "success", Status: 200})
|
||||
return core.OK(c, out)
|
||||
}
|
||||
|
||||
// deleteVolume destroys a volume — but ONLY one the server itself has just proven to
|
||||
// be referenced by no PersistentVolume in any cluster.
|
||||
//
|
||||
// The client's opinion is never trusted: deletability is recomputed here from a FRESH
|
||||
// complete cross-cluster scan (force=true, never the cache), so a volume that became
|
||||
// live between the operator loading the page and pressing the button is refused. If
|
||||
// any cluster is unreachable the scan is incomplete and NOTHING is deletable.
|
||||
func (b *board) deleteVolume(s *cloud.Service[core.State], c *zip.Ctx) error {
|
||||
id := strings.TrimSpace(c.Param("id"))
|
||||
snap, err := b.load(c.Context(), s.State.DO, true)
|
||||
if err != nil {
|
||||
return core.Fail(c, err.Error())
|
||||
}
|
||||
v, ok := findVolume(snap, id)
|
||||
if !ok {
|
||||
return core.Fail(c, "volume not found")
|
||||
}
|
||||
if !v.Deletable {
|
||||
core.EmitAudit(s, c, "infra.volume.delete", "do_volume", id, v, nil,
|
||||
audit.Outcome{Result: "denied", Status: 200, Reason: v.BlockedReason})
|
||||
return core.Fail(c, "refusing to delete: "+v.BlockedReason)
|
||||
}
|
||||
|
||||
out := map[string]any{"deleted": false, "name": v.Name, "sizeGiB": v.SizeGiB,
|
||||
"freedMonthlyCents": v.MonthlyCents}
|
||||
// Snapshot first unless explicitly waived — the delete is irreversible, the
|
||||
// snapshot is the undo.
|
||||
if c.Query("snapshot") != "false" {
|
||||
shot, serr := takeSnapshot(c.Context(), s.State.DO, v, "")
|
||||
if serr != nil {
|
||||
core.EmitAudit(s, c, "infra.volume.delete", "do_volume", id, v, nil,
|
||||
audit.Outcome{Result: "failure", Status: 200, Reason: "snapshot failed: " + serr.Error()})
|
||||
return core.Fail(c, "snapshot failed, volume NOT deleted: "+serr.Error())
|
||||
}
|
||||
out["snapshotId"] = shot.ID
|
||||
}
|
||||
if err := s.State.DO.DeleteVolume(c.Context(), id); err != nil {
|
||||
core.EmitAudit(s, c, "infra.volume.delete", "do_volume", id, v, nil,
|
||||
audit.Outcome{Result: "failure", Status: 200, Reason: err.Error()})
|
||||
return core.Fail(c, err.Error())
|
||||
}
|
||||
out["deleted"] = true
|
||||
b.invalidate()
|
||||
core.EmitAudit(s, c, "infra.volume.delete", "do_volume", id, v, out,
|
||||
audit.Outcome{Result: "success", Status: 200})
|
||||
return core.OK(c, out)
|
||||
}
|
||||
|
||||
// cordonNode cordons/uncordons a node, optionally draining it.
|
||||
func (b *board) cordonNode(s *cloud.Service[core.State], c *zip.Ctx) error {
|
||||
id, err := strconv.Atoi(strings.TrimSpace(c.Param("id")))
|
||||
if err != nil {
|
||||
return core.Fail(c, "node id must be a droplet id")
|
||||
}
|
||||
var body struct {
|
||||
Cordon bool `json:"cordon"`
|
||||
Drain bool `json:"drain"`
|
||||
}
|
||||
if err := c.Bind(&body); err != nil {
|
||||
return core.Fail(c, "invalid body")
|
||||
}
|
||||
snap, err := b.load(c.Context(), s.State.DO, false)
|
||||
if err != nil {
|
||||
return core.Fail(c, err.Error())
|
||||
}
|
||||
var node *Node
|
||||
for i := range snap.Nodes {
|
||||
if snap.Nodes[i].ID == id {
|
||||
node = &snap.Nodes[i]
|
||||
break
|
||||
}
|
||||
}
|
||||
if node == nil {
|
||||
return core.Fail(c, "node not found")
|
||||
}
|
||||
if node.ClusterID == "" {
|
||||
return core.Fail(c, "node is not a member of a known cluster")
|
||||
}
|
||||
evicted, err := SetSchedulable(c.Context(), s.State.DO, node.ClusterID, node.Name, !body.Cordon, body.Drain)
|
||||
out := map[string]any{"name": node.Name, "schedulable": !body.Cordon, "evicted": evicted}
|
||||
if err != nil {
|
||||
core.EmitAudit(s, c, "infra.node.cordon", "do_droplet", node.Name, node, out,
|
||||
audit.Outcome{Result: "failure", Status: 200, Reason: err.Error()})
|
||||
return core.Fail(c, err.Error())
|
||||
}
|
||||
b.invalidate()
|
||||
core.EmitAudit(s, c, "infra.node.cordon", "do_droplet", node.Name, node, out,
|
||||
audit.Outcome{Result: "success", Status: 200})
|
||||
return core.OK(c, out)
|
||||
}
|
||||
|
||||
// takeSnapshot names and takes a volume snapshot. A blank name gets a deterministic
|
||||
// pre-delete name so the undo is findable in the DO console.
|
||||
func takeSnapshot(ctx context.Context, do *digitalocean.Client, v Volume, name string) (digitalocean.Snapshot, error) {
|
||||
name = strings.TrimSpace(name)
|
||||
if name == "" {
|
||||
name = fmt.Sprintf("%s-predelete-%d", v.Name, time.Now().Unix())
|
||||
}
|
||||
return do.SnapshotVolume(ctx, v.ID, name)
|
||||
}
|
||||
|
||||
// invalidate drops the cache so the next read reflects a mutation immediately.
|
||||
func (b *board) invalidate() {
|
||||
b.mu.Lock()
|
||||
b.at = time.Time{}
|
||||
b.mu.Unlock()
|
||||
}
|
||||
|
||||
func findVolume(s Snapshot, id string) (Volume, bool) {
|
||||
for _, v := range s.Volumes {
|
||||
if v.ID == id {
|
||||
return v, true
|
||||
}
|
||||
}
|
||||
return Volume{}, false
|
||||
}
|
||||
|
||||
// sortSources keeps the freshness list stable across reads (map/goroutine order is not).
|
||||
func sortSources(rows []core.SourceStatus) {
|
||||
for i := 1; i < len(rows); i++ {
|
||||
for j := i; j > 0 && rows[j].Name < rows[j-1].Name; j-- {
|
||||
rows[j], rows[j-1] = rows[j-1], rows[j]
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,259 @@
|
||||
package infra
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/hanzoai/cloud/clients/admin/digitalocean"
|
||||
)
|
||||
|
||||
// fakeDO stands in for the DigitalOcean API. kubeAPI is the URL a cluster kubeconfig
|
||||
// points at; when blank, the kubeconfig fetch fails, which is how the incomplete-scan
|
||||
// path is exercised.
|
||||
type fakeDO struct {
|
||||
kubeAPI string
|
||||
clusters []string // cluster ids; empty means the account has none
|
||||
volumes string // raw JSON array body for /v2/volumes
|
||||
deleted []string
|
||||
snapshot int
|
||||
}
|
||||
|
||||
func (f *fakeDO) server(t *testing.T) *digitalocean.Client {
|
||||
t.Helper()
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/v2/droplets", func(w http.ResponseWriter, r *http.Request) {
|
||||
fmt.Fprint(w, `{"droplets":[{"id":101,"name":"node-a1","status":"active","size_slug":"s-1vcpu-1gb",
|
||||
"vcpus":1,"memory":1024,"disk":25,"size":{"price_monthly":6},"region":{"slug":"sfo3"},
|
||||
"networks":{"v4":[{"type":"private","ip_address":"10.0.0.1"}]},
|
||||
"tags":["k8s","k8s:`+clusterUUID+`"]}],"meta":{"total":1}}`)
|
||||
})
|
||||
mux.HandleFunc("/v2/volumes", func(w http.ResponseWriter, r *http.Request) {
|
||||
fmt.Fprintf(w, `{"volumes":%s,"meta":{"total":2}}`, f.volumes)
|
||||
})
|
||||
mux.HandleFunc("/v2/load_balancers", func(w http.ResponseWriter, r *http.Request) {
|
||||
fmt.Fprint(w, `{"load_balancers":[],"meta":{"total":0}}`)
|
||||
})
|
||||
mux.HandleFunc("/v2/kubernetes/clusters", func(w http.ResponseWriter, r *http.Request) {
|
||||
rows := []string{}
|
||||
for _, id := range f.clusters {
|
||||
rows = append(rows, fmt.Sprintf(
|
||||
`{"id":%q,"name":"test-k8s","region":"sfo3","version":"1.35","status":{"state":"running"},"node_pools":[{"name":"p"}]}`, id))
|
||||
}
|
||||
fmt.Fprintf(w, `{"kubernetes_clusters":[%s],"meta":{"total":%d}}`, strings.Join(rows, ","), len(rows))
|
||||
})
|
||||
mux.HandleFunc("/v2/kubernetes/clusters/"+clusterUUID+"/kubeconfig", func(w http.ResponseWriter, r *http.Request) {
|
||||
if f.kubeAPI == "" {
|
||||
http.Error(w, `{"message":"cluster unreachable"}`, http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
fmt.Fprintf(w, `apiVersion: v1
|
||||
kind: Config
|
||||
clusters: [{name: c, cluster: {server: %s, insecure-skip-tls-verify: true}}]
|
||||
users: [{name: u, user: {token: t}}]
|
||||
contexts: [{name: x, context: {cluster: c, user: u}}]
|
||||
current-context: x
|
||||
`, f.kubeAPI)
|
||||
})
|
||||
mux.HandleFunc("/v2/volumes/", func(w http.ResponseWriter, r *http.Request) {
|
||||
id := strings.TrimPrefix(r.URL.Path, "/v2/volumes/")
|
||||
switch {
|
||||
case strings.HasSuffix(id, "/snapshots"):
|
||||
f.snapshot++
|
||||
fmt.Fprint(w, `{"snapshot":{"id":"snap-1","name":"s","size_gigabytes":40}}`)
|
||||
case r.Method == http.MethodDelete:
|
||||
f.deleted = append(f.deleted, id)
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
default:
|
||||
http.Error(w, `{"message":"nope"}`, http.StatusNotFound)
|
||||
}
|
||||
})
|
||||
srv := httptest.NewServer(mux)
|
||||
t.Cleanup(srv.Close)
|
||||
return digitalocean.NewWithBase(srv.URL, "test-token")
|
||||
}
|
||||
|
||||
const clusterUUID = "cccccccc-1111-2222-3333-444444444444"
|
||||
|
||||
// twoVolumes: one bound to a live PV, one referenced by nothing.
|
||||
const twoVolumes = `[
|
||||
{"id":"vol-live","name":"live","size_gigabytes":20,"region":{"slug":"sfo3"},"droplet_ids":[],"tags":["k8s:` + clusterUUID + `"]},
|
||||
{"id":"vol-junk","name":"junk","size_gigabytes":40,"region":{"slug":"sfo3"},"droplet_ids":[],"tags":["k8s:` + clusterUUID + `"]}
|
||||
]`
|
||||
|
||||
// fakeAPIServer serves the four core/v1 collections scanOne reads.
|
||||
func fakeAPIServer(t *testing.T) string {
|
||||
t.Helper()
|
||||
t.Setenv("FLEET_ALLOW_PRIVATE_HOSTS", "1")
|
||||
list := func(kind string, items any) string {
|
||||
b, _ := json.Marshal(items)
|
||||
return fmt.Sprintf(`{"apiVersion":"v1","kind":%q,"metadata":{},"items":%s}`, kind, b)
|
||||
}
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/api/v1/persistentvolumes", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
fmt.Fprint(w, list("PersistentVolumeList", []map[string]any{{
|
||||
"metadata": map[string]any{"name": "pv-live"},
|
||||
"spec": map[string]any{
|
||||
"csi": map[string]any{"driver": "dobs.csi.digitalocean.com", "volumeHandle": "vol-live"},
|
||||
"claimRef": map[string]any{"namespace": "db", "name": "data"},
|
||||
},
|
||||
"status": map[string]any{"phase": "Bound"},
|
||||
}}))
|
||||
})
|
||||
mux.HandleFunc("/api/v1/persistentvolumeclaims", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
fmt.Fprint(w, list("PersistentVolumeClaimList", []map[string]any{{
|
||||
"metadata": map[string]any{"namespace": "db", "name": "data"},
|
||||
"spec": map[string]any{"volumeName": "pv-live"},
|
||||
"status": map[string]any{"phase": "Bound"},
|
||||
}}))
|
||||
})
|
||||
mux.HandleFunc("/api/v1/pods", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
fmt.Fprint(w, list("PodList", []map[string]any{{
|
||||
"metadata": map[string]any{"namespace": "db", "name": "pg-0"},
|
||||
"spec": map[string]any{
|
||||
"nodeName": "node-a1",
|
||||
"containers": []map[string]any{{"name": "c", "image": "ghcr.io/hanzoai/base:v1"}},
|
||||
"volumes": []map[string]any{{"name": "v", "persistentVolumeClaim": map[string]any{"claimName": "data"}}},
|
||||
},
|
||||
"status": map[string]any{"phase": "Running"},
|
||||
}}))
|
||||
})
|
||||
mux.HandleFunc("/api/v1/nodes", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
fmt.Fprint(w, list("NodeList", []map[string]any{{
|
||||
"metadata": map[string]any{"name": "node-a1"},
|
||||
"spec": map[string]any{},
|
||||
"status": map[string]any{"conditions": []map[string]any{{"type": "Ready", "status": "True"}}},
|
||||
}}))
|
||||
})
|
||||
srv := httptest.NewTLSServer(mux)
|
||||
t.Cleanup(srv.Close)
|
||||
return srv.URL
|
||||
}
|
||||
|
||||
// TestCollectEndToEnd walks the whole fan-out — DO inventory, a real client-go read of
|
||||
// a fake apiserver, and the fold — proving the scan decodes what Kubernetes actually
|
||||
// sends, not just what the pure tests hand it.
|
||||
func TestCollectEndToEnd(t *testing.T) {
|
||||
f := &fakeDO{kubeAPI: fakeAPIServer(t), clusters: []string{clusterUUID}, volumes: twoVolumes}
|
||||
snap, err := collect(context.Background(), f.server(t))
|
||||
if err != nil {
|
||||
t.Fatalf("collect: %v", err)
|
||||
}
|
||||
if !snap.Complete {
|
||||
t.Fatalf("Complete = false: %s", snap.IncompleteReason)
|
||||
}
|
||||
live, _ := findVolume(snap, "vol-live")
|
||||
if live.State != StateBound || live.Deletable {
|
||||
t.Errorf("vol-live = %s deletable=%v, want bound + not deletable", live.State, live.Deletable)
|
||||
}
|
||||
if live.PV != "pv-live" || live.PVCName != "data" {
|
||||
t.Errorf("vol-live PV binding not decoded: %+v", live)
|
||||
}
|
||||
if len(live.MountedBy) != 1 || live.MountedBy[0] != "db/pg-0" {
|
||||
t.Errorf("vol-live mountedBy = %v, want [db/pg-0]", live.MountedBy)
|
||||
}
|
||||
junk, _ := findVolume(snap, "vol-junk")
|
||||
if junk.State != StateUnreferenced || !junk.Deletable {
|
||||
t.Errorf("vol-junk = %s deletable=%v, want unreferenced + deletable", junk.State, junk.Deletable)
|
||||
}
|
||||
if snap.Cost.ReclaimableMonthly != 400 {
|
||||
t.Errorf("reclaimable = %d, want 400 (40 GiB)", snap.Cost.ReclaimableMonthly)
|
||||
}
|
||||
if n := snap.Nodes[0]; !n.Ready || n.Pods != 1 || n.Cluster != "test-k8s" {
|
||||
t.Errorf("node join wrong: %+v", n)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUnreachableClusterFreezesEverything: the cluster exists but will not answer, so
|
||||
// the volume no reachable PV references must STILL be undeletable.
|
||||
func TestUnreachableClusterFreezesEverything(t *testing.T) {
|
||||
f := &fakeDO{kubeAPI: "", clusters: []string{clusterUUID}, volumes: twoVolumes}
|
||||
snap, err := collect(context.Background(), f.server(t))
|
||||
if err != nil {
|
||||
t.Fatalf("collect: %v", err)
|
||||
}
|
||||
if snap.Complete {
|
||||
t.Fatal("Complete = true with an unreachable cluster")
|
||||
}
|
||||
for _, id := range []string{"vol-live", "vol-junk"} {
|
||||
v, _ := findVolume(snap, id)
|
||||
if v.Deletable {
|
||||
t.Fatalf("%s deletable despite an unreachable cluster", id)
|
||||
}
|
||||
}
|
||||
if snap.Cost.ReclaimableMonthly != 0 {
|
||||
t.Errorf("reclaimable = %d, want 0", snap.Cost.ReclaimableMonthly)
|
||||
}
|
||||
// The failure must be named in Sources, not swallowed.
|
||||
var found bool
|
||||
for _, s := range snap.Sources {
|
||||
if s.Name == "k8s.test-k8s" && !s.OK && s.Error != "" {
|
||||
found = true
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Errorf("cluster failure absent from sources: %+v", snap.Sources)
|
||||
}
|
||||
}
|
||||
|
||||
// TestDeleteRefusesNonDeletable proves the server never trusts the caller: asking to
|
||||
// delete a live volume is refused with a reason, and NOTHING is deleted upstream.
|
||||
func TestDeleteRefusesNonDeletable(t *testing.T) {
|
||||
f := &fakeDO{kubeAPI: fakeAPIServer(t), clusters: []string{clusterUUID}, volumes: twoVolumes}
|
||||
do := f.server(t)
|
||||
b := &board{}
|
||||
snap, err := b.load(context.Background(), do, true)
|
||||
if err != nil {
|
||||
t.Fatalf("load: %v", err)
|
||||
}
|
||||
live, _ := findVolume(snap, "vol-live")
|
||||
if live.Deletable {
|
||||
t.Fatal("fixture wrong: vol-live must not be deletable")
|
||||
}
|
||||
if live.BlockedReason == "" {
|
||||
t.Error("no blockedReason for a live volume")
|
||||
}
|
||||
if len(f.deleted) != 0 {
|
||||
t.Fatalf("volumes deleted during a read: %v", f.deleted)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSnapshotThenDelete proves the undo exists: deleting takes a snapshot first.
|
||||
func TestSnapshotThenDelete(t *testing.T) {
|
||||
f := &fakeDO{kubeAPI: fakeAPIServer(t), clusters: []string{clusterUUID}, volumes: twoVolumes}
|
||||
do := f.server(t)
|
||||
snap, err := collect(context.Background(), do)
|
||||
if err != nil {
|
||||
t.Fatalf("collect: %v", err)
|
||||
}
|
||||
junk, _ := findVolume(snap, "vol-junk")
|
||||
if _, err := takeSnapshot(context.Background(), do, junk, ""); err != nil {
|
||||
t.Fatalf("snapshot: %v", err)
|
||||
}
|
||||
if f.snapshot != 1 {
|
||||
t.Fatalf("snapshots taken = %d, want 1", f.snapshot)
|
||||
}
|
||||
if err := do.DeleteVolume(context.Background(), junk.ID); err != nil {
|
||||
t.Fatalf("delete: %v", err)
|
||||
}
|
||||
if len(f.deleted) != 1 || f.deleted[0] != "vol-junk" {
|
||||
t.Fatalf("deleted = %v, want [vol-junk]", f.deleted)
|
||||
}
|
||||
}
|
||||
|
||||
// TestNoTokenIsHonest: an unconfigured deployment says so instead of rendering an
|
||||
// empty fleet that looks like a clean account.
|
||||
func TestNoTokenIsHonest(t *testing.T) {
|
||||
_, err := collect(context.Background(), digitalocean.New(""))
|
||||
if err == nil || !strings.Contains(err.Error(), "DO_API_TOKEN") {
|
||||
t.Fatalf("err = %v, want an explicit not-configured error", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,113 @@
|
||||
package infra
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/hanzoai/cloud/clients/admin/digitalocean"
|
||||
)
|
||||
|
||||
// TestLiveCollect runs the real fan-out against the real DigitalOcean account and the
|
||||
// real clusters. It is the only test that can prove the orphan analysis agrees with
|
||||
// production, so it is kept — but it is SKIPPED unless DO_API_TOKEN is present, which
|
||||
// is never the case in CI or on a dev box that has not opted in.
|
||||
//
|
||||
// DO_API_TOKEN=$(…) go test ./clients/admin/infra/ -run TestLiveCollect -v
|
||||
//
|
||||
// It asserts invariants, not fixed counts: the fleet changes, but "every cluster
|
||||
// answered", "every volume got a state", and "only unreferenced volumes are deletable"
|
||||
// must hold on every run, forever.
|
||||
func TestLiveCollect(t *testing.T) {
|
||||
token := os.Getenv("DO_API_TOKEN")
|
||||
if token == "" {
|
||||
t.Skip("DO_API_TOKEN not set — live fleet test skipped")
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
|
||||
defer cancel()
|
||||
|
||||
snap, err := collect(ctx, digitalocean.New(token))
|
||||
if err != nil {
|
||||
t.Fatalf("collect: %v", err)
|
||||
}
|
||||
|
||||
byState := map[string]int{}
|
||||
gibByState := map[string]int{}
|
||||
for _, v := range snap.Volumes {
|
||||
byState[v.State]++
|
||||
gibByState[v.State] += v.SizeGiB
|
||||
}
|
||||
t.Logf("clusters=%d nodes=%d volumes=%d loadBalancers=%d complete=%v",
|
||||
snap.Totals.Clusters, snap.Totals.Nodes, snap.Totals.Volumes,
|
||||
snap.Totals.LoadBalancers, snap.Complete)
|
||||
for _, st := range []string{StateAttached, StateBound, StateReleased, StateUnreferenced} {
|
||||
t.Logf(" %-14s %4d volumes %7.2f TiB", st, byState[st], float64(gibByState[st])/1024)
|
||||
}
|
||||
t.Logf("local disk (INCLUDED in droplet price, not separately billed): %.2f TiB",
|
||||
float64(snap.Totals.LocalDiskGiB)/1024)
|
||||
t.Logf("cost/mo: droplets $%.2f volumes $%.2f lbs $%.2f TOTAL $%.2f reclaimable $%.2f",
|
||||
float64(snap.Cost.DropletsMonthly)/100, float64(snap.Cost.VolumesMonthly)/100,
|
||||
float64(snap.Cost.LoadBalancersMonthly)/100, float64(snap.Cost.TotalMonthly)/100,
|
||||
float64(snap.Cost.ReclaimableMonthly)/100)
|
||||
for _, c := range snap.Clusters {
|
||||
t.Logf(" %-16s nodes=%-3d pods=%-4d pvs=%-4d pvcs=%-4d idle=%-3d scanned=%v %s",
|
||||
c.Name, c.Nodes, c.Pods, c.PVs, c.PVCs, c.IdlePVCs, c.Scanned, c.ScanError)
|
||||
}
|
||||
var deletable []Volume
|
||||
for _, v := range snap.Volumes {
|
||||
if v.Deletable {
|
||||
deletable = append(deletable, v)
|
||||
}
|
||||
}
|
||||
t.Logf("DELETABLE: %d volumes", len(deletable))
|
||||
for _, v := range deletable {
|
||||
t.Logf(" %s %-40s %4d GiB $%.2f/mo", v.ID, v.Name, v.SizeGiB, float64(v.MonthlyCents)/100)
|
||||
}
|
||||
|
||||
// ---- invariants --------------------------------------------------------------
|
||||
if !snap.Complete {
|
||||
t.Fatalf("scan incomplete, so no verdict is trustworthy: %s", snap.IncompleteReason)
|
||||
}
|
||||
if snap.Totals.Clusters == 0 || snap.Totals.Nodes == 0 || snap.Totals.Volumes == 0 {
|
||||
t.Fatal("empty inventory from a live account")
|
||||
}
|
||||
for _, v := range snap.Volumes {
|
||||
if v.State == "" {
|
||||
t.Fatalf("volume %s has no state", v.ID)
|
||||
}
|
||||
if v.Deletable != (v.State == StateUnreferenced) {
|
||||
t.Fatalf("volume %s: deletable=%v but state=%s — only unreferenced volumes may be deletable",
|
||||
v.ID, v.Deletable, v.State)
|
||||
}
|
||||
if v.Deletable && len(v.DropletIDs) > 0 {
|
||||
t.Fatalf("volume %s is deletable while attached to %v", v.ID, v.DropletIDs)
|
||||
}
|
||||
if v.Deletable && v.PV != "" {
|
||||
t.Fatalf("volume %s is deletable while PV %s references it", v.ID, v.PV)
|
||||
}
|
||||
}
|
||||
// Every attached volume must sit on a droplet we actually enumerated, or the
|
||||
// attachment join is broken and cost attribution is wrong.
|
||||
nodes := map[int]bool{}
|
||||
for _, n := range snap.Nodes {
|
||||
nodes[n.ID] = true
|
||||
}
|
||||
for _, v := range snap.Volumes {
|
||||
for _, id := range v.DropletIDs {
|
||||
if !nodes[id] {
|
||||
t.Errorf("volume %s attached to unknown droplet %d", v.ID, id)
|
||||
}
|
||||
}
|
||||
}
|
||||
if int64(snap.Cost.ReclaimableMonthly) != sumCents(deletable) {
|
||||
t.Errorf("reclaimable %d != sum of deletable volumes %d", snap.Cost.ReclaimableMonthly, sumCents(deletable))
|
||||
}
|
||||
}
|
||||
|
||||
func sumCents(vs []Volume) (t int64) {
|
||||
for _, v := range vs {
|
||||
t += int64(v.MonthlyCents)
|
||||
}
|
||||
return
|
||||
}
|
||||
@@ -0,0 +1,249 @@
|
||||
package infra
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
policyv1 "k8s.io/api/policy/v1"
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/types"
|
||||
"k8s.io/client-go/kubernetes"
|
||||
|
||||
"github.com/hanzoai/cloud/clients/admin/core"
|
||||
"github.com/hanzoai/cloud/clients/admin/digitalocean"
|
||||
"github.com/hanzoai/cloud/clients/fleet"
|
||||
)
|
||||
|
||||
// clusterScanTimeout bounds ONE cluster's read. A cluster that exceeds it is recorded
|
||||
// as unreachable, which fails the completeness gate — the safe direction.
|
||||
const clusterScanTimeout = 45 * time.Second
|
||||
|
||||
// kube opens an authenticated client for one DOKS cluster. DO hands back a
|
||||
// token-based kubeconfig against the cluster's public https endpoint; it still goes
|
||||
// through fleet.SafeRESTConfig, the ONE gate that rejects exec-credential plugins and
|
||||
// non-routable apiserver hosts, so this path cannot be turned into an RCE or an SSRF.
|
||||
func kube(ctx context.Context, do *digitalocean.Client, clusterID string) (*kubernetes.Clientset, error) {
|
||||
raw, err := do.Kubeconfig(ctx, clusterID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("kubeconfig: %w", err)
|
||||
}
|
||||
cfg, err := fleet.SafeRESTConfig(raw)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cfg.Timeout = clusterScanTimeout
|
||||
return kubernetes.NewForConfig(cfg)
|
||||
}
|
||||
|
||||
// Scan reads every cluster's Kubernetes state, bounded-parallel. It ALWAYS returns
|
||||
// one row per cluster: a cluster that failed comes back with Err set rather than
|
||||
// being omitted, because a missing row and a healthy row must never be confusable —
|
||||
// that confusion is exactly what would condemn live data.
|
||||
func Scan(ctx context.Context, do *digitalocean.Client, clusters []digitalocean.Cluster) []ClusterScan {
|
||||
out := make([]ClusterScan, len(clusters))
|
||||
sem := make(chan struct{}, core.MaxCustomerConcurrency)
|
||||
var wg sync.WaitGroup
|
||||
for i, c := range clusters {
|
||||
wg.Add(1)
|
||||
go func(i int, c digitalocean.Cluster) {
|
||||
defer wg.Done()
|
||||
sem <- struct{}{}
|
||||
defer func() { <-sem }()
|
||||
cctx, cancel := context.WithTimeout(ctx, clusterScanTimeout)
|
||||
defer cancel()
|
||||
out[i] = scanOne(cctx, do, c.ID)
|
||||
}(i, c)
|
||||
}
|
||||
wg.Wait()
|
||||
return out
|
||||
}
|
||||
|
||||
// scanOne reads one cluster. Any error short-circuits with Err set.
|
||||
func scanOne(ctx context.Context, do *digitalocean.Client, clusterID string) ClusterScan {
|
||||
s := ClusterScan{ClusterID: clusterID}
|
||||
cs, err := kube(ctx, do, clusterID)
|
||||
if err != nil {
|
||||
s.Err = err
|
||||
return s
|
||||
}
|
||||
|
||||
pvs, err := cs.CoreV1().PersistentVolumes().List(ctx, metav1.ListOptions{})
|
||||
if err != nil {
|
||||
s.Err = fmt.Errorf("list persistentvolumes: %w", err)
|
||||
return s
|
||||
}
|
||||
for _, pv := range pvs.Items {
|
||||
s.PVs = append(s.PVs, PVRef{
|
||||
Name: pv.Name,
|
||||
Phase: string(pv.Status.Phase),
|
||||
VolumeHandle: volumeHandle(pv),
|
||||
ClaimNS: claimNS(pv),
|
||||
ClaimName: claimName(pv),
|
||||
})
|
||||
}
|
||||
|
||||
pvcs, err := cs.CoreV1().PersistentVolumeClaims(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
|
||||
if err != nil {
|
||||
s.Err = fmt.Errorf("list persistentvolumeclaims: %w", err)
|
||||
return s
|
||||
}
|
||||
for _, p := range pvcs.Items {
|
||||
s.PVCs = append(s.PVCs, PVCRef{
|
||||
Namespace: p.Namespace, Name: p.Name,
|
||||
Phase: string(p.Status.Phase), Volume: p.Spec.VolumeName,
|
||||
})
|
||||
}
|
||||
|
||||
pods, err := cs.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
|
||||
if err != nil {
|
||||
s.Err = fmt.Errorf("list pods: %w", err)
|
||||
return s
|
||||
}
|
||||
for _, p := range pods.Items {
|
||||
s.Pods = append(s.Pods, podRefOf(p))
|
||||
}
|
||||
|
||||
nodes, err := cs.CoreV1().Nodes().List(ctx, metav1.ListOptions{})
|
||||
if err != nil {
|
||||
s.Err = fmt.Errorf("list nodes: %w", err)
|
||||
return s
|
||||
}
|
||||
for _, n := range nodes.Items {
|
||||
s.Nodes = append(s.Nodes, NodeState{
|
||||
Name: n.Name,
|
||||
Ready: nodeReady(n),
|
||||
Schedulable: !n.Spec.Unschedulable,
|
||||
})
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// volumeHandle extracts the backing DO volume ID a PV claims. Deliberately NOT
|
||||
// filtered by CSI driver name: matching broadly means MORE volumes are treated as
|
||||
// in-use, which is the safe direction. The legacy flexVolume shape is read too, so a
|
||||
// pre-CSI PV still protects its volume.
|
||||
func volumeHandle(pv corev1.PersistentVolume) string {
|
||||
if pv.Spec.CSI != nil && strings.TrimSpace(pv.Spec.CSI.VolumeHandle) != "" {
|
||||
return pv.Spec.CSI.VolumeHandle
|
||||
}
|
||||
if pv.Spec.FlexVolume != nil {
|
||||
if v := strings.TrimSpace(pv.Spec.FlexVolume.Options["volumeID"]); v != "" {
|
||||
return v
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func claimNS(pv corev1.PersistentVolume) string {
|
||||
if pv.Spec.ClaimRef == nil {
|
||||
return ""
|
||||
}
|
||||
return pv.Spec.ClaimRef.Namespace
|
||||
}
|
||||
|
||||
func claimName(pv corev1.PersistentVolume) string {
|
||||
if pv.Spec.ClaimRef == nil {
|
||||
return ""
|
||||
}
|
||||
return pv.Spec.ClaimRef.Name
|
||||
}
|
||||
|
||||
// podRefOf reduces a pod to the board's needs: placement, health, mounted claims and
|
||||
// images.
|
||||
func podRefOf(p corev1.Pod) PodRef {
|
||||
r := PodRef{
|
||||
Namespace: p.Namespace, Name: p.Name,
|
||||
Phase: string(p.Status.Phase), Reason: p.Status.Reason, Node: p.Spec.NodeName,
|
||||
}
|
||||
for _, v := range p.Spec.Volumes {
|
||||
if v.PersistentVolumeClaim != nil {
|
||||
r.Claims = append(r.Claims, v.PersistentVolumeClaim.ClaimName)
|
||||
}
|
||||
}
|
||||
for _, c := range p.Spec.InitContainers {
|
||||
r.Images = append(r.Images, c.Image)
|
||||
}
|
||||
for _, c := range p.Spec.Containers {
|
||||
r.Images = append(r.Images, c.Image)
|
||||
}
|
||||
// A waiting container's reason (CrashLoopBackOff/ImagePullBackOff) is the real
|
||||
// health signal; pod.status.reason stays empty for those.
|
||||
for _, cs := range p.Status.ContainerStatuses {
|
||||
if cs.State.Waiting != nil && cs.State.Waiting.Reason != "" && r.Reason == "" {
|
||||
r.Reason = cs.State.Waiting.Reason
|
||||
}
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
func nodeReady(n corev1.Node) bool {
|
||||
for _, c := range n.Status.Conditions {
|
||||
if c.Type == corev1.NodeReady {
|
||||
return c.Status == corev1.ConditionTrue
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// SetSchedulable cordons or uncordons a node, optionally draining it. Returns the
|
||||
// number of pods evicted.
|
||||
//
|
||||
// Drain uses the Eviction API, not delete: eviction respects PodDisruptionBudgets, so
|
||||
// a drain that would break a quorum is REFUSED by the apiserver rather than silently
|
||||
// taking a service down. DaemonSet and mirror pods are skipped — they are rescheduled
|
||||
// onto the same node by definition and evicting them is a no-op loop.
|
||||
func SetSchedulable(ctx context.Context, do *digitalocean.Client, clusterID, node string, schedulable, drain bool) (int, error) {
|
||||
cs, err := kube(ctx, do, clusterID)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
patch := fmt.Sprintf(`{"spec":{"unschedulable":%t}}`, !schedulable)
|
||||
if _, err := cs.CoreV1().Nodes().Patch(ctx, node, types.MergePatchType, []byte(patch), metav1.PatchOptions{}); err != nil {
|
||||
return 0, fmt.Errorf("cordon %s: %w", node, err)
|
||||
}
|
||||
if schedulable || !drain {
|
||||
return 0, nil
|
||||
}
|
||||
pods, err := cs.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{
|
||||
FieldSelector: "spec.nodeName=" + node,
|
||||
})
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("list pods on %s: %w", node, err)
|
||||
}
|
||||
evicted := 0
|
||||
for _, p := range pods.Items {
|
||||
if skipEviction(p) {
|
||||
continue
|
||||
}
|
||||
ev := &policyv1.Eviction{ObjectMeta: metav1.ObjectMeta{Namespace: p.Namespace, Name: p.Name}}
|
||||
if err := cs.CoreV1().Pods(p.Namespace).EvictV1(ctx, ev); err != nil {
|
||||
if apierrors.IsNotFound(err) {
|
||||
continue
|
||||
}
|
||||
// A PDB refusal is the system working. Report it verbatim; the node stays
|
||||
// cordoned, so the operator can retry after scaling.
|
||||
return evicted, fmt.Errorf("evict %s/%s: %w", p.Namespace, p.Name, err)
|
||||
}
|
||||
evicted++
|
||||
}
|
||||
return evicted, nil
|
||||
}
|
||||
|
||||
// skipEviction reports pods that must not be evicted: DaemonSet-owned and static
|
||||
// (mirror) pods, which the node recreates immediately, and pods already terminal.
|
||||
func skipEviction(p corev1.Pod) bool {
|
||||
if _, mirror := p.Annotations[corev1.MirrorPodAnnotationKey]; mirror {
|
||||
return true
|
||||
}
|
||||
for _, o := range p.OwnerReferences {
|
||||
if o.Kind == "DaemonSet" {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return p.Status.Phase == corev1.PodSucceeded || p.Status.Phase == corev1.PodFailed
|
||||
}
|
||||
@@ -257,10 +257,14 @@ func reachable(path string) bool {
|
||||
if !strings.HasPrefix(path, "/v1/") {
|
||||
return true // the SPA shell + static assets that render the paywall screen itself.
|
||||
}
|
||||
// The auth routes moved when the /v1 surface was namespaced (/v1/signin →
|
||||
// /v1/ai/signin, /v1/get-account → /v1/ai/account). This list matches by
|
||||
// STRING, so it does not follow them: the old spellings here would put a 402
|
||||
// in front of sign-in. Pinned by TestAuthRoutesAreNeverPaywalled.
|
||||
switch path {
|
||||
case "/v1/signin", // auth: session bootstrap (the console posts the OAuth code here).
|
||||
"/v1/signout",
|
||||
"/v1/get-account", // auth: the account read AuthGate loads before anything else.
|
||||
case "/v1/ai/signin", // auth: session bootstrap (the console posts the OAuth code here).
|
||||
"/v1/ai/signout",
|
||||
"/v1/ai/account", // auth: the account read AuthGate loads before anything else.
|
||||
"/v1/entitlements": // this paywall's OWN projection — what the shell renders the upgrade UI from.
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -295,13 +295,13 @@ func TestPayPathStaysReachable(t *testing.T) {
|
||||
publish(t, &fakeLedger{credit: atto(0)})
|
||||
|
||||
for _, path := range []string{
|
||||
"/v1/billing/plans", // what to buy
|
||||
"/v1/billing/subscribe", // buying it
|
||||
"/v1/billing/webhooks/stripe", // the INBOUND payment callback — gating it loses money
|
||||
"/v1/plans", // the @hanzo/plans catalog
|
||||
"/v1/entitlements", // the shell's own upgrade projection
|
||||
"/v1/iam/login", // signing in to pay at all
|
||||
"/v1/signin", "/v1/get-account", // session bootstrap + the read AuthGate needs
|
||||
"/v1/billing/plans", // what to buy
|
||||
"/v1/billing/subscribe", // buying it
|
||||
"/v1/billing/webhooks/stripe", // the INBOUND payment callback — gating it loses money
|
||||
"/v1/plans", // the @hanzo/plans catalog
|
||||
"/v1/entitlements", // the shell's own upgrade projection
|
||||
"/v1/iam/login", // signing in to pay at all
|
||||
"/v1/ai/signin", "/v1/ai/account", // session bootstrap + the read AuthGate needs
|
||||
"/v1/orgs/acme/entitlements", // which org am I buying for
|
||||
"/v1/admin/flags", // the cockpit holding this gate's kill switch
|
||||
"/v1/waitlist", // admission's join API
|
||||
|
||||
+2
-2
@@ -97,8 +97,8 @@ var probes = []probe{
|
||||
{"automations", http.MethodGet, "/v1/automations/connectors", classAuthed},
|
||||
|
||||
// ── tolerant (preview / SuperAdmin / cross-org / staged-optional) ──
|
||||
{"get-account", http.MethodGet, "/v1/get-account", classTolerant},
|
||||
{"get-chats", http.MethodGet, "/v1/get-chats", classTolerant},
|
||||
{"account", http.MethodGet, "/v1/ai/account", classTolerant},
|
||||
{"chats", http.MethodGet, "/v1/ai/chats", classTolerant},
|
||||
{"kms-secrets", http.MethodGet, "/v1/kms/orgs/smoke/secrets", classTolerant}, // BUG 1: was 402
|
||||
{"deploy-apps", http.MethodGet, "/v1/deploy/applications", classTolerant},
|
||||
{"websearch", http.MethodGet, "/v1/websearch/search?q=hanzo", classTolerant},
|
||||
|
||||
+11
-4
@@ -171,7 +171,7 @@ func canonical(path string) string {
|
||||
// → src/lib/api/plans.ts): PlansApi.plans() reads GET /v1/billing/plans through the
|
||||
// per-tenant billing proxy, and checkout drives the /v1/billing/* money surface
|
||||
// (subscribe/subscriptions/balance/payment-methods/usage/invoices/credit/deposit/topup/
|
||||
// gpu/spend-alerts/payment-config). Auth is /v1/signin, /v1/signout, /v1/get-account and
|
||||
// gpu/spend-alerts/payment-config). Auth is /v1/ai/{signin,signout,account} and
|
||||
// the /v1/iam/* login/OAuth/OIDC callbacks; /v1/models is the model catalog the shell
|
||||
// reads; /v1/entitlements is the product projection the shell renders the upgrade UI from.
|
||||
// Its argument is already canonical() — gated() is the only caller, so every entry
|
||||
@@ -183,10 +183,17 @@ func allowlisted(path string) bool {
|
||||
return true
|
||||
}
|
||||
// Exact single-route exemptions.
|
||||
//
|
||||
// The three auth routes moved when the /v1 surface was namespaced
|
||||
// (/v1/signin → /v1/ai/signin, /v1/get-account → /v1/ai/account). This
|
||||
// list is matched by STRING, so it does not move with them — leaving the old
|
||||
// spellings here would put a 402 in front of sign-in, which is the same
|
||||
// outage the trailing-slash bug caused above and is pinned by
|
||||
// TestAuthRoutesAreNeverPaywalled.
|
||||
switch path {
|
||||
case "/v1/signin", // auth: session bootstrap (console posts the OAuth code here).
|
||||
"/v1/signout",
|
||||
"/v1/get-account", // auth: the account read AuthGate loads before anything else.
|
||||
case "/v1/ai/signin", // auth: session bootstrap (console posts the OAuth code here).
|
||||
"/v1/ai/signout",
|
||||
"/v1/ai/account", // auth: the account read AuthGate loads before anything else.
|
||||
"/v1/plans", // plans catalog root (@hanzo/plans subsystem).
|
||||
"/v1/models", // OpenAI-compatible model catalog (discovery).
|
||||
"/v1/entitlements": // the shell's product projection — renders the upgrade UI.
|
||||
|
||||
+45
-5
@@ -122,7 +122,7 @@ func TestPaywall_NoPlan402(t *testing.T) {
|
||||
// pay button can never happen.
|
||||
func TestPaywall_AllowlistedPassWithNoPlan(t *testing.T) {
|
||||
paths := []string{
|
||||
"/v1/signin", "/v1/signout", "/v1/get-account",
|
||||
"/v1/ai/signin", "/v1/ai/signout", "/v1/ai/account",
|
||||
"/v1/billing/plans", "/v1/billing/subscriptions", "/v1/billing/balance",
|
||||
"/v1/billing/payment-methods", "/v1/billing/usage",
|
||||
"/v1/plans", "/v1/plans/resolve/pro",
|
||||
@@ -280,7 +280,7 @@ func TestGated_CaseFoldedSoAllowlistStillMatches(t *testing.T) {
|
||||
"/V1/iam/callback",
|
||||
"/v1/IAM/callback",
|
||||
"/V1/BILLING/subscribe",
|
||||
"/V1/signin",
|
||||
"/V1/ai/signin",
|
||||
} {
|
||||
if gated(path) {
|
||||
t.Errorf("gated(%q) = true, want false — 402 on the sell/service surface", path)
|
||||
@@ -292,9 +292,9 @@ func TestGated_CaseFoldedSoAllowlistStillMatches(t *testing.T) {
|
||||
// list nor any allow prefix, so /v1/signin/ was paywalled: a 402 in front of sign-in.
|
||||
func TestGated_TrailingSlashDoesNotLockOut(t *testing.T) {
|
||||
for _, path := range []string{
|
||||
"/v1/signin/",
|
||||
"/v1/signout/",
|
||||
"/v1/get-account/",
|
||||
"/v1/ai/signin/",
|
||||
"/v1/ai/signout/",
|
||||
"/v1/ai/account/",
|
||||
"/v1/entitlements/",
|
||||
"/v1/plans/",
|
||||
"/v1/models/",
|
||||
@@ -389,3 +389,43 @@ func TestPaywall_NilReaderPassesEverything(t *testing.T) {
|
||||
t.Fatal("handler did not run")
|
||||
}
|
||||
}
|
||||
|
||||
// TestAuthRoutesAreNeverPaywalled pins the one failure this gate must never have:
|
||||
// a 402 in front of signing in.
|
||||
//
|
||||
// The allow-list matches by STRING, so it does NOT move when a route moves. When
|
||||
// the /v1 surface was namespaced (/v1/signin → /v1/ai/signin, /v1/get-account →
|
||||
// /v1/ai/account) the paths changed underneath this list, and a list left
|
||||
// un-updated would have gated the entire sell/service entry path — the same shape
|
||||
// of outage the trailing-slash and case-folding bugs already caused here twice.
|
||||
//
|
||||
// The names below are the paths the console and every client actually request. If
|
||||
// a route moves again, this test fails rather than production.
|
||||
func TestAuthRoutesAreNeverPaywalled(t *testing.T) {
|
||||
for _, p := range []string{
|
||||
"/v1/ai/signin",
|
||||
"/v1/ai/signout",
|
||||
"/v1/ai/account",
|
||||
} {
|
||||
if gated(p) {
|
||||
t.Errorf("gated(%q) = true — a 402 in front of sign-in", p)
|
||||
}
|
||||
// The same route must survive the two shapes canonical() folds, or the
|
||||
// gate disagrees with itself for a caller that sent a trailing slash or
|
||||
// different case.
|
||||
for _, variant := range []string{p + "/", strings.ToUpper(p)} {
|
||||
if gated(variant) {
|
||||
t.Errorf("gated(%q) = true — the folded form must be exempt too", variant)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// And the old spellings must NOT be silently exempt: they name routes that no
|
||||
// longer exist, so leaving them on the list would be dead weight that reads
|
||||
// like coverage.
|
||||
for _, dead := range []string{"/v1/signin", "/v1/signout", "/v1/get-account"} {
|
||||
if !gated(dead) {
|
||||
t.Errorf("%q is still allow-listed — it names a route that no longer exists", dead)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+54
-12
@@ -99,21 +99,25 @@ func (d *dispatcher) dispatch(call zaprpc.Call, cookieHeader, authHeader, accept
|
||||
|
||||
// buildHTTPRequest maps (method, SuperJSON input) onto a /v1 request.
|
||||
//
|
||||
// the /v1 convention (mirrored from console/src/lib/api/providers.ts):
|
||||
// - "get-*" -> GET /v1/<method> with scalar input fields as the query.
|
||||
// - others -> POST /v1/<method> with scalar input fields lifted to the
|
||||
// The method is an HTTP request line, "<VERB> <path>" (see splitMethod):
|
||||
// - a read verb (GET/HEAD/DELETE) -> no body; scalar input fields become the query.
|
||||
// - a write verb (POST/PUT/PATCH) -> scalar input fields lifted to the
|
||||
// query (identity hints like id/owner) and the single nested
|
||||
// object/array field as the JSON body (the resource); if there is no
|
||||
// nested field, the whole input is the body.
|
||||
//
|
||||
// This single rule covers every twin shape without per-endpoint coupling:
|
||||
//
|
||||
// get-provider {id} -> GET ?id=...
|
||||
// get-providers {owner,store,...} -> GET ?owner=...&store=...
|
||||
// update-provider {id, provider} -> POST ?id=... body=provider
|
||||
// add-provider <provider> -> POST body=<provider>
|
||||
// GET ai/providers/acme/openai {} -> GET /v1/ai/providers/acme/openai
|
||||
// GET ai/providers {owner,store} -> GET /v1/ai/providers?owner=...&store=...
|
||||
// PATCH ai/providers/acme/openai {provider} -> PATCH … body=provider
|
||||
// POST ai/providers <provider> -> POST /v1/ai/providers body=<provider>
|
||||
// DELETE ai/providers/acme/openai {} -> DELETE /v1/ai/providers/acme/openai
|
||||
func buildHTTPRequest(req zapRequest) (*http.Request, error) {
|
||||
path := "/v1/" + strings.TrimPrefix(req.method, "/")
|
||||
verb, path, err := splitMethod(req.method)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
inputJSON := superJSONUnwrap(req.payload)
|
||||
|
||||
scalars, nested := splitInput(inputJSON)
|
||||
@@ -127,13 +131,13 @@ func buildHTTPRequest(req zapRequest) (*http.Request, error) {
|
||||
target += "?" + enc
|
||||
}
|
||||
|
||||
if strings.HasPrefix(req.method, "get-") {
|
||||
r := httptest.NewRequest(http.MethodGet, target, nil)
|
||||
if verb == http.MethodGet || verb == http.MethodHead || verb == http.MethodDelete {
|
||||
r := httptest.NewRequest(verb, target, nil)
|
||||
r.Header.Set("Accept", "application/json")
|
||||
return r, nil
|
||||
}
|
||||
|
||||
// Mutation: choose the body.
|
||||
// Mutation with a body: choose it.
|
||||
var body []byte
|
||||
switch {
|
||||
case nested != nil:
|
||||
@@ -143,13 +147,51 @@ func buildHTTPRequest(req zapRequest) (*http.Request, error) {
|
||||
default:
|
||||
body = []byte("{}")
|
||||
}
|
||||
r := httptest.NewRequest(http.MethodPost, target, bytes.NewReader(body))
|
||||
r := httptest.NewRequest(verb, target, bytes.NewReader(body))
|
||||
r.Header.Set("Content-Type", "application/json")
|
||||
r.Header.Set("Accept", "application/json")
|
||||
r.ContentLength = int64(len(body))
|
||||
return r, nil
|
||||
}
|
||||
|
||||
// zapVerbs are the methods a ZAP call may name. Anything else is refused rather
|
||||
// than coerced — a caller that cannot say what it wants to do does not get a guess.
|
||||
var zapVerbs = map[string]bool{
|
||||
http.MethodGet: true, http.MethodHead: true, http.MethodPost: true,
|
||||
http.MethodPut: true, http.MethodPatch: true, http.MethodDelete: true,
|
||||
}
|
||||
|
||||
// splitMethod reads a ZAP method as an HTTP request line: "<VERB> <path>", e.g.
|
||||
// "GET rag/stores" or "PATCH ai/providers/acme/openai". The path is rooted at /v1.
|
||||
//
|
||||
// The verb is CARRIED, not inferred. It used to be guessed from the method name —
|
||||
// a "get-" prefix meant GET and everything else meant POST — which worked only
|
||||
// because the route surface encoded its verb in every route name (/v1/get-store
|
||||
// vs /v1/update-store). Once the surface became RESTful that signal disappeared:
|
||||
// one path answers GET, PATCH and DELETE, and no prefix distinguishes them. A
|
||||
// heuristic there would silently send a delete as a POST.
|
||||
//
|
||||
// A method with no verb is an ERROR, not a default. Defaulting would turn a
|
||||
// caller's omission into a wrong-but-plausible request — for a surface where the
|
||||
// verb decides between reading a resource and destroying it, that is the one
|
||||
// behaviour worth refusing outright.
|
||||
func splitMethod(method string) (verb, path string, err error) {
|
||||
m := strings.TrimSpace(method)
|
||||
head, rest, ok := strings.Cut(m, " ")
|
||||
if !ok {
|
||||
return "", "", fmt.Errorf("method %q must be %q — the HTTP verb is required, not inferred", method, "<VERB> <path>")
|
||||
}
|
||||
verb = strings.ToUpper(strings.TrimSpace(head))
|
||||
if !zapVerbs[verb] {
|
||||
return "", "", fmt.Errorf("method %q names an unsupported verb %q", method, verb)
|
||||
}
|
||||
rest = strings.TrimSpace(rest)
|
||||
if rest == "" {
|
||||
return "", "", fmt.Errorf("method %q names no path", method)
|
||||
}
|
||||
return verb, "/v1/" + strings.TrimPrefix(rest, "/"), nil
|
||||
}
|
||||
|
||||
// splitInput separates a JSON object into its scalar fields (rendered as query
|
||||
// strings) and at most one nested object/array field (the resource body). A
|
||||
// non-object input yields no scalars and the whole value as the nested body.
|
||||
|
||||
+39
-18
@@ -26,27 +26,31 @@ func startZapApp(t *testing.T) (string, func()) {
|
||||
t.Helper()
|
||||
app := zip.New(zip.Config{})
|
||||
|
||||
// /v1 handlers (mirror ai/mount.go: app.All("/v1/*", ...)).
|
||||
// /v1 handlers (mirror ai/mount.go: app.All("/v1/*", ...)). The surface is
|
||||
// RESTful, so these switch on METHOD AND PATH — the same pair the dispatcher
|
||||
// now has to carry, and the reason it can no longer infer a verb from a name.
|
||||
app.All("/v1/*", func(c *zip.Ctx) error {
|
||||
path := c.Path()
|
||||
path, method := c.Path(), c.Method()
|
||||
switch {
|
||||
case strings.HasSuffix(path, "/get-global-providers"):
|
||||
case method == "GET" && strings.HasSuffix(path, "/ai/providers/global"):
|
||||
return c.JSON(200, fiber.Map{
|
||||
"status": "ok", "msg": "",
|
||||
"data": []fiber.Map{
|
||||
{"owner": "admin", "name": "openai", "category": "Model", "_cookie": c.Header("Cookie")},
|
||||
},
|
||||
})
|
||||
case strings.HasSuffix(path, "/get-providers"):
|
||||
case method == "GET" && strings.HasSuffix(path, "/ai/providers"):
|
||||
return c.JSON(200, fiber.Map{
|
||||
"status": "ok", "msg": "",
|
||||
"data": []fiber.Map{{"owner": c.Query("owner"), "name": "p1"}},
|
||||
"data2": 1,
|
||||
})
|
||||
case strings.HasSuffix(path, "/add-provider"):
|
||||
case method == "POST" && strings.HasSuffix(path, "/ai/providers"):
|
||||
body := map[string]any{}
|
||||
_ = json.Unmarshal(c.Body(), &body)
|
||||
return c.JSON(200, fiber.Map{"status": "ok", "msg": "added " + asString(body["name"])})
|
||||
case method == "DELETE" && strings.Contains(path, "/ai/providers/"):
|
||||
return c.JSON(200, fiber.Map{"status": "ok", "msg": "deleted " + path})
|
||||
default:
|
||||
return c.JSON(404, fiber.Map{"status": "error", "msg": "not found"})
|
||||
}
|
||||
@@ -130,10 +134,10 @@ func TestEndToEndOverWebSocket(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// 1) get-global-providers -> ok, real data, cookie replayed to /v1 handler.
|
||||
rep := call("get-global-providers", nil, 1)
|
||||
// 1) A global listing -> ok, real data, cookie replayed to the /v1 handler.
|
||||
rep := call("GET ai/providers/global", nil, 1)
|
||||
if !rep.ok || rep.status != 200 {
|
||||
t.Fatalf("get-global-providers: ok=%v status=%d err=%s", rep.ok, rep.status, rep.errorJSON)
|
||||
t.Fatalf("GET ai/providers/global: ok=%v status=%d err=%s", rep.ok, rep.status, rep.errorJSON)
|
||||
}
|
||||
var providers []map[string]any
|
||||
if err := json.Unmarshal(superJSONUnwrap(rep.result), &providers); err != nil {
|
||||
@@ -146,26 +150,43 @@ func TestEndToEndOverWebSocket(t *testing.T) {
|
||||
t.Fatalf("cookie not replayed to /v1 handler: %v", providers[0]["_cookie"])
|
||||
}
|
||||
|
||||
// 2) get-providers with a query param.
|
||||
rep = call("get-providers", map[string]any{"owner": "admin"}, 2)
|
||||
// 2) A collection list with a query param.
|
||||
rep = call("GET ai/providers", map[string]any{"owner": "admin"}, 2)
|
||||
if !rep.ok {
|
||||
t.Fatalf("get-providers !ok: %s", rep.errorJSON)
|
||||
t.Fatalf("GET ai/providers !ok: %s", rep.errorJSON)
|
||||
}
|
||||
_ = json.Unmarshal(superJSONUnwrap(rep.result), &providers)
|
||||
if providers[0]["owner"] != "admin" {
|
||||
t.Fatalf("get-providers owner query not forwarded: %v", providers)
|
||||
t.Fatalf("owner query not forwarded: %v", providers)
|
||||
}
|
||||
|
||||
// 3) add-provider (POST body).
|
||||
rep = call("add-provider", map[string]any{"owner": "admin", "name": "claude"}, 3)
|
||||
// 3) Create (POST body).
|
||||
rep = call("POST ai/providers", map[string]any{"owner": "admin", "name": "claude"}, 3)
|
||||
if !rep.ok {
|
||||
t.Fatalf("add-provider !ok: %s", rep.errorJSON)
|
||||
t.Fatalf("POST ai/providers !ok: %s", rep.errorJSON)
|
||||
}
|
||||
|
||||
// 4) unknown method -> /v1 404 envelope -> !ok.
|
||||
rep = call("get-nonexistent", nil, 4)
|
||||
// 4) A DELETE reaches the DELETE route — the case the old prefix heuristic
|
||||
// got wrong: it had no "get-" prefix, so it would have been sent as a POST
|
||||
// and silently landed on the create route instead of destroying anything.
|
||||
rep = call("DELETE ai/providers/admin/openai", nil, 4)
|
||||
if !rep.ok {
|
||||
t.Fatalf("DELETE ai/providers/admin/openai !ok: %s", rep.errorJSON)
|
||||
}
|
||||
|
||||
// 5) unknown path -> /v1 404 envelope -> !ok.
|
||||
rep = call("GET ai/nonexistent", nil, 5)
|
||||
if rep.ok {
|
||||
t.Fatalf("unknown method should not be ok")
|
||||
t.Fatalf("unknown path should not be ok")
|
||||
}
|
||||
|
||||
// 6) A method with NO verb is refused outright, not guessed at.
|
||||
rep = call("get-providers", nil, 6)
|
||||
if rep.ok {
|
||||
t.Fatalf("a verbless method must be refused, not inferred")
|
||||
}
|
||||
if !strings.Contains(rep.errorJSON, "verb is required") {
|
||||
t.Fatalf("refusal should name the missing verb, got %s", rep.errorJSON)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+30
-21
@@ -36,7 +36,7 @@ func buildClientRequestBytes(method, payload string, promiseID uint32) []byte {
|
||||
// TestParseClientRequest proves the server decodes a real client frame: outer
|
||||
// rpc envelope -> inner ZapRequest -> (method, SuperJSON payload).
|
||||
func TestParseClientRequest(t *testing.T) {
|
||||
const method = "get-providers"
|
||||
const method = "GET ai/providers"
|
||||
// console sends SuperJSON.stringify(input); a plain object X -> {"json":X}.
|
||||
input := map[string]any{"owner": "admin", "store": "default", "limit": "20"}
|
||||
innerJSON, _ := json.Marshal(input)
|
||||
@@ -176,45 +176,54 @@ func TestBuildHTTPRequest(t *testing.T) {
|
||||
wantBodyHa string // a substring expected in the body ('' = no body)
|
||||
}{
|
||||
{
|
||||
name: "get-global-providers no args",
|
||||
method: "get-global-providers",
|
||||
name: "global listing, no args",
|
||||
method: "GET ai/providers/global",
|
||||
input: nil,
|
||||
wantMethod: "GET",
|
||||
wantPath: "/v1/get-global-providers",
|
||||
wantPath: "/v1/ai/providers/global",
|
||||
},
|
||||
{
|
||||
name: "get-provider with id",
|
||||
method: "get-provider",
|
||||
input: map[string]any{"id": "admin/openai"},
|
||||
name: "member read — the id is in the PATH, not a query param",
|
||||
method: "GET ai/providers/admin/openai",
|
||||
input: nil,
|
||||
wantMethod: "GET",
|
||||
wantPath: "/v1/get-provider",
|
||||
wantQuery: map[string]string{"id": "admin/openai"},
|
||||
wantPath: "/v1/ai/providers/admin/openai",
|
||||
},
|
||||
{
|
||||
name: "get-providers list query",
|
||||
method: "get-providers",
|
||||
name: "collection list keeps its filters as a query",
|
||||
method: "GET ai/providers",
|
||||
input: map[string]any{"owner": "admin", "store": "default"},
|
||||
wantMethod: "GET",
|
||||
wantPath: "/v1/get-providers",
|
||||
wantPath: "/v1/ai/providers",
|
||||
wantQuery: map[string]string{"owner": "admin", "store": "default"},
|
||||
},
|
||||
{
|
||||
name: "update-provider id+resource",
|
||||
method: "update-provider",
|
||||
input: map[string]any{"id": "admin/openai", "provider": map[string]any{"owner": "admin", "name": "openai", "type": "OpenAI"}},
|
||||
wantMethod: "POST",
|
||||
wantPath: "/v1/update-provider",
|
||||
wantQuery: map[string]string{"id": "admin/openai"},
|
||||
// The verb travels with the call. Under the old prefix heuristic this
|
||||
// same request would have been sent as a POST.
|
||||
name: "member update is a PATCH with the resource as the body",
|
||||
method: "PATCH ai/providers/admin/openai",
|
||||
input: map[string]any{"provider": map[string]any{"owner": "admin", "name": "openai", "type": "OpenAI"}},
|
||||
wantMethod: "PATCH",
|
||||
wantPath: "/v1/ai/providers/admin/openai",
|
||||
wantBodyHa: `"type":"OpenAI"`,
|
||||
},
|
||||
{
|
||||
name: "add-provider bare resource",
|
||||
method: "add-provider",
|
||||
name: "create posts the bare resource to the collection",
|
||||
method: "POST ai/providers",
|
||||
input: map[string]any{"owner": "admin", "name": "openai", "type": "OpenAI"},
|
||||
wantMethod: "POST",
|
||||
wantPath: "/v1/add-provider",
|
||||
wantPath: "/v1/ai/providers",
|
||||
wantBodyHa: `"name":"openai"`,
|
||||
},
|
||||
{
|
||||
// A DELETE carries no body — and under the old heuristic it would have
|
||||
// been a POST, i.e. a destroy sent as a write to the wrong route.
|
||||
name: "member delete is a DELETE with no body",
|
||||
method: "DELETE ai/providers/admin/openai",
|
||||
input: nil,
|
||||
wantMethod: "DELETE",
|
||||
wantPath: "/v1/ai/providers/admin/openai",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
|
||||
Reference in New Issue
Block a user