From 17ed0192c13c555a204c17eab64162031d3fcf85 Mon Sep 17 00:00:00 2001 From: mrhid6 Date: Thu, 17 Sep 2026 08:27:34 +0000 Subject: [PATCH] feat(monitors): public heartbeat ping endpoints, header token, log masking and sweeper --- deploy/docker/nginx/vantage.conf | 13 ++ ...6-09-17-metric-alerts-heartbeats-design.md | 8 +- server/cmd/main.go | 14 +- server/internal/api/handlers.go | 8 + server/internal/api/heartbeats.go | 158 ++++++++++++++++++ server/internal/api/heartbeats_test.go | 70 ++++++++ server/internal/api/monitors.go | 10 ++ server/internal/api/scopes.go | 17 +- server/internal/api/serverscope.go | 3 + server/internal/metricsched/scheduler.go | 40 +++++ 10 files changed, 329 insertions(+), 12 deletions(-) create mode 100644 server/internal/api/heartbeats.go create mode 100644 server/internal/api/heartbeats_test.go create mode 100644 server/internal/metricsched/scheduler.go diff --git a/deploy/docker/nginx/vantage.conf b/deploy/docker/nginx/vantage.conf index 65b2cb3..729ef64 100644 --- a/deploy/docker/nginx/vantage.conf +++ b/deploy/docker/nginx/vantage.conf @@ -12,6 +12,17 @@ map $http_upgrade $connection_upgrade { '' close; } +# Heartbeat ping URLs carry a credential. Log them with the token replaced; +# the header form (X-Vantage-Token) is never logged by this format. +map $request_uri $vantage_log_uri { + "~^/public/hb/(?!start(?:[/?]|$)|fail(?:[/?]|$))[^/?]+(?.*)$" "/public/hb/***$hb_rest"; + default $request_uri; +} + +log_format vantage '$remote_addr - $remote_user [$time_local] ' + '"$request_method $vantage_log_uri $server_protocol" ' + '$status $body_bytes_sent "$http_referer" "$http_user_agent"'; + upstream vantage_server { server server:8080; keepalive 16; @@ -27,6 +38,8 @@ server { listen [::]:80; server_name _; + access_log /var/log/nginx/access.log vantage; + # Step imports and licence pastes are the largest request bodies. client_max_body_size 10m; diff --git a/docs/superpowers/specs/2026-09-17-metric-alerts-heartbeats-design.md b/docs/superpowers/specs/2026-09-17-metric-alerts-heartbeats-design.md index ef3fde7..cc7046e 100644 --- a/docs/superpowers/specs/2026-09-17-metric-alerts-heartbeats-design.md +++ b/docs/superpowers/specs/2026-09-17-metric-alerts-heartbeats-design.md @@ -114,9 +114,11 @@ Evaluators are pure functions `func(kind string, t models.MonitorTarget, srv mod Registered outside `/api`, unauthenticated, no scope declarations needed: -- `GET|POST /hb/:token`: success ping -- `GET|POST /hb/:token/start`: run started -- `GET|POST /hb/:token/fail`: run failed +- `GET|POST /public/hb/:token`: success ping +- `GET|POST /public/hb/:token/start`: run started +- `GET|POST /public/hb/:token/fail`: run failed + +Mounted under /public because every deployment already routes that prefix to the server. The token may instead be sent in the `X-Vantage-Token` header to `/public/hb`, `/public/hb/start` or `/public/hb/fail`; a URL token wins when both are present. The server's request log and the bundled nginx access log mask the URL token. Lookup is by SHA-256 of the token. Unknown token or disabled monitor gives 404. Rate limit is one accepted request per second per token (in-process, Redis-backed if a limiter helper already exists); excess gives 429. The response body is `OK`. Request bodies over 1 KB are truncated; only `/fail` uses the body. diff --git a/server/cmd/main.go b/server/cmd/main.go index 74f0f18..4f86806 100644 --- a/server/cmd/main.go +++ b/server/cmd/main.go @@ -3,6 +3,7 @@ package main import ( "context" "errors" + "fmt" "log" "net/http" "os" @@ -23,6 +24,7 @@ import ( "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/db" grpcserver "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/grpc" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/mcp" + "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/metricsched" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/models" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/monitorsched" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/patchsched" @@ -259,6 +261,7 @@ func serve() { services.StartAuditSweeper(jobCtx) services.StartReaper(jobCtx) monitorsched.Start(jobCtx) + metricsched.Start(jobCtx) workflowsched.Start(jobCtx, workflowsched.Deps{ TriggerWorkflow: services.TriggerWorkflow, LogEvent: services.LogEvent, @@ -305,7 +308,16 @@ func serve() { log.Fatalf("trusted proxies: %v", err) } r.Use(gin.Recovery()) - r.Use(gin.LoggerWithConfig(gin.LoggerConfig{SkipPaths: []string{"/api/console/tunnel"}})) + r.Use(gin.LoggerWithConfig(gin.LoggerConfig{ + SkipPaths: []string{"/api/console/tunnel"}, + // Heartbeat URLs carry a credential; the request log must not. + Formatter: func(p gin.LogFormatterParams) string { + return fmt.Sprintf("[GIN] %v | %3d | %13v | %15s | %-7s %#v\n%s", + p.TimeStamp.Format("2006/01/02 - 15:04:05"), + p.StatusCode, p.Latency, p.ClientIP, p.Method, + api.MaskLogPath(p.Path), p.ErrorMessage) + }, + })) r.Use(corsMiddleware()) services.SetStatusRedis(auth.Redis()) diff --git a/server/internal/api/handlers.go b/server/internal/api/handlers.go index f3d7a14..76f2f37 100644 --- a/server/internal/api/handlers.go +++ b/server/internal/api/handlers.go @@ -71,6 +71,14 @@ func RegisterRoutes(r *gin.Engine) { // rather than under /api precisely so that none of those apply. r.GET("/public/status/:pageId", RateLimitPublicStatus(), getPublicStatusPage) + // Ping endpoints for heartbeat monitors. The token is in the URL or the + // X-Vantage-Token header; see resolveHeartbeat for the shapes. Rate + // limiting is per token inside the handler. + for _, p := range []string{"/public/hb", "/public/hb/:a", "/public/hb/:a/:b"} { + r.GET(p, handleHeartbeat) + r.POST(p, handleHeartbeat) + } + apiGroup := r.Group("/api") apiGroup.Use(auth.Middleware()) // Scope enforcement sits between authentication and the licence gate, and diff --git a/server/internal/api/heartbeats.go b/server/internal/api/heartbeats.go new file mode 100644 index 0000000..c93b166 --- /dev/null +++ b/server/internal/api/heartbeats.go @@ -0,0 +1,158 @@ +package api + +import ( + "errors" + "io" + "log" + "net/http" + "strconv" + "strings" + "time" + + "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/auth" + "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/services" + "github.com/gin-gonic/gin" +) + +// HeartbeatTokenHeader carries the ping token for callers who want it out of +// URLs, and so out of their own proxies' access logs. +const HeartbeatTokenHeader = "X-Vantage-Token" + +const heartbeatPathPrefix = "/public/hb/" + +func heartbeatKind(seg string) string { + switch seg { + case "": + return services.HeartbeatPing + case "start": + return services.HeartbeatStart + case "fail": + return services.HeartbeatFail + } + return "" +} + +// resolveHeartbeat maps the three route shapes onto a token and a kind. a and b +// are the first and second path segments after /public/hb. A token in the URL +// wins over the header, so a URL copied from the UI behaves the same no matter +// what headers a client adds. +func resolveHeartbeat(header, a, b string) (string, string, bool) { + var token, kindSeg string + switch { + // a is a URL token unless a header is present, no second segment follows + // and a is itself a kind word (the header routes /public/hb/start|fail). + case a != "" && (header == "" || b != "" || heartbeatKind(a) == ""): + token, kindSeg = a, b + case header != "": + token, kindSeg = header, a + default: + return "", "", false + } + kind := heartbeatKind(kindSeg) + if token == "" || kind == "" { + return "", "", false + } + return token, kind, true +} + +// MaskLogPath hides a heartbeat token in a request path before it is logged. +// "start" and "fail" directly under the prefix are the header-token routes and +// carry no secret, so they are left readable. +func MaskLogPath(path string) string { + rest, found := strings.CutPrefix(path, heartbeatPathPrefix) + if !found { + return path + } + end := strings.IndexAny(rest, "/?") + if end < 0 { + end = len(rest) + } + seg := rest[:end] + if seg == "" || seg == "start" || seg == "fail" { + return path + } + return heartbeatPathPrefix + "***" + rest[end:] +} + +// heartbeatAllowed admits one request per token per second. A cron job pinging +// in a loop should not become a write per request, and a leaked URL should not +// be a way to hammer Mongo. Like the status page limiter it allows when Redis +// is down: a missed ping pages someone. It runs inside the handler rather than +// as middleware because the token may come from a header. +func heartbeatAllowed(c *gin.Context, token string) bool { + rdb := auth.Redis() + if rdb == nil { + return true + } + key := "vantage:hbrl:" + services.HashHeartbeatToken(token) + ":" + strconv.FormatInt(time.Now().Unix(), 10) + count, err := rdb.Incr(c.Request.Context(), key).Result() + if err != nil { + return true + } + if count == 1 { + rdb.Expire(c.Request.Context(), key, 2*time.Second) + } + return count <= 1 +} + +// handleHeartbeat records a push from a job. It is mounted on the gin root +// under /public for the same reasons as the status page (see +// getPublicStatusPage): no session, no token, no licence gate, and /public is +// already routed to this server by every deployment. +// +// Unknown token, disabled monitor and an unresolvable path all answer the same +// 404. +func handleHeartbeat(c *gin.Context) { + token, kind, ok := resolveHeartbeat(c.GetHeader(HeartbeatTokenHeader), c.Param("a"), c.Param("b")) + if !ok { + c.String(http.StatusNotFound, "not found") + return + } + if !heartbeatAllowed(c, token) { + c.Header("Retry-After", "1") + c.String(http.StatusTooManyRequests, "too many requests") + return + } + var body string + if kind == services.HeartbeatFail && c.Request.Body != nil { + b, _ := io.ReadAll(io.LimitReader(c.Request.Body, services.MaxHeartbeatBody)) + body = string(b) + } + err := services.RecordHeartbeat(token, kind, body, time.Now()) + if errors.Is(err, services.ErrHeartbeatNotFound) { + c.String(http.StatusNotFound, "not found") + return + } + if err != nil { + log.Printf("heartbeat: %v", err) + c.String(http.StatusInternalServerError, "error") + return + } + c.String(http.StatusOK, "OK") +} + +// rotateHeartbeatToken godoc +// +// @Summary Rotate a heartbeat monitor's ping token +// @Description Issues a new token and invalidates the old ping URL immediately. The token is returned only in this response. +// @Tags monitors +// @Produce json +// @Param id path string true "Monitor ID" +// @Success 200 {object} object{heartbeat_token=string} +// @Failure 404 {object} ErrorResponse +// @Failure 500 {object} ErrorResponse +// @Security cookieAuth +// @Security bearerAuth +// @Router /monitors/{id}/rotate-token [post] +func rotateHeartbeatToken(c *gin.Context) { + tok, err := services.RotateHeartbeatToken(auth.InstanceID(c), c.Param("id")) + if errors.Is(err, services.ErrHeartbeatNotFound) { + c.JSON(http.StatusNotFound, gin.H{"error": "heartbeat monitor not found"}) + return + } + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"heartbeat_token": tok}) +} diff --git a/server/internal/api/heartbeats_test.go b/server/internal/api/heartbeats_test.go new file mode 100644 index 0000000..84b5906 --- /dev/null +++ b/server/internal/api/heartbeats_test.go @@ -0,0 +1,70 @@ +package api + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/gin-gonic/gin" +) + +func TestResolveHeartbeat(t *testing.T) { + cases := []struct { + name, header, a, b string + token, kind string + ok bool + }{ + {"url ping", "", "tok", "", "tok", "ping", true}, + {"url start", "", "tok", "start", "tok", "start", true}, + {"url fail", "", "tok", "fail", "tok", "fail", true}, + {"url bad kind", "", "tok", "explode", "", "", false}, + {"header ping", "htok", "", "", "htok", "ping", true}, + {"header start", "htok", "start", "", "htok", "start", true}, + {"header fail", "htok", "fail", "", "htok", "fail", true}, + {"url token wins over header", "htok", "tok", "", "tok", "ping", true}, + {"url token and kind win over header", "htok", "tok", "fail", "tok", "fail", true}, + {"nothing", "", "", "", "", "", false}, + // Without a header, /public/hb/start is a token called "start": it + // resolves, and the lookup answers 404 like any unknown token. + {"bare start without header", "", "start", "", "start", "ping", true}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + tok, kind, ok := resolveHeartbeat(c.header, c.a, c.b) + if tok != c.token || kind != c.kind || ok != c.ok { + t.Fatalf("got (%q,%q,%v), want (%q,%q,%v)", tok, kind, ok, c.token, c.kind, c.ok) + } + }) + } +} + +// A ping URL is a credential. Anything that logs request paths must see a +// masked one. +func TestMaskLogPath(t *testing.T) { + cases := map[string]string{ + "/public/hb/abc123": "/public/hb/***", + "/public/hb/abc123/fail": "/public/hb/***/fail", + "/public/hb/abc123?x=1": "/public/hb/***?x=1", + "/public/hb/start": "/public/hb/start", + "/public/hb/fail": "/public/hb/fail", + "/public/hb": "/public/hb", + "/public/status/page": "/public/status/page", + "/api/monitors/abc/uptime": "/api/monitors/abc/uptime", + } + for in, want := range cases { + if got := MaskLogPath(in); got != want { + t.Errorf("MaskLogPath(%q) = %q, want %q", in, got, want) + } + } +} + +func TestHeartbeatUnresolvableIs404(t *testing.T) { + gin.SetMode(gin.TestMode) + r := gin.New() + r.POST("/public/hb/:a/:b", handleHeartbeat) + w := httptest.NewRecorder() + r.ServeHTTP(w, httptest.NewRequest(http.MethodPost, "/public/hb/abc/explode", nil)) + if w.Code != http.StatusNotFound { + t.Fatalf("code = %d, want 404", w.Code) + } +} diff --git a/server/internal/api/monitors.go b/server/internal/api/monitors.go index 7428bda..67393ce 100644 --- a/server/internal/api/monitors.go +++ b/server/internal/api/monitors.go @@ -1,6 +1,7 @@ package api import ( + "errors" "net/http" "strconv" "time" @@ -21,6 +22,7 @@ func registerMonitorRoutes(g *gin.RouterGroup) { g.GET("/monitors/:id/incidents", getMonitorIncidents) g.GET("/monitors/:id/uptime", getMonitorUptime) g.GET("/monitors/:id/samples", getMonitorSamples) + g.POST("/monitors/:id/rotate-token", rotateHeartbeatToken) } // listMonitors godoc @@ -85,6 +87,10 @@ func createMonitor(c *gin.Context) { } created, err := services.CreateMonitor(auth.InstanceID(c), &m, auth.ServerScope(c)) if err != nil { + if errors.Is(err, services.ErrInvalidMonitor) { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } @@ -187,6 +193,10 @@ func updateMonitor(c *gin.Context) { return } if err := services.UpdateMonitor(auth.InstanceID(c), c.Param("id"), upd, auth.ServerScope(c)); err != nil { + if errors.Is(err, services.ErrInvalidMonitor) { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } diff --git a/server/internal/api/scopes.go b/server/internal/api/scopes.go index e7ad2e3..2930339 100644 --- a/server/internal/api/scopes.go +++ b/server/internal/api/scopes.go @@ -99,14 +99,15 @@ var routeScopes = map[string]string{ "GET /api/runs/:runId/servers/:serverId/logs/stream": "workflows:read", // Monitor and incident routes, registered by registerMonitorRoutes. - "GET /api/monitors": "monitors:read", - "POST /api/monitors": "monitors:write", - "GET /api/monitors/:id": "monitors:read", - "PUT /api/monitors/:id": "monitors:write", - "DELETE /api/monitors/:id": "monitors:write", - "GET /api/monitors/:id/incidents": "monitors:read", - "GET /api/monitors/:id/uptime": "monitors:read", - "GET /api/monitors/:id/samples": "monitors:read", + "GET /api/monitors": "monitors:read", + "POST /api/monitors": "monitors:write", + "GET /api/monitors/:id": "monitors:read", + "PUT /api/monitors/:id": "monitors:write", + "DELETE /api/monitors/:id": "monitors:write", + "GET /api/monitors/:id/incidents": "monitors:read", + "GET /api/monitors/:id/uptime": "monitors:read", + "GET /api/monitors/:id/samples": "monitors:read", + "POST /api/monitors/:id/rotate-token": "monitors:write", // Channel routes, registered by registerChannelRoutes. Channels exist to // serve alerts, so they share the monitors scope rather than getting their diff --git a/server/internal/api/serverscope.go b/server/internal/api/serverscope.go index 166f4e8..2f6141a 100644 --- a/server/internal/api/serverscope.go +++ b/server/internal/api/serverscope.go @@ -292,6 +292,9 @@ var serverScopedRoutes = map[string]scopeDecl{ // then act on. "DELETE /api/monitors/:id": fleetWide, + // Rotating a ping token touches no server and returns only the token. + "POST /api/monitors/:id/rotate-token": fleetWide, + // A monitor's incidents, uptime rollups and recent samples are all about // the monitored endpoint - status, latency, timestamps - and carry no // server identifier at all; the runner is a field of the monitor diff --git a/server/internal/metricsched/scheduler.go b/server/internal/metricsched/scheduler.go new file mode 100644 index 0000000..24da575 --- /dev/null +++ b/server/internal/metricsched/scheduler.go @@ -0,0 +1,40 @@ +// Package metricsched sweeps passive monitors - heartbeats and, from phase 2, +// metric monitors - on a fixed tick. Nothing here runs a check: it reads what +// pings and agents already delivered and decides what is overdue or breaching. +package metricsched + +import ( + "context" + "log" + "time" + + "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/services" +) + +const tick = 30 * time.Second + +func Start(ctx context.Context) { + go func() { + t := time.NewTicker(tick) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case now := <-t.C: + sweep(now) + } + } + }() +} + +// sweep recovers per sweep so one bad document cannot stop alerting for the +// whole instance until the next deploy. +func sweep(now time.Time) { + defer func() { + if r := recover(); r != nil { + log.Printf("metricsched: sweep panic: %v", r) + } + }() + services.SweepHeartbeats(now) +}