generated from mathias/template-go-web
feat(atlas): Phase B — substrate from the live cluster
The substrate machines are now rendered from the live k3s nodes (arch, cpu, memory, GPU, k3s version), read in-cluster via the mounted ServiceAccount with a stdlib HTTP client (no client-go) and a 30s cache; falls back to the authored substrate whenever the cluster is unreachable. Placement per the homelab decision "in-cluster only for in-workload ops" — a running web app rendering live data fits. HostsFromNodes parser built test-first. RBAC (read-only node-reader SA) shipped in infra k3s/apps/cad-atlas/. Verified: go build/vet/lint(0)/test green. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,79 @@
|
||||
package atlas
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"math"
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// HostsFromNodes parses a Kubernetes `/api/v1/nodes` list response into
|
||||
// substrate Host entries, so the atlas machines reflect the live cluster.
|
||||
func HostsFromNodes(nodesJSON []byte) ([]Host, error) {
|
||||
var list struct {
|
||||
Items []struct {
|
||||
Metadata struct {
|
||||
Name string `json:"name"`
|
||||
} `json:"metadata"`
|
||||
Status struct {
|
||||
Capacity map[string]string `json:"capacity"`
|
||||
NodeInfo struct {
|
||||
Architecture string `json:"architecture"`
|
||||
KubeletVersion string `json:"kubeletVersion"`
|
||||
} `json:"nodeInfo"`
|
||||
} `json:"status"`
|
||||
} `json:"items"`
|
||||
}
|
||||
if err := json.Unmarshal(nodesJSON, &list); err != nil {
|
||||
return nil, fmt.Errorf("parse nodes: %w", err)
|
||||
}
|
||||
|
||||
hosts := make([]Host, 0, len(list.Items))
|
||||
for _, it := range list.Items {
|
||||
var parts []string
|
||||
if a := it.Status.NodeInfo.Architecture; a != "" {
|
||||
parts = append(parts, a)
|
||||
}
|
||||
if c := it.Status.Capacity["cpu"]; c != "" {
|
||||
parts = append(parts, c+" cpu")
|
||||
}
|
||||
if gi := memGi(it.Status.Capacity["memory"]); gi != "" {
|
||||
parts = append(parts, gi+"Gi")
|
||||
}
|
||||
if g := it.Status.Capacity["nvidia.com/gpu"]; g != "" && g != "0" {
|
||||
parts = append(parts, g+"× GPU")
|
||||
}
|
||||
if v := k3sVersion(it.Status.NodeInfo.KubeletVersion); v != "" {
|
||||
parts = append(parts, v)
|
||||
}
|
||||
hosts = append(hosts, Host{Name: it.Metadata.Name, Spec: strings.Join(parts, " · ")})
|
||||
}
|
||||
return hosts, nil
|
||||
}
|
||||
|
||||
// memGi converts a Kubernetes memory quantity in Ki (e.g. "67108864Ki") to a
|
||||
// rounded Gi string. Returns "" if unparseable.
|
||||
func memGi(ki string) string {
|
||||
n, err := strconv.ParseFloat(strings.TrimSuffix(ki, "Ki"), 64)
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
return strconv.Itoa(int(math.Round(n / 1048576)))
|
||||
}
|
||||
|
||||
// k3sVersion trims a kubeletVersion's build metadata, labelling k3s builds.
|
||||
// "v1.31.4+k3s1" → "k3s v1.31.4"; "v1.31.4" → "v1.31.4".
|
||||
func k3sVersion(kubelet string) string {
|
||||
if kubelet == "" {
|
||||
return ""
|
||||
}
|
||||
if i := strings.Index(kubelet, "+"); i >= 0 {
|
||||
ver, suffix := kubelet[:i], kubelet[i+1:]
|
||||
if strings.Contains(suffix, "k3s") {
|
||||
return "k3s " + ver
|
||||
}
|
||||
return ver
|
||||
}
|
||||
return kubelet
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
package atlas_test
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
"git.d-ma.be/mathias/cad-atlas/internal/atlas"
|
||||
)
|
||||
|
||||
func TestHostsFromNodes_DerivesSpecFromNodeStatus(t *testing.T) {
|
||||
nodes := []byte(`{"items":[
|
||||
{"metadata":{"name":"koala"},
|
||||
"status":{"capacity":{"cpu":"16","memory":"67108864Ki","nvidia.com/gpu":"1"},
|
||||
"nodeInfo":{"architecture":"amd64","kubeletVersion":"v1.31.4+k3s1"}}},
|
||||
{"metadata":{"name":"worker2"},
|
||||
"status":{"capacity":{"cpu":"8","memory":"33554432Ki"},
|
||||
"nodeInfo":{"architecture":"arm64","kubeletVersion":"v1.30.0+k3s1"}}}
|
||||
]}`)
|
||||
|
||||
hosts, err := atlas.HostsFromNodes(nodes)
|
||||
if err != nil {
|
||||
t.Fatalf("HostsFromNodes: %v", err)
|
||||
}
|
||||
|
||||
want := []atlas.Host{
|
||||
{Name: "koala", Spec: "amd64 · 16 cpu · 64Gi · 1× GPU · k3s v1.31.4"},
|
||||
{Name: "worker2", Spec: "arm64 · 8 cpu · 32Gi · k3s v1.30.0"},
|
||||
}
|
||||
if !reflect.DeepEqual(hosts, want) {
|
||||
t.Fatalf("hosts = %+v\nwant %+v", hosts, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHostsFromNodes_ErrorsOnBadJSON(t *testing.T) {
|
||||
if _, err := atlas.HostsFromNodes([]byte("{not json")); err == nil {
|
||||
t.Fatal("expected error on bad JSON, got nil")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,66 @@
|
||||
// Package cluster reads live k3s state from inside a pod via the Kubernetes
|
||||
// API, using the mounted ServiceAccount credentials. No client-go: the queries
|
||||
// are read-only and few, so stdlib net/http keeps the dependency surface small.
|
||||
package cluster
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
"crypto/x509"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"os"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
tokenPath = "/var/run/secrets/kubernetes.io/serviceaccount/token" //nolint:gosec // well-known in-cluster path, not a secret literal
|
||||
caPath = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
|
||||
)
|
||||
|
||||
// Nodes returns the raw /api/v1/nodes JSON from the in-cluster API server.
|
||||
func Nodes() ([]byte, error) { return get("/api/v1/nodes") }
|
||||
|
||||
func get(path string) ([]byte, error) {
|
||||
host, port := os.Getenv("KUBERNETES_SERVICE_HOST"), os.Getenv("KUBERNETES_SERVICE_PORT")
|
||||
if host == "" || port == "" {
|
||||
return nil, fmt.Errorf("not in-cluster: KUBERNETES_SERVICE_HOST unset")
|
||||
}
|
||||
token, err := os.ReadFile(tokenPath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("read sa token: %w", err)
|
||||
}
|
||||
ca, err := os.ReadFile(caPath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("read ca: %w", err)
|
||||
}
|
||||
pool := x509.NewCertPool()
|
||||
if !pool.AppendCertsFromPEM(ca) {
|
||||
return nil, fmt.Errorf("invalid cluster CA cert")
|
||||
}
|
||||
client := &http.Client{
|
||||
Timeout: 5 * time.Second,
|
||||
Transport: &http.Transport{
|
||||
TLSClientConfig: &tls.Config{RootCAs: pool, MinVersion: tls.VersionTLS12},
|
||||
},
|
||||
}
|
||||
req, err := http.NewRequest(http.MethodGet, "https://"+host+":"+port+path, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req.Header.Set("Authorization", "Bearer "+string(token))
|
||||
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, fmt.Errorf("k8s API GET %s: %s", path, resp.Status)
|
||||
}
|
||||
return body, nil
|
||||
}
|
||||
+39
-1
@@ -4,9 +4,12 @@ import (
|
||||
_ "embed"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
cadatlas "git.d-ma.be/mathias/cad-atlas"
|
||||
"git.d-ma.be/mathias/cad-atlas/internal/atlas"
|
||||
"git.d-ma.be/mathias/cad-atlas/internal/cluster"
|
||||
)
|
||||
|
||||
// atlasHTML is the Phase-A/B static shell. It fetches /api/atlas.json at load
|
||||
@@ -16,7 +19,8 @@ import (
|
||||
var atlasHTML []byte
|
||||
|
||||
// NewHandler serves the CAD Atlas: the shell at "/", and the sourced data at
|
||||
// "/api/atlas.json" (authored data + CI stage generated from the real cd.yml).
|
||||
// "/api/atlas.json" (authored data + CI stage generated from the real cd.yml +
|
||||
// substrate overlaid from the live cluster when reachable).
|
||||
func NewHandler() http.Handler {
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -33,8 +37,42 @@ func NewHandler() http.Handler {
|
||||
http.Error(w, "atlas build failed", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
if live := liveSubstrate(); len(live) > 0 {
|
||||
a.Substrate = live
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json; charset=utf-8")
|
||||
_ = json.NewEncoder(w).Encode(a)
|
||||
})
|
||||
return mux
|
||||
}
|
||||
|
||||
// substrate cache: query the cluster at most once per TTL; fall back to the
|
||||
// authored substrate (returns nil) whenever the cluster is unreachable.
|
||||
var (
|
||||
subMu sync.Mutex
|
||||
subCache []atlas.Host
|
||||
subAt time.Time
|
||||
)
|
||||
|
||||
const substrateTTL = 30 * time.Second
|
||||
|
||||
func liveSubstrate() []atlas.Host {
|
||||
subMu.Lock()
|
||||
defer subMu.Unlock()
|
||||
if !subAt.IsZero() && time.Since(subAt) < substrateTTL {
|
||||
return subCache
|
||||
}
|
||||
subAt = time.Now()
|
||||
raw, err := cluster.Nodes()
|
||||
if err != nil {
|
||||
subCache = nil
|
||||
return nil
|
||||
}
|
||||
hosts, err := atlas.HostsFromNodes(raw)
|
||||
if err != nil {
|
||||
subCache = nil
|
||||
return nil
|
||||
}
|
||||
subCache = hosts
|
||||
return hosts
|
||||
}
|
||||
|
||||
@@ -114,7 +114,7 @@
|
||||
<body>
|
||||
<header>
|
||||
<h1><b>CAD</b> Atlas · From Signal to Pod</h1>
|
||||
<span class="sub mono">one human gate · everything up- and downstream is agents · <em>v0.4 · data-driven (CI stage generated from cd.yml)</em></span>
|
||||
<span class="sub mono">one human gate · everything up- and downstream is agents · <em>v0.5 · data-driven (CI + substrate from the live cluster)</em></span>
|
||||
<div class="controls">
|
||||
<button id="replay"><span class="dot"></span> Replay</button>
|
||||
<button id="slowmo">Slow-mo · <span id="slowState">off</span></button>
|
||||
@@ -145,7 +145,7 @@
|
||||
<footer class="mono">
|
||||
CAD → CI → CD · intent→specify→dispatch · build→test→validate · deploy→ship.
|
||||
Dashed violet = feedback bus (stage 08 → TELOS: deploy outcome scored vs originating goal).
|
||||
Data served from <code>/api/atlas.json</code> (authored <code>atlas.json</code> + CI stage generated from the live <code>cd.yml</code>).
|
||||
Data served from <code>/api/atlas.json</code> (authored <code>atlas.json</code> + CI stage from the live <code>cd.yml</code> + substrate from the live cluster nodes).
|
||||
Phase C plugs in live reads of <code>assessor-loop</code> ledger · <code>session_log</code> · Gitea run API · Flux events.
|
||||
</footer>
|
||||
|
||||
|
||||
Reference in New Issue
Block a user