From fef886c93b500931260c96b05dad02249b0b2e27 Mon Sep 17 00:00:00 2001 From: mrhid6 Date: Tue, 15 Sep 2026 09:04:23 +0000 Subject: [PATCH] feat: patch run service - dispatch, results, reboot verification, cancel, alerts, retention --- server/internal/notify/dispatch.go | 6 +- server/internal/notify/patch_test.go | 17 + server/internal/services/patch_runs.go | 465 +++++++++++++++++++++++++ server/internal/services/steplogs.go | 2 + 4 files changed, 489 insertions(+), 1 deletion(-) create mode 100644 server/internal/notify/patch_test.go create mode 100644 server/internal/services/patch_runs.go diff --git a/server/internal/notify/dispatch.go b/server/internal/notify/dispatch.go index bc63971..bd247f4 100644 --- a/server/internal/notify/dispatch.go +++ b/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) diff --git a/server/internal/notify/patch_test.go b/server/internal/notify/patch_test.go new file mode 100644 index 0000000..222a72a --- /dev/null +++ b/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) + } +} diff --git a/server/internal/services/patch_runs.go b/server/internal/services/patch_runs.go new file mode 100644 index 0000000..e997cf0 --- /dev/null +++ b/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}) + } +} diff --git a/server/internal/services/steplogs.go b/server/internal/services/steplogs.go index fa0e761..cf96b04 100644 --- a/server/internal/services/steplogs.go +++ b/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() } } }()