llmgw/internal/web/node_page.go

package web

import (
	"context"
	"fmt"
	"log/slog"
	"net/http"
	"sort"
	"strings"
	"time"

	"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
)

// nodePageData is one node: what it is, how busy it is now, and how busy it
// has been.
type nodePageData struct {
	Nav             Nav
	OnAdminListener bool
	Node            adminapi.NodeView
	Engines         []nodeEngine
	Charts          *nodeCharts
	HasHistory      bool
	Back, Logs      string
	// Self is this page's address, for the in-place refresh.
	Self         string
	Version, Now string
	RefreshSec   int
}

// nodeEngine is one engine running on the node, and its share of it.
type nodeEngine struct {
	Replica, Model, Shard, State string
	CPUCores                     float64
	MemoryBytes                  int64
	// CPUShare is its cores as a percentage of the machine, for the bar.
	CPUShare float64
}

// nodeCharts are the node's recent history, from the metrics store.
type nodeCharts struct {
	CPU, Memory, Load, EngineCPU, EngineMemory, Queue TimeSeries
	Window                                            string
	Err                                               string
}

// nodePanels are the charts, each a query scoped to this node. %s is the
// node's ID, quoted.
var nodePanels = []struct {
	expr, nameBy, unit string
	into               func(*nodeCharts, TimeSeries)
}{
	// Divided to a fraction because that is what the "%" unit draws.
	{`llmgw_node_cpu_utilization_percent{node=%s} / 100`, "node", "%",
		func(c *nodeCharts, t TimeSeries) { c.CPU = t }},
	{`llmgw_node_memory_used_bytes{node=%s}`, "node", "bytes",
		func(c *nodeCharts, t TimeSeries) { c.Memory = t }},
	{`llmgw_node_load{node=%s}`, "window", "",
		func(c *nodeCharts, t TimeSeries) { c.Load = t }},
	{`sum by (model) (llmgw_shard_cpu_cores{node=%s})`, "model", "",
		func(c *nodeCharts, t TimeSeries) { c.EngineCPU = t }},
	{`sum by (model) (llmgw_shard_memory_bytes{node=%s})`, "model", "bytes",
		func(c *nodeCharts, t TimeSeries) { c.EngineMemory = t }},
}

// NodePage shows one node. Its charts come from the metrics store when one
// is configured; the current figures come from the node's last heartbeat
// either way.
func NodePage(src Source, history History, window time.Duration, onAdminListener bool,
	version string, log *slog.Logger,
) http.HandlerFunc {
	if window <= 0 {
		window = time.Hour
	}
	return func(w http.ResponseWriter, r *http.Request) {
		id := r.PathValue("id")
		d := nodePageData{
			Nav: navFrom(r, "cluster"), OnAdminListener: onAdminListener,
			Back: "/admin", Logs: "/admin/nodes/" + id + "/logs", Self: "/admin/nodes/" + id,
			HasHistory: history != nil, Version: version, RefreshSec: 15,
			Now: time.Now().UTC().Format(time.RFC3339),
		}
		if onAdminListener {
			d.Back, d.Logs, d.Self = "/", "/nodes/"+id+"/logs", "/nodes/"+id
		}
		found := false
		for _, n := range src.Nodes() {
			if n.ID == id {
				d.Node, found = n, true
				break
			}
		}
		// Only a node the control plane knows reaches a query below, which
		// is what keeps the ID a name rather than something typed into PromQL.
		if !found {
			http.NotFound(w, r)
			return
		}

		var replicas []string
		for _, rep := range src.Replicas("") {
			for _, sh := range rep.Shards {
				if sh.Node != id {
					continue
				}
				e := nodeEngine{Replica: rep.ID, Model: rep.Model, Shard: sh.ID, State: sh.State,
					CPUCores: sh.CPUCores, MemoryBytes: sh.MemoryBytes}
				if l := d.Node.Load; l != nil && l.CPUs > 0 {
					e.CPUShare = sh.CPUCores * 100 / float64(l.CPUs)
				}
				d.Engines = append(d.Engines, e)
				replicas = append(replicas, rep.ID)
			}
		}
		sort.Slice(d.Engines, func(i, j int) bool { return d.Engines[i].CPUCores > d.Engines[j].CPUCores })

		if history != nil {
			d.Charts = loadNodeCharts(r.Context(), history, id, replicas, window, log)
		}
		noStore(w)
		render(w, "node", d)
	}
}

func loadNodeCharts(ctx context.Context, h History, node string, replicas []string,
	window time.Duration, log *slog.Logger,
) *nodeCharts {
	step := window / 60
	out := &nodeCharts{Window: windowWord(window)}
	quoted := fmt.Sprintf("%q", node)
	for _, p := range nodePanels {
		series, err := h.RangeQuery(ctx, fmt.Sprintf(p.expr, quoted), p.nameBy, window, step)
		if err != nil {
			if log != nil {
				log.Warn("could not read a node's history", "node", node, "err", err)
			}
			out.Err = err.Error()
			return out
		}
		p.into(out, LineChart(series, p.unit, 4))
	}
	// The engines' queues, for the replicas that are here now.
	if len(replicas) > 0 {
		re := make([]string, len(replicas))
		for i, r := range replicas {
			re[i] = regexpQuote(r)
		}
		expr := fmt.Sprintf(`llmgw_endpoint_waiting{replica=~%q}`, strings.Join(re, "|"))
		if series, err := h.RangeQuery(ctx, expr, "replica", window, step); err == nil {
			out.Queue = LineChart(series, "", 4)
		}
	}
	return out
}

// regexpQuote escapes what a replica ID could carry into a PromQL regex.
// IDs are letters, digits and dashes; this is for the day one is not.
func regexpQuote(s string) string {
	var b strings.Builder
	for _, r := range s {
		if strings.ContainsRune(`\.+*?()|[]{}^$`, r) {
			b.WriteRune('\\')
		}
		b.WriteRune(r)
	}
	return b.String()
}