diff --git a/server/cmd/main.go b/server/cmd/main.go index 7375a74..d7a599e 100644 --- a/server/cmd/main.go +++ b/server/cmd/main.go @@ -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, diff --git a/server/internal/grpc/server.go b/server/internal/grpc/server.go index ef780ce..f9fbe84 100644 --- a/server/internal/grpc/server.go +++ b/server/internal/grpc/server.go @@ -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) } diff --git a/server/internal/patchsched/sched.go b/server/internal/patchsched/sched.go new file mode 100644 index 0000000..b9597b5 --- /dev/null +++ b/server/internal/patchsched/sched.go @@ -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) +}