Files
rollcall/cmd/rollcall/rackd.go
T
Ilker ManapandClaude Opus 5.5 bbe45804c7
CI / agent (push) Successful in 2s
CI / site (push) Successful in 3s
CI / go (push) Successful in 38s
Rack controllers update themselves to the region's version
The region offers its binary (version, arch, sha256) in each rack's state and
serves it over mutual TLS; the rack stages it after checking checksum and
version; a root path unit installs it, restarts the rack and rolls back if it
does not register within a minute.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-28 13:59:52 +01:00

643 lines
20 KiB
Go

package main
// `rollcall rack`: a rack controller next to the machines. It registers with
// the region, long-polls its desired state and writes it (PXE files, dnsmasq
// configuration, host files), runs BMC commands the region queues for it,
// serves boot and OS images from a local cache filled from the region on
// first use, and proxies the node API (commissioning, deploy jobs, overlays,
// phone-home) to the region, marking requests with its name.
import (
"bytes"
"context"
"crypto/ecdsa"
"crypto/elliptic"
"crypto/rand"
"crypto/tls"
"crypto/x509"
"crypto/x509/pkix"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"io"
"log"
"net"
"net/http"
"net/http/httputil"
"net/url"
"os"
"os/user"
"path"
"path/filepath"
"runtime"
"strconv"
"strings"
"sync"
"time"
)
type RackConfig struct {
Name string `json:"name,omitempty"` // default: hostname
Region string `json:"region,omitempty"` // the region's URL, e.g. http://10.0.0.234
Secret string `json:"secret,omitempty"` // the region's rack_secret
URL string `json:"url,omitempty"` // how machines reach this rack, e.g. http://10.213.78.1
Listen string `json:"listen,omitempty"` // default ":80"
TFTPRoot string `json:"tftp_root,omitempty"` // default /srv/tftp
DnsmasqDir string `json:"dnsmasq_dir,omitempty"` // default /var/lib/rollcall/dnsmasq
CacheDir string `json:"cache_dir,omitempty"` // default /var/cache/rollcall
IPMITool string `json:"ipmitool,omitempty"` // default ipmitool
DHCPInterfaces []string `json:"dhcp_interfaces,omitempty"` // default: the interfaces with an address in a served subnet
CAFile string `json:"ca_file,omitempty"` // set by rollcall rack join: the region's CA,
CertFile string `json:"cert_file,omitempty"` // this rack controller's certificate
KeyFile string `json:"key_file,omitempty"` // and key; with them the secret is not needed
AutoUpdate *bool `json:"auto_update,omitempty"` // follow the region's version (default true)
UpdateDir string `json:"update_dir,omitempty"` // default /var/lib/rollcall/update
}
type rackd struct {
cfg RackConfig
region *url.URL
hc *http.Client // long polls
fast *http.Client
big *http.Client // image downloads, no timeout
gen int64
tlsc *tls.Config
certMu sync.Mutex
upd rackUpdater
cert *tls.Certificate
headMu sync.Mutex
heads map[string]headInfo
fetch sync.Mutex // one boot-file prefetch at a time
}
type headInfo struct {
size int64
mod time.Time
at time.Time
}
func loadRackConfig(p string) (RackConfig, error) {
var c RackConfig
b, err := os.ReadFile(p)
if err != nil {
return c, err
}
if err := json.Unmarshal(b, &c); err != nil {
return c, fmt.Errorf("%s: %w", p, err)
}
if c.Name == "" {
c.Name, _ = os.Hostname()
}
if c.Listen == "" {
c.Listen = ":80"
}
if c.TFTPRoot == "" {
c.TFTPRoot = "/srv/tftp"
}
if c.DnsmasqDir == "" {
c.DnsmasqDir = "/var/lib/rollcall/dnsmasq"
}
if c.CacheDir == "" {
c.CacheDir = "/var/cache/rollcall"
}
if c.IPMITool == "" {
c.IPMITool = "ipmitool"
}
c.Region, c.URL = strings.TrimRight(c.Region, "/"), strings.TrimRight(c.URL, "/")
for _, k := range []struct{ v, n string }{{c.Region, "region"}, {c.URL, "url"}} {
if k.v == "" {
return c, fmt.Errorf("%s: %s is required", p, k.n)
}
}
if c.Secret == "" && c.CertFile == "" {
return c, fmt.Errorf("%s: not enrolled: run rollcall rack join CODE (or set the legacy secret)", p)
}
return c, nil
}
// clients builds the HTTP clients; with a certificate they speak mutual TLS
// and pick up a renewed certificate without being rebuilt.
func (r *rackd) clients() error {
r.tlsc = &tls.Config{MinVersion: tls.VersionTLS12}
if r.cfg.CertFile != "" {
cab, err := os.ReadFile(r.cfg.CAFile)
if err != nil {
return err
}
pool := x509.NewCertPool()
if !pool.AppendCertsFromPEM(cab) {
return fmt.Errorf("%s: no certificate", r.cfg.CAFile)
}
pair, err := tls.LoadX509KeyPair(r.cfg.CertFile, r.cfg.KeyFile)
if err != nil {
return err
}
pair.Leaf, _ = x509.ParseCertificate(pair.Certificate[0])
r.cert = &pair
r.tlsc.RootCAs = pool
r.tlsc.GetClientCertificate = func(*tls.CertificateRequestInfo) (*tls.Certificate, error) {
r.certMu.Lock()
defer r.certMu.Unlock()
return r.cert, nil
}
}
tr := func() *http.Transport {
return &http.Transport{TLSClientConfig: r.tlsc, Proxy: http.ProxyFromEnvironment, MaxIdleConnsPerHost: 8,
DialContext: (&net.Dialer{Timeout: 10 * time.Second}).DialContext}
}
r.hc = &http.Client{Timeout: 45 * time.Second, Transport: tr()}
r.fast = &http.Client{Timeout: 30 * time.Second, Transport: tr()}
r.big = &http.Client{Transport: tr()}
return nil
}
// renewLoop asks for a new certificate a month before the current one ends.
func (r *rackd) renewLoop() {
for {
r.certMu.Lock()
leaf := r.cert.Leaf
r.certMu.Unlock()
if leaf != nil && time.Until(leaf.NotAfter) < renewBefore {
if err := r.renew(); err != nil {
log.Printf("certificate renewal: %v", err)
}
}
time.Sleep(12 * time.Hour)
}
}
func (r *rackd) renew() error {
key, csr, err := newCSR("rack:" + r.cfg.Name)
if err != nil {
return err
}
var resp struct {
Cert string `json:"cert"`
}
if _, err := r.call(r.fast, "POST", "/api/v1/rack/renew", enrollRequest{CSR: csr}, &resp); err != nil {
return err
}
pair, err := tls.X509KeyPair([]byte(resp.Cert), key)
if err != nil {
return err
}
pair.Leaf, _ = x509.ParseCertificate(pair.Certificate[0])
if err := os.WriteFile(r.cfg.KeyFile, key, 0o640); err != nil {
return err
}
if err := os.WriteFile(r.cfg.CertFile, []byte(resp.Cert), 0o644); err != nil {
return err
}
r.certMu.Lock()
r.cert = &pair
r.certMu.Unlock()
log.Printf("renewed the rack certificate, valid until %s", pair.Leaf.NotAfter.Format("2006-01-02"))
return nil
}
func newCSR(cn string) (keyPEM []byte, csrPEM string, err error) {
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
if err != nil {
return nil, "", err
}
der, err := x509.CreateCertificateRequest(rand.Reader, &x509.CertificateRequest{Subject: pkix.Name{CommonName: cn}}, key)
if err != nil {
return nil, "", err
}
kder, _ := x509.MarshalECPrivateKey(key)
return pemEncode("EC PRIVATE KEY", kder), string(pemEncode("CERTIFICATE REQUEST", der)), nil
}
// rackJoin enrolls this host as a rack controller with a code from
// `rollcall controller join-code` and writes rack.json and the certificate.
func rackJoin(args []string) error {
if len(args) < 1 || strings.HasPrefix(args[0], "-") {
return errors.New("usage: rollcall rack join CODE -url http://ADDR [-name N] [-listen ADDR:80] [-config /etc/rollcall/rack.json]")
}
code, err := parseJoinCode(args[0])
if err != nil {
return err
}
fs := newFlags("rack join")
host, _ := os.Hostname()
name := fs.String("name", host, "rack controller name")
u := fs.String("url", "", "how machines reach this rack controller, e.g. http://10.213.78.1 (required)")
listen := fs.String("listen", "", "listen address (default: the url's host, port 80)")
cfgPath := fs.String("config", "/etc/rollcall/rack.json", "configuration to write")
dir := fs.String("dir", "/etc/rollcall/rack", "where the certificate goes")
_ = fs.Parse(args[1:])
if *u == "" {
return errors.New("-url is required: the address machines on this segment reach this host at")
}
cfg := RackConfig{}
if b, err := os.ReadFile(*cfgPath); err == nil {
_ = json.Unmarshal(b, &cfg)
}
cfg.Name, cfg.URL, cfg.Region, cfg.Secret = *name, strings.TrimRight(*u, "/"), "https://"+code.Addr, ""
if *listen != "" {
cfg.Listen = *listen
} else if cfg.Listen == "" {
if pu, err := url.Parse(cfg.URL); err == nil {
cfg.Listen = net.JoinHostPort(pu.Hostname(), "80")
}
}
cfg.CAFile, cfg.CertFile, cfg.KeyFile = filepath.Join(*dir, "ca.pem"), filepath.Join(*dir, "rack.pem"), filepath.Join(*dir, "rack-key.pem")
if cfg.TFTPRoot == "" {
cfg.TFTPRoot = "/srv/tftp"
}
key, csr, err := newCSR("rack:" + cfg.Name)
if err != nil {
return err
}
req := enrollRequest{Secret: base64.RawURLEncoding.EncodeToString(code.Secret[:]), Name: cfg.Name, CSR: csr,
Rack: RackController{URL: cfg.URL, Version: version, Interfaces: localInterfaces(), DHCPInterfaces: cfg.DHCPInterfaces, TFTPRoot: cfg.TFTPRoot}}
body, _ := json.Marshal(req)
hc := &http.Client{Timeout: 30 * time.Second, Transport: &http.Transport{TLSClientConfig: pinnedTLS(code.Fingerprint)}}
resp, err := hc.Post(cfg.Region+"/api/v1/rack/enroll", "application/json", bytes.NewReader(body))
if err != nil {
return fmt.Errorf("cannot reach the region at %s: %w", cfg.Region, err)
}
defer resp.Body.Close()
data, _ := io.ReadAll(resp.Body)
if resp.StatusCode != 200 {
var e apiError
_ = json.Unmarshal(data, &e)
return fmt.Errorf("the region refused: %s %s", resp.Status, e.Error)
}
var out struct {
Name, Cert, CA string
}
if err := json.Unmarshal(data, &out); err != nil {
return err
}
cfg.Name = out.Name
if err := os.MkdirAll(*dir, 0o750); err != nil {
return err
}
for _, f := range []struct {
p string
b []byte
mode os.FileMode
}{{cfg.CAFile, []byte(out.CA), 0o644}, {cfg.CertFile, []byte(out.Cert), 0o644}, {cfg.KeyFile, key, 0o640}} {
if err := os.WriteFile(f.p, f.b, f.mode); err != nil {
return err
}
}
cb, _ := json.MarshalIndent(cfg, "", " ")
if err := os.MkdirAll(filepath.Dir(*cfgPath), 0o755); err != nil {
return err
}
if err := os.WriteFile(*cfgPath, append(cb, '\n'), 0o640); err != nil {
return err
}
// the service runs as user rollcall: give its group the key and the configuration
if g, err := user.LookupGroup("rollcall"); err == nil {
if gid, err := strconv.Atoi(g.Gid); err == nil {
for _, p := range []string{*dir, cfg.CAFile, cfg.CertFile, cfg.KeyFile, *cfgPath} {
_ = os.Chown(p, 0, gid)
}
}
}
ca, _ := parseCertPEM([]byte(out.CA))
fmt.Printf("joined the region at %s as rack controller %s\n CA %s\n wrote %s and %s\nnext: sh install-rack.sh (first time) or systemctl restart rollcall-rack\n",
cfg.Region, cfg.Name, fpString(certFingerprint(ca)), *cfgPath, *dir)
return nil
}
func runRack(args []string) error {
fs := newFlags("rack")
cfgPath := fs.String("config", "/etc/rollcall/rack.json", "rack controller configuration")
_ = fs.Parse(args)
cfg, err := loadRackConfig(*cfgPath)
if err != nil {
return err
}
ru, err := url.Parse(cfg.Region)
if err != nil {
return fmt.Errorf("region: %w", err)
}
r := &rackd{cfg: cfg, region: ru, heads: map[string]headInfo{}, upd: rackUpdater{tried: map[string]time.Time{}}}
if err := r.clients(); err != nil {
return fmt.Errorf("tls: %w", err)
}
if r.cert != nil {
go r.renewLoop()
} else {
log.Printf("warning: using the legacy shared secret; enroll with rollcall rack join for mutual TLS")
}
for _, d := range []string{filepath.Join(cfg.TFTPRoot, "pxelinux.cfg"), cfg.CacheDir, filepath.Join(cfg.DnsmasqDir, "conf.d")} {
if err := os.MkdirAll(d, 0o755); err != nil {
return err
}
}
go r.loop()
mux := http.NewServeMux()
for _, p := range []string{"/rollcall/", "/images/", "/os/"} {
mux.HandleFunc("GET "+p, r.serveCached)
mux.HandleFunc("HEAD "+p, r.serveCached)
}
proxy := &httputil.ReverseProxy{
Rewrite: func(pr *httputil.ProxyRequest) {
pr.SetURL(ru)
pr.SetXForwarded()
pr.Out.Header.Del("X-Rollcall-Rack-Secret")
pr.Out.Header.Set("X-Rollcall-Rack", cfg.Name)
if cfg.Secret != "" {
pr.Out.Header.Set("X-Rollcall-Rack-Secret", cfg.Secret)
}
},
Transport: r.big.Transport,
FlushInterval: 100 * time.Millisecond,
}
for _, p := range []string{"/api/v1/commission", "/api/v1/bmc-hint", "/api/v1/jobs/self", "/api/v1/jobs/self/progress",
"/api/v1/nodes/self/overlay", "/api/v1/phone-home/"} {
mux.Handle(p, proxy)
}
mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) {
writeJSON(w, 200, obj{"rack": cfg.Name, "gen": r.gen, "version": version})
})
srv := &http.Server{Addr: cfg.Listen, Handler: logging(mux), ReadHeaderTimeout: 10 * time.Second}
log.Printf("rack controller %s: region %s, serving %s on %s (tftp %s, cache %s)", cfg.Name, cfg.Region, cfg.URL, cfg.Listen, cfg.TFTPRoot, cfg.CacheDir)
return srv.ListenAndServe()
}
func (r *rackd) req(ctx context.Context, method, p string, body any) (*http.Request, error) {
var rd io.Reader
if body != nil {
b, _ := json.Marshal(body)
rd = bytes.NewReader(b)
}
req, err := http.NewRequestWithContext(ctx, method, r.cfg.Region+p, rd)
if err != nil {
return nil, err
}
req.Header.Set("X-Rollcall-Rack", r.cfg.Name)
if r.cfg.Secret != "" {
req.Header.Set("X-Rollcall-Rack-Secret", r.cfg.Secret)
}
req.Header.Set("Content-Type", "application/json")
return req, nil
}
func (r *rackd) call(c *http.Client, method, p string, body, out any) (int, error) {
req, err := r.req(context.Background(), method, p, body)
if err != nil {
return 0, err
}
resp, err := c.Do(req)
if err != nil {
return 0, err
}
defer resp.Body.Close()
data, _ := io.ReadAll(resp.Body)
if resp.StatusCode >= 300 {
var e apiError
_ = json.Unmarshal(data, &e)
if e.Error == "" {
e.Error = strings.TrimSpace(string(data))
}
return resp.StatusCode, fmt.Errorf("%s %s: %s", method, p, e.Error)
}
if out != nil {
return resp.StatusCode, json.Unmarshal(data, out)
}
return resp.StatusCode, nil
}
func (r *rackd) register() error {
_, err := r.call(r.fast, "POST", "/api/v1/rack/register", RackController{URL: r.cfg.URL, Version: version, Arch: runtime.GOARCH,
Interfaces: localInterfaces(), DHCPInterfaces: r.cfg.DHCPInterfaces, TFTPRoot: r.cfg.TFTPRoot}, nil)
return err
}
func (r *rackd) loop() {
backoff := time.Second
for {
if err := r.register(); err != nil {
log.Printf("register: %v (retry in %s)", err, backoff)
time.Sleep(backoff)
backoff = min(backoff*2, 30*time.Second)
continue
}
break
}
log.Printf("registered with %s as %s", r.cfg.Region, r.cfg.Name)
go r.workLoop()
go func() { // interfaces change; registration also refreshes last_seen
for range time.Tick(5 * time.Minute) {
if err := r.register(); err != nil {
log.Printf("re-register: %v", err)
}
}
}()
for {
var st rackState
code, err := r.call(r.hc, "GET", "/api/v1/rack/state?since="+strconv.FormatInt(r.gen, 10), nil, &st)
if err != nil {
log.Printf("state: %v", err)
if code == 409 || code == 401 {
_ = r.register()
}
time.Sleep(3 * time.Second)
continue
}
go r.maybeUpdate(st.Update)
if st.Gen == r.gen {
continue
}
ch, err := applyRackState(&st, r.cfg.TFTPRoot, r.cfg.DnsmasqDir)
if err != nil {
log.Printf("apply: %v", err)
} else if len(ch) > 0 {
log.Printf("state %d: wrote %s", st.Gen, strings.Join(ch, " "))
}
r.gen = st.Gen
go r.prefetch(st.BootFiles)
}
}
func (r *rackd) workLoop() {
for {
var jobs []rackJob
if _, err := r.call(r.hc, "GET", "/api/v1/rack/work", nil, &jobs); err != nil {
log.Printf("work: %v", err)
time.Sleep(3 * time.Second)
continue
}
for _, j := range jobs {
go func(j rackJob) {
res := rackResult{ID: j.ID}
switch j.Kind {
case "power":
out, err := powerRun(context.Background(), r.cfg.IPMITool, &bmcCreds{Driver: j.Driver, Address: j.Address, Port: j.Port,
User: j.User, Password: j.Password, Extra: j.Extra}, j.Op, j.Arg)
res.Output = out
if err != nil {
res.Error = err.Error()
if out != "" {
res.Error = out + " (" + err.Error() + ")"
}
}
default:
res.Error = "unknown job kind " + j.Kind + " (update this rack controller)"
}
log.Printf("job %s %s %s %s %s: %s", j.ID, j.Kind, j.Driver, j.Op, j.Arg, oneLine(res.Output+" "+res.Error))
if _, err := r.call(r.fast, "POST", "/api/v1/rack/result", res, nil); err != nil {
log.Printf("result %s: %v", j.ID, err)
}
}(j)
}
}
}
// ---------------------------------------------------------------- image cache
// head asks the region for a file's size and modification time (cached briefly).
func (r *rackd) head(rel string) (headInfo, error) {
r.headMu.Lock()
if h, ok := r.heads[rel]; ok && time.Since(h.at) < 30*time.Second {
r.headMu.Unlock()
return h, nil
}
r.headMu.Unlock()
req, _ := http.NewRequest("HEAD", r.cfg.Region+rel, nil)
resp, err := r.fast.Do(req)
if err != nil {
return headInfo{}, err
}
resp.Body.Close()
if resp.StatusCode != 200 {
return headInfo{}, fmt.Errorf("%s: %s", rel, resp.Status)
}
mod, _ := http.ParseTime(resp.Header.Get("Last-Modified"))
h := headInfo{size: resp.ContentLength, mod: mod, at: time.Now()}
r.headMu.Lock()
r.heads[rel] = h
r.headMu.Unlock()
return h, nil
}
// fresh says whether the cached copy matches the region's file. OS images live
// under versioned paths and never change; the rest is checked by size and time.
func (r *rackd) fresh(rel string, fi os.FileInfo) bool {
if strings.HasPrefix(rel, "/os/") && !strings.Contains(rel, "/current/") {
return true
}
h, err := r.head(rel)
if err != nil {
return true // region unreachable: serve what we have
}
return h.size == fi.Size() && h.mod.Unix() == fi.ModTime().Unix()
}
func (r *rackd) localPath(rel string) (string, bool) {
clean := path.Clean("/" + rel)
if clean == "/" || strings.Contains(clean, "..") {
return "", false
}
return filepath.Join(r.cfg.CacheDir, filepath.FromSlash(clean)), true
}
// fill downloads rel from the region into the cache, copying to w on the way when given.
func (r *rackd) fill(rel string, w http.ResponseWriter) error {
local, ok := r.localPath(rel)
if !ok {
return errors.New("bad path")
}
resp, err := r.big.Get(r.cfg.Region + rel) // no timeout: images are large
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != 200 {
if w != nil {
w.WriteHeader(resp.StatusCode)
}
return fmt.Errorf("%s: %s", rel, resp.Status)
}
if err := os.MkdirAll(filepath.Dir(local), 0o755); err != nil {
return err
}
tmp, err := os.CreateTemp(filepath.Dir(local), ".fill-*")
if err != nil {
return err
}
defer os.Remove(tmp.Name())
var dst io.Writer = tmp
if w != nil {
for _, h := range []string{"Content-Length", "Content-Type", "Last-Modified"} {
if v := resp.Header.Get(h); v != "" {
w.Header().Set(h, v)
}
}
w.WriteHeader(200)
dst = io.MultiWriter(tmp, w)
}
n, err := io.Copy(dst, resp.Body)
if cerr := tmp.Close(); err == nil {
err = cerr
}
if err != nil {
return err
}
if resp.ContentLength >= 0 && n != resp.ContentLength {
return fmt.Errorf("%s: short read %d/%d", rel, n, resp.ContentLength)
}
if mod, err := http.ParseTime(resp.Header.Get("Last-Modified")); err == nil {
_ = os.Chtimes(tmp.Name(), mod, mod)
}
_ = os.Chmod(tmp.Name(), 0o644)
if err := os.Rename(tmp.Name(), local); err != nil {
return err
}
log.Printf("cached %s (%s)", rel, human(float64(n)))
return nil
}
func (r *rackd) serveCached(w http.ResponseWriter, req *http.Request) {
rel := path.Clean(req.URL.Path)
local, ok := r.localPath(rel)
if !ok {
http.NotFound(w, req)
return
}
if fi, err := os.Stat(local); err == nil && !fi.IsDir() && r.fresh(rel, fi) {
http.ServeFile(w, req, local)
return
}
if req.Method == http.MethodHead {
h, err := r.head(rel)
if err != nil {
http.NotFound(w, req)
return
}
w.Header().Set("Content-Length", strconv.FormatInt(h.size, 10))
w.Header().Set("Last-Modified", h.mod.UTC().Format(http.TimeFormat))
return
}
if err := r.fill(rel, w); err != nil {
log.Printf("cache %s: %v", rel, err)
}
}
// prefetch keeps kernels and initrds cached: TFTP cannot fetch on demand.
func (r *rackd) prefetch(files []string) {
r.fetch.Lock()
defer r.fetch.Unlock()
for _, f := range files {
rel := "/" + strings.TrimPrefix(f, "/")
local, ok := r.localPath(rel)
if !ok {
continue
}
if fi, err := os.Stat(local); err == nil && r.fresh(rel, fi) {
continue
}
if err := r.fill(rel, nil); err != nil {
log.Printf("prefetch %s: %v", rel, err)
}
}
}