Ilker ManapandClaude Opus 5.5
b01d8e2b4d
A firmware repository on the region (firmware_dir, served under /firmware/<id>/<file>, cached by rack controllers for BMCs to pull), each machine's firmware inventory (FirmwareInventory, classified into bios/bmc/nic/raid/cpld/psu/drive), and update jobs (kind "firmware") that the region drives with short rack commands: fw-start hands the package over, fw-poll watches the task (tolerating a BMC that restarts), fw-finish releases what was claimed, fw-inventory verifies the version. Running updates resume after a region restart; they cannot be cancelled once the BMC has the package. Standard methods (multipart, SimpleUpdate pull, HttpPushUri) with known Content-Length, plus vendor plans from their documentation or their own tools: Dell MultipartUpload (OnReset), HPE SimpleUpdate with .fwpkg extracted at import, Lenovo /mfwupdate with HttpPushUriTargetsBusy, Supermicro /upload with targets and preserve options, GIGABYTE/AMI/ ASRock Rack /upload with OemParameters ImageType, OpenBMC update-multipart, xFusion upload to /tmp/web then SimpleUpdate, KAYTUS upload then OEM SimpleUpdate with fixed tasks, Fujitsu iRMC OEM actions; Cisco CIMC is refused clearly (SCP only). fw_params and fw_method on the BMC record override a plan. CLI rollcall firmware list|import|rm|show|update, API, a Firmware tab and a repository card in the UI. The lab Redfish emulator gains an UpdateService (BMC applies at once, BIOS at the next reset). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
827 lines
29 KiB
Go
827 lines
29 KiB
Go
package main
|
|
|
|
// Disk-based OS deployment: a job boots the commissioning image in deploy
|
|
// mode; the agent streams <os>/<version>/disk.raw.zst onto the chosen disk,
|
|
// seeds cloud-init (NoCloud) in the written root filesystem and reports
|
|
// progress. On success the machine boots from disk and cloud-init phones home.
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log"
|
|
"net/http"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
type Job struct {
|
|
ID string `json:"id"`
|
|
MachineID string `json:"machine_id"`
|
|
Kind string `json:"kind"` // deploy | release
|
|
State string `json:"state"`
|
|
Spec json.RawMessage `json:"spec"`
|
|
Progress json.RawMessage `json:"progress"`
|
|
Created time.Time `json:"created"`
|
|
Started *time.Time `json:"started"`
|
|
Finished *time.Time `json:"finished"`
|
|
Error *string `json:"error"`
|
|
}
|
|
|
|
// DeploySpec is what the agent receives (plus the rendered seed files).
|
|
type DeploySpec struct {
|
|
OS string `json:"os"`
|
|
Version string `json:"version"`
|
|
ImageURL string `json:"image_url"`
|
|
SHA256 string `json:"sha256"`
|
|
Size int64 `json:"size"`
|
|
RawSize int64 `json:"raw_size"`
|
|
Disk string `json:"disk"`
|
|
DiskSerial string `json:"disk_serial,omitempty"`
|
|
Hostname string `json:"hostname"`
|
|
Domain string `json:"domain,omitempty"`
|
|
SSHKeys []string `json:"ssh_keys"`
|
|
PackageUpgrade bool `json:"package_upgrade"`
|
|
Packages []string `json:"packages,omitempty"`
|
|
UserData string `json:"user_data,omitempty"` // full #cloud-config override
|
|
Wipe string `json:"wipe,omitempty"` // release: quick | full
|
|
NetworkConfig string `json:"network_config,omitempty"`
|
|
LinkRules string `json:"link_rules,omitempty"`
|
|
}
|
|
|
|
type seedFile struct {
|
|
Path string `json:"path"`
|
|
Mode string `json:"mode"`
|
|
Content string `json:"content"`
|
|
}
|
|
|
|
func (j *Job) spec() DeploySpec {
|
|
var s DeploySpec
|
|
_ = json.Unmarshal(j.Spec, &s)
|
|
return s
|
|
}
|
|
|
|
// ---------------------------------------------------------------- store
|
|
|
|
const jobCols = `id, machine_id, kind, state, spec, progress, created, started, finished, error`
|
|
|
|
func scanJob(row interface{ Scan(...any) error }) (*Job, error) {
|
|
j := &Job{}
|
|
err := row.Scan(&j.ID, &j.MachineID, &j.Kind, &j.State, &j.Spec, &j.Progress, &j.Created, &j.Started, &j.Finished, &j.Error)
|
|
if err != nil {
|
|
if strings.Contains(err.Error(), "no rows") {
|
|
return nil, errNotFound
|
|
}
|
|
return nil, err
|
|
}
|
|
return j, nil
|
|
}
|
|
|
|
func (s *Store) CreateJob(ctx context.Context, j *Job, tokenHash string, sealed []byte) error {
|
|
_, err := s.pool.Exec(ctx, `INSERT INTO jobs(id, machine_id, kind, state, spec, token_hash, token) VALUES ($1,$2,$3,'pending',$4,$5,$6)`,
|
|
j.ID, j.MachineID, j.Kind, j.Spec, tokenHash, sealed)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) JobByToken(ctx context.Context, hash string) (*Job, error) {
|
|
return scanJob(s.pool.QueryRow(ctx, `SELECT `+jobCols+` FROM jobs WHERE token_hash=$1`, hash))
|
|
}
|
|
|
|
func (s *Store) Job(ctx context.Context, id string) (*Job, error) {
|
|
return scanJob(s.pool.QueryRow(ctx, `SELECT `+jobCols+` FROM jobs WHERE id=$1`, id))
|
|
}
|
|
|
|
func (s *Store) ActiveJob(ctx context.Context, machineID string) (*Job, error) {
|
|
return scanJob(s.pool.QueryRow(ctx, `SELECT `+jobCols+` FROM jobs WHERE machine_id=$1 AND state IN ('pending','running') ORDER BY created DESC LIMIT 1`, machineID))
|
|
}
|
|
|
|
func (s *Store) ActiveJobs(ctx context.Context) (map[string]*Job, error) {
|
|
rows, err := s.pool.Query(ctx, `SELECT `+jobCols+` FROM jobs WHERE state IN ('pending','running') ORDER BY created`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
out := map[string]*Job{}
|
|
for rows.Next() {
|
|
j, err := scanJob(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out[j.MachineID] = j
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (s *Store) Jobs(ctx context.Context, machineID string, limit int) ([]*Job, error) {
|
|
rows, err := s.pool.Query(ctx, `SELECT `+jobCols+` FROM jobs WHERE machine_id=$1 ORDER BY created DESC LIMIT $2`, machineID, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
out := []*Job{}
|
|
for rows.Next() {
|
|
j, err := scanJob(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, j)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// RecentJobs lists jobs across all machines, newest first (active ones first when activeFirst).
|
|
func (s *Store) RecentJobs(ctx context.Context, limit int) ([]*Job, error) {
|
|
rows, err := s.pool.Query(ctx, `SELECT `+jobCols+` FROM jobs ORDER BY (state IN ('pending','running')) DESC, created DESC LIMIT $1`, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
out := []*Job{}
|
|
for rows.Next() {
|
|
j, err := scanJob(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, j)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (s *Store) JobProgress(ctx context.Context, id string, progress []byte) error {
|
|
_, err := s.pool.Exec(ctx, `UPDATE jobs SET state='running', started=coalesce(started, now()), progress=$2 WHERE id=$1 AND state IN ('pending','running')`, id, progress)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) FinishJob(ctx context.Context, id, state, errMsg string) error {
|
|
_, err := s.pool.Exec(ctx, `UPDATE jobs SET state=$2, finished=now(), started=coalesce(started, now()), error=nullif($3,'') WHERE id=$1`, id, state, errMsg)
|
|
return err
|
|
}
|
|
|
|
// FailPendingJob fails a job that has not started; false when it started meanwhile.
|
|
func (s *Store) FailPendingJob(ctx context.Context, id, errMsg string) (bool, error) {
|
|
tag, err := s.pool.Exec(ctx, `UPDATE jobs SET state='failed', finished=now(), started=coalesce(started, now()), error=nullif($2,'') WHERE id=$1 AND state='pending'`, id, errMsg)
|
|
return err == nil && tag.RowsAffected() == 1, err
|
|
}
|
|
|
|
func (s *Store) SetDeployed(ctx context.Context, id, os, version string) error {
|
|
_, err := s.pool.Exec(ctx, `UPDATE machines SET state='deployed', os=$2, os_version=$3, deployed_at=now(), booted_at=NULL, last_seen=now() WHERE id=$1`, id, os, version)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) SetReleased(ctx context.Context, id string) error {
|
|
if _, err := s.pool.Exec(ctx, `DELETE FROM ip_addresses WHERE machine_id=$1 AND kind='auto'`, id); err != nil {
|
|
return err
|
|
}
|
|
_, err := s.pool.Exec(ctx, `UPDATE machines SET state='ready', os=NULL, os_version=NULL, deployed_at=NULL, booted_at=NULL, last_seen=now(),
|
|
mgmt_ip=(SELECT ip FROM ip_addresses a WHERE a.machine_id=machines.id ORDER BY ip LIMIT 1), ib_ip=NULL WHERE id=$1`, id)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) SetNodeAddr(ctx context.Context, id, name, mgmt, ib string) error {
|
|
if name != "" && !validDomain(name) {
|
|
return fmt.Errorf("node name %q is not a host name (letters, digits, hyphens)", name)
|
|
}
|
|
_, err := s.pool.Exec(ctx, `UPDATE machines SET node_name=$2, mgmt_ip=$3, ib_ip=$4 WHERE id=$1`, id, name, nilIfEmpty(mgmt), nilIfEmpty(ib))
|
|
return err
|
|
}
|
|
|
|
func (s *Store) MarkPhoneHome(ctx context.Context, id string) error {
|
|
_, err := s.pool.Exec(ctx, `UPDATE machines SET booted_at=now(), last_seen=now() WHERE id=$1`, id)
|
|
return err
|
|
}
|
|
|
|
// ---------------------------------------------------------------- disk selection
|
|
|
|
type invDisk struct {
|
|
Name, Path, Serial, Tran, Model string
|
|
Size int64
|
|
Rota, Removable bool
|
|
}
|
|
|
|
func inventoryDisks(m *Machine) []invDisk {
|
|
var inv obj
|
|
_ = json.Unmarshal(m.Inventory, &inv)
|
|
var out []invDisk
|
|
for _, dv := range toList(dig(inv, "storage", "disks")) {
|
|
d, _ := dv.(obj)
|
|
rm, _ := d["rm"].(bool)
|
|
rota, _ := d["rota"].(bool)
|
|
id := invDisk{Name: str(d["name"]), Path: str(d["path"]), Serial: str(d["serial"]), Tran: str(d["tran"]),
|
|
Model: str(d["model"]), Size: int64(num(d["size"])), Rota: rota, Removable: rm}
|
|
if id.Path == "" {
|
|
id.Path = "/dev/" + id.Name
|
|
}
|
|
out = append(out, id)
|
|
}
|
|
return out
|
|
}
|
|
|
|
// pickDisk: the smallest non-removable, non-USB disk that fits the image (boot/OS disks are
|
|
// usually the small ones on servers; data disks stay untouched). Explicit choice wins.
|
|
// remoteTran: transports of disks that are not the machine's own (lsblk TRAN).
|
|
var remoteTran = map[string]bool{"usb": true, "iscsi": true, "fc": true, "fcoe": true, "srp": true, "iser": true}
|
|
|
|
func pickDisk(disks []invDisk, want string, need int64) (*invDisk, error) {
|
|
if want != "" {
|
|
for i := range disks {
|
|
if disks[i].Path == want || disks[i].Name == want || "/dev/"+disks[i].Name == want {
|
|
if disks[i].Size > 0 && disks[i].Size < need {
|
|
return nil, fmt.Errorf("disk %s (%s) is smaller than the image (%s)", want, human(float64(disks[i].Size)), human(float64(need)))
|
|
}
|
|
return &disks[i], nil
|
|
}
|
|
}
|
|
if len(disks) == 0 {
|
|
return &invDisk{Path: want, Name: strings.TrimPrefix(want, "/dev/")}, nil
|
|
}
|
|
return nil, fmt.Errorf("disk %s not in inventory", want)
|
|
}
|
|
var cands []invDisk
|
|
for _, d := range disks {
|
|
if d.Removable || remoteTran[d.Tran] || d.Size < need { // never a USB stick or a SAN LUN by itself
|
|
continue
|
|
}
|
|
cands = append(cands, d)
|
|
}
|
|
if len(cands) == 0 {
|
|
return nil, fmt.Errorf("no disk >= %s in inventory (pass disk explicitly)", human(float64(need)))
|
|
}
|
|
sort.Slice(cands, func(i, j int) bool {
|
|
if cands[i].Size != cands[j].Size {
|
|
return cands[i].Size < cands[j].Size
|
|
}
|
|
return cands[i].Name < cands[j].Name
|
|
})
|
|
return &cands[0], nil
|
|
}
|
|
|
|
// ---------------------------------------------------------------- cloud-init seed
|
|
|
|
func yamlStr(s string) string { b, _ := json.Marshal(s); return string(b) }
|
|
func yamlList(l []string) string {
|
|
if l == nil {
|
|
l = []string{}
|
|
}
|
|
b, _ := json.Marshal(l)
|
|
return string(b)
|
|
}
|
|
|
|
func (a *App) renderSeed(j *Job, sp DeploySpec, token string) []seedFile {
|
|
base := a.cfg.ServerURL
|
|
m, merr := a.store.Get(context.Background(), j.MachineID)
|
|
if merr == nil {
|
|
base = a.urlFor(m)
|
|
}
|
|
meta := fmt.Sprintf("instance-id: %s\nlocal-hostname: %s\n", j.ID, sp.Hostname)
|
|
ud := sp.UserData
|
|
if ud == "" {
|
|
var b strings.Builder
|
|
b.WriteString("#cloud-config\n# generated by rollcall for " + j.MachineID + " (" + j.ID + ")\n")
|
|
fmt.Fprintf(&b, "hostname: %s\n", yamlStr(sp.Hostname))
|
|
if sp.Domain != "" {
|
|
fmt.Fprintf(&b, "fqdn: %s\n", yamlStr(sp.Hostname+"."+sp.Domain))
|
|
}
|
|
b.WriteString("manage_etc_hosts: true\npreserve_hostname: false\ndisable_root: false\nssh_pwauth: false\n")
|
|
fmt.Fprintf(&b, "ssh_authorized_keys: %s\n", yamlList(sp.SSHKeys))
|
|
fmt.Fprintf(&b, "users:\n - default\n - name: root\n ssh_authorized_keys: %s\n", yamlList(sp.SSHKeys))
|
|
if sp.PackageUpgrade || len(sp.Packages) > 0 {
|
|
b.WriteString("package_update: true\n")
|
|
}
|
|
if sp.PackageUpgrade {
|
|
b.WriteString("package_upgrade: true\npackage_reboot_if_required: true\n")
|
|
}
|
|
if len(sp.Packages) > 0 {
|
|
fmt.Fprintf(&b, "packages: %s\n", yamlList(sp.Packages))
|
|
}
|
|
fmt.Fprintf(&b, "phone_home:\n url: %s/api/v1/phone-home/%s\n post: [instance_id, hostname, fqdn, pub_key_ed25519]\n tries: 10\n", base, token)
|
|
b.WriteString("final_message: \"rollcall: deployed, up after $UPTIME s\"\n") // only $UPTIME/$TIMESTAMP/$DATASOURCE/$VERSION are expanded
|
|
ud = b.String()
|
|
}
|
|
var frags []TagDef
|
|
if a.ca != nil {
|
|
// The CA file itself is written into the image (seed files below); this rebuilds
|
|
// the trust store on every boot, early, with whatever the distribution has.
|
|
// cloud-init's ca_certs runs update-ca-certificates even on RHEL-family systems
|
|
// and fails there, so it is not used.
|
|
frags = append(frags, TagDef{Name: "rollcall-ca", CloudConfig: "#cloud-config\n# rollcall: trust the installation's certificate authority\nbootcmd:\n" +
|
|
" - [sh, -c, 'if command -v update-ca-trust >/dev/null 2>&1; then update-ca-trust extract; elif command -v update-ca-certificates >/dev/null 2>&1; then update-ca-certificates; fi']\n"})
|
|
}
|
|
if merr == nil {
|
|
if tf, err := a.tagFragments(context.Background(), m); err == nil {
|
|
frags = append(frags, tf...)
|
|
}
|
|
}
|
|
if len(frags) > 0 {
|
|
ud = multipartUserData(ud, frags)
|
|
}
|
|
files := []seedFile{
|
|
{Path: "var/lib/cloud/seed/nocloud/meta-data", Mode: "0600", Content: meta},
|
|
{Path: "var/lib/cloud/seed/nocloud/user-data", Mode: "0600", Content: ud},
|
|
{Path: "etc/cloud/cloud.cfg.d/99-rollcall.cfg", Mode: "0644", Content: "# written by rollcall at deploy time\ndatasource_list: [ NoCloud, None ]\n"},
|
|
}
|
|
if a.ca != nil { // both layouts; the bootcmd above picks the one the distribution uses
|
|
files = append(files,
|
|
seedFile{Path: "usr/local/share/ca-certificates/rollcall-ca.crt", Mode: "0644", Content: string(a.ca.PEM)},
|
|
seedFile{Path: "etc/pki/ca-trust/source/anchors/rollcall-ca.crt", Mode: "0644", Content: string(a.ca.PEM)})
|
|
}
|
|
if sp.NetworkConfig != "" {
|
|
files = append(files, seedFile{Path: "var/lib/cloud/seed/nocloud/network-config", Mode: "0600", Content: sp.NetworkConfig})
|
|
}
|
|
if sp.LinkRules != "" {
|
|
files = append(files, seedFile{Path: "etc/udev/rules.d/70-rollcall-net.rules", Mode: "0644", Content: sp.LinkRules})
|
|
}
|
|
return files
|
|
}
|
|
|
|
// ---------------------------------------------------------------- PXE
|
|
|
|
const deployTmpl = `# generated by rollcall %s
|
|
# machine %s job %s %s %s/%s -> %s
|
|
DEFAULT menu.c32
|
|
PROMPT 0
|
|
TIMEOUT 30
|
|
MENU TITLE %s - %s %s (job %s)
|
|
|
|
LABEL deploy
|
|
MENU LABEL Run deployment job (default in 3s)
|
|
MENU DEFAULT
|
|
KERNEL rollcall/vmlinuz
|
|
INITRD rollcall/initrd.img
|
|
APPEND %s
|
|
|
|
LABEL recommission
|
|
MENU LABEL Re-run commissioning
|
|
KERNEL rollcall/vmlinuz
|
|
INITRD rollcall/initrd.img
|
|
APPEND %s
|
|
|
|
LABEL local
|
|
MENU LABEL Boot from local disk
|
|
LOCALBOOT 0
|
|
`
|
|
|
|
const deployedTmpl = `# generated by rollcall %s
|
|
# machine %s deployed %s/%s
|
|
DEFAULT menu.c32
|
|
PROMPT 0
|
|
TIMEOUT 30
|
|
MENU TITLE %s - deployed %s (%s)
|
|
|
|
LABEL local
|
|
MENU LABEL Boot %s from local disk (default in 3s)
|
|
MENU DEFAULT
|
|
LOCALBOOT 0
|
|
|
|
LABEL recommission
|
|
MENU LABEL Re-run commissioning
|
|
KERNEL rollcall/vmlinuz
|
|
INITRD rollcall/initrd.img
|
|
APPEND %s
|
|
`
|
|
|
|
// deployAppend derives the deploy-mode cmdline from commission_append (same image, no agent token).
|
|
func (a *App) deployAppend(base, token string) string {
|
|
var parts []string
|
|
for _, p := range strings.Fields(a.appendFor(base)) {
|
|
if strings.HasPrefix(p, "rollcall.token=") || strings.HasPrefix(p, "rollcall.nopoweroff") {
|
|
continue
|
|
}
|
|
parts = append(parts, p)
|
|
}
|
|
parts = append(parts, "rollcall.mode=deploy", "rollcall.job="+token)
|
|
return strings.Join(parts, " ")
|
|
}
|
|
|
|
func (a *App) writePXEDeploy(m *Machine, j *Job, sp DeploySpec, token string) error {
|
|
body := fmt.Sprintf(deployTmpl, time.Now().Format(time.RFC3339), m.ID, j.ID, j.Kind, sp.OS, sp.Version, sp.Disk,
|
|
m.ID, j.Kind, sp.OS, j.ID, a.deployAppend(a.urlFor(m), token), a.appendFor(a.urlFor(m)))
|
|
return a.writePXEBody(m, body)
|
|
}
|
|
|
|
func (a *App) writePXEDeployed(m *Machine) error {
|
|
body := fmt.Sprintf(deployedTmpl, time.Now().Format(time.RFC3339), m.ID, deref(m.OS), deref(m.OSVersion),
|
|
deref(m.NodeName), deref(m.OS), m.ID, deref(m.OS), a.appendFor(a.urlFor(m)))
|
|
return a.writePXEBody(m, body)
|
|
}
|
|
|
|
// ---------------------------------------------------------------- job creation
|
|
|
|
type deployRequest struct {
|
|
OS string `json:"os"`
|
|
Version string `json:"version"`
|
|
Disk string `json:"disk"`
|
|
Hostname string `json:"hostname"`
|
|
SSHKeys []string `json:"ssh_keys"`
|
|
PackageUpgrade bool `json:"package_upgrade"`
|
|
Packages []string `json:"packages"`
|
|
UserData string `json:"user_data"`
|
|
Boot bool `json:"boot"`
|
|
Wipe string `json:"wipe"`
|
|
}
|
|
|
|
func (a *App) configSSHKeys() []string {
|
|
var keys []string
|
|
for _, k := range toList(a.cfg.ClusterVars["ssh_keys"]) {
|
|
if s := str(k); s != "" {
|
|
keys = append(keys, s)
|
|
}
|
|
}
|
|
return keys
|
|
}
|
|
|
|
// startJob creates the job row, PXE entry and (optionally) power-cycles the machine.
|
|
func (a *App) startJob(ctx context.Context, m *Machine, kind string, sp DeploySpec, boot bool) (obj, error) {
|
|
if m.Profile != nil {
|
|
return nil, fmt.Errorf("machine is a stateless node (profile %s); unassign it first", *m.Profile)
|
|
}
|
|
if j, err := a.store.ActiveJob(ctx, m.ID); err == nil {
|
|
return nil, fmt.Errorf("job %s (%s, %s) is still active; cancel it first", j.ID, j.Kind, j.State)
|
|
}
|
|
token := randomToken(16)
|
|
sum := sha256.Sum256([]byte(token))
|
|
sealed, err := a.sealer.Seal(token, m.ID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
specB, _ := json.Marshal(sp)
|
|
j := &Job{ID: "j-" + randomToken(5), MachineID: m.ID, Kind: kind, State: "pending", Spec: specB}
|
|
if err := a.store.CreateJob(ctx, j, hex.EncodeToString(sum[:]), sealed); err != nil {
|
|
return nil, err
|
|
}
|
|
state := "deploying"
|
|
if kind == "release" {
|
|
state = "releasing"
|
|
}
|
|
_ = a.store.SetState(ctx, m.ID, state)
|
|
if err := a.writePXEDeploy(m, j, sp, token); err != nil {
|
|
return nil, err
|
|
}
|
|
_ = a.store.AddEvent(ctx, m.ID, kind+" requested", fmt.Sprintf("job=%s os=%s/%s disk=%s hostname=%s", j.ID, sp.OS, sp.Version, sp.Disk, sp.Hostname))
|
|
out := obj{"id": m.ID, "job": j.ID, "state": state, "os": sp.OS, "version": sp.Version, "disk": sp.Disk, "hostname": sp.Hostname}
|
|
if boot {
|
|
out["ipmi"], out["ipmi_ok"] = a.pxeBoot(ctx, m)
|
|
}
|
|
log.Printf("%s %s: job %s os=%s/%s disk=%s", kind, m.ID, j.ID, sp.OS, sp.Version, sp.Disk)
|
|
return out, nil
|
|
}
|
|
|
|
func (a *App) handleDeploy(w http.ResponseWriter, r *http.Request) {
|
|
m, err := a.store.Get(r.Context(), r.PathValue("id"))
|
|
if err != nil {
|
|
fail(w, 500, err)
|
|
return
|
|
}
|
|
var req deployRequest
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.OS == "" {
|
|
fail(w, 400, fmt.Errorf("os is required"))
|
|
return
|
|
}
|
|
req.Hostname = strings.ToLower(req.Hostname)
|
|
if req.Hostname != "" && !validDomain(req.Hostname) {
|
|
fail(w, 400, fmt.Errorf("hostname %q is not a host name (letters, digits, hyphens)", req.Hostname))
|
|
return
|
|
}
|
|
// refuse before anything is allocated or renamed
|
|
if m.Profile != nil {
|
|
fail(w, 409, fmt.Errorf("machine is a stateless node (profile %s); unassign it first", *m.Profile))
|
|
return
|
|
}
|
|
if j, err := a.store.ActiveJob(r.Context(), m.ID); err == nil {
|
|
fail(w, 409, fmt.Errorf("job %s (%s, %s) is still active; cancel it first", j.ID, j.Kind, j.State))
|
|
return
|
|
}
|
|
meta, err := osResolve(a.cfg.OSDir, req.OS, req.Version)
|
|
if err != nil {
|
|
fail(w, 404, err)
|
|
return
|
|
}
|
|
disk, err := pickDisk(inventoryDisks(m), req.Disk, meta.RawSize)
|
|
if err != nil {
|
|
fail(w, 409, err)
|
|
return
|
|
}
|
|
// node name: keep the existing one unless another is asked for
|
|
name, mgmt, ib := deref(m.NodeName), deref(m.MgmtIP), deref(m.IBIP)
|
|
if req.Hostname != "" && req.Hostname != name {
|
|
name = ""
|
|
}
|
|
if name == "" {
|
|
if name, _, _, err = a.allocatePools(r.Context(), str(a.cfg.ClusterVars["name_format"]), req.Hostname, false); err != nil {
|
|
fail(w, 409, err)
|
|
return
|
|
}
|
|
m.NodeName = &name
|
|
}
|
|
// addresses: the network plan (layout, IPAM); without a network model, the legacy pools
|
|
netcfg, links := "", ""
|
|
var warnings []string
|
|
if len(a.subnetsSnapshot()) > 0 {
|
|
plan, err := a.netPlan(r.Context(), m, true)
|
|
if err != nil {
|
|
fail(w, 409, err)
|
|
return
|
|
}
|
|
warnings = plan.Warnings
|
|
if pm, pib := plan.primaryIPs(); pm != "" {
|
|
mgmt, ib = pm, pib
|
|
}
|
|
nb, _ := json.MarshalIndent(networkConfig(plan), "", " ")
|
|
netcfg = string(nb) + "\n"
|
|
links = linkRules(plan)
|
|
} else if mgmt == "" {
|
|
if _, mgmt, ib, err = a.allocatePools(r.Context(), "", name, true); err != nil {
|
|
fail(w, 409, err)
|
|
return
|
|
}
|
|
}
|
|
if err := a.store.SetNodeAddr(r.Context(), m.ID, name, mgmt, ib); err != nil {
|
|
fail(w, 500, err)
|
|
return
|
|
}
|
|
m, _ = a.store.Get(r.Context(), m.ID)
|
|
keys := req.SSHKeys
|
|
if keys == nil {
|
|
keys = a.configSSHKeys()
|
|
}
|
|
sp := DeploySpec{OS: meta.Name, Version: meta.Version,
|
|
ImageURL: a.urlFor(m) + "/os/" + meta.Name + "/" + meta.Version + "/disk.raw.zst",
|
|
SHA256: meta.SHA256, Size: meta.Size, RawSize: meta.RawSize, Disk: disk.Path, DiskSerial: disk.Serial,
|
|
Hostname: name, Domain: str(a.cfg.ClusterVars["domain"]), SSHKeys: keys, PackageUpgrade: req.PackageUpgrade,
|
|
Packages: req.Packages, UserData: req.UserData, NetworkConfig: netcfg, LinkRules: links}
|
|
out, err := a.startJob(r.Context(), m, "deploy", sp, req.Boot)
|
|
if err != nil {
|
|
fail(w, 409, err)
|
|
return
|
|
}
|
|
out["mgmt_ip"] = mgmt
|
|
out["rack"] = a.rackOf(m)
|
|
if len(warnings) > 0 {
|
|
out["warnings"] = warnings
|
|
}
|
|
writeJSON(w, 200, out)
|
|
}
|
|
|
|
func (a *App) handleRelease(w http.ResponseWriter, r *http.Request) {
|
|
m, err := a.store.Get(r.Context(), r.PathValue("id"))
|
|
if err != nil {
|
|
fail(w, 500, err)
|
|
return
|
|
}
|
|
var req deployRequest
|
|
_ = json.NewDecoder(r.Body).Decode(&req)
|
|
if req.Wipe == "" {
|
|
req.Wipe = "quick"
|
|
}
|
|
if req.Disk == "" { // the disk the OS went to, not "the smallest one"
|
|
if jobs, err := a.store.Jobs(r.Context(), m.ID, 50); err == nil {
|
|
for _, j := range jobs {
|
|
var sp DeploySpec
|
|
if j.Kind == "deploy" && j.State == "done" && json.Unmarshal(j.Spec, &sp) == nil && sp.Disk != "" {
|
|
req.Disk = sp.Disk
|
|
break
|
|
}
|
|
}
|
|
}
|
|
}
|
|
disk, err := pickDisk(inventoryDisks(m), req.Disk, 0)
|
|
if err != nil {
|
|
fail(w, 409, err)
|
|
return
|
|
}
|
|
sp := DeploySpec{OS: deref(m.OS), Version: deref(m.OSVersion), Disk: disk.Path, DiskSerial: disk.Serial, Wipe: req.Wipe, Hostname: deref(m.NodeName)}
|
|
out, err := a.startJob(r.Context(), m, "release", sp, req.Boot)
|
|
if err != nil {
|
|
fail(w, 409, err)
|
|
return
|
|
}
|
|
writeJSON(w, 200, out)
|
|
}
|
|
|
|
func (a *App) handleJobs(w http.ResponseWriter, r *http.Request) {
|
|
m, err := a.store.Get(r.Context(), r.PathValue("id"))
|
|
if err != nil {
|
|
fail(w, 500, err)
|
|
return
|
|
}
|
|
jobs, err := a.store.Jobs(r.Context(), m.ID, 20)
|
|
if err != nil {
|
|
fail(w, 500, err)
|
|
return
|
|
}
|
|
writeJSON(w, 200, jobs)
|
|
}
|
|
|
|
func (a *App) handleJobCancel(w http.ResponseWriter, r *http.Request) {
|
|
m, err := a.store.Get(r.Context(), r.PathValue("id"))
|
|
if err != nil {
|
|
fail(w, 500, err)
|
|
return
|
|
}
|
|
j, err := a.store.ActiveJob(r.Context(), m.ID)
|
|
if err != nil {
|
|
fail(w, 404, fmt.Errorf("no active job"))
|
|
return
|
|
}
|
|
if j.Kind == "firmware" {
|
|
fail(w, 409, errors.New("a firmware update cannot be stopped from here once the BMC has it (it would leave the component half-flashed); it ends by itself"))
|
|
return
|
|
}
|
|
_ = a.store.FinishJob(r.Context(), j.ID, "cancelled", "cancelled by operator")
|
|
a.afterJob(r.Context(), m, j, "cancelled", "cancelled by operator")
|
|
writeJSON(w, 200, obj{"id": m.ID, "job": j.ID, "state": "cancelled"})
|
|
}
|
|
|
|
// afterJob settles machine state and the PXE entry once a job stopped (done/failed/cancelled).
|
|
func (a *App) afterJob(ctx context.Context, m *Machine, j *Job, state, errMsg string) {
|
|
sp := j.spec()
|
|
switch {
|
|
case state == "done" && j.Kind == "deploy":
|
|
_ = a.store.SetDeployed(ctx, m.ID, sp.OS, sp.Version)
|
|
m, _ = a.store.Get(ctx, m.ID)
|
|
if err := a.writePXEDeployed(m); err != nil {
|
|
log.Printf("pxe write %s: %v", m.ID, err)
|
|
}
|
|
_ = a.store.AddEvent(ctx, m.ID, "deployed", fmt.Sprintf("job=%s os=%s/%s disk=%s", j.ID, sp.OS, sp.Version, sp.Disk))
|
|
case state == "done" && j.Kind == "release":
|
|
_ = a.store.SetReleased(ctx, m.ID)
|
|
a.kickIPAM()
|
|
m, _ = a.store.Get(ctx, m.ID)
|
|
_ = a.writePXE(m)
|
|
_ = a.store.AddEvent(ctx, m.ID, "released", fmt.Sprintf("job=%s disk=%s wipe=%s", j.ID, sp.Disk, sp.Wipe))
|
|
case state == "cancelled":
|
|
// nothing happened on the machine (as far as we know): back to the previous state
|
|
if m.OS != nil {
|
|
_ = a.store.SetState(ctx, m.ID, "deployed")
|
|
m, _ = a.store.Get(ctx, m.ID)
|
|
_ = a.writePXEDeployed(m)
|
|
} else {
|
|
_ = a.store.SetState(ctx, m.ID, "ready")
|
|
_ = a.writePXE(m)
|
|
}
|
|
_ = a.store.AddEvent(ctx, m.ID, j.Kind+" cancelled", "job="+j.ID)
|
|
default:
|
|
_ = a.store.SetState(ctx, m.ID, j.Kind+"-failed")
|
|
m, _ = a.store.Get(ctx, m.ID)
|
|
_ = a.writePXE(m) // next PXE boot: commissioned menu (power off), no retry loop
|
|
_ = a.store.AddEvent(ctx, m.ID, j.Kind+" "+state, "job="+j.ID+" "+errMsg)
|
|
}
|
|
log.Printf("job %s (%s %s): %s %s", j.ID, j.Kind, m.ID, state, errMsg)
|
|
}
|
|
|
|
// ---------------------------------------------------------------- agent side (job token)
|
|
|
|
func (a *App) jobFromRequest(w http.ResponseWriter, r *http.Request) (*Job, string) {
|
|
tok := r.Header.Get("X-Rollcall-Job")
|
|
if tok == "" {
|
|
fail(w, 401, errors.New("X-Rollcall-Job required"))
|
|
return nil, ""
|
|
}
|
|
sum := sha256.Sum256([]byte(tok))
|
|
j, err := a.store.JobByToken(r.Context(), hex.EncodeToString(sum[:]))
|
|
if err != nil {
|
|
fail(w, 401, errors.New("unknown job token"))
|
|
return nil, ""
|
|
}
|
|
return j, tok
|
|
}
|
|
|
|
func (a *App) handleJobSelf(w http.ResponseWriter, r *http.Request) {
|
|
j, tok := a.jobFromRequest(w, r)
|
|
if j == nil {
|
|
return
|
|
}
|
|
if j.State != "pending" && j.State != "running" {
|
|
fail(w, 409, fmt.Errorf("job %s is %s", j.ID, j.State))
|
|
return
|
|
}
|
|
sp := j.spec()
|
|
out := obj{"id": j.ID, "machine_id": j.MachineID, "kind": j.Kind, "spec": sp}
|
|
if j.Kind == "deploy" {
|
|
out["files"] = a.renderSeed(j, sp, tok)
|
|
}
|
|
if j.State == "pending" {
|
|
_ = a.store.JobProgress(r.Context(), j.ID, []byte(`{"phase":"started","pct":0}`))
|
|
_ = a.store.AddEvent(r.Context(), j.MachineID, j.Kind+" started", "job="+j.ID+" from "+r.RemoteAddr)
|
|
}
|
|
writeJSON(w, 200, out)
|
|
}
|
|
|
|
func (a *App) handleJobProgress(w http.ResponseWriter, r *http.Request) {
|
|
j, _ := a.jobFromRequest(w, r)
|
|
if j == nil {
|
|
return
|
|
}
|
|
var p struct {
|
|
State string `json:"state"` // "" (progress), done, failed
|
|
Phase string `json:"phase"`
|
|
Pct float64 `json:"pct"`
|
|
Msg string `json:"msg"`
|
|
Error string `json:"error"`
|
|
Details map[string]any `json:"details"`
|
|
}
|
|
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&p); err != nil {
|
|
fail(w, 400, err)
|
|
return
|
|
}
|
|
if j.State != "pending" && j.State != "running" {
|
|
writeJSON(w, 200, obj{"job": j.ID, "state": j.State, "next": "poweroff"})
|
|
return
|
|
}
|
|
prog, _ := json.Marshal(obj{"phase": p.Phase, "pct": p.Pct, "msg": p.Msg, "details": p.Details, "updated": time.Now()})
|
|
_ = a.store.JobProgress(r.Context(), j.ID, prog)
|
|
m, err := a.store.Get(r.Context(), j.MachineID)
|
|
if err != nil {
|
|
fail(w, 500, err)
|
|
return
|
|
}
|
|
next := "continue"
|
|
switch p.State {
|
|
case "done":
|
|
_ = a.store.FinishJob(r.Context(), j.ID, "done", "")
|
|
a.afterJob(r.Context(), m, j, "done", "")
|
|
next = "reboot"
|
|
if j.Kind == "release" {
|
|
next = "poweroff"
|
|
}
|
|
case "failed":
|
|
if p.Error == "" {
|
|
p.Error = p.Msg
|
|
}
|
|
_ = a.store.FinishJob(r.Context(), j.ID, "failed", p.Error)
|
|
a.afterJob(r.Context(), m, j, "failed", p.Error)
|
|
next = "stay"
|
|
}
|
|
writeJSON(w, 200, obj{"job": j.ID, "state": p.State, "next": next})
|
|
}
|
|
|
|
// handlePhoneHome is cloud-init's phone_home from the freshly deployed OS (form-encoded POST).
|
|
func (a *App) handlePhoneHome(w http.ResponseWriter, r *http.Request) {
|
|
sum := sha256.Sum256([]byte(r.PathValue("token")))
|
|
j, err := a.store.JobByToken(r.Context(), hex.EncodeToString(sum[:]))
|
|
if err != nil || j.Kind != "deploy" {
|
|
fail(w, 404, errNotFound)
|
|
return
|
|
}
|
|
_ = r.ParseForm()
|
|
_ = a.store.MarkPhoneHome(r.Context(), j.MachineID)
|
|
detail := fmt.Sprintf("job=%s from %s hostname=%s fqdn=%s", j.ID, r.RemoteAddr, r.Form.Get("hostname"), r.Form.Get("fqdn"))
|
|
if k := strings.TrimSpace(r.Form.Get("pub_key_ed25519")); k != "" {
|
|
detail += " hostkey=" + k
|
|
}
|
|
_ = a.store.AddEvent(r.Context(), j.MachineID, "phone-home", detail)
|
|
log.Printf("phone-home %s: %s", j.MachineID, detail)
|
|
writeJSON(w, 200, obj{"ok": true})
|
|
}
|
|
|
|
// jobStartTimeout: a deploy or release job the machine has not picked up by then
|
|
// is failed, so a lost power command or a machine that did not PXE-boot is
|
|
// visible instead of an endless "pending". Real servers spend 4-5 minutes in
|
|
// POST on a cold start (more after a firmware or memory change), and many
|
|
// machines booting at once share the PXE and image servers: the default leaves
|
|
// room for that; job_start_timeout in config.json changes it.
|
|
const jobStartTimeout = 30 * time.Minute
|
|
|
|
func (a *App) jobStartTimeout() time.Duration {
|
|
if d, err := time.ParseDuration(a.cfg.JobStartTimeout); err == nil && d >= time.Minute {
|
|
return d
|
|
}
|
|
return jobStartTimeout
|
|
}
|
|
|
|
// staleJobs picks the pending jobs created before cutoff.
|
|
func staleJobs(active map[string]*Job, cutoff time.Time) []*Job {
|
|
var out []*Job
|
|
for _, j := range active {
|
|
if j.State == "pending" && j.Kind != "firmware" && j.Created.Before(cutoff) { // firmware jobs have their own watch
|
|
out = append(out, j)
|
|
}
|
|
}
|
|
sort.Slice(out, func(i, k int) bool { return out[i].Created.Before(out[k].Created) })
|
|
return out
|
|
}
|
|
|
|
func (a *App) jobWatchdog() {
|
|
for range time.Tick(time.Minute) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
active, err := a.store.ActiveJobs(ctx)
|
|
if err == nil {
|
|
limit := a.jobStartTimeout()
|
|
for _, j := range staleJobs(active, time.Now().Add(-limit)) {
|
|
m, err := a.store.Get(ctx, j.MachineID)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
msg := fmt.Sprintf("the machine did not start the job within %s: check its power, that it PXE-boots on the provisioning network, and its console for a hung POST (events show the power command's result; job_start_timeout raises the limit)", limit)
|
|
if ok, _ := a.store.FailPendingJob(ctx, j.ID, msg); ok { // not if it reported progress since the snapshot
|
|
a.afterJob(ctx, m, j, "failed", msg)
|
|
}
|
|
}
|
|
}
|
|
cancel()
|
|
}
|
|
}
|