crossbar/internal/admin/admin.go

// Package admin serves the operator's view of crossbar: the hosts view with
// slots and drain state, the routes view with leases and pins, the pin/release
// and drain controls, usage accounting, and Prometheus metrics, all under
// /_crossbar/.
package admin

import (
	"net/http"
	"sort"
	"time"

	"git.wntrmute.dev/kyle/crossbar/internal/config"
	"git.wntrmute.dev/kyle/crossbar/internal/health"
	"git.wntrmute.dev/kyle/crossbar/internal/lease"
	"git.wntrmute.dev/kyle/crossbar/internal/limiter"
	"git.wntrmute.dev/kyle/crossbar/internal/store"
)

// Hosts is what the admin handler needs from the health table.
type Hosts interface {
	All() map[string]health.Status
}

// Drainer is what the admin handler needs to steer draining; *proxy.Hosts
// satisfies it.
type Drainer interface {
	Draining(name string) bool
	SetDraining(name string, on bool)
}

// HostView is one host's row in the hosts view.
type HostView struct {
	Healthy   bool     `json:"healthy"`
	Loaded    []string `json:"loaded"`  // never null
	LastOK    string   `json:"last_ok"` // RFC 3339 UTC or ""
	LastErr   string   `json:"last_err"`
	FreeSlots int      `json:"free_slots"` // lim.FreeSlots(host)
	InFlight  int      `json:"in_flight"`  // sum over the host's configured models
	Queued    int      `json:"queued"`     // same
	Draining  bool     `json:"draining"`
}

// LeaseView is one lease's row in a route's leases.
type LeaseView struct {
	FP       string `json:"fp"`
	Model    string `json:"model"`
	Host     string `json:"host"`
	State    string `json:"state"`
	Created  string `json:"created"`   // RFC 3339 UTC
	LastUsed string `json:"last_used"` // RFC 3339 UTC
}

// RouteView is one route's row in the routes view.
type RouteView struct {
	Hosts        []string    `json:"hosts"`
	DefaultModel string      `json:"default_model"`
	Pinned       string      `json:"pinned"` // "" when not pinned
	Leases       []LeaseView `json:"leases"` // never null
}

// handler implements the operator's endpoints under /_crossbar/.
type handler struct {
	cfg *config.Config
	h   Hosts
	lt  *lease.Table
	lim *limiter.Limiter
	st  *store.Store
	d   Drainer
}

// Handler builds the operator's HTTP handler.
func Handler(cfg *config.Config, h Hosts, lt *lease.Table, lim *limiter.Limiter, st *store.Store, d Drainer) http.Handler {
	hx := &handler{cfg: cfg, h: h, lt: lt, lim: lim, st: st, d: d}
	mux := http.NewServeMux()
	mux.HandleFunc("/_crossbar/hosts", hx.hostsGet)
	mux.HandleFunc("/_crossbar/hosts/{host}", hx.hostDrain)
	mux.HandleFunc("/_crossbar/routes", hx.routesGet)
	mux.HandleFunc("/_crossbar/routes/{route}", hx.routePin)
	mux.HandleFunc("/_crossbar/usage", hx.usageGet)
	mux.HandleFunc("/_crossbar/metrics", hx.metricsGet)
	mux.HandleFunc("/_crossbar/", hx.unknown)
	return mux
}

func (hx *handler) hostsGet(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		wrongMethod(w, "GET")
		return
	}
	writeJSON(w, http.StatusOK, hx.hostViews())
}

// hostViews builds every host's row, keyed by host name.
func (hx *handler) hostViews() map[string]HostView {
	all := hx.h.All()
	out := make(map[string]HostView, len(all))
	for name, s := range all {
		out[name] = hx.hostView(name, s)
	}
	return out
}

// hostView builds one host's row: concurrency from the limiter summed over the
// models the host serves, draining from the drainer, and the health snapshot.
func (hx *handler) hostView(name string, s health.Status) HostView {
	inflight, queued := 0, 0
	for _, m := range configuredModels(hx.cfg, name) {
		inflight += hx.lim.InFlight(name, m)
		queued += hx.lim.Queued(name, m)
	}
	loaded := s.Loaded
	if loaded == nil {
		loaded = []string{}
	}
	lastOK := ""
	if !s.LastOK.IsZero() {
		lastOK = s.LastOK.UTC().Format(time.RFC3339)
	}
	return HostView{
		Healthy:   s.Healthy,
		Loaded:    loaded,
		LastOK:    lastOK,
		LastErr:   s.LastErr,
		FreeSlots: hx.lim.FreeSlots(name),
		InFlight:  inflight,
		Queued:    queued,
		Draining:  hx.d.Draining(name),
	}
}

// configuredModels returns the sorted model ids the host serves, or nil when
// the host is unknown.
func configuredModels(cfg *config.Config, name string) []string {
	h, ok := cfg.Hosts[name]
	if !ok {
		return nil
	}
	models := make([]string, 0, len(h.Models))
	for m := range h.Models {
		models = append(models, m)
	}
	sort.Strings(models)
	return models
}

func (hx *handler) routesGet(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		wrongMethod(w, "GET")
		return
	}
	snap := hx.lt.Snapshot()
	views := make(map[string]RouteView, len(hx.cfg.Routes))
	for name, route := range hx.cfg.Routes {
		hosts := make([]string, len(route.Hosts))
		copy(hosts, route.Hosts)
		views[name] = hx.routeView(name, hosts, route.DefaultModel, snap)
	}
	writeJSON(w, http.StatusOK, views)
}

// routeView builds one route's row: the pinned host (empty if none) and the
// active leases on it, in snapshot order.
func (hx *handler) routeView(route string, hosts []string, defaultModel string, snap []lease.Lease) RouteView {
	leases := make([]LeaseView, 0)
	for _, l := range snap {
		if l.Route == route {
			leases = append(leases, leaseView(l))
		}
	}
	return RouteView{
		Hosts:        hosts,
		DefaultModel: defaultModel,
		Pinned:       hx.lt.Pinned(route),
		Leases:       leases,
	}
}

// leaseView maps a lease to its operator view.
func leaseView(l lease.Lease) LeaseView {
	return LeaseView{
		FP:       l.FP,
		Model:    l.Model,
		Host:     l.Host,
		State:    string(l.State),
		Created:  formatTime(l.Created),
		LastUsed: formatTime(l.LastUsed),
	}
}

// formatTime renders t as RFC 3339 in UTC, or "" for the zero time.
func formatTime(t time.Time) string {
	if t.IsZero() {
		return ""
	}
	return t.UTC().Format(time.RFC3339)
}