feat: patch scheduler loop; record patch results and verify reboots from the agent stream

2026-09-15 09:17:27 +00:00
parent 1bb2ba7f2b
commit 139658864b
3 changed files with 155 additions and 0 deletions
@@ -23,7 +23,9 @@ 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/models"
 	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/monitorsched"
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/patchsched"
 	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/services"
 	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/vulnsched"
 	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/workflowsched"
@@ -257,6 +259,17 @@ func serve() {
 			LogEvent:        services.LogEvent,
 		})

+		patchsched.Start(jobCtx, patchsched.Deps{
+			LookupWindow: services.LookupWindow,
+			CountTargets: services.CountPolicyTargets,
+			StartPolicyRun: func(p models.PatchPolicy, windowEnd time.Time) error {
+				_, err := services.StartPolicyRun(p, windowEnd, models.PatchSourceSchedule, "schedule")
+				return err
+			},
+			AdvanceRuns: services.AdvancePatchRuns,
+			LogEvent:    services.LogEvent,
+		})
+
 		vulnsched.Start(jobCtx, vulnsched.Deps{
 			LogEvent:   services.LogEvent,
 			SendDigest: services.SendVulnDigest,
@@ -236,6 +236,12 @@ func (s *vantageServer) ReportInventory(ctx context.Context, req *pb.InventoryRe
 	if err := services.StoreInventory(srv.ServerID, req); err != nil {
 		log.Printf("store inventory for %s: %v", srv.ServerID, err)
 	}
+	// Only static snapshots compute reboot_required, so only they can settle a
+	// reboot. The agent sends one at start, so the first report after a
+	// reboot qualifies.
+	if req.IncludeStatic && req.BootTimeUnix > 0 {
+		services.VerifyPatchReboots(srv.InstanceID, srv.ServerID, time.Unix(req.BootTimeUnix, 0), req.RebootRequired)
+	}
 	return &pb.InventoryReportResponse{}, nil
 }

@@ -330,6 +336,9 @@ func (s *vantageServer) CommandStream(stream pb.Vantage_CommandStreamServer) err
 			if m.WorkloadLogsResult != nil {
 				services.WorkloadResults.Deliver(m.WorkloadLogsResult)
 			}
+			if m.PatchResult != nil {
+				services.RecordPatchResult(srv.InstanceID, srv.ServerID, m.PatchResult)
+			}
 			if m.StepResult != nil {
 				services.StepResults.Deliver(m.StepResult)
 			}
@@ -0,0 +1,133 @@
+package patchsched
+
+import (
+	"context"
+	"log"
+	"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/options"
+)
+
+const tickInterval = 30 * time.Second
+
+// Deps are injected from main.go: services imports this package, so this
+// package cannot import services.
+type Deps struct {
+	// LookupWindow returns nil, nil when the window no longer exists.
+	LookupWindow   func(instanceID, windowID string) (*models.MaintenanceWindow, error)
+	CountTargets   func(p models.PatchPolicy) (int, error)
+	StartPolicyRun func(p models.PatchPolicy, windowEnd time.Time) error
+	AdvanceRuns    func(ctx context.Context)
+	LogEvent       func(instanceID, eventType, actor, serverID, keyID, details string)
+}
+
+// Start runs until ctx is cancelled, inside bus.RunAsLeader("housekeeping").
+// Each tick fires due policies, then advances every running run.
+func Start(ctx context.Context, deps Deps) {
+	go func() {
+		t := time.NewTicker(tickInterval)
+		defer t.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-t.C:
+				fireDue(ctx, deps, time.Now())
+				deps.AdvanceRuns(ctx)
+			}
+		}
+	}()
+}
+
+func fireDue(ctx context.Context, deps Deps, now time.Time) {
+	cur, err := db.Col("patch_policies").Find(ctx, bson.M{"enabled": true, "next_run_at": bson.M{"$lte": now}})
+	if err != nil {
+		log.Printf("patchsched: find due: %v", err)
+		return
+	}
+	var due []models.PatchPolicy
+	if err := cur.All(ctx, &due); err != nil {
+		log.Printf("patchsched: decode due: %v", err)
+		return
+	}
+	for _, p := range due {
+		if ctx.Err() != nil {
+			return
+		}
+		process(ctx, deps, p, now)
+	}
+}
+
+func process(ctx context.Context, deps Deps, p models.PatchPolicy, now time.Time) {
+	if p.NextRunAt == nil {
+		return
+	}
+	due := *p.NextRunAt
+
+	w, err := deps.LookupWindow(p.InstanceID, p.WindowID)
+	if err != nil {
+		log.Printf("patchsched: policy %s: load window: %v", p.PolicyID, err)
+		return // a database error is retried next tick, not treated as "gone"
+	}
+	if w == nil {
+		disable(ctx, deps, p, "its maintenance window no longer exists")
+		return
+	}
+	end := WindowEnd(due, w.DurationMinutes)
+	next, err := NextStart(w.Cron, w.TZ, Later(now, end))
+	if err != nil {
+		disable(ctx, deps, p, "its maintenance window schedule is no longer valid: "+err.Error())
+		return
+	}
+
+	// The claim, as in workflowsched: matching the current next_run_at means a
+	// second process reaching this policy matches nothing.
+	res, err := db.Col("patch_policies").UpdateOne(ctx,
+		bson.M{"policy_id": p.PolicyID, "next_run_at": due},
+		bson.M{"$set": bson.M{"next_run_at": next}})
+	if err != nil || res.MatchedCount == 0 {
+		return
+	}
+
+	n, err := deps.CountTargets(p)
+	if err != nil {
+		recordSkip(ctx, deps, p, "error: "+err.Error(), due, now)
+		return
+	}
+	switch d := Decide(due, end, now, hasActiveRun(ctx, p), n); d {
+	case Fire:
+		if err := deps.StartPolicyRun(p, end); err != nil {
+			recordSkip(ctx, deps, p, "error: "+err.Error(), due, now)
+			return
+		}
+		_, _ = db.Col("patch_policies").UpdateOne(ctx, bson.M{"policy_id": p.PolicyID},
+			bson.M{"$set": bson.M{"last_run_at": now}, "$unset": bson.M{"last_skipped": ""}})
+	default:
+		recordSkip(ctx, deps, p, string(d), due, now)
+	}
+}
+
+func hasActiveRun(ctx context.Context, p models.PatchPolicy) bool {
+	err := db.Col("patch_runs").FindOne(ctx,
+		bson.M{"instance_id": p.InstanceID, "policy_id": p.PolicyID, "status": models.PatchRunRunning},
+		options.FindOne().SetProjection(bson.M{"_id": 1})).Err()
+	return err == nil
+}
+
+func recordSkip(ctx context.Context, deps Deps, p models.PatchPolicy, reason string, due, at time.Time) {
+	_, _ = db.Col("patch_policies").UpdateOne(ctx, bson.M{"policy_id": p.PolicyID},
+		bson.M{"$set": bson.M{"last_skipped": models.Skip{Reason: reason, Due: due, At: at}}})
+	deps.LogEvent(p.InstanceID, "patch.skipped", "schedule", "", "",
+		"patch policy "+p.Name+" skipped "+due.Format(time.RFC3339)+": "+reason)
+}
+
+func disable(ctx context.Context, deps Deps, p models.PatchPolicy, reason string) {
+	_, _ = db.Col("patch_policies").UpdateOne(ctx, bson.M{"policy_id": p.PolicyID}, bson.M{
+		"$set":   bson.M{"enabled": false, "disabled_reason": reason},
+		"$unset": bson.M{"next_run_at": ""},
+	})
+	deps.LogEvent(p.InstanceID, "patch.policy_disabled", "schedule", "", "", "patch policy "+p.Name+" disabled: "+reason)
+}