diff --git a/server/internal/models/monitor.go b/server/internal/models/monitor.go index 1dd7920..87da815 100644 --- a/server/internal/models/monitor.go +++ b/server/internal/models/monitor.go @@ -92,6 +92,8 @@ type Monitor struct { // HeartbeatTokenHash is the SHA-256 of the ping token. The token itself is // shown once, on create or rotate, and never stored. HeartbeatTokenHash string `bson:"heartbeat_token_hash,omitempty" json:"-"` + // HeartbeatToken is the plaintext token, set only on the create response. + HeartbeatToken string `bson:"-" json:"heartbeat_token,omitempty"` } type Incident struct { diff --git a/server/internal/services/heartbeats.go b/server/internal/services/heartbeats.go new file mode 100644 index 0000000..36f07ac --- /dev/null +++ b/server/internal/services/heartbeats.go @@ -0,0 +1,213 @@ +package services + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "errors" + "fmt" + "log" + "strings" + "time" + + "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/db" + "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/models" + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" +) + +const ( + HeartbeatPing = "ping" + HeartbeatStart = "start" + HeartbeatFail = "fail" +) + +// MaxHeartbeatBody bounds what a /fail request can put into an incident cause +// and a notification. A job's stderr can be megabytes; the first kilobyte is +// what a human reads. +const MaxHeartbeatBody = 1024 + +const defaultHeartbeatGraceSec = 300 + +var ErrHeartbeatNotFound = errors.New("heartbeat not found") + +// ErrInvalidMonitor marks a validation failure, which handlers answer with 400. +var ErrInvalidMonitor = errors.New("invalid monitor") + +func NewHeartbeatToken() (string, string, error) { + raw := make([]byte, 24) + if _, err := rand.Read(raw); err != nil { + return "", "", err + } + tok := base64.RawURLEncoding.EncodeToString(raw) + return tok, HashHeartbeatToken(tok), nil +} + +func HashHeartbeatToken(token string) string { + sum := sha256.Sum256([]byte(token)) + return hex.EncodeToString(sum[:]) +} + +func validateHeartbeat(t *models.MonitorTarget) error { + if t.PeriodSec < 60 { + return fmt.Errorf("%w: period_sec must be at least 60", ErrInvalidMonitor) + } + if t.GraceSec < 0 { + return fmt.Errorf("%w: grace_sec must not be negative", ErrInvalidMonitor) + } + if t.GraceSec == 0 { + t.GraceSec = defaultHeartbeatGraceSec + } + return nil +} + +func failMessage(body string) string { + body = strings.TrimSpace(body) + if len(body) > MaxHeartbeatBody { + body = body[:MaxHeartbeatBody] + } + if body == "" { + return "reported failure" + } + return "reported failure: " + body +} + +// heartbeatVerdict decides whether a heartbeat is overdue. A heartbeat that has +// never pinged is never down: the clock starts at the first ping, so creating +// one before the job is deployed does not page anyone. +func heartbeatVerdict(m models.Monitor, now time.Time) (bool, string) { + grace := time.Duration(m.Target.GraceSec) * time.Second + if m.State.StartedAt != nil && now.After(m.State.StartedAt.Add(grace)) { + return true, fmt.Sprintf("started %s, never finished", m.State.StartedAt.UTC().Format(time.RFC3339)) + } + if m.State.LastPingAt == nil { + return false, "" + } + deadline := m.State.LastPingAt.Add(time.Duration(m.Target.PeriodSec)*time.Second + grace) + if now.After(deadline) { + return true, fmt.Sprintf("no ping since %s", m.State.LastPingAt.UTC().Format(time.RFC3339)) + } + return false, "" +} + +// RecordHeartbeat applies one ping. The state change is a single +// FindOneAndUpdate on the token hash, so two pings racing each other cannot +// both read the old state and lose one of the writes. +func RecordHeartbeat(token, kind, body string, now time.Time) error { + ctx, cancel := monCtx() + defer cancel() + + filter := bson.M{"heartbeat_token_hash": HashHeartbeatToken(token), "type": models.MonitorHeartbeat, "enabled": true} + set := bson.M{"state.last_check_at": now} + unset := bson.M{} + var next, message string + switch kind { + case HeartbeatStart: + set["state.started_at"] = now + case HeartbeatPing: + set["state.last_ping_at"] = now + set["state.status"] = models.StatusUp + set["state.message"] = "" + unset["state.started_at"] = "" + next = models.StatusUp + case HeartbeatFail: + message = failMessage(body) + set["state.last_ping_at"] = now + set["state.status"] = models.StatusDown + set["state.message"] = message + unset["state.started_at"] = "" + next = models.StatusDown + default: + return fmt.Errorf("unknown heartbeat kind %q", kind) + } + upd := bson.M{"$set": set} + if len(unset) > 0 { + upd["$unset"] = unset + } + + // ReturnDocument Before: the previous status and started_at are what the + // transition and the duration need. + var before models.Monitor + err := db.Col("monitors").FindOneAndUpdate(ctx, filter, upd, + options.FindOneAndUpdate().SetReturnDocument(options.Before)).Decode(&before) + if errors.Is(err, mongo.ErrNoDocuments) { + return ErrHeartbeatNotFound + } + if err != nil { + return err + } + if kind == HeartbeatStart { + return nil + } + + latency := 0 + if kind == HeartbeatPing && before.State.StartedAt != nil { + latency = int(now.Sub(*before.State.StartedAt).Milliseconds()) + } + if latency > 0 { + db.Col("monitors").UpdateOne(ctx, bson.M{"monitor_id": before.MonitorID}, bson.M{"$set": bson.M{"state.latency_ms": latency}}) + } + recordSample(ctx, &before, kind == HeartbeatPing, latency, now) + applyTransition(ctx, &before, "", "", before.State.Status, next, message, now) + return nil +} + +// SweepHeartbeats marks overdue heartbeats down. Recovery only ever comes from +// a ping, so the sweep never moves anything up. +func SweepHeartbeats(now time.Time) { + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + cur, err := db.Col("monitors").Find(ctx, bson.M{ + "type": models.MonitorHeartbeat, "enabled": true, + "state.status": bson.M{"$ne": models.StatusDown}, + }) + if err != nil { + log.Printf("heartbeats: list: %v", err) + return + } + var monitors []models.Monitor + if err := cur.All(ctx, &monitors); err != nil { + log.Printf("heartbeats: decode: %v", err) + return + } + for i := range monitors { + m := &monitors[i] + down, msg := heartbeatVerdict(*m, now) + if !down { + continue + } + // Conditional on status so a ping landing between the read and this + // write is not overwritten by a stale "down". + res, err := db.Col("monitors").UpdateOne(ctx, + bson.M{"monitor_id": m.MonitorID, "state.status": m.State.Status}, + bson.M{"$set": bson.M{"state.status": models.StatusDown, "state.message": msg, "state.last_check_at": now}, + "$unset": bson.M{"state.started_at": ""}}) + if err != nil || res.ModifiedCount == 0 { + continue + } + recordSample(ctx, m, false, 0, now) + applyTransition(ctx, m, "", "", m.State.Status, models.StatusDown, msg, now) + } +} + +func RotateHeartbeatToken(instanceID, monitorID string) (string, error) { + ctx, cancel := monCtx() + defer cancel() + tok, hash, err := NewHeartbeatToken() + if err != nil { + return "", err + } + res, err := db.Col("monitors").UpdateOne(ctx, + bson.M{"monitor_id": monitorID, "instance_id": instanceID, "type": models.MonitorHeartbeat}, + bson.M{"$set": bson.M{"heartbeat_token_hash": hash}}) + if err != nil { + return "", err + } + if res.MatchedCount == 0 { + return "", ErrHeartbeatNotFound + } + return tok, nil +} diff --git a/server/internal/services/heartbeats_test.go b/server/internal/services/heartbeats_test.go new file mode 100644 index 0000000..6bcabb4 --- /dev/null +++ b/server/internal/services/heartbeats_test.go @@ -0,0 +1,82 @@ +package services + +import ( + "strings" + "testing" + "time" + + "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/models" +) + +func hbMonitor(status string, lastPing, started *time.Time) models.Monitor { + return models.Monitor{ + Type: models.MonitorHeartbeat, + Target: models.MonitorTarget{PeriodSec: 3600, GraceSec: 300}, + State: models.MonitorState{Status: status, LastPingAt: lastPing, StartedAt: started}, + } +} + +func TestHeartbeatVerdict(t *testing.T) { + now := time.Date(2026, 9, 17, 12, 0, 0, 0, time.UTC) + at := func(d time.Duration) *time.Time { v := now.Add(-d); return &v } + + cases := []struct { + name string + m models.Monitor + wantDown bool + wantMsg string + }{ + {"never pinged stays pending", hbMonitor(models.StatusPending, nil, nil), false, ""}, + {"within period", hbMonitor(models.StatusUp, at(30*time.Minute), nil), false, ""}, + {"inside grace", hbMonitor(models.StatusUp, at(62*time.Minute), nil), false, ""}, + {"overdue", hbMonitor(models.StatusUp, at(66*time.Minute), nil), true, "no ping since"}, + {"started inside grace", hbMonitor(models.StatusUp, at(10*time.Minute), at(4*time.Minute)), false, ""}, + {"started never finished", hbMonitor(models.StatusUp, at(10*time.Minute), at(6*time.Minute)), true, "never finished"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + down, msg := heartbeatVerdict(c.m, now) + if down != c.wantDown || !strings.Contains(msg, c.wantMsg) { + t.Fatalf("got (%v,%q), want (%v, contains %q)", down, msg, c.wantDown, c.wantMsg) + } + }) + } +} + +func TestNewHeartbeatTokenHashes(t *testing.T) { + tok, hash, err := NewHeartbeatToken() + if err != nil { + t.Fatal(err) + } + if len(tok) < 32 { + t.Fatalf("token too short: %d", len(tok)) + } + if hash != HashHeartbeatToken(tok) || hash == tok { + t.Fatal("hash must be the SHA-256 of the token and differ from it") + } +} + +func TestValidateHeartbeat(t *testing.T) { + tg := models.MonitorTarget{PeriodSec: 59} + if err := validateHeartbeat(&tg); err == nil { + t.Fatal("period under 60 must be rejected") + } + tg = models.MonitorTarget{PeriodSec: 60} + if err := validateHeartbeat(&tg); err != nil || tg.GraceSec != 300 { + t.Fatalf("grace should default to 300, got %d err %v", tg.GraceSec, err) + } + tg = models.MonitorTarget{PeriodSec: 60, GraceSec: -1} + if err := validateHeartbeat(&tg); err == nil { + t.Fatal("negative grace must be rejected") + } +} + +func TestTruncateHeartbeatBody(t *testing.T) { + long := strings.Repeat("x", MaxHeartbeatBody+50) + if got := failMessage(long); len(got) != len("reported failure: ")+MaxHeartbeatBody { + t.Fatalf("len = %d", len(got)) + } + if got := failMessage(" "); got != "reported failure" { + t.Fatalf("empty body message = %q", got) + } +} diff --git a/server/internal/services/monitors.go b/server/internal/services/monitors.go index 07515bb..50aaa24 100644 --- a/server/internal/services/monitors.go +++ b/server/internal/services/monitors.go @@ -188,6 +188,20 @@ func CreateMonitor(instanceID string, m *models.Monitor, tokenScope map[string]s if m.Runner == "" { m.Runner = models.RunnerServer } + if m.Type == models.MonitorHeartbeat { + if err := validateHeartbeat(&m.Target); err != nil { + return nil, err + } + tok, hash, err := NewHeartbeatToken() + if err != nil { + return nil, err + } + m.HeartbeatTokenHash = hash + m.HeartbeatToken = tok + // A heartbeat is never run, so neither where nor how often applies. + m.Runner = models.RunnerServer + m.IntervalSec = 0 + } m.State = models.MonitorState{Status: models.StatusPending} if _, err := db.Col("monitors").InsertOne(ctx, m); err != nil { return nil, err @@ -199,6 +213,33 @@ func UpdateMonitor(instanceID, monitorID string, upd bson.M, tokenScope map[stri ctx, cancel := monCtx() defer cancel() + existing, err := GetMonitor(instanceID, monitorID) + if err != nil { + return err + } + if existing == nil { + return fmt.Errorf("monitor not found") + } + if raw, present := upd["type"]; present { + if t, _ := raw.(string); t != existing.Type && (models.IsPassiveMonitor(t) || models.IsPassiveMonitor(existing.Type)) { + bad := existing.Type + if models.IsPassiveMonitor(t) { + bad = t + } + return fmt.Errorf("type cannot be changed to or from %s", bad) + } + } + if raw, present := upd["target"]; present && existing.Type == models.MonitorHeartbeat { + tg, ok := raw.(models.MonitorTarget) + if !ok { + return fmt.Errorf("target must be an object") + } + if err := validateHeartbeat(&tg); err != nil { + return err + } + upd["target"] = tg + } + if raw, present := upd["channel_ids"]; present { ids, ok := raw.([]string) if !ok { @@ -232,7 +273,7 @@ func UpdateMonitor(instanceID, monitorID string, upd bson.M, tokenScope map[stri upd["runner"] = models.RunnerServer } } - _, err := db.Col("monitors").UpdateOne(ctx, bson.M{"monitor_id": monitorID, "instance_id": instanceID}, bson.M{"$set": upd}) + _, err = db.Col("monitors").UpdateOne(ctx, bson.M{"monitor_id": monitorID, "instance_id": instanceID}, bson.M{"$set": upd}) return err } @@ -359,35 +400,35 @@ func ingestResult(instanceID, runner, monitorID string, res checker.Result) erro return err } - bucket := now.Truncate(time.Hour) - up := 0 - if res.Up { - up = 1 - } - - /* The sample is the same result at full resolution, expiring by TTL. It is - written next to the rollup rather than instead of it: the rollup is what - survives, the sample is what the sub-hour views read. */ - db.Col("monitor_samples").InsertOne(ctx, models.MonitorSample{ - InstanceID: m.InstanceID, - MonitorID: monitorID, - At: now, - Up: res.Up, - LatencyMs: res.LatencyMs, - }) - - db.Col("monitor_rollups").UpdateOne(ctx, - bson.M{"monitor_id": monitorID, "period_start": bucket}, - bson.M{ - "$inc": bson.M{"checks": 1, "up_count": up, "sum_latency": int64(res.LatencyMs)}, - "$setOnInsert": bson.M{"instance_id": m.InstanceID}, - }, - options.UpdateOne().SetUpsert(true)) + recordSample(ctx, m, res.Up, res.LatencyMs, now) applyTransition(ctx, m, "", "", prev, newStatus, res.Message, now) return nil } +// recordSample writes one result at full resolution and folds it into the +// hourly rollup, the two records every history view reads. +func recordSample(ctx context.Context, m *models.Monitor, up bool, latencyMs int, now time.Time) { + u := 0 + if up { + u = 1 + } + db.Col("monitor_samples").InsertOne(ctx, models.MonitorSample{ + InstanceID: m.InstanceID, + MonitorID: m.MonitorID, + At: now, + Up: up, + LatencyMs: latencyMs, + }) + db.Col("monitor_rollups").UpdateOne(ctx, + bson.M{"monitor_id": m.MonitorID, "period_start": now.Truncate(time.Hour)}, + bson.M{ + "$inc": bson.M{"checks": 1, "up_count": u, "sum_latency": int64(latencyMs)}, + "$setOnInsert": bson.M{"instance_id": m.InstanceID}, + }, + options.UpdateOne().SetUpsert(true)) +} + func notifyTransition(m *models.Monitor, serverName, oldStatus, newStatus, message string) { if len(m.ChannelIDs) == 0 { return