From 062cfa74d553eef534a58dc72df19e775bc0432c Mon Sep 17 00:00:00 2001 From: prop4n Date: Sun, 30 Aug 2026 18:46:15 +0200 Subject: [PATCH] Record and expose per-resource history --- internal/app/app.go | 26 +++++++- internal/reconcile/dispatcher.go | 14 ++++ internal/reconcile/engine.go | 41 +++++++++++- internal/reconcile/events.go | 31 +++++++++ internal/reconcile/events_test.go | 88 ++++++++++++++++++++++++++ internal/reconcile/plan.go | 3 + internal/server/resources.go | 27 ++++++++ internal/server/routes.go | 1 + internal/server/server.go | 3 + internal/store/events.go | 57 +++++++++++++++++ internal/store/store.go | 13 ++++ internal/store/store_test.go | 37 +++++++++++ web/ui/src/components/ResourceCard.vue | 68 +++++++++++++++++++- web/ui/src/lib/api.ts | 12 ++++ 14 files changed, 415 insertions(+), 6 deletions(-) create mode 100644 internal/reconcile/events.go create mode 100644 internal/reconcile/events_test.go create mode 100644 internal/store/events.go diff --git a/internal/app/app.go b/internal/app/app.go index 70fbcce..18e2f6a 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -23,9 +23,25 @@ import ( const shutdownTimeout = 10 * time.Second type App struct { - cfg config.Config - log *slog.Logger - set *settings.Service + cfg config.Config + log *slog.Logger + set *settings.Service + events reconcile.EventSink +} + +// eventSink adapts the store to reconcile.EventSink, recording resource history. +type eventSink struct { + store *store.Store + log *slog.Logger +} + +func (s *eventSink) Record(ctx context.Context, e reconcile.Event) { + err := s.store.AppendEvent(ctx, store.ResourceEvent{ + Kind: string(e.Kind), Name: e.Name, Type: e.Type, Reason: e.Reason, Commit: e.Commit, + }) + if err != nil { + s.log.Error("record event", "err", err) + } } func New(cfg config.Config, log *slog.Logger) *App { @@ -99,6 +115,7 @@ func (a *App) Run(ctx context.Context, addr string) error { return fmt.Errorf("load encryption key: %w", err) } a.set = settings.New(st, key) + a.events = &eventSink{store: st, log: a.log} authSvc := auth.New(st, a.log) if err := authSvc.Init(ctx); err != nil { @@ -112,6 +129,7 @@ func (a *App) Run(ctx context.Context, addr string) error { Status: statusStore, Settings: a.set, Deleter: a, + Events: st, CookieSecure: a.cfg.Server.CookieSecure, }, a.log) @@ -180,8 +198,10 @@ func (a *App) reconcileLoop(ctx context.Context, statusStore *status.Store) erro if complete { if dispatcher == nil { dispatcher = reconcile.NewDispatcher(cfg.Reconcile.Concurrency, a.log) + dispatcher.SetEventSink(a.events) } engine := a.newEngine(cfg, statusStore) + engine.SetEventSink(a.events) if err := engine.Scan(ctx, dispatcher); err != nil && !errors.Is(err, context.Canceled) { a.log.Error("reconcile failed", "err", err) } diff --git a/internal/reconcile/dispatcher.go b/internal/reconcile/dispatcher.go index 7670b6b..e9ba582 100644 --- a/internal/reconcile/dispatcher.go +++ b/internal/reconcile/dispatcher.go @@ -14,6 +14,7 @@ import ( // dropped and picked up on a later scan. type Dispatcher struct { log *slog.Logger + events EventSink sem chan struct{} mu sync.Mutex inflight map[actionKey]struct{} @@ -37,6 +38,10 @@ func NewDispatcher(limit int, log *slog.Logger) *Dispatcher { } } +// SetEventSink attaches a sink so applied/failed actions are recorded in the +// resource history. Optional; nil means no history is recorded. +func (d *Dispatcher) SetEventSink(s EventSink) { d.events = s } + // Submit starts action a in the background unless it is already running or the // pool is full. It never blocks; a dropped action is retried on the next scan. func (d *Dispatcher) Submit(ctx context.Context, a Action) { @@ -66,12 +71,21 @@ func (d *Dispatcher) Submit(ctx context.Context, a Action) { d.log.Info("applying", "action", a.String(), "reason", a.Reason) if err := a.Apply(ctx); err != nil { d.log.Error("apply failed", "action", a.String(), "err", err) + d.record(Event{Kind: a.Kind, Name: a.Name, Type: EventFailed, Reason: err.Error(), Commit: a.Commit}) return } d.log.Info("applied", "action", a.String()) + d.record(Event{Kind: a.Kind, Name: a.Name, Type: EventApplied, Reason: string(a.Type), Commit: a.Commit}) }) } +// record forwards an event to the sink when one is attached. +func (d *Dispatcher) record(ev Event) { + if d.events != nil { + d.events.Record(context.Background(), ev) + } +} + // Wait blocks until every in-flight action has finished. func (d *Dispatcher) Wait() { d.wg.Wait() diff --git a/internal/reconcile/engine.go b/internal/reconcile/engine.go index 1b95b99..89bfc6d 100644 --- a/internal/reconcile/engine.go +++ b/internal/reconcile/engine.go @@ -44,6 +44,7 @@ type Engine struct { opts Options log *slog.Logger status *status.Store + events EventSink } // NewEngine wires an Engine from its collaborators. status may be nil (e.g. for @@ -52,6 +53,10 @@ func NewEngine(source Source, reconcilers []Reconciler, opts Options, log *slog. return &Engine{source: source, reconcilers: reconcilers, opts: opts, log: log, status: status} } +// SetEventSink attaches a sink for resource history events. Optional; nil means +// no history is recorded. +func (e *Engine) SetEventSink(s EventSink) { e.events = s } + // computePlan loads the desired state and aggregates every reconciler's plan. func (e *Engine) computePlan(ctx context.Context) ([]manifest.Resource, Plan, error) { desired, err := e.source.Desired(ctx) @@ -90,6 +95,7 @@ func (e *Engine) Scan(ctx context.Context, d *Dispatcher) error { if plan.Empty() || !e.opts.AutoSync { return nil } + commit := e.commit() for _, a := range plan.Actions { switch { case a.Informational: @@ -99,6 +105,7 @@ func (e *Engine) Scan(ctx context.Context, d *Dispatcher) error { case e.opts.DryRun: e.log.Info("dry-run", "action", a.String(), "reason", a.Reason) default: + a.Commit = commit d.Submit(ctx, a) } } @@ -179,13 +186,45 @@ func (e *Engine) recordStatus(desired []manifest.Resource, plan Plan) { resources = append(resources, res) } + commit := e.commit() e.status.Set(status.Snapshot{ UpdatedAt: time.Now(), InSync: plan.Empty(), Configured: true, - Commit: e.commit(), + Commit: commit, Resources: resources, }) + + // History: record state transitions against the previous snapshot. A brand + // new resource is only logged when it starts OutOfSync, to avoid a burst of + // "synced" events for everything already in sync on boot. + if e.events == nil { + return + } + current := make(map[key]bool, len(resources)) + for _, r := range resources { + k := key{manifest.Kind(r.Kind), r.Name} + current[k] = true + prev, existed := prevState[k] + switch { + case existed && prev != r.State && r.State == status.StateSynced: + e.record(Event{Kind: k.kind, Name: k.name, Type: EventSynced, Commit: commit}) + case prev != r.State && r.State == status.StateOutOfSync: + e.record(Event{Kind: k.kind, Name: k.name, Type: EventDrifted, Reason: r.Reason, Commit: commit}) + } + } + for k, prev := range prevState { + if _, ok := current[k]; !ok && prev != "" { + e.record(Event{Kind: k.kind, Name: k.name, Type: EventRemoved, Commit: commit}) + } + } +} + +// record forwards an event to the sink when one is attached. +func (e *Engine) record(ev Event) { + if e.events != nil { + e.events.Record(context.Background(), ev) + } } // committer is implemented by sources that track a Git commit (the Git source); diff --git a/internal/reconcile/events.go b/internal/reconcile/events.go new file mode 100644 index 0000000..7206056 --- /dev/null +++ b/internal/reconcile/events.go @@ -0,0 +1,31 @@ +package reconcile + +import ( + "context" + + "github.com/prop4n/proxmops/internal/manifest" +) + +// Event types recorded in a resource's history. +const ( + EventDrifted = "drifted" // went OutOfSync + EventSynced = "synced" // returned to Synced + EventRemoved = "removed" // disappeared from the desired/observed set + EventApplied = "applied" // an action ran successfully + EventFailed = "failed" // an action failed +) + +// Event is one entry in a resource's history. +type Event struct { + Kind manifest.Kind + Name string + Type string + Reason string + Commit string +} + +// EventSink records resource events. Implementations must be safe for +// concurrent use, since the dispatcher records from background goroutines. +type EventSink interface { + Record(ctx context.Context, e Event) +} diff --git a/internal/reconcile/events_test.go b/internal/reconcile/events_test.go new file mode 100644 index 0000000..b9dc008 --- /dev/null +++ b/internal/reconcile/events_test.go @@ -0,0 +1,88 @@ +package reconcile + +import ( + "context" + "sync" + "testing" + + "github.com/prop4n/proxmops/internal/manifest" + "github.com/prop4n/proxmops/internal/status" +) + +// recordingSink collects events for assertions, safe for concurrent use. +type recordingSink struct { + mu sync.Mutex + events []Event +} + +func (s *recordingSink) Record(_ context.Context, e Event) { + s.mu.Lock() + s.events = append(s.events, e) + s.mu.Unlock() +} + +func (s *recordingSink) types() []string { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]string, len(s.events)) + for i, e := range s.events { + out[i] = e.Type + } + return out +} + +func TestEngineRecordsDriftThenSynced(t *testing.T) { + st := status.NewStore() + rec := &staticReconciler{} + sink := &recordingSink{} + src := staticSource{res: []manifest.Resource{vmResource("web-01")}} + eng := NewEngine(src, []Reconciler{rec}, Options{AutoSync: false}, testLogger(), st) + eng.SetEventSink(sink) + d := NewDispatcher(1, testLogger()) + + // Pass 1: a create is planned -> the resource is OutOfSync -> drifted event. + rec.plan = Plan{Actions: []Action{{Type: ActionCreate, Kind: manifest.KindVirtualMachine, Name: "web-01"}}} + if err := eng.Scan(context.Background(), d); err != nil { + t.Fatal(err) + } + // Pass 2: plan empty -> resource returns to Synced -> synced event. + rec.plan = Plan{} + if err := eng.Scan(context.Background(), d); err != nil { + t.Fatal(err) + } + + got := sink.types() + if len(got) != 2 || got[0] != EventDrifted || got[1] != EventSynced { + t.Fatalf("events = %v, want [drifted synced]", got) + } +} + +func TestDispatcherRecordsAppliedAndFailed(t *testing.T) { + sink := &recordingSink{} + d := NewDispatcher(2, testLogger()) + d.SetEventSink(sink) + + ok := Action{Type: ActionCreate, Kind: manifest.KindIso, Name: "ok", Commit: "c1", + Apply: func(context.Context) error { return nil }} + bad := Action{Type: ActionCreate, Kind: manifest.KindIso, Name: "bad", Commit: "c1", + Apply: func(context.Context) error { return context.DeadlineExceeded }} + d.Submit(context.Background(), ok) + d.Submit(context.Background(), bad) + d.Wait() + + var applied, failed int + for _, e := range sink.events { + switch e.Type { + case EventApplied: + applied++ + if e.Commit != "c1" || e.Reason != string(ActionCreate) { + t.Errorf("applied event fields: %+v", e) + } + case EventFailed: + failed++ + } + } + if applied != 1 || failed != 1 { + t.Fatalf("applied=%d failed=%d, want 1/1", applied, failed) + } +} diff --git a/internal/reconcile/plan.go b/internal/reconcile/plan.go index 18141b1..32f0444 100644 --- a/internal/reconcile/plan.go +++ b/internal/reconcile/plan.go @@ -37,6 +37,9 @@ type Action struct { // Informational actions report a condition (e.g. "reboot required") in the // status without being applied. The engine never dispatches them. Informational bool + // Commit is the Git commit this action was planned from, set by the engine + // before dispatch so applied/failed events are attributed to it. + Commit string } // String renders an action for display, e.g. "create VirtualMachine/web-01". diff --git a/internal/server/resources.go b/internal/server/resources.go index fc19174..ba83e59 100644 --- a/internal/server/resources.go +++ b/internal/server/resources.go @@ -6,8 +6,15 @@ import ( "net/http" "github.com/go-chi/chi/v5" + + "github.com/prop4n/proxmops/internal/store" ) +// EventsReader reads a resource's recorded history. +type EventsReader interface { + EventsFor(ctx context.Context, kind, name string, limit int) ([]store.ResourceEvent, error) +} + // ResourceDeleter removes a managed resource from the cluster. It is implemented // outside the transport layer and injected through Options. type ResourceDeleter interface { @@ -43,3 +50,23 @@ func (s *Server) handleDeleteResource(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusInternalServerError, "internal error") } } + +// handleResourceEvents returns a resource's recorded history, newest first. +func (s *Server) handleResourceEvents(w http.ResponseWriter, r *http.Request) { + if s.events == nil { + writeJSON(w, http.StatusOK, []store.ResourceEvent{}) + return + } + kind := chi.URLParam(r, "kind") + name := chi.URLParam(r, "name") + events, err := s.events.EventsFor(r.Context(), kind, name, 200) + if err != nil { + s.log.Error("read events", "kind", kind, "name", name, "err", err) + writeError(w, http.StatusInternalServerError, "internal error") + return + } + if events == nil { + events = []store.ResourceEvent{} + } + writeJSON(w, http.StatusOK, events) +} diff --git a/internal/server/routes.go b/internal/server/routes.go index 0baa4b6..cd341ce 100644 --- a/internal/server/routes.go +++ b/internal/server/routes.go @@ -31,6 +31,7 @@ func (s *Server) routes() http.Handler { r.Get("/me", s.handleMe) r.Get("/resources", s.handleResources) r.Delete("/resources/{kind}/{name}", s.handleDeleteResource) + r.Get("/resources/{kind}/{name}/events", s.handleResourceEvents) r.Get("/events", s.handleEvents) r.Get("/settings", s.handleGetSettings) r.Put("/settings", s.handlePutSettings) diff --git a/internal/server/server.go b/internal/server/server.go index f0386d3..2c25a54 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -22,6 +22,7 @@ type Server struct { status *status.Store settings *settings.Service deleter ResourceDeleter + events EventsReader cookieSecure bool } @@ -32,6 +33,7 @@ type Options struct { Status *status.Store Settings *settings.Service Deleter ResourceDeleter + Events EventsReader CookieSecure bool } @@ -43,6 +45,7 @@ func New(opts Options, log *slog.Logger) *Server { status: opts.Status, settings: opts.Settings, deleter: opts.Deleter, + events: opts.Events, cookieSecure: opts.CookieSecure, } s.http = &http.Server{ diff --git a/internal/store/events.go b/internal/store/events.go new file mode 100644 index 0000000..98a8d62 --- /dev/null +++ b/internal/store/events.go @@ -0,0 +1,57 @@ +package store + +import ( + "context" + "fmt" + "time" +) + +// ResourceEvent is one recorded change in a resource's history. +type ResourceEvent struct { + ID int64 `json:"id"` + Kind string `json:"kind"` + Name string `json:"name"` + Type string `json:"type"` + Reason string `json:"reason,omitempty"` + Commit string `json:"commit,omitempty"` + At time.Time `json:"at"` +} + +// AppendEvent records one resource event. +func (s *Store) AppendEvent(ctx context.Context, e ResourceEvent) error { + at := e.At + if at.IsZero() { + at = time.Now() + } + _, err := s.db.ExecContext(ctx, + "INSERT INTO resource_events (kind, name, type, reason, commit_, at) VALUES (?, ?, ?, ?, ?, ?)", + e.Kind, e.Name, e.Type, e.Reason, e.Commit, at) + if err != nil { + return fmt.Errorf("append event: %w", err) + } + return nil +} + +// EventsFor returns a resource's events, newest first, capped at limit. +func (s *Store) EventsFor(ctx context.Context, kind, name string, limit int) ([]ResourceEvent, error) { + if limit <= 0 { + limit = 100 + } + rows, err := s.db.QueryContext(ctx, + "SELECT id, kind, name, type, reason, commit_, at FROM resource_events WHERE kind = ? AND name = ? ORDER BY id DESC LIMIT ?", + kind, name, limit) + if err != nil { + return nil, fmt.Errorf("query events: %w", err) + } + defer rows.Close() + + var events []ResourceEvent + for rows.Next() { + var e ResourceEvent + if err := rows.Scan(&e.ID, &e.Kind, &e.Name, &e.Type, &e.Reason, &e.Commit, &e.At); err != nil { + return nil, fmt.Errorf("scan event: %w", err) + } + events = append(events, e) + } + return events, rows.Err() +} diff --git a/internal/store/store.go b/internal/store/store.go index 6211143..002b3e7 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -31,6 +31,19 @@ CREATE TABLE IF NOT EXISTS settings ( data BLOB NOT NULL, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ); + +CREATE TABLE IF NOT EXISTS resource_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + kind TEXT NOT NULL, + name TEXT NOT NULL, + type TEXT NOT NULL, + reason TEXT NOT NULL DEFAULT '', + commit_ TEXT NOT NULL DEFAULT '', + at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP +); + +CREATE INDEX IF NOT EXISTS idx_resource_events_resource + ON resource_events (kind, name, id); ` // Store is a handle to the application database. diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 2774b91..525724b 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -149,6 +149,43 @@ func TestSessionCascadeOnAccountDelete(t *testing.T) { } } +func TestResourceEvents(t *testing.T) { + st := open(t) + ctx := context.Background() + + if evs, err := st.EventsFor(ctx, "Iso", "nix", 10); err != nil || len(evs) != 0 { + t.Fatalf("EventsFor empty = %v, %v; want none", evs, err) + } + + for _, e := range []ResourceEvent{ + {Kind: "Iso", Name: "nix", Type: "drifted", Commit: "abc"}, + {Kind: "Iso", Name: "nix", Type: "applied", Reason: "create", Commit: "abc"}, + {Kind: "Iso", Name: "other", Type: "drifted"}, + } { + if err := st.AppendEvent(ctx, e); err != nil { + t.Fatalf("AppendEvent: %v", err) + } + } + + evs, err := st.EventsFor(ctx, "Iso", "nix", 10) + if err != nil { + t.Fatalf("EventsFor: %v", err) + } + if len(evs) != 2 { + t.Fatalf("got %d events for nix, want 2", len(evs)) + } + // Newest first. + if evs[0].Type != "applied" || evs[1].Type != "drifted" { + t.Fatalf("order wrong: %+v", evs) + } + if evs[0].Reason != "create" || evs[0].Commit != "abc" { + t.Errorf("fields not persisted: %+v", evs[0]) + } + if evs[0].At.IsZero() { + t.Error("At not stamped") + } +} + func TestSettingsRoundTrip(t *testing.T) { st := open(t) ctx := context.Background() diff --git a/web/ui/src/components/ResourceCard.vue b/web/ui/src/components/ResourceCard.vue index c0f3c62..e3a5f34 100644 --- a/web/ui/src/components/ResourceCard.vue +++ b/web/ui/src/components/ResourceCard.vue @@ -1,10 +1,10 @@