Vantage/vantage-app
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(monitors): public heartbeat ping endpoints, header token, log masking and sweeper

Browse Source
This commit is contained in:
mrhid6
2026-09-17 08:27:34 +00:00
parent 839d716a47
commit 17ed0192c1
10 changed files with 329 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/docker/nginx/vantage.conf
+13
View File
@@ -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(?:[/?]|$))[^/?]+(?<hb_rest>.*)$" "/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;
docs/superpowers/specs/2026-09-17-metric-alerts-heartbeats-design.md
+5 -3
View File
@@ -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.
server/cmd/main.go
+13 -1
View File
@@ -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())
server/internal/api/handlers.go
+8
View File
@@ -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
server/internal/api/heartbeats.go
+158
View File
@@ -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})
}
server/internal/api/heartbeats_test.go
+70
View File
@@ -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)
}
}
server/internal/api/monitors.go
+10
View File
@@ -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
}
server/internal/api/scopes.go
+9 -8
View File
@@ -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
server/internal/api/serverscope.go
+3
View File
@@ -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
server/internal/metricsched/scheduler.go
+40
View File
@@ -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)
}