internal/server/api/events.go
Ref: Size: 10.9 KiB History
package api
import (
"database/sql"
"encoding/json"
"errors"
"fmt"
"net/http"
"strconv"
"strings"
"time"
"github.com/a73x/eitri/internal/server/api/types"
"github.com/a73x/eitri/internal/server/store"
"github.com/a73x/eitri/internal/version"
)
// handleDecommissionHost begins graceful host decommission: its VMs are
// tombstoned and reaped by the agent, then the sweeper removes the host and
// frees its CIDR. The agent only learns of the tombstones when its desired-state
// stream is poked, so we poke here exactly as the VM mutation handlers do —
// without it a healthy host never drains and stalls in `decommissioning`.
//
// ?force=true is the escape hatch for dead hardware whose agent will never
// report: it purges the VM rows and removes the host immediately (their compute
// is gone with the box), reclaiming the CIDR without waiting for a drain that
// can never happen.
func (a *API) handleDecommissionHost(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
// Resolve the host's tenant up front: it is both the ownership gate (a
// foreign-tenant host answers exactly like a missing one — no existence
// leak) and the scope for the decommission audit row. Both paths below
// destroy the host row, so its tenant must be read before then.
h, err := a.st.GetHost(id)
switch {
case errors.Is(err, sql.ErrNoRows):
http.Error(w, "host not found", http.StatusNotFound)
return
case err != nil:
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
if !mayActAs(principalFromContext(r), h.Tenant) {
http.Error(w, "host not found", http.StatusNotFound)
return
}
if forceParam(r) {
gone, err := a.st.ForceRemoveHost(id)
switch {
case errors.Is(err, sql.ErrNoRows):
http.Error(w, "host not found", http.StatusNotFound)
return
case err != nil:
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
// Force is the one path that destroys data on purpose, so its audit row
// is the only record of WHAT it destroyed: the volumes are deleted here
// and their ids exist nowhere afterwards. An operator reconciling a lost
// host against backups has this row and nothing else.
a.audit(h.Tenant, "host.decommission", map[string]string{
"host_id": id, "remote": clientIP(r),
"force": "true", "vms_purged": strconv.Itoa(gone.VMsPurged),
"volumes_destroyed": strconv.Itoa(gone.VolumesDestroyed),
"claims_unbound": strconv.Itoa(gone.ClaimsUnbound),
"volume_ids": strings.Join(gone.VolumeIDs, ","),
})
if a.upgrader != nil {
a.upgrader.ClearAgentUpgrade(id)
}
a.hub.Poke(id)
a.notif.notify()
w.WriteHeader(http.StatusOK)
return
}
switch err := a.st.DecommissionHost(id); {
case errors.Is(err, sql.ErrNoRows):
http.Error(w, "host not found", http.StatusNotFound)
return
case err != nil:
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
a.audit(h.Tenant, "host.decommission", map[string]string{"host_id": id, "remote": clientIP(r)})
if a.upgrader != nil {
a.upgrader.ClearAgentUpgrade(id)
}
a.hub.Poke(id)
a.notif.notify()
w.WriteHeader(http.StatusAccepted)
}
// forceParam reports whether the request opts into forced removal via ?force,
// accepting a bare ?force or ?force=true. An absent key, or any other value, is
// false (the default graceful path).
func forceParam(r *http.Request) bool {
q := r.URL.Query()
if !q.Has("force") {
return false
}
v := q.Get("force")
return v == "" || v == "true"
}
// handleRevokeCredential bumps the host's credential generation, revoking its
// outstanding credential WITHOUT rotating the fleet secret. The agent's live
// session is closed by syncsvc within one report tick; the host stays dark
// until the operator re-enrolls it with a fresh join blob.
func (a *API) handleRevokeCredential(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
h, err := a.st.GetHost(id)
switch {
case errors.Is(err, sql.ErrNoRows):
http.Error(w, "host not found", http.StatusNotFound)
return
case err != nil:
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
if !mayActAs(principalFromContext(r), h.Tenant) {
http.Error(w, "host not found", http.StatusNotFound)
return
}
// Audit is written inside the bump transaction (a security action must
// not be able to happen unrecorded).
if _, err := a.st.BumpCredGeneration(id, clientIP(r)); err != nil {
if errors.Is(err, sql.ErrNoRows) {
http.Error(w, "host not found", http.StatusNotFound)
return
}
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
w.WriteHeader(http.StatusNoContent)
}
// parseLimit reads the shared ?limit=N query param (default 100, capped to
// 1..1000). On a bad value it writes the 400 and returns ok=false so the caller
// just returns.
func parseLimit(w http.ResponseWriter, r *http.Request) (int, bool) {
limit := 100
if v := r.URL.Query().Get("limit"); v != "" {
n, err := strconv.Atoi(v)
if err != nil || n < 1 || n > 1000 {
http.Error(w, "limit must be 1..1000", http.StatusBadRequest)
return 0, false
}
limit = n
}
return limit, true
}
// auditRowsToResponse maps store rows to the wire shape, embedding each detail
// as raw JSON and defensively re-marshalling anything that isn't valid JSON so
// the endpoint never emits a malformed body (detail is always a marshaled
// object — see store.AppendAudit callers).
func auditRowsToResponse(rows []store.AuditEntry) []types.AuditEvent {
out := make([]types.AuditEvent, len(rows))
for i, e := range rows {
detail := json.RawMessage(e.Detail)
if !json.Valid(detail) { // defensive: never emit invalid JSON
detail, _ = json.Marshal(e.Detail)
}
out[i] = types.AuditEvent{At: e.At, Action: e.Action, Detail: detail}
}
return out
}
// handleListAudit returns the newest audit rows for the caller's tenant
// (default 100, ?limit=N caps at 1000). Completes the forensic story: rows were
// previously reachable only by opening the SQLite file.
//
// Scoped to the principal's tenant: audit_log now carries a tenant column, and
// each row is filed under the tenant it concerns (rows written before any
// tenant is known — e.g. a denied enroll attempt — fall to store.SystemTenant,
// which no principal holds).
func (a *API) handleListAudit(w http.ResponseWriter, r *http.Request) {
limit, ok := parseLimit(w, r)
if !ok {
return
}
rows, err := a.st.ListAudit(principalFromContext(r).Tenant, limit)
if err != nil {
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
writeJSON(w, http.StatusOK, auditRowsToResponse(rows))
}
// handleListVMEvents returns one VM's lifecycle timeline: the audit rows whose
// detail carries this {id} as "vm_id", newest first (default 100, ?limit=N caps
// at 1000). It deliberately does NOT 404 when the VM row is gone — a reaped VM's
// history (ending in vm.reap) must stay retrievable. Same wire shape as
// handleListAudit.
func (a *API) handleListVMEvents(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
limit, ok := parseLimit(w, r)
if !ok {
return
}
rows, err := a.st.ListVMEvents(principalFromContext(r).Tenant, id, limit)
if err != nil {
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
writeJSON(w, http.StatusOK, auditRowsToResponse(rows))
}
// handleMintStreamTicket issues a one-time short-TTL ticket for the SSE stream
// or console WS (user-authenticated; the ticket is the only thing that ever
// appears in a URL). The ticket is stamped with the caller's tenant so the
// stream/console it later unlocks is scoped to that tenant, not the fleet.
func (a *API) handleMintStreamTicket(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusCreated, types.StreamTicketResponse{Ticket: a.tickets.mint(principalFromContext(r).Tenant)})
}
// handleEvents streams the caller tenant's fleet-view snapshot as Server-Sent
// Events. It subscribes to the central snapshot hub — which reads the store ONCE
// on a 1s tick / desired-state wake and marshals one filtered payload per
// subscribed tenant, fanning each connection its own tenant's bytes — and writes
// each delivered snapshot as an `event: state` frame. The tenant is the one the
// consumed ticket was minted for, so a connection never sees another tenant's
// hosts/VMs. The hub delivers the current snapshot immediately on subscribe, so a
// new client gets initial state without doing its own marshal, and suppresses
// unchanged snapshots so no frame is pushed when nothing changed. A periodic
// comment keeps the connection alive.
func (a *API) handleEvents(w http.ResponseWriter, r *http.Request) {
// Assert streaming support BEFORE consuming the one-time ticket, so a 500 on
// a non-flushing ResponseWriter doesn't burn the client's ticket.
flusher, ok := w.(http.Flusher)
if !ok {
http.Error(w, "streaming unsupported", http.StatusInternalServerError)
return
}
tenant, ok := a.tickets.consume(r.URL.Query().Get("ticket"))
if !ok {
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
snaps, unsub := a.snap.subscribe(tenant)
defer unsub()
heartbeat := time.NewTicker(15 * time.Second)
defer heartbeat.Stop()
sendState := func(payload []byte) bool {
if _, err := fmt.Fprintf(w, "event: state\ndata: %s\n\n", payload); err != nil {
return false
}
flusher.Flush()
return true
}
ctx := r.Context()
for {
select {
case <-ctx.Done():
return
case payload := <-snaps:
if !sendState(payload) {
return
}
case <-heartbeat.C:
if _, err := fmt.Fprint(w, ": ping\n\n"); err != nil {
return
}
flusher.Flush()
}
}
}
// marshalSnapshots reads the fleet snapshot ONCE and JSON-encodes one payload
// per requested tenant, each filtered to that tenant's hosts/VMs. The store
// reads happen in one transaction (store.Snapshot) so hosts, allocation and VMs
// can never mix state from two different epochs; the registry states are fetched
// ONCE over the union of hosts (registry.Get deep-clones the whole host report
// per call) and shared across every tenant's marshal. The hub calls this with
// the distinct set of currently-subscribed tenants, so N connected tenants cost
// one store read and N marshals per tick — not N store reads.
func (a *API) marshalSnapshots(tenants []string) (map[string][]byte, error) {
hosts, alloc, vms, err := a.st.Snapshot()
if err != nil {
return nil, err
}
states := a.fetchStates(hostIDs(hosts), vmHostIDs(vms))
latest := a.latestVersion()
out := make(map[string][]byte, len(tenants))
for _, tenant := range tenants {
p := Principal{Tenant: tenant}
th := filterHosts(p, hosts)
tv := filterVMs(p, vms)
b, err := json.Marshal(types.StateSnapshot{
Hosts: a.buildHostResponses(th, alloc, states),
VMs: a.buildVMResponses(tv, states),
ServerVersion: version.Version,
LatestVersion: latest,
})
if err != nil {
return nil, err
}
out[tenant] = b
}
return out, nil
}