feat: patch run service - dispatch, results, reboot verification, cancel, alerts, retention

server/internal/notify/dispatch.go

@@ -11,6 +11,10 @@ import (
 // monitor check. MonitorName carries the hostname in that case.
 const TypeServer = "server"
 
+// TypePatch marks a patch run summary. Like a vulnerability digest it is a
+// headline, not a transition, so title() adds no verb.
+const TypePatch = "patch"
+
 type Event struct {
 	MonitorName string
 	Type        string
@@ -26,7 +30,7 @@ func (e Event) title() string {
 		verb = "is DOWN"
 	}
 	var s string
-	if e.Type == TypeVuln {
+	if e.Type == TypeVuln || e.Type == TypePatch {
 		// A digest is not a transition. MonitorName already carries the whole
 		// headline ("12 new critical across 4 servers"), so no verb applies.
 		s = fmt.Sprintf("[Vantage] %s", e.MonitorName)

server/internal/notify/patch_test.go

@@ -0,0 +1,17 @@
+package notify
+
+import (
+	"testing"
+
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/models"
+)
+
+// A patch summary is not a transition: it must not read "is DOWN" or
+// "recovered".
+func TestPatchEventTitle(t *testing.T) {
+	ev := Event{MonitorName: `Patch policy "Sunday prod" partial`, Type: TypePatch, NewStatus: models.PatchRunPartial, Message: "38 succeeded, 2 failed"}
+	want := `[Vantage] Patch policy "Sunday prod" partial: 38 succeeded, 2 failed`
+	if got := ev.title(); got != want {
+		t.Fatalf("got %q, want %q", got, want)
+	}
+}

server/internal/services/patch_runs.go

@@ -0,0 +1,465 @@
+package services
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"time"
+
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/db"
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/models"
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/notify"
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/patchrun"
+	"gitea.hostxtra.co.uk/vantage/vantage-shared/grpc/pb"
+	"github.com/google/uuid"
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+	"go.mongodb.org/mongo-driver/v2/mongo/options"
+)
+
+// ErrAgentOffline is declared in consoleproxy.go and reused here: both mean
+// the same thing, the target's agent is not on the command stream.
+var (
+	ErrPatchRunNotFound = errors.New("patch run not found")
+	ErrPatchRunFinished = errors.New("patch run has already finished")
+)
+
+const patchRunsCol = "patch_runs"
+
+func newServerRun(s models.Server, now time.Time) models.PatchServerRun {
+	r := models.PatchServerRun{ServerID: s.ServerID, Hostname: s.Hostname, Status: models.PatchSrvQueued, PendingBefore: len(s.AvailableUpdates)}
+	if !patchrun.AgentSupportsPatchResults(s.AgentVersion) {
+		v := s.AgentVersion
+		if v == "" {
+			v = "unknown"
+		}
+		r.Status = models.PatchSrvAgentTooOld
+		r.Error = fmt.Sprintf("agent %s predates patch results; update it to %s or later", v, patchrun.MinAgentVersion)
+		r.FinishedAt = &now
+	}
+	return r
+}
+
+// StartPolicyRun records a run for a policy and dispatches its first batch
+// immediately. The run document is written before any command leaves, so a
+// fast agent's result always finds it.
+func StartPolicyRun(p models.PatchPolicy, windowEnd time.Time, source, actor string) (*models.PatchRun, error) {
+	targets, err := ResolveTargets(p.InstanceID, p.TargetServerIDs, p.TargetTags)
+	if err != nil {
+		return nil, err
+	}
+	now := time.Now()
+	run := models.PatchRun{
+		InstanceID: p.InstanceID, RunID: uuid.New().String(), PolicyID: p.PolicyID, PolicyName: p.Name,
+		TriggeredBy: actor, Source: source, Scope: p.Scope, Reboot: p.Reboot, MaxConcurrent: p.MaxConcurrent,
+		WindowEnd: &windowEnd, Status: models.PatchRunRunning, StartedAt: now,
+	}
+	for _, s := range targets {
+		run.Servers = append(run.Servers, newServerRun(s, now))
+	}
+	ctx, cancel := patchCtx()
+	defer cancel()
+	if _, err := db.Col(patchRunsCol).InsertOne(ctx, run); err != nil {
+		return nil, err
+	}
+	LogEvent(p.InstanceID, "patch.run_started", actor, "", "",
+		fmt.Sprintf("patch policy %s started on %d servers (%s, reboot %s)", p.Name, len(run.Servers), p.Scope, p.Reboot))
+	advanceRun(ctx, run.RunID)
+	return &run, nil
+}
+
+// StartManualRun is Apply updates on one server. An agent too old to answer
+// still gets the command it always got, and the run says it cannot know the
+// outcome rather than claiming one.
+func StartManualRun(instanceID string, srv *models.Server, actor, source string) (*models.PatchRun, error) {
+	now := time.Now()
+	sr := newServerRun(*srv, now)
+	legacy := sr.Status == models.PatchSrvAgentTooOld
+	sr.Status, sr.Error, sr.FinishedAt, sr.StartedAt = models.PatchSrvPatching, "", nil, &now
+	run := models.PatchRun{
+		InstanceID: instanceID, RunID: uuid.New().String(), TriggeredBy: actor, Source: source,
+		Scope: models.PatchScopeAll, Reboot: models.PatchRebootNever, Status: models.PatchRunRunning, StartedAt: now,
+	}
+
+	ctx, cancel := patchCtx()
+	defer cancel()
+
+	var dispatchErr error
+	if !Dispatcher.IsConnected(srv.ServerID) {
+		dispatchErr = ErrAgentOffline
+	}
+	if dispatchErr == nil && legacy {
+		dispatchErr = DispatchApplyUpdates(srv.ServerID)
+		if dispatchErr == nil {
+			sr.Status = models.PatchSrvSucceeded
+			sr.Error = "no result reported: agent predates patch results"
+			sr.FinishedAt = &now
+		}
+	}
+	if dispatchErr != nil {
+		sr.Status, sr.Error, sr.FinishedAt = models.PatchSrvFailed, dispatchErr.Error(), &now
+	} else if !legacy {
+		sr.CommandID = uuid.New().String()
+	}
+	run.Servers = []models.PatchServerRun{sr}
+	if _, err := db.Col(patchRunsCol).InsertOne(ctx, run); err != nil {
+		return nil, err
+	}
+	if dispatchErr == nil && !legacy {
+		if err := dispatchPatch(srv.ServerID, sr.CommandID, run); err != nil {
+			_, _ = setServer(ctx, run.RunID, srv.ServerID, models.PatchSrvPatching,
+				bson.M{"status": models.PatchSrvFailed, "error": err.Error(), "finished_at": time.Now()})
+			dispatchErr = ErrAgentOffline
+		}
+	}
+	finalizeRun(ctx, run.RunID)
+	if dispatchErr != nil {
+		return &run, ErrAgentOffline
+	}
+	return &run, nil
+}
+
+func dispatchPatch(serverID, commandID string, run models.PatchRun) error {
+	cmd := &pb.ApplyUpdatesCmd{Scope: run.Scope, RebootIfRequired: run.Reboot == models.PatchRebootIfRequired}
+	if run.WindowEnd != nil {
+		cmd.DeadlineUnix = run.WindowEnd.Unix()
+	}
+	return Dispatcher.dispatch(serverID, &pb.ServerCommand{CommandId: commandID, ApplyUpdates: cmd})
+}
+
+// setServer updates one server run, but only if it is still in status from.
+// That guard is what stops a tick from overwriting a result that landed while
+// the tick was working.
+func setServer(ctx context.Context, runID, serverID, from string, set bson.M) (bool, error) {
+	fields := bson.M{}
+	for k, v := range set {
+		fields["servers.$."+k] = v
+	}
+	res, err := db.Col(patchRunsCol).UpdateOne(ctx,
+		bson.M{"run_id": runID, "servers": bson.M{"$elemMatch": bson.M{"server_id": serverID, "status": from}}},
+		bson.M{"$set": fields})
+	if err != nil {
+		return false, err
+	}
+	return res.MatchedCount > 0, nil
+}
+
+func loadRun(ctx context.Context, filter bson.M) (*models.PatchRun, error) {
+	var run models.PatchRun
+	err := db.Col(patchRunsCol).FindOne(ctx, filter).Decode(&run)
+	if errors.Is(err, mongo.ErrNoDocuments) {
+		return nil, ErrPatchRunNotFound
+	}
+	return &run, err
+}
+
+// AdvancePatchRuns is the scheduler's second job each tick.
+func AdvancePatchRuns(ctx context.Context) {
+	cur, err := db.Col(patchRunsCol).Find(ctx, bson.M{"status": models.PatchRunRunning}, options.Find().SetProjection(bson.M{"run_id": 1}))
+	if err != nil {
+		log.Printf("patch runs: find running: %v", err)
+		return
+	}
+	var ids []struct {
+		RunID string `bson:"run_id"`
+	}
+	if err := cur.All(ctx, &ids); err != nil {
+		log.Printf("patch runs: decode running: %v", err)
+		return
+	}
+	for _, r := range ids {
+		if ctx.Err() != nil {
+			return
+		}
+		advanceRun(ctx, r.RunID)
+	}
+}
+
+func advanceRun(ctx context.Context, runID string) {
+	run, err := loadRun(ctx, bson.M{"run_id": runID})
+	if err != nil {
+		log.Printf("patch run %s: load: %v", runID, err)
+		return
+	}
+	now := time.Now()
+	connected := map[string]bool{}
+	for _, s := range run.Servers {
+		if !patchrun.IsTerminal(s.Status) {
+			connected[s.ServerID] = Dispatcher.IsConnected(s.ServerID)
+		}
+	}
+	for _, tr := range patchrun.Advance(*run, now, connected) {
+		if tr.Dispatch {
+			cmdID := uuid.New().String()
+			ok, err := setServer(ctx, runID, tr.ServerID, tr.From,
+				bson.M{"status": models.PatchSrvPatching, "command_id": cmdID, "started_at": now})
+			if err != nil || !ok {
+				continue
+			}
+			if err := dispatchPatch(tr.ServerID, cmdID, *run); err != nil {
+				// The agent dropped between the connection check and the send.
+				// Back to waiting: the next tick tries again while the window
+				// is open.
+				_, _ = setServer(ctx, runID, tr.ServerID, models.PatchSrvPatching,
+					bson.M{"status": models.PatchSrvWaitingOffline, "command_id": ""})
+			}
+			continue
+		}
+		set := bson.M{"status": tr.To}
+		if tr.Error != "" {
+			set["error"] = tr.Error
+		}
+		if patchrun.IsTerminal(tr.To) {
+			set["finished_at"] = now
+		}
+		if _, err := setServer(ctx, runID, tr.ServerID, tr.From, set); err != nil {
+			log.Printf("patch run %s: server %s: %v", runID, tr.ServerID, err)
+		}
+	}
+	finalizeRun(ctx, runID)
+}
+
+func finalizeRun(ctx context.Context, runID string) {
+	run, err := loadRun(ctx, bson.M{"run_id": runID})
+	if err != nil || run.Status != models.PatchRunRunning {
+		return
+	}
+	status, done := patchrun.Finalize(*run)
+	if !done {
+		return
+	}
+	now := time.Now()
+	res, err := db.Col(patchRunsCol).UpdateOne(ctx,
+		bson.M{"run_id": runID, "status": models.PatchRunRunning},
+		bson.M{"$set": bson.M{"status": status, "finished_at": now}})
+	if err != nil || res.MatchedCount == 0 {
+		return // finalised by someone else
+	}
+	run.Status = status
+	name := run.PolicyName
+	if name == "" {
+		name = "manual update"
+	}
+	LogEvent(run.InstanceID, "patch.run_finished", run.TriggeredBy, "", "",
+		fmt.Sprintf("patch run %s (%s) %s: %s", runID, name, status, patchrun.Summary(*run)))
+	if status == models.PatchRunPartial || status == models.PatchRunFailed {
+		notifyPatchRun(*run)
+	}
+}
+
+func notifyPatchRun(run models.PatchRun) {
+	if run.PolicyID == "" {
+		return // a manual run was watched by the person who clicked
+	}
+	p, err := GetPolicy(run.InstanceID, run.PolicyID)
+	if err != nil || len(p.NotifyChannelIDs) == 0 {
+		return
+	}
+	chs, err := GetChannels(run.InstanceID, p.NotifyChannelIDs)
+	if err != nil {
+		log.Printf("patch run %s: channels: %v", run.RunID, err)
+		return
+	}
+	ev := notify.Event{
+		MonitorName: fmt.Sprintf("Patch policy %q %s", run.PolicyName, run.Status),
+		Type:        notify.TypePatch,
+		NewStatus:   run.Status,
+		Message:     fmt.Sprintf("%s (run %s)", patchrun.Summary(run), run.RunID),
+		Time:        time.Now(),
+	}
+	for _, ch := range chs {
+		if err := notify.Dispatch(ch, ev); err != nil {
+			log.Printf("patch run %s: notify %s: %v", run.RunID, ch.Name, err)
+		}
+	}
+}
+
+// RecordPatchResult is called by whichever pod holds the agent's stream. The
+// filter names this agent's own server, so one agent cannot answer for
+// another's command.
+func RecordPatchResult(instanceID, serverID string, r *pb.PatchResult) {
+	ctx, cancel := patchCtx()
+	defer cancel()
+	run, err := loadRun(ctx, bson.M{"instance_id": instanceID,
+		"servers": bson.M{"$elemMatch": bson.M{"server_id": serverID, "command_id": r.CommandId}}})
+	if err != nil {
+		log.Printf("patch result %s from %s matches no run: %v", r.CommandId, serverID, err)
+		return
+	}
+	for _, s := range run.Servers {
+		if s.ServerID != serverID || s.CommandID != r.CommandId {
+			continue
+		}
+		updated, ok := patchrun.ApplyResult(s, r, time.Now())
+		if !ok {
+			return
+		}
+		set := bson.M{"status": updated.Status, "output": updated.Output, "error": updated.Error}
+		if updated.PendingAfter != nil {
+			set["pending_after"] = *updated.PendingAfter
+		}
+		if updated.RebootedAt != nil {
+			set["rebooted_at"] = *updated.RebootedAt
+		}
+		if updated.FinishedAt != nil {
+			set["finished_at"] = *updated.FinishedAt
+		}
+		if ok, _ := setServer(ctx, run.RunID, serverID, models.PatchSrvPatching, set); ok && updated.Status == models.PatchSrvRebooting {
+			LogEvent(instanceID, "patch.reboot", "schedule", serverID, "",
+				fmt.Sprintf("%s rebooting for patch policy %s", s.Hostname, run.PolicyName))
+		}
+	}
+	finalizeRun(ctx, run.RunID)
+}
+
+// VerifyPatchReboots settles any rebooting server run for this server from a
+// static inventory report.
+func VerifyPatchReboots(instanceID, serverID string, bootTime time.Time, rebootRequired bool) {
+	ctx, cancel := patchCtx()
+	defer cancel()
+	cur, err := db.Col(patchRunsCol).Find(ctx, bson.M{"instance_id": instanceID, "status": models.PatchRunRunning,
+		"servers": bson.M{"$elemMatch": bson.M{"server_id": serverID, "status": models.PatchSrvRebooting}}})
+	if err != nil {
+		return
+	}
+	var runs []models.PatchRun
+	if err := cur.All(ctx, &runs); err != nil {
+		return
+	}
+	now := time.Now()
+	for _, run := range runs {
+		for _, s := range run.Servers {
+			if s.ServerID != serverID {
+				continue
+			}
+			updated, ok := patchrun.VerifyReboot(s, bootTime, rebootRequired, now)
+			if !ok {
+				continue
+			}
+			set := bson.M{"status": updated.Status, "error": updated.Error, "finished_at": now}
+			if updated.VerifiedAt != nil {
+				set["verified_at"] = *updated.VerifiedAt
+			}
+			_, _ = setServer(ctx, run.RunID, serverID, models.PatchSrvRebooting, set)
+		}
+		finalizeRun(ctx, run.RunID)
+	}
+}
+
+// CancelPatchRun stops further dispatch. Servers already patching finish:
+// killing a package manager mid-transaction is worse than letting it end.
+func CancelPatchRun(instanceID, runID string) error {
+	ctx, cancel := patchCtx()
+	defer cancel()
+	res, err := db.Col(patchRunsCol).UpdateOne(ctx,
+		bson.M{"instance_id": instanceID, "run_id": runID, "status": models.PatchRunRunning, "cancelled_at": bson.M{"$exists": false}},
+		bson.M{"$set": bson.M{"cancelled_at": time.Now()}})
+	if err != nil {
+		return err
+	}
+	if res.MatchedCount == 0 {
+		if _, err := GetPatchRun(instanceID, runID); err != nil {
+			return err
+		}
+		return ErrPatchRunFinished
+	}
+	advanceRun(ctx, runID)
+	return nil
+}
+
+func GetPatchRun(instanceID, runID string) (*models.PatchRun, error) {
+	ctx, cancel := patchCtx()
+	defer cancel()
+	return loadRun(ctx, bson.M{"instance_id": instanceID, "run_id": runID})
+}
+
+// ListPatchRuns omits output: a list of fifty runs would otherwise carry up
+// to 64KB per server.
+func ListPatchRuns(instanceID, policyID, serverID string, limit int64) ([]models.PatchRun, error) {
+	ctx, cancel := patchCtx()
+	defer cancel()
+	filter := bson.M{"instance_id": instanceID}
+	if policyID != "" {
+		filter["policy_id"] = policyID
+	}
+	if serverID != "" {
+		filter["servers.server_id"] = serverID
+	}
+	if limit <= 0 || limit > 200 {
+		limit = 50
+	}
+	cur, err := db.Col(patchRunsCol).Find(ctx, filter, options.Find().
+		SetSort(bson.M{"started_at": -1}).SetLimit(limit).SetProjection(bson.M{"servers.output": 0}))
+	if err != nil {
+		return nil, err
+	}
+	out := []models.PatchRun{}
+	return out, cur.All(ctx, &out)
+}
+
+// ScopePatchRun removes servers outside a tag-restricted token's reach, so a
+// run record cannot name a host the caller could not otherwise see.
+func ScopePatchRun(instanceID string, run *models.PatchRun, tokenScope map[string]string) error {
+	if len(tokenScope) == 0 {
+		return nil
+	}
+	ids := make([]string, 0, len(run.Servers))
+	for _, s := range run.Servers {
+		ids = append(ids, s.ServerID)
+	}
+	visible, err := ResolveTargetsScoped(instanceID, ids, nil, tokenScope)
+	if err != nil && !errors.Is(err, ErrNoTargets) {
+		return err
+	}
+	keep := map[string]bool{}
+	for _, s := range visible {
+		keep[s.ServerID] = true
+	}
+	kept := run.Servers[:0]
+	for _, s := range run.Servers {
+		if keep[s.ServerID] {
+			kept = append(kept, s)
+		}
+	}
+	run.Servers = kept
+	return nil
+}
+
+// sweepPatchRuns deletes finished runs past their instance's workflow log
+// retention: they are the same kind of record, and one setting governs both.
+func sweepPatchRuns() {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	defer cancel()
+	cur, err := db.Col(patchRunsCol).Find(ctx, bson.M{"finished_at": bson.M{"$ne": nil}},
+		options.Find().SetProjection(bson.M{"run_id": 1, "instance_id": 1, "finished_at": 1}))
+	if err != nil {
+		log.Printf("patch run sweep: %v", err)
+		return
+	}
+	defer cur.Close(ctx)
+	cache := map[string]int{}
+	now := time.Now()
+	for cur.Next(ctx) {
+		var r struct {
+			RunID      string     `bson:"run_id"`
+			InstanceID string     `bson:"instance_id"`
+			FinishedAt *time.Time `bson:"finished_at"`
+		}
+		if cur.Decode(&r) != nil || r.FinishedAt == nil {
+			continue
+		}
+		days, ok := cache[r.InstanceID]
+		if !ok {
+			days = defaultRetentionDays
+			if v, err := GetWorkflowLogRetentionDays(r.InstanceID); err == nil {
+				days = v
+			}
+			cache[r.InstanceID] = days
+		}
+		if days <= 0 || !r.FinishedAt.Before(now.AddDate(0, 0, -days)) {
+			continue
+		}
+		_, _ = db.Col(patchRunsCol).DeleteOne(ctx, bson.M{"run_id": r.RunID})
+	}
+}

server/internal/services/steplogs.go

@@ -332,6 +332,7 @@ func HasServerRunLog(runID, serverID string) bool {
 func StartLogSweeper(ctx context.Context) {
 	go func() {
 		sweepLogs()
+		sweepPatchRuns()
 		t := time.NewTicker(time.Hour)
 		defer t.Stop()
 		for {
@@ -340,6 +341,7 @@ func StartLogSweeper(ctx context.Context) {
 				return
 			case <-t.C:
 				sweepLogs()
+				sweepPatchRuns()
 			}
 		}
 	}()