feat(world/cloud): real DigitalOcean fleet on the map + overview
Enumerate the live DO fleet (DOKS clusters + their nodes + standalone droplets) via the DO REST API and surface it as geo-located data: - /v1/world/cloud/nodes: per-region rollup + per-cluster detail, each placed via the shared regionCatalog (resolveRegion) with a supplement for DO regions the 8-city catalog lacks (tor1, atl1). Public, cached 60s with last-known fail-soft — never 5xxes, never blanks the map. - cloud-pulse: folds the real fleet into the overview counts + region breakdown when the multi-cloud visor plane isn't wired (source "digitalocean"), so the signed-out dashboard shows live infra. Honesty: every DOKS node IS a droplet (tagged k8s) — those are counted once as nodes; "droplets" means standalone only, never double-counted. GPUs stay 0 (DOKS has no GPU pools). Token read at boot from hanzo KMS /world-secrets (DIGITALOCEAN_ACCESS_TOKEN) — never committed. Tests: fixtures cover aggregation, k8s-droplet de-dup, pagination, catalog+supplement placement, online/degraded status, and fail-soft (unconfigured/unavailable/last-known).
This commit is contained in:
hanzo-dev
@@ -178,8 +178,17 @@ func (s *Server) producePulse(ctx context.Context, auth map[string]string) cloud
|
||||
real := false
|
||||
volSrc := ""
|
||||
|
||||
// 1) Fleet COUNTS + REGION breakdown (auth → visor + ai catalog).
|
||||
// 1) Fleet COUNTS + REGION breakdown. Best source is the multi-cloud visor plane
|
||||
// (auth → /v1/machines + /v1/gpus). When that is not wired, fall back to the
|
||||
// REAL DigitalOcean fleet enumerated directly (DOKS clusters + nodes + droplets)
|
||||
// so the public dashboard shows the live infra either way — never an empty fleet.
|
||||
fleetSrc := ""
|
||||
countsReal := s.applyServiceCounts(ctx, &p, auth)
|
||||
if countsReal {
|
||||
fleetSrc = "service"
|
||||
} else if s.applyDOFleet(ctx, &p) {
|
||||
countsReal, fleetSrc = true, "digitalocean"
|
||||
}
|
||||
if countsReal {
|
||||
real = true
|
||||
}
|
||||
@@ -201,8 +210,8 @@ func (s *Server) producePulse(ctx context.Context, auth map[string]string) cloud
|
||||
|
||||
p.Demo = !real
|
||||
switch {
|
||||
case countsReal:
|
||||
p.Source = "service" // the service-token plane resolved (counts, and usually ledger volume)
|
||||
case fleetSrc != "":
|
||||
p.Source = fleetSrc // "service" (visor plane) or "digitalocean" (direct DO fleet)
|
||||
case real:
|
||||
p.Source = "public" // tokenless, but real public volume/uptime landed
|
||||
default:
|
||||
|
||||
@@ -0,0 +1,410 @@
|
||||
package world
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Real DigitalOcean fleet — the live infrastructure behind Hanzo Cloud, placed on
|
||||
// the world map and summed into the platform overview. It enumerates the actual DO
|
||||
// resources (DOKS clusters + their nodes, and any standalone droplets) via the DO
|
||||
// REST API and geo-locates each by its region slug through the shared region
|
||||
// catalog (resolveRegion). NOTHING here is fabricated: counts are whatever the DO
|
||||
// API returns right now, or an honest empty when the token is unset / DO is
|
||||
// unreachable.
|
||||
//
|
||||
// HONESTY / DE-DUP CONTRACT: every DOKS worker node IS a droplet (DO backs each
|
||||
// node with a droplet tagged `k8s` / `k8s:<clusterID>`). Reporting nodes AND those
|
||||
// same droplets would double-count the same machines, so droplets tagged as k8s
|
||||
// workers are EXCLUDED from the standalone-droplet count — `droplets` means only
|
||||
// droplets that are NOT part of a cluster. A DOKS node is counted once, as a node.
|
||||
//
|
||||
// FAIL-SOFT: the enumeration is cached (doFleetTTL) with last-known fallback. A DO
|
||||
// API hiccup returns the previous good fleet (or an honest empty on a cold miss) —
|
||||
// it never errors, never blocks the pulse, never blanks the map.
|
||||
|
||||
// ── DO API shapes (only the fields we read) ──────────────────────────────────
|
||||
|
||||
type doAPICluster struct {
|
||||
Name string `json:"name"`
|
||||
Region string `json:"region"`
|
||||
Version string `json:"version"`
|
||||
Status struct {
|
||||
State string `json:"state"`
|
||||
} `json:"status"`
|
||||
NodePools []struct {
|
||||
Count int `json:"count"`
|
||||
Nodes []struct {
|
||||
Status struct {
|
||||
State string `json:"state"`
|
||||
} `json:"status"`
|
||||
} `json:"nodes"`
|
||||
} `json:"node_pools"`
|
||||
}
|
||||
|
||||
type doAPIClustersResp struct {
|
||||
Clusters []doAPICluster `json:"kubernetes_clusters"`
|
||||
}
|
||||
|
||||
type doAPIDroplet struct {
|
||||
Name string `json:"name"`
|
||||
Status string `json:"status"`
|
||||
Tags []string `json:"tags"`
|
||||
Region struct {
|
||||
Slug string `json:"slug"`
|
||||
} `json:"region"`
|
||||
SizeSlug string `json:"size_slug"`
|
||||
}
|
||||
|
||||
type doAPIDropletsResp struct {
|
||||
Droplets []doAPIDroplet `json:"droplets"`
|
||||
Links struct {
|
||||
Pages struct {
|
||||
Next string `json:"next"`
|
||||
} `json:"pages"`
|
||||
} `json:"links"`
|
||||
}
|
||||
|
||||
// ── world-facing fleet shapes (the /v1/world/cloud/nodes contract) ────────────
|
||||
|
||||
// doCluster is one DOKS cluster placed on the globe: its region coords, node count
|
||||
// and how many of those nodes report running.
|
||||
type doCluster struct {
|
||||
Name string `json:"name"`
|
||||
Region string `json:"region"` // DO region slug (e.g. "sfo3")
|
||||
City string `json:"city"`
|
||||
Lat float64 `json:"lat"`
|
||||
Lon float64 `json:"lon"`
|
||||
Nodes int `json:"nodes"`
|
||||
Online int `json:"online"`
|
||||
Status string `json:"status"` // cluster state (running/…)
|
||||
Version string `json:"version"` // k8s version
|
||||
}
|
||||
|
||||
// doRegion is the per-region rollup the map's cloud-region dots render: real coords
|
||||
// (from the catalog) + node / droplet / cluster counts placed at that datacenter.
|
||||
type doRegion struct {
|
||||
ID string `json:"id"` // catalog id (e.g. "sfo") for map alignment
|
||||
Slug string `json:"slug"` // DO region slug (e.g. "sfo3")
|
||||
Name string `json:"name"`
|
||||
City string `json:"city"`
|
||||
Country string `json:"country"`
|
||||
Lat float64 `json:"lat"`
|
||||
Lon float64 `json:"lon"`
|
||||
Nodes int `json:"nodes"` // DOKS worker nodes in this region
|
||||
Droplets int `json:"droplets"` // STANDALONE droplets (never k8s workers)
|
||||
Clusters int `json:"clusters"`
|
||||
Online int `json:"online"` // running nodes + active standalone droplets
|
||||
Status string `json:"status"`
|
||||
}
|
||||
|
||||
type doFleetTotals struct {
|
||||
Clusters int `json:"clusters"`
|
||||
Nodes int `json:"nodes"`
|
||||
NodesOnline int `json:"nodesOnline"`
|
||||
Droplets int `json:"droplets"` // standalone droplets only (de-duped from nodes)
|
||||
Regions int `json:"regions"`
|
||||
}
|
||||
|
||||
// doFleet is the whole real DO fleet: totals + per-region rollups (for map dots and
|
||||
// the overview) + per-cluster detail. source is "digitalocean" when live, or
|
||||
// "unconfigured"/"unavailable" for the two honest empty states.
|
||||
type doFleet struct {
|
||||
UpdatedAt string `json:"updatedAt"`
|
||||
Source string `json:"source"`
|
||||
Live bool `json:"live"`
|
||||
Totals doFleetTotals `json:"totals"`
|
||||
Regions []doRegion `json:"regions"`
|
||||
Clusters []doCluster `json:"clusters"`
|
||||
}
|
||||
|
||||
// ── config ───────────────────────────────────────────────────────────────────
|
||||
|
||||
const (
|
||||
doDefaultAPIBase = "https://api.digitalocean.com"
|
||||
doFleetTTL = 60 * time.Second
|
||||
doFleetTimeout = 10 * time.Second
|
||||
doMaxPages = 20 // droplet pagination safety cap
|
||||
)
|
||||
|
||||
// doAPIBase is the DO REST base (override with DIGITALOCEAN_API_BASE, e.g. in tests).
|
||||
func doAPIBase() string {
|
||||
if v := strings.TrimRight(strings.TrimSpace(env("DIGITALOCEAN_API_BASE")), "/"); v != "" {
|
||||
return v
|
||||
}
|
||||
return doDefaultAPIBase
|
||||
}
|
||||
|
||||
// doToken is the DO read token, injected at boot from KMS (org=hanzo, /world-secrets).
|
||||
// Empty ⇒ the fleet enumeration is honestly "unconfigured" (no map dots, no fake).
|
||||
func doToken() string { return strings.TrimSpace(env("DIGITALOCEAN_ACCESS_TOKEN")) }
|
||||
|
||||
// doRegionSupplement covers DO regions the 8-city regionCatalog() does not carry, so
|
||||
// EVERY DO region still geo-locates (never dropped just because it's outside the
|
||||
// catalog). resolveRegion (the shared placement) is tried first for catalog
|
||||
// alignment; this is the fallback only.
|
||||
var doRegionSupplement = map[string]cloudRegion{
|
||||
"tor1": {ID: "tor", Name: "Toronto", City: "Toronto", Country: "Canada", Lat: 43.6532, Lon: -79.3832, Status: "online"},
|
||||
"atl1": {ID: "atl", Name: "Atlanta", City: "Atlanta", Country: "USA", Lat: 33.7490, Lon: -84.3880, Status: "online"},
|
||||
}
|
||||
|
||||
// doRegionGeo resolves a DO region slug to placement coords. It prefers the shared
|
||||
// catalog (so DO dots sit exactly where the rest of the map places that region),
|
||||
// then the supplement, and finally reports ok=false (counted in totals, but no dot)
|
||||
// rather than inventing coordinates.
|
||||
func doRegionGeo(slug string) (cloudRegion, bool) {
|
||||
if rg, ok := resolveRegion(slug); ok {
|
||||
return rg, true
|
||||
}
|
||||
if rg, ok := doRegionSupplement[strings.ToLower(strings.TrimSpace(slug))]; ok {
|
||||
return rg, true
|
||||
}
|
||||
return cloudRegion{}, false
|
||||
}
|
||||
|
||||
// ── enumeration ──────────────────────────────────────────────────────────────
|
||||
|
||||
// dropletIsK8s reports whether a droplet is a DOKS worker (tagged `k8s` or
|
||||
// `k8s:<clusterID>`) — those are counted as cluster NODES, never as standalone
|
||||
// droplets, so the same machine is never counted twice.
|
||||
func dropletIsK8s(d doAPIDroplet) bool {
|
||||
for _, t := range d.Tags {
|
||||
if t == "k8s" || strings.HasPrefix(t, "k8s:") {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// regionAcc accumulates one region's rollup during enumeration. Node liveness and
|
||||
// droplet liveness are kept separate so a provisioning node is never masked by an
|
||||
// active droplet (and vice-versa) when deriving the region's health.
|
||||
type regionAcc struct {
|
||||
geo cloudRegion
|
||||
slug string
|
||||
placed bool
|
||||
nodes int // DOKS worker nodes
|
||||
nodeOn int // running nodes
|
||||
droplets int // standalone droplets (never k8s workers)
|
||||
dropOn int // active standalone droplets
|
||||
clusters int
|
||||
}
|
||||
|
||||
// fetchDOFleet enumerates the real DO fleet from base with token. It is pure (no
|
||||
// cache, no package state) so tests drive it against an httptest server. Returns an
|
||||
// error only on a hard API failure; the caller (getDOFleet) turns that into the
|
||||
// fail-soft last-known/empty response.
|
||||
func (s *Server) fetchDOFleet(ctx context.Context, base, token string) (doFleet, error) {
|
||||
hdr := map[string]string{"Authorization": "Bearer " + token, "Accept": "application/json"}
|
||||
|
||||
// 1) DOKS clusters + their nodes.
|
||||
var cr doAPIClustersResp
|
||||
if err := s.getJSON(ctx, base+"/v2/kubernetes/clusters?per_page=100", hdr, &cr); err != nil {
|
||||
return doFleet{}, err
|
||||
}
|
||||
|
||||
acc := map[string]*regionAcc{}
|
||||
var order []string
|
||||
region := func(slug string) *regionAcc {
|
||||
slug = strings.TrimSpace(slug)
|
||||
key := strings.ToLower(slug)
|
||||
if a := acc[key]; a != nil {
|
||||
return a
|
||||
}
|
||||
a := ®ionAcc{slug: slug}
|
||||
if geo, ok := doRegionGeo(slug); ok {
|
||||
a.geo, a.placed = geo, true
|
||||
} else {
|
||||
a.geo = cloudRegion{ID: key, Name: slug, City: slug, Status: "online"}
|
||||
}
|
||||
acc[key] = a
|
||||
order = append(order, key)
|
||||
return a
|
||||
}
|
||||
|
||||
clusters := make([]doCluster, 0, len(cr.Clusters))
|
||||
for _, c := range cr.Clusters {
|
||||
nodes, online := 0, 0
|
||||
for _, p := range c.NodePools {
|
||||
// nodes[] is the authoritative per-node list; fall back to the pool count
|
||||
// when the provisioner has not yet populated nodes[] (freshly scaling pool).
|
||||
if len(p.Nodes) > 0 {
|
||||
for _, n := range p.Nodes {
|
||||
nodes++
|
||||
if n.Status.State == "running" {
|
||||
online++
|
||||
}
|
||||
}
|
||||
} else {
|
||||
nodes += p.Count
|
||||
online += p.Count
|
||||
}
|
||||
}
|
||||
ra := region(c.Region)
|
||||
ra.nodes += nodes
|
||||
ra.nodeOn += online
|
||||
ra.clusters++
|
||||
clusters = append(clusters, doCluster{
|
||||
Name: c.Name, Region: c.Region, City: ra.geo.City, Lat: ra.geo.Lat, Lon: ra.geo.Lon,
|
||||
Nodes: nodes, Online: online, Status: firstNonEmpty(c.Status.State, "unknown"), Version: c.Version,
|
||||
})
|
||||
}
|
||||
|
||||
// 2) Droplets — paginated; exclude DOKS workers (counted above as nodes).
|
||||
next := base + "/v2/droplets?per_page=200"
|
||||
allowedHost := hostOf(base)
|
||||
for page := 0; next != "" && page < doMaxPages; page++ {
|
||||
var dr doAPIDropletsResp
|
||||
if err := s.getJSON(ctx, next, hdr, &dr); err != nil {
|
||||
return doFleet{}, err
|
||||
}
|
||||
for _, d := range dr.Droplets {
|
||||
if dropletIsK8s(d) {
|
||||
continue // a cluster node, already counted
|
||||
}
|
||||
ra := region(d.Region.Slug)
|
||||
ra.droplets++
|
||||
if d.Status == "active" {
|
||||
ra.dropOn++
|
||||
}
|
||||
}
|
||||
next = dr.Links.Pages.Next
|
||||
// SSRF guard: only follow a next-link on the SAME host we were pointed at.
|
||||
if next != "" && hostOf(next) != allowedHost {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
// 3) Materialize regions + totals (stable order = discovery order). A region's
|
||||
// status is derived from its machine online ratio: all up ⇒ online, some up ⇒
|
||||
// degraded, none up ⇒ offline.
|
||||
f := doFleet{UpdatedAt: nowRFC(), Source: "digitalocean", Live: true, Clusters: clusters}
|
||||
for _, key := range order {
|
||||
a := acc[key]
|
||||
total := a.nodes + a.droplets
|
||||
online := a.nodeOn + a.dropOn
|
||||
status := "online"
|
||||
switch {
|
||||
case total > 0 && online == 0:
|
||||
status = "offline"
|
||||
case online < total:
|
||||
status = "degraded"
|
||||
}
|
||||
if a.placed {
|
||||
f.Regions = append(f.Regions, doRegion{
|
||||
ID: a.geo.ID, Slug: a.slug, Name: a.geo.Name, City: a.geo.City, Country: a.geo.Country,
|
||||
Lat: a.geo.Lat, Lon: a.geo.Lon, Nodes: a.nodes, Droplets: a.droplets, Clusters: a.clusters,
|
||||
Online: online, Status: status,
|
||||
})
|
||||
f.Totals.Regions++
|
||||
}
|
||||
f.Totals.Nodes += a.nodes
|
||||
f.Totals.NodesOnline += online
|
||||
f.Totals.Droplets += a.droplets
|
||||
f.Totals.Clusters += a.clusters
|
||||
}
|
||||
return f, nil
|
||||
}
|
||||
|
||||
// ── cache (60s TTL + last-known, fail-soft) ──────────────────────────────────
|
||||
|
||||
var (
|
||||
doFleetMu sync.Mutex
|
||||
doFleetVal *doFleet
|
||||
doFleetAt time.Time
|
||||
)
|
||||
|
||||
// getDOFleet returns the cached real fleet, refreshing past doFleetTTL. It never
|
||||
// errors: a DO failure returns the last good fleet (or an honest empty on a cold
|
||||
// miss), and a missing token returns the "unconfigured" empty. This is the ONE
|
||||
// place both the /nodes endpoint and the cloud-pulse overview read, so a single DO
|
||||
// round-trip feeds both surfaces (DRY).
|
||||
func (s *Server) getDOFleet(ctx context.Context) doFleet {
|
||||
doFleetMu.Lock()
|
||||
fresh := doFleetVal != nil && time.Since(doFleetAt) < doFleetTTL
|
||||
last := doFleetVal
|
||||
doFleetMu.Unlock()
|
||||
if fresh {
|
||||
return *last
|
||||
}
|
||||
|
||||
token := doToken()
|
||||
if token == "" {
|
||||
if last != nil {
|
||||
return *last
|
||||
}
|
||||
return doFleet{UpdatedAt: nowRFC(), Source: "unconfigured", Regions: []doRegion{}, Clusters: []doCluster{}}
|
||||
}
|
||||
|
||||
cctx, cancel := context.WithTimeout(ctx, doFleetTimeout)
|
||||
defer cancel()
|
||||
f, err := s.fetchDOFleet(cctx, doAPIBase(), token)
|
||||
if err != nil {
|
||||
if last != nil {
|
||||
return *last // fail-soft: serve the previous good fleet
|
||||
}
|
||||
return doFleet{UpdatedAt: nowRFC(), Source: "unavailable", Regions: []doRegion{}, Clusters: []doCluster{}}
|
||||
}
|
||||
doFleetMu.Lock()
|
||||
doFleetVal = &f
|
||||
doFleetAt = time.Now()
|
||||
doFleetMu.Unlock()
|
||||
return f
|
||||
}
|
||||
|
||||
// applyDOFleet folds the real DO fleet into the cloud-pulse overview + region
|
||||
// breakdown (the source of the map's cloud-region dots and the overview's node
|
||||
// counts). Called on the fallback rung of producePulse when the multi-cloud visor
|
||||
// plane did not resolve — so the public dashboard shows the real infra either way.
|
||||
// Returns false (leaving the pulse untouched) when no real fleet resolved. GPUs stay
|
||||
// 0: DOKS carries no GPU pools, and inventing a GPU count would be a lie.
|
||||
func (s *Server) applyDOFleet(ctx context.Context, p *cloudPulse) bool {
|
||||
f := s.getDOFleet(ctx)
|
||||
if !f.Live || (f.Totals.Nodes == 0 && f.Totals.Droplets == 0) {
|
||||
return false
|
||||
}
|
||||
regions := make([]cloudRegion, 0, len(f.Regions))
|
||||
for _, r := range f.Regions {
|
||||
regions = append(regions, cloudRegion{
|
||||
ID: r.ID, Name: r.Name, City: r.City, Country: r.Country, Lat: r.Lat, Lon: r.Lon,
|
||||
Nodes: r.Nodes + r.Droplets, Gpus: 0, Status: r.Status,
|
||||
})
|
||||
}
|
||||
p.Regions = regions
|
||||
p.Overview.NodesTotal = f.Totals.Nodes + f.Totals.Droplets
|
||||
p.Overview.NodesOnline = f.Totals.NodesOnline
|
||||
p.Overview.Regions = f.Totals.Regions
|
||||
return true
|
||||
}
|
||||
|
||||
// handleCloudNodes serves the real DO fleet (DOKS clusters + nodes + standalone
|
||||
// droplets) as the map's cloud-region / node feed. Public, cached, fail-soft: it
|
||||
// never 5xxes — an unreachable DO or unset token degrades to an honest empty body.
|
||||
func (s *Server) handleCloudNodes(w http.ResponseWriter, r *http.Request) {
|
||||
if preflight(w, r, "GET, OPTIONS") || methodNotGet(w, r) {
|
||||
return
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(r.Context(), doFleetTimeout+2*time.Second)
|
||||
defer cancel()
|
||||
f := s.getDOFleet(ctx) // already cached + fail-soft (never errors)
|
||||
if f.Regions == nil {
|
||||
f.Regions = []doRegion{}
|
||||
}
|
||||
if f.Clusters == nil {
|
||||
f.Clusters = []doCluster{}
|
||||
}
|
||||
writeJSON(w, http.StatusOK, "public, max-age=30, s-maxage=30, stale-while-revalidate=120", f)
|
||||
}
|
||||
|
||||
// hostOf returns the lowercased host[:port] of rawURL, or "" if unparseable.
|
||||
func hostOf(rawURL string) string {
|
||||
u, err := url.Parse(rawURL)
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
return strings.ToLower(u.Host)
|
||||
}
|
||||
@@ -0,0 +1,278 @@
|
||||
package world
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// doAPIStub serves the two DO REST endpoints fetchDOFleet reads, from canned
|
||||
// fixtures. Droplets paginate: page 2 is served when ?page=2 is present, and the
|
||||
// page-1 body advertises the next link pointing back at this same server (mirrors
|
||||
// DO's absolute next-URL, exercising the SSRF same-host guard on the happy path).
|
||||
func doAPIStub(t *testing.T, clusters string, dropletsP1, dropletsP2 func(self string) string) *httptest.Server {
|
||||
t.Helper()
|
||||
var srv *httptest.Server
|
||||
srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if got := r.Header.Get("Authorization"); got != "Bearer test-token" {
|
||||
t.Errorf("missing/wrong auth header: %q", got)
|
||||
}
|
||||
switch {
|
||||
case strings.HasPrefix(r.URL.Path, "/v2/kubernetes/clusters"):
|
||||
_, _ = w.Write([]byte(clusters))
|
||||
case strings.HasPrefix(r.URL.Path, "/v2/droplets"):
|
||||
if r.URL.Query().Get("page") == "2" {
|
||||
_, _ = w.Write([]byte(dropletsP2(srv.URL)))
|
||||
} else {
|
||||
_, _ = w.Write([]byte(dropletsP1(srv.URL)))
|
||||
}
|
||||
default:
|
||||
http.NotFound(w, r)
|
||||
}
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
return srv
|
||||
}
|
||||
|
||||
// Fixtures: 3 clusters (2× sfo3, 1× tor1) totalling 6 nodes, 5 running (one node in
|
||||
// the second sfo3 cluster is still provisioning). Droplets: 2 k8s workers (MUST be
|
||||
// excluded — they are already nodes) + 2 standalone (1 sfo3 active, 1 tor1 active),
|
||||
// split across two pages to exercise pagination.
|
||||
const fxClusters = `{"kubernetes_clusters":[
|
||||
{"name":"hanzo-k8s","region":"sfo3","version":"1.34.1-do.4","status":{"state":"running"},
|
||||
"node_pools":[{"count":2,"nodes":[{"status":{"state":"running"}},{"status":{"state":"running"}}]},
|
||||
{"count":1,"nodes":[{"status":{"state":"running"}}]}]},
|
||||
{"name":"lux-k8s","region":"sfo3","version":"1.34.1-do.4","status":{"state":"running"},
|
||||
"node_pools":[{"count":2,"nodes":[{"status":{"state":"running"}},{"status":{"state":"provisioning"}}]}]},
|
||||
{"name":"edge-k8s","region":"tor1","version":"1.35.5-do.2","status":{"state":"running"},
|
||||
"node_pools":[{"count":1,"nodes":[{"status":{"state":"running"}}]}]}
|
||||
]}`
|
||||
|
||||
func fxDropletsP1(self string) string {
|
||||
return `{"droplets":[
|
||||
{"name":"worker-a","status":"active","region":{"slug":"sfo3"},"size_slug":"s-4vcpu-8gb","tags":["k8s","k8s:worker","k8s:uuid"]},
|
||||
{"name":"worker-b","status":"active","region":{"slug":"sfo3"},"size_slug":"s-4vcpu-8gb","tags":["k8s:uuid"]},
|
||||
{"name":"bastion","status":"active","region":{"slug":"sfo3"},"size_slug":"s-1vcpu-1gb","tags":["infra"]}
|
||||
],"links":{"pages":{"next":"` + self + `/v2/droplets?per_page=200&page=2"}}}`
|
||||
}
|
||||
func fxDropletsP2(self string) string {
|
||||
return `{"droplets":[
|
||||
{"name":"relay","status":"active","region":{"slug":"tor1"},"size_slug":"s-1vcpu-1gb","tags":[]}
|
||||
],"links":{"pages":{}}}`
|
||||
}
|
||||
|
||||
func TestFetchDOFleet_AggregatesDedupesPlaces(t *testing.T) {
|
||||
s := NewServer()
|
||||
stub := doAPIStub(t, fxClusters, fxDropletsP1, fxDropletsP2)
|
||||
|
||||
f, err := s.fetchDOFleet(context.Background(), stub.URL, "test-token")
|
||||
if err != nil {
|
||||
t.Fatalf("fetchDOFleet: %v", err)
|
||||
}
|
||||
if f.Source != "digitalocean" || !f.Live {
|
||||
t.Fatalf("source/live = %q/%v, want digitalocean/true", f.Source, f.Live)
|
||||
}
|
||||
|
||||
// Totals: 6 nodes (3+2+1), 5 running nodes + 2 active standalone droplets = 7
|
||||
// machines online; 2 standalone droplets (the 2 k8s-tagged are NOT droplets here);
|
||||
// 3 clusters; 2 regions.
|
||||
want := doFleetTotals{Clusters: 3, Nodes: 6, NodesOnline: 7, Droplets: 2, Regions: 2}
|
||||
if f.Totals != want {
|
||||
t.Fatalf("totals = %+v, want %+v", f.Totals, want)
|
||||
}
|
||||
|
||||
byID := map[string]doRegion{}
|
||||
for _, r := range f.Regions {
|
||||
byID[r.Slug] = r
|
||||
}
|
||||
|
||||
// sfo3 → catalog "sfo" coords (shared resolveRegion, so dots align with the map).
|
||||
sfo, ok := byID["sfo3"]
|
||||
if !ok {
|
||||
t.Fatal("missing sfo3 region")
|
||||
}
|
||||
if sfo.ID != "sfo" || sfo.City != "San Francisco" {
|
||||
t.Fatalf("sfo3 placed as id=%q city=%q, want sfo/San Francisco", sfo.ID, sfo.City)
|
||||
}
|
||||
if sfo.Lat < 37 || sfo.Lat > 38 || sfo.Lon > -122 || sfo.Lon < -123 {
|
||||
t.Fatalf("sfo3 coords (%.4f,%.4f) not near San Francisco", sfo.Lat, sfo.Lon)
|
||||
}
|
||||
// sfo3: 5 nodes (hanzo 3 + lux 2), 4 running (lux has one provisioning); 1
|
||||
// standalone active droplet → 4+1 = 5 machines online of 6, so degraded.
|
||||
if sfo.Nodes != 5 || sfo.Droplets != 1 || sfo.Clusters != 2 || sfo.Online != 5 {
|
||||
t.Fatalf("sfo3 rollup = nodes:%d droplets:%d clusters:%d online:%d, want 5/1/2/5", sfo.Nodes, sfo.Droplets, sfo.Clusters, sfo.Online)
|
||||
}
|
||||
if sfo.Status != "degraded" {
|
||||
t.Fatalf("sfo3 status = %q, want degraded (one node provisioning)", sfo.Status)
|
||||
}
|
||||
|
||||
// tor1 → placed via the supplement (NOT in the 8-city catalog): all up ⇒ online.
|
||||
tor, ok := byID["tor1"]
|
||||
if !ok {
|
||||
t.Fatal("missing tor1 region — supplement placement failed")
|
||||
}
|
||||
if tor.City != "Toronto" || tor.Lat < 43 || tor.Lat > 44 {
|
||||
t.Fatalf("tor1 placed as city=%q lat=%.4f, want Toronto ~43.65", tor.City, tor.Lat)
|
||||
}
|
||||
if tor.Nodes != 1 || tor.Droplets != 1 || tor.Online != 2 || tor.Status != "online" {
|
||||
t.Fatalf("tor1 rollup = nodes:%d droplets:%d online:%d status:%q, want 1/1/2/online", tor.Nodes, tor.Droplets, tor.Online, tor.Status)
|
||||
}
|
||||
|
||||
// Per-cluster detail carries real coords + version + online counts.
|
||||
var lux *doCluster
|
||||
for i := range f.Clusters {
|
||||
if f.Clusters[i].Name == "lux-k8s" {
|
||||
lux = &f.Clusters[i]
|
||||
}
|
||||
}
|
||||
if lux == nil {
|
||||
t.Fatal("missing lux-k8s cluster")
|
||||
}
|
||||
if lux.Nodes != 2 || lux.Online != 1 || lux.Version != "1.34.1-do.4" || lux.City != "San Francisco" {
|
||||
t.Fatalf("lux-k8s = nodes:%d online:%d ver:%q city:%q, want 2/1/1.34.1-do.4/San Francisco", lux.Nodes, lux.Online, lux.Version, lux.City)
|
||||
}
|
||||
}
|
||||
|
||||
// The exact double-count trap: a DOKS worker droplet must never be added to the
|
||||
// standalone droplet count. With ONLY k8s-tagged droplets, standalone droplets = 0.
|
||||
func TestFetchDOFleet_ExcludesK8sWorkerDroplets(t *testing.T) {
|
||||
s := NewServer()
|
||||
onlyK8s := func(self string) string {
|
||||
return `{"droplets":[
|
||||
{"name":"w1","status":"active","region":{"slug":"sfo3"},"tags":["k8s","k8s:worker"]},
|
||||
{"name":"w2","status":"active","region":{"slug":"sfo3"},"tags":["k8s:abc"]}
|
||||
],"links":{"pages":{}}}`
|
||||
}
|
||||
stub := doAPIStub(t, fxClusters, onlyK8s, onlyK8s)
|
||||
f, err := s.fetchDOFleet(context.Background(), stub.URL, "test-token")
|
||||
if err != nil {
|
||||
t.Fatalf("fetchDOFleet: %v", err)
|
||||
}
|
||||
if f.Totals.Droplets != 0 {
|
||||
t.Fatalf("standalone droplets = %d, want 0 (all droplets are k8s workers)", f.Totals.Droplets)
|
||||
}
|
||||
if f.Totals.Nodes != 6 {
|
||||
t.Fatalf("nodes = %d, want 6 (unchanged — droplets are not double counted)", f.Totals.Nodes)
|
||||
}
|
||||
}
|
||||
|
||||
// applyDOFleet folds the real fleet into the cloud-pulse overview + region breakdown
|
||||
// (what the overview panel counts and the map's cloud-region dots render).
|
||||
func TestApplyDOFleet_FoldsIntoPulse(t *testing.T) {
|
||||
s := NewServer()
|
||||
stub := doAPIStub(t, fxClusters, fxDropletsP1, fxDropletsP2)
|
||||
t.Setenv("DIGITALOCEAN_API_BASE", stub.URL)
|
||||
t.Setenv("DIGITALOCEAN_ACCESS_TOKEN", "test-token")
|
||||
resetDOFleetCache()
|
||||
|
||||
p := emptyPulse()
|
||||
if !s.applyDOFleet(context.Background(), &p) {
|
||||
t.Fatal("applyDOFleet returned false with a live fleet")
|
||||
}
|
||||
// NodesTotal = nodes + standalone droplets = 6 + 2 = 8; online machines = 7.
|
||||
if p.Overview.NodesTotal != 8 || p.Overview.NodesOnline != 7 {
|
||||
t.Fatalf("overview nodes = %d/%d, want 7/8", p.Overview.NodesOnline, p.Overview.NodesTotal)
|
||||
}
|
||||
if p.Overview.Regions != 2 || len(p.Regions) != 2 {
|
||||
t.Fatalf("regions = overview:%d list:%d, want 2/2", p.Overview.Regions, len(p.Regions))
|
||||
}
|
||||
if p.Overview.GpusOnline != 0 {
|
||||
t.Fatalf("gpusOnline = %d, want 0 (DOKS has no GPU pools — never invented)", p.Overview.GpusOnline)
|
||||
}
|
||||
// Region dots carry real coords for the map.
|
||||
for _, r := range p.Regions {
|
||||
if r.Lat == 0 && r.Lon == 0 {
|
||||
t.Fatalf("region %q has no coords — would not place on the map", r.ID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// getDOFleet is fail-soft: an unset token is the honest "unconfigured" empty, and a
|
||||
// DO failure with no last-known is the honest "unavailable" empty — never a panic,
|
||||
// never a fabricated number, never an error to the caller.
|
||||
func TestGetDOFleet_FailSoft(t *testing.T) {
|
||||
s := NewServer()
|
||||
|
||||
// Unset token → unconfigured, empty (not nil) collections.
|
||||
t.Setenv("DIGITALOCEAN_ACCESS_TOKEN", "")
|
||||
resetDOFleetCache()
|
||||
f := s.getDOFleet(context.Background())
|
||||
if f.Source != "unconfigured" || f.Live {
|
||||
t.Fatalf("no-token fleet = %q/live=%v, want unconfigured/false", f.Source, f.Live)
|
||||
}
|
||||
if f.Regions == nil || f.Clusters == nil {
|
||||
t.Fatal("empty fleet must carry empty (non-nil) slices")
|
||||
}
|
||||
|
||||
// Token set but DO unreachable (bad base) and no last-known → unavailable.
|
||||
t.Setenv("DIGITALOCEAN_ACCESS_TOKEN", "test-token")
|
||||
t.Setenv("DIGITALOCEAN_API_BASE", "http://127.0.0.1:1") // connection refused
|
||||
resetDOFleetCache()
|
||||
f = s.getDOFleet(context.Background())
|
||||
if f.Source != "unavailable" || f.Live {
|
||||
t.Fatalf("unreachable fleet = %q/live=%v, want unavailable/false", f.Source, f.Live)
|
||||
}
|
||||
}
|
||||
|
||||
// getDOFleet serves the last-known good fleet when a later refresh fails (a DO API
|
||||
// hiccup must never blank the map).
|
||||
func TestGetDOFleet_LastKnownOnError(t *testing.T) {
|
||||
s := NewServer()
|
||||
stub := doAPIStub(t, fxClusters, fxDropletsP1, fxDropletsP2)
|
||||
t.Setenv("DIGITALOCEAN_ACCESS_TOKEN", "test-token")
|
||||
t.Setenv("DIGITALOCEAN_API_BASE", stub.URL)
|
||||
resetDOFleetCache()
|
||||
|
||||
good := s.getDOFleet(context.Background())
|
||||
if !good.Live || good.Totals.Nodes != 6 {
|
||||
t.Fatalf("first fetch not live: %+v", good.Totals)
|
||||
}
|
||||
// Now point at a dead host and force past the TTL: the cache must return the
|
||||
// previous good fleet, not an empty.
|
||||
t.Setenv("DIGITALOCEAN_API_BASE", "http://127.0.0.1:1")
|
||||
doFleetMu.Lock()
|
||||
doFleetAt = doFleetAt.Add(-2 * doFleetTTL) // expire
|
||||
doFleetMu.Unlock()
|
||||
again := s.getDOFleet(context.Background())
|
||||
if again.Source != "digitalocean" || again.Totals.Nodes != 6 {
|
||||
t.Fatalf("last-known lost on error: source=%q nodes=%d", again.Source, again.Totals.Nodes)
|
||||
}
|
||||
}
|
||||
|
||||
// The public endpoint never 5xxes and always emits non-nil arrays.
|
||||
func TestHandleCloudNodes_NeverErrors(t *testing.T) {
|
||||
s := NewServer()
|
||||
t.Setenv("DIGITALOCEAN_ACCESS_TOKEN", "") // unconfigured path
|
||||
resetDOFleetCache()
|
||||
mux := http.NewServeMux()
|
||||
s.Mount(mux)
|
||||
ts := httptest.NewServer(mux)
|
||||
t.Cleanup(ts.Close)
|
||||
|
||||
resp, err := http.Get(ts.URL + "/v1/world/cloud/nodes")
|
||||
if err != nil {
|
||||
t.Fatalf("request: %v", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200", resp.StatusCode)
|
||||
}
|
||||
var f doFleet
|
||||
if err := json.NewDecoder(resp.Body).Decode(&f); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
if f.Regions == nil || f.Clusters == nil {
|
||||
t.Fatal("endpoint must emit [] not null for regions/clusters")
|
||||
}
|
||||
}
|
||||
|
||||
// resetDOFleetCache clears the package-level fleet cache between tests.
|
||||
func resetDOFleetCache() {
|
||||
doFleetMu.Lock()
|
||||
doFleetVal = nil
|
||||
doFleetAt = doFleetAt.Add(0)
|
||||
doFleetMu.Unlock()
|
||||
}
|
||||
+14
-13
@@ -59,19 +59,20 @@ var kmsBootTimeout = 5 * time.Second
|
||||
// are resolved by env() at read time and are NOT fetched — only the canonical
|
||||
// name is stored in KMS. A key absent from KMS simply 404s and is skipped.
|
||||
var worldSecretKeys = []string{
|
||||
"HANZO_CLOUD_PULSE_TOKEN", // cloud-map pulse backend service token
|
||||
"HANZO_AI_KEY", // AI endpoints (gateway key)
|
||||
"HANZO_AI_BASE", // AI gateway base URL override
|
||||
"HANZO_AI_MODEL", // AI default model override
|
||||
"YOUTUBE_API_KEY", // live-video reliability
|
||||
"FRED_API_KEY", // macro / econ series
|
||||
"FINNHUB_API_KEY", // market quotes
|
||||
"EIA_API_KEY", // energy data
|
||||
"NASA_FIRMS_API_KEY", // wildfire hotspots
|
||||
"ACLED_ACCESS_TOKEN", // conflict / risk events
|
||||
"CLOUDFLARE_API_TOKEN", // infra telemetry
|
||||
"WINGBITS_API_KEY", // ADS-B feed
|
||||
"WS_RELAY_URL", // live websocket relay
|
||||
"HANZO_CLOUD_PULSE_TOKEN", // cloud-map pulse backend service token
|
||||
"DIGITALOCEAN_ACCESS_TOKEN", // real DO fleet enumeration (DOKS clusters + nodes + droplets)
|
||||
"HANZO_AI_KEY", // AI endpoints (gateway key)
|
||||
"HANZO_AI_BASE", // AI gateway base URL override
|
||||
"HANZO_AI_MODEL", // AI default model override
|
||||
"YOUTUBE_API_KEY", // live-video reliability
|
||||
"FRED_API_KEY", // macro / econ series
|
||||
"FINNHUB_API_KEY", // market quotes
|
||||
"EIA_API_KEY", // energy data
|
||||
"NASA_FIRMS_API_KEY", // wildfire hotspots
|
||||
"ACLED_ACCESS_TOKEN", // conflict / risk events
|
||||
"CLOUDFLARE_API_TOKEN", // infra telemetry
|
||||
"WINGBITS_API_KEY", // ADS-B feed
|
||||
"WS_RELAY_URL", // live websocket relay
|
||||
}
|
||||
|
||||
// kmsConfig is the resolved fetch scope. host has no trailing slash; path has
|
||||
|
||||
@@ -134,6 +134,8 @@ func (s *Server) mount(mux registrar) {
|
||||
// PUBLIC map layers (real telemetry when reachable; modeled/demo carries a flag):
|
||||
mux.HandleFunc("/v1/world/cloud/chain-nodes", s.handleCloudChainNodes)
|
||||
mux.HandleFunc("/v1/world/cloud/byo-gpu", s.handleCloudBYOGPU)
|
||||
// Real DigitalOcean fleet: DOKS clusters + nodes + standalone droplets, geo-located.
|
||||
mux.HandleFunc("/v1/world/cloud/nodes", s.handleCloudNodes)
|
||||
mux.HandleFunc("/v1/world/cloud/traffic", s.handleCloudTraffic)
|
||||
// Native LB request-geo aggregate (points + throughput) for the Hanzo-mode globe.
|
||||
// Proxies the ai backend's public /v1/traffic/globe; honest empty state, no demo.
|
||||
|
||||
Reference in New Issue
Block a user