diff --git a/server/internal/agentapi/push_test.go b/server/internal/agentapi/push_test.go new file mode 100644 index 0000000..74f9c81 --- /dev/null +++ b/server/internal/agentapi/push_test.go @@ -0,0 +1,129 @@ +// SPDX-License-Identifier: LicenseRef-DDI-Proprietary + +package agentapi + +import ( + "net/netip" + "strconv" + "testing" + "time" + + "google.golang.org/grpc/codes" + "google.golang.org/protobuf/types/known/timestamppb" + + agentv1 "git.beugsiud.de/beug.k/DDI/proto/gen/ddi/agent/v1" + "git.beugsiud.de/beug.k/DDI/server/internal/ipam" + "git.beugsiud.de/beug.k/DDI/server/internal/service" +) + +// connect opens a stream, sends Hello and returns the Welcome. +func connect(t *testing.T, a *testAgent, applied uint64) (agentv1.AgentService_ConnectClient, *agentv1.Welcome) { + t.Helper() + stream, err := a.client.Connect(ctxT(t)) + if err != nil { + t.Fatal(err) + } + if err := stream.Send(&agentv1.ConnectRequest{Message: &agentv1.ConnectRequest_Hello{Hello: &agentv1.Hello{ + AppliedRevision: applied, LastGoodRevision: applied}}}); err != nil { + t.Fatal(err) + } + msg, err := stream.Recv() + if err != nil || msg.GetWelcome() == nil { + t.Fatalf("welcome %v %v", msg, err) + } + return stream, msg.GetWelcome() +} + +func TestPushModesAndReports(t *testing.T) { + e := newTestEnv(t) + ctx := ctxT(t) + a := e.enroll("dns") + e.configureGenerator(3600) + if _, _, err := e.svc.CreateRevision(ctx, service.SystemActor, ""); err != nil { + t.Fatal(err) + } + if ag, err := e.svc.SetAgentMode(ctx, service.SystemActor, a.id, "manual"); err != nil || ag.TargetRevision != 1 { + t.Fatalf("mode %+v %v", ag, err) + } + + // A push while the agent is offline is announced by polling ... + targets, err := e.svc.DeployRevision(ctx, service.SystemActor, 1, []string{a.id}) + if err != nil || len(targets) != 1 || targets[0].Sent { + t.Fatalf("deploy %+v %v", targets, err) + } + id := strconv.FormatInt(targets[0].DeploymentID, 10) + st, err := a.client.GetDesiredState(ctx, &agentv1.GetDesiredStateRequest{}) + if err != nil || st.GetDeploymentId() != id || st.GetDesiredRevision() != 1 || st.GetMode() != agentv1.OperatingMode_OPERATING_MODE_MANUAL { + t.Fatalf("desired state %v %v", st, err) + } + // ... and sent again when it connects. + stream, w := connect(t, a, 0) + if w.GetMode() != agentv1.OperatingMode_OPERATING_MODE_MANUAL || w.GetDesiredRevision() != 1 { + t.Errorf("welcome %v", w) + } + msg, err := stream.Recv() + if err != nil || msg.GetCommand().GetApplyRevision().GetRevision() != 1 || msg.GetCommand().GetApplyRevision().GetDeploymentId() != id { + t.Fatalf("pending push %v %v", msg, err) + } + + // The report completes the deployment; it is not announced any more and + // a resent report is accepted without changes. + report := &agentv1.ReportDeploymentRequest{DeploymentId: id, Revision: 1, + Trigger: agentv1.DeploymentTrigger_DEPLOYMENT_TRIGGER_PUSH, Outcome: agentv1.DeploymentOutcome_DEPLOYMENT_OUTCOME_SUCCEEDED, + ActiveRevision: 1, StartedAt: timestamppb.New(time.Now().Add(-time.Second)), FinishedAt: timestamppb.Now()} + for range 2 { + if _, err := a.client.ReportDeployment(ctx, report); err != nil { + t.Fatal(err) + } + } + deps, err := e.svc.ListDeployments(ctx, service.DeploymentFilter{AgentID: a.id}) + if err != nil || len(deps.Items) != 1 || deps.Items[0].Outcome != "succeeded" || deps.Items[0].FinishedAt == nil || + deps.Items[0].RequestedByName != "system" { + t.Fatalf("deployments %+v %v", deps, err) + } + if st, err := a.client.GetDesiredState(ctx, &agentv1.GetDesiredStateRequest{}); err != nil || st.GetDeploymentId() != "" { + t.Errorf("desired state after report %v %v", st, err) + } + + // A manual agent gets no new revisions, but a push while connected. + var netID string + if err := e.st.Pool().QueryRow(ctx, `SELECT id FROM networks`).Scan(&netID); err != nil { + t.Fatal(err) + } + if _, err := e.svc.CreateAddress(ctx, service.SystemActor, service.AddressInput{NetworkID: netID, + IP: netip.MustParseAddr("192.0.2.10"), Type: ipam.TypeStatic, Hostname: "www"}); err != nil { + t.Fatal(err) + } + if rev, created, err := e.svc.CreateRevision(ctx, service.SystemActor, ""); err != nil || !created || rev.Number != 2 { + t.Fatalf("revision 2 %+v %v %v", rev, created, err) + } + targets, err = e.svc.DeployRevision(ctx, service.SystemActor, 2, nil) + if err != nil || len(targets) != 1 || !targets[0].Sent { + t.Fatalf("push %+v %v", targets, err) + } + msg, err = stream.Recv() + if err != nil || msg.GetCommand().GetApplyRevision().GetRevision() != 2 || + msg.GetCommand().GetApplyRevision().GetDeploymentId() != strconv.FormatInt(targets[0].DeploymentID, 10) { + t.Fatalf("push command %v %v", msg, err) + } + + // A mode change reconnects the agent with the new mode. + if _, err := e.svc.SetAgentMode(ctx, service.SystemActor, a.id, "auto_pull"); err != nil { + t.Fatal(err) + } + _, err = stream.Recv() + wantCode(t, err, codes.Unavailable) + stream, w = connect(t, a, 1) + if w.GetMode() != agentv1.OperatingMode_OPERATING_MODE_AUTO_PULL || w.GetDesiredRevision() != 2 { + t.Errorf("welcome after mode change %v", w) + } + // The open push of revision 2 is still pending (the desired revision). + msg, err = stream.Recv() + if err != nil || msg.GetCommand().GetApplyRevision().GetRevision() != 2 { + t.Errorf("open push after reconnect %v %v", msg, err) + } + eventually(t, "connected", func() bool { return e.hub.Connected(a.id) }) + if err := stream.CloseSend(); err != nil { + t.Fatal(err) + } +} diff --git a/server/internal/agentapi/rpcs.go b/server/internal/agentapi/rpcs.go index 059a5a0..ef01587 100644 --- a/server/internal/agentapi/rpcs.go +++ b/server/internal/agentapi/rpcs.go @@ -116,6 +116,7 @@ func (s *Server) GetDesiredState(ctx context.Context, _ *agentv1.GetDesiredState } return &agentv1.GetDesiredStateResponse{ DesiredRevision: st.DesiredRevision, Mode: modeProto(st.Mode), PullInterval: durationpb.New(service.PullInterval), + DeploymentId: st.PushDeploymentID, }, nil } @@ -240,7 +241,10 @@ func (s *Server) Connect(stream agentv1.AgentService_ConnectServer) error { // Register before reading the desired revision, so that a revision // created in between is pushed rather than missed. conn := s.hub.Attach(c.AgentID, c.Role, c.Mode) - defer s.hub.Detach(conn) + defer func() { + s.hub.Detach(conn) + s.svc.AgentDisconnected(c.AgentID) + }() settings, err := s.svc.AgentHello(ctx, c.AgentID, service.HelloInput{ Hostname: hello.GetHostname(), AgentVersion: hello.GetAgentVersion(), ServiceVersion: hello.GetServiceVersion(), AppliedRevision: hello.GetAppliedRevision(), LastGoodRevision: hello.GetLastGoodRevision(), @@ -249,6 +253,9 @@ func (s *Server) Connect(stream agentv1.AgentService_ConnectServer) error { if err != nil { return err } + // The mode read with Hello is the one Welcome announces; it may be + // newer than the one from the authentication (a concurrent mode change). + s.hub.SetMode(c.AgentID, settings.Mode) if err := stream.Send(&agentv1.ConnectResponse{Message: &agentv1.ConnectResponse_Welcome{Welcome: &agentv1.Welcome{ Mode: modeProto(settings.Mode), PullInterval: durationpb.New(service.PullInterval), HeartbeatInterval: durationpb.New(service.HeartbeatInterval), DesiredRevision: settings.DesiredRevision, @@ -259,6 +266,18 @@ func (s *Server) Connect(stream agentv1.AgentService_ConnectServer) error { slog.String("role", c.Role), slog.Uint64("applied_revision", hello.GetAppliedRevision()), slog.Uint64("desired_revision", settings.DesiredRevision)) + // An open push is sent again: the agent may have missed it while it was + // offline (ADR 0022). Only the latest one is still pending. + if settings.PushDeploymentID != "" { + cmd := agenthub.ApplyCommand(settings.DesiredRevision, settings.PushDeploymentID) + if err := stream.Send(&agentv1.ConnectResponse{Message: &agentv1.ConnectResponse_Command{Command: cmd}}); err != nil { + return err + } + s.log.Info("pending push sent", slog.String("request_id", c.RequestID), slog.String("agent_id", c.AgentID), + slog.String("command_id", cmd.GetCommandId()), slog.Uint64("revision", settings.DesiredRevision), + slog.String("deployment_id", settings.PushDeploymentID)) + } + recvErr := make(chan error, 1) go func() { for { diff --git a/server/internal/agenthub/hub.go b/server/internal/agenthub/hub.go index cc33853..90074c2 100644 --- a/server/internal/agenthub/hub.go +++ b/server/internal/agenthub/hub.go @@ -29,7 +29,10 @@ const ( ReasonDeleted = "agent deleted" ReasonReplaced = "replaced by a newer connection" ReasonShutdown = "server shutting down" - reasonEnded = "stream ended" + // ReasonModeChanged closes the stream after a mode change; the agent + // reconnects and receives the new mode in Welcome (ADR 0022). + ReasonModeChanged = "operating mode changed" + reasonEnded = "stream ended" ) // Hub is the registry of open streams. It is safe for concurrent use. @@ -126,6 +129,16 @@ func (h *Hub) Disconnect(agentID, reason string) { } } +// SetMode updates the mode of the agent's open stream (the mode the agent +// was told in Welcome), which decides whether new revisions are pushed. +func (h *Hub) SetMode(agentID, mode string) { + h.mu.Lock() + defer h.mu.Unlock() + if c := h.conns[agentID]; c != nil { + c.Mode = mode + } +} + // Count returns the number of open streams per role. func (h *Hub) Count() map[string]int { h.mu.Lock() @@ -150,21 +163,42 @@ func (h *Hub) NotifyRevision(revision uint64) int { h.mu.Unlock() n := 0 for _, c := range targets { - cmd := &agentv1.Command{ - CommandId: NewID(), - Action: &agentv1.Command_ApplyRevision{ApplyRevision: &agentv1.ApplyRevisionCommand{ - Revision: revision, DeploymentId: NewID(), - }}, - } - select { - case c.cmds <- cmd: + if c.enqueue(ApplyCommand(revision, NewID())) { n++ - default: // queue full: the agent will pull } } return n } +// ApplyCommand builds an ApplyRevisionCommand with a new command ID. +func ApplyCommand(revision uint64, deploymentID string) *agentv1.Command { + return &agentv1.Command{ + CommandId: NewID(), + Action: &agentv1.Command_ApplyRevision{ApplyRevision: &agentv1.ApplyRevisionCommand{ + Revision: revision, DeploymentId: deploymentID, + }}, + } +} + +// Send queues cmd for the agent's stream and reports whether it has one +// (and its queue had room). Without, the agent picks the work up when it +// reconnects or polls. +func (h *Hub) Send(agentID string, cmd *agentv1.Command) bool { + h.mu.Lock() + c := h.conns[agentID] + h.mu.Unlock() + return c != nil && c.enqueue(cmd) +} + +func (c *Conn) enqueue(cmd *agentv1.Command) bool { + select { + case c.cmds <- cmd: + return true + default: // queue full: the agent will pull + return false + } +} + // NewID returns a random UUID (version 4) in canonical lowercase form. func NewID() string { var b [16]byte diff --git a/server/internal/agenthub/hub_test.go b/server/internal/agenthub/hub_test.go index fc399b4..6f3acf7 100644 --- a/server/internal/agenthub/hub_test.go +++ b/server/internal/agenthub/hub_test.go @@ -61,3 +61,33 @@ func TestHub(t *testing.T) { } h.Disconnect("unknown", "x") // no-op } + +func TestSend(t *testing.T) { + h := New() + if h.Send("m", ApplyCommand(3, "12")) { + t.Error("sent to an agent without stream") + } + m := h.Attach("m", "dns", ModeManual) + if !h.Send("m", ApplyCommand(3, "12")) { + t.Fatal("not sent") + } + cmd := <-m.Commands() + if cmd.GetApplyRevision().GetRevision() != 3 || cmd.GetApplyRevision().GetDeploymentId() != "12" || + !agentpki.ValidAgentID(cmd.GetCommandId()) { + t.Errorf("command %v", cmd) + } + for range queueSize { + h.Send("m", ApplyCommand(4, "13")) + } + if h.Send("m", ApplyCommand(5, "14")) { + t.Error("sent to a full queue") + } + + // Only streams in auto_pull mode get new revisions. + x := h.Attach("x", "dhcp", ModeAutoPull) + h.SetMode("x", ModeManual) + h.SetMode("unknown", ModeManual) // no-op + if n := h.NotifyRevision(9); n != 0 || len(x.Commands()) != 0 { + t.Errorf("manual stream notified: %d", n) + } +} diff --git a/server/internal/app/app.go b/server/internal/app/app.go index a26ce98..d1e89a7 100644 --- a/server/internal/app/app.go +++ b/server/internal/app/app.go @@ -20,6 +20,7 @@ import ( "git.beugsiud.de/beug.k/DDI/server/internal/agentapi" "git.beugsiud.de/beug.k/DDI/server/internal/agenthub" "git.beugsiud.de/beug.k/DDI/server/internal/config" + "git.beugsiud.de/beug.k/DDI/server/internal/events" "git.beugsiud.de/beug.k/DDI/server/internal/httpapi" "git.beugsiud.de/beug.k/DDI/server/internal/httpserver" "git.beugsiud.de/beug.k/DDI/server/internal/license" @@ -66,6 +67,7 @@ type Core struct { Service *service.Service PKI *pki.Authority Hub *agenthub.Hub + Events *events.Bus Log *slog.Logger } @@ -104,10 +106,12 @@ func Open(ctx context.Context, cfg config.Config, log *slog.Logger, migrate bool return nil, fmt.Errorf("agent PKI: %w", err) } core.Hub = agenthub.New() + core.Events = events.New() core.Service = service.New(service.Options{ Store: st, Keyring: kr, License: core.License, ProductName: brand.DisplayName, SessionLifetime: cfg.Session.Lifetime.Std(), SessionIdle: cfg.Session.IdleTimeout.Std(), Logger: log, PKI: core.PKI, Hub: core.Hub, AgentAPIAddress: cfg.AgentAPI.PublicAddress, AgentServerNames: cfg.AgentAPI.ServerNames, + Events: core.Events, }) return core, nil } @@ -212,6 +216,9 @@ func Serve(ctx context.Context, core *Core) error { log.Info("shutting down", slog.Duration("timeout", cfg.ShutdownTimeout.Std())) shutdownCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), cfg.ShutdownTimeout.Std()) defer cancel() + // Event streams never become idle: end them so that the HTTP server + // can shut down gracefully. + core.Events.Close() agentSrv.Shutdown(shutdownCtx) if err := srv.Shutdown(shutdownCtx); err != nil && runErr == nil { runErr = fmt.Errorf("graceful shutdown: %w", err) diff --git a/server/internal/events/events.go b/server/internal/events/events.go new file mode 100644 index 0000000..50bb6da --- /dev/null +++ b/server/internal/events/events.go @@ -0,0 +1,161 @@ +// SPDX-License-Identifier: LicenseRef-DDI-Proprietary + +// Package events is the in-process event bus behind the live updates of the +// web UI (GET /api/v1/events, ADR 0022). Events carry IDs only; the UI +// reloads the affected data. Publishers never block: a subscriber whose +// buffer is full misses events (it falls back to reloading everything when +// it reconnects or polls). +package events + +import ( + "encoding/json" + "sync" + "sync/atomic" +) + +// Event types (the SSE "event:" field). +const ( + TypeAgent = "agent" + TypeDeployment = "deployment" + TypeRevision = "revision" + TypePending = "pending" +) + +// Agent says that an agent changed (status, connection, mode, revocation, +// deletion). +type Agent struct { + ID string `json:"id"` +} + +// Deployment says that a deployment was created or completed. +type Deployment struct { + ID int64 `json:"id"` + AgentID string `json:"agent_id"` + Revision int64 `json:"revision"` +} + +// Revision says that a revision was created (also by a rollback). +type Revision struct { + Number int64 `json:"number"` +} + +// Pending says that data used by the generator changed. +type Pending struct{} + +// Event is one published event with its JSON payload. +type Event struct { + Type string + Data json.RawMessage +} + +// DefaultBuffer is the buffer of a subscription. +const DefaultBuffer = 64 + +// Bus fans events out to subscribers. The zero value is not usable; a nil +// *Bus accepts and drops everything. +type Bus struct { + mu sync.Mutex + subs map[*Subscription]struct{} + closed bool +} + +// New creates a bus. +func New() *Bus { return &Bus{subs: map[*Subscription]struct{}{}} } + +// Subscription receives events until it is closed. +type Subscription struct { + bus *Bus + ch chan Event + once sync.Once + dropped atomic.Int64 +} + +// C delivers the events. It is closed when the subscription or the bus is +// closed. +func (s *Subscription) C() <-chan Event { return s.ch } + +// Dropped returns the number of events missed because the buffer was full. +func (s *Subscription) Dropped() int64 { return s.dropped.Load() } + +// Close ends the subscription. +func (s *Subscription) Close() { + if s.bus != nil { + s.bus.mu.Lock() + delete(s.bus.subs, s) + s.bus.mu.Unlock() + } + s.once.Do(func() { close(s.ch) }) +} + +// Subscribe registers a subscriber with a buffer of the given size +// (DefaultBuffer if not positive). On a closed or nil bus the subscription +// is closed already. +func (b *Bus) Subscribe(buffer int) *Subscription { + if buffer <= 0 { + buffer = DefaultBuffer + } + s := &Subscription{bus: b, ch: make(chan Event, buffer)} + if b == nil { + s.Close() + return s + } + b.mu.Lock() + if b.closed { + b.mu.Unlock() + s.Close() + return s + } + b.subs[s] = struct{}{} + b.mu.Unlock() + return s +} + +// Publish sends an event to every subscriber without blocking. data is +// encoded as JSON once. +func (b *Bus) Publish(typ string, data any) { + if b == nil { + return + } + raw, err := json.Marshal(data) + if err != nil { + return // our own payload types always encode + } + ev := Event{Type: typ, Data: raw} + b.mu.Lock() + defer b.mu.Unlock() + // Sending under the lock keeps Close from closing a channel that is + // being written to; the sends never block. + for s := range b.subs { + select { + case s.ch <- ev: + default: + s.dropped.Add(1) + } + } +} + +// Subscribers returns the number of open subscriptions. +func (b *Bus) Subscribers() int { + if b == nil { + return 0 + } + b.mu.Lock() + defer b.mu.Unlock() + return len(b.subs) +} + +// Close ends all subscriptions (server shutdown); later subscriptions are +// closed immediately and publishing does nothing. +func (b *Bus) Close() { + if b == nil { + return + } + b.mu.Lock() + subs := b.subs + b.subs = map[*Subscription]struct{}{} + b.closed = true + b.mu.Unlock() + for s := range subs { + s.once.Do(func() { close(s.ch) }) + } +} diff --git a/server/internal/events/events_test.go b/server/internal/events/events_test.go new file mode 100644 index 0000000..0592b8a --- /dev/null +++ b/server/internal/events/events_test.go @@ -0,0 +1,105 @@ +// SPDX-License-Identifier: LicenseRef-DDI-Proprietary + +package events + +import ( + "sync" + "testing" +) + +func TestFanOut(t *testing.T) { + b := New() + s1 := b.Subscribe(4) + s2 := b.Subscribe(4) + if b.Subscribers() != 2 { + t.Fatalf("subscribers %d", b.Subscribers()) + } + b.Publish(TypeAgent, Agent{ID: "a1"}) + b.Publish(TypeDeployment, Deployment{ID: 7, AgentID: "a1", Revision: 3}) + for _, s := range []*Subscription{s1, s2} { + ev := <-s.C() + if ev.Type != TypeAgent || string(ev.Data) != `{"id":"a1"}` { + t.Errorf("first event %s %s", ev.Type, ev.Data) + } + ev = <-s.C() + if ev.Type != TypeDeployment || string(ev.Data) != `{"id":7,"agent_id":"a1","revision":3}` { + t.Errorf("second event %s %s", ev.Type, ev.Data) + } + } + b.Publish(TypeRevision, Revision{Number: 2}) + b.Publish(TypePending, Pending{}) + if ev := <-s1.C(); string(ev.Data) != `{"number":2}` { + t.Errorf("revision %s", ev.Data) + } + if ev := <-s1.C(); ev.Type != TypePending || string(ev.Data) != `{}` { + t.Errorf("pending %s %s", ev.Type, ev.Data) + } + + s1.Close() + s1.Close() // idempotent + if _, ok := <-s1.C(); ok { + t.Error("closed subscription delivers") + } + if b.Subscribers() != 1 { + t.Errorf("subscribers after close %d", b.Subscribers()) + } +} + +func TestSlowSubscriberDropsEvents(t *testing.T) { + b := New() + slow := b.Subscribe(2) + fast := b.Subscribe(100) + for range 10 { + b.Publish(TypePending, Pending{}) // must not block + } + if slow.Dropped() != 8 || len(slow.C()) != 2 { + t.Errorf("slow: dropped %d, buffered %d", slow.Dropped(), len(slow.C())) + } + if fast.Dropped() != 0 || len(fast.C()) != 10 { + t.Errorf("fast: dropped %d, buffered %d", fast.Dropped(), len(fast.C())) + } +} + +func TestCloseBus(t *testing.T) { + b := New() + s := b.Subscribe(0) + b.Close() + for range s.C() { + t.Error("event after close") + } + b.Publish(TypePending, Pending{}) + late := b.Subscribe(1) + if _, ok := <-late.C(); ok { + t.Error("subscription on a closed bus is open") + } + s.Close() + + var nilBus *Bus + nilBus.Publish(TypePending, Pending{}) + nilBus.Close() + if _, ok := <-nilBus.Subscribe(1).C(); ok || nilBus.Subscribers() != 0 { + t.Error("nil bus subscription is open") + } +} + +func TestConcurrentPublishAndClose(t *testing.T) { + b := New() + var wg sync.WaitGroup + for range 4 { + wg.Go(func() { + for range 1000 { + b.Publish(TypePending, Pending{}) + } + }) + } + for range 4 { + wg.Go(func() { + for range 100 { + s := b.Subscribe(1) + s.Close() + } + }) + } + wg.Wait() + b.Close() +} diff --git a/server/internal/httpapi/api.go b/server/internal/httpapi/api.go index 1a44591..c8f9457 100644 --- a/server/internal/httpapi/api.go +++ b/server/internal/httpapi/api.go @@ -39,6 +39,9 @@ type Options struct { Now func() time.Time // OIDCHTTPClient is used for discovery and token requests (tests). OIDCHTTPClient *http.Client + // EventKeepalive is the keepalive and session check interval of the + // event stream (default DefaultEventKeepalive; shorter in tests). + EventKeepalive time.Duration } // API serves /api/v1. diff --git a/server/internal/httpapi/h_deploy.go b/server/internal/httpapi/h_deploy.go new file mode 100644 index 0000000..f54eafd --- /dev/null +++ b/server/internal/httpapi/h_deploy.go @@ -0,0 +1,210 @@ +// SPDX-License-Identifier: LicenseRef-DDI-Proprietary + +package httpapi + +import ( + "bytes" + "errors" + "io" + "log/slog" + "net/http" + "strconv" + "time" + + "git.beugsiud.de/beug.k/DDI/server/internal/httpserver" + "git.beugsiud.de/beug.k/DDI/server/internal/service" + "git.beugsiud.de/beug.k/DDI/server/internal/store" +) + +// DefaultEventKeepalive is the interval of keepalive comments on the event +// stream; the session is checked at the same interval (ADR 0022). +const DefaultEventKeepalive = 25 * time.Second + +// decodeOptional decodes an optional JSON body into dst. +func decodeOptional(w http.ResponseWriter, r *http.Request, dst any) error { + body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxBody)) + if err != nil { + return errBadRequest("body_too_large", "request body too large") + } + if len(bytes.TrimSpace(body)) == 0 { + return nil + } + r.Body = io.NopCloser(bytes.NewReader(body)) + return decode(w, r, dst) +} + +// revisionNumber parses the {number} path parameter; anything invalid is a +// revision that does not exist. +func revisionNumber(r *http.Request) (int64, error) { + n, err := strconv.ParseInt(r.PathValue("number"), 10, 64) + if err != nil || n < 1 || n > store.MaxRevision { + return 0, service.ErrNotFound + } + return n, nil +} + +type agentUpdateRequest struct { + Mode string `json:"mode"` +} + +func (a *API) updateAgent(w http.ResponseWriter, r *http.Request, rc *reqCtx) error { + var req agentUpdateRequest + if err := decode(w, r, &req); err != nil { + return err + } + ag, err := a.svc.SetAgentMode(r.Context(), rc.Actor, r.PathValue("id"), req.Mode) + if err != nil { + return err + } + return writeJSON(w, http.StatusOK, ag) +} + +type deployRequest struct { + AgentIDs []string `json:"agent_ids"` +} + +func (a *API) deployRevision(w http.ResponseWriter, r *http.Request, rc *reqCtx) error { + n, err := revisionNumber(r) + if err != nil { + return err + } + var req deployRequest + if err := decodeOptional(w, r, &req); err != nil { + return err + } + targets, err := a.svc.DeployRevision(r.Context(), rc.Actor, n, req.AgentIDs) + if err != nil { + return err + } + return writeJSON(w, http.StatusOK, listResponse[service.DeployTarget]{Items: targets}) +} + +func (a *API) rollbackRevision(w http.ResponseWriter, r *http.Request, rc *reqCtx) error { + n, err := revisionNumber(r) + if err != nil { + return err + } + var req revisionRequest + if err := decodeOptional(w, r, &req); err != nil { + return err + } + rev, err := a.svc.RollbackRevision(r.Context(), rc.Actor, n, req.Comment) + if err != nil { + return err + } + return writeJSON(w, http.StatusCreated, rev) +} + +func (a *API) diffRevision(w http.ResponseWriter, r *http.Request, _ *reqCtx) error { + n, err := revisionNumber(r) + if err != nil { + return err + } + var against *int64 + if v := r.URL.Query().Get("against"); v != "" { + m, err := strconv.ParseInt(v, 10, 64) + if err != nil || m < 0 { + return errBadRequest("invalid_parameter", "invalid against") + } + against = &m + } + d, err := a.svc.DiffRevisions(r.Context(), n, against) + if err != nil { + return err + } + return writeJSON(w, http.StatusOK, d) +} + +func (a *API) pendingChanges(w http.ResponseWriter, r *http.Request, _ *reqCtx) error { + p, err := a.svc.PendingChanges(r.Context()) + if err != nil { + return err + } + return writeJSON(w, http.StatusOK, p) +} + +var errEventsUnavailable = &service.Error{Kind: service.KindUnavailable, Code: "events_unavailable", + Message: "live events are not available"} + +// streamEvents serves the Server-Sent Events stream (ADR 0022) until the +// client goes away, the session ends or the server shuts down. +func (a *API) streamEvents(w http.ResponseWriter, r *http.Request, rc *reqCtx) error { + bus := a.svc.Events() + if bus == nil { + return errEventsUnavailable + } + ctl := http.NewResponseController(w) + // The server's read and write timeouts are meant for ordinary + // requests: lift the read deadline (it would cancel the request) and + // move the write deadline ahead before every write. + if err := ctl.SetReadDeadline(time.Time{}); err != nil && !errors.Is(err, http.ErrNotSupported) { + return err + } + keepalive := a.opts.EventKeepalive + if keepalive <= 0 { + keepalive = DefaultEventKeepalive + } + sub := bus.Subscribe(0) + defer sub.Close() + + write := func(b []byte) error { + if err := ctl.SetWriteDeadline(time.Now().Add(keepalive + 30*time.Second)); err != nil && !errors.Is(err, http.ErrNotSupported) { + return err + } + if _, err := w.Write(b); err != nil { + return err + } + return ctl.Flush() + } + + h := w.Header() + h.Set("Content-Type", "text/event-stream") + h.Set("Cache-Control", "no-store") + h.Set("X-Accel-Buffering", "no") // reverse proxies must not buffer + w.WriteHeader(http.StatusOK) + // The retry hint makes EventSource reconnect after 5 s. + if err := write([]byte("retry: 5000\n: connected\n\n")); err != nil { + return nil //nolint:nilerr // the client is gone; nothing to report + } + + ticker := time.NewTicker(keepalive) + defer ticker.Stop() + expiry := time.NewTimer(time.Until(rc.Session.ExpiresAt)) + defer expiry.Stop() + log := a.log.With(slog.String("request_id", httpserver.RequestIDFromContext(r.Context()))) + for { + select { + case <-r.Context().Done(): + return nil + case <-expiry.C: + return nil + case ev, ok := <-sub.C(): + if !ok { + return nil // server shutdown + } + msg := make([]byte, 0, len(ev.Type)+len(ev.Data)+16) + msg = append(msg, "event: "...) + msg = append(msg, ev.Type...) + msg = append(msg, "\ndata: "...) + msg = append(msg, ev.Data...) + msg = append(msg, "\n\n"...) + if err := write(msg); err != nil { + return nil //nolint:nilerr // the client is gone + } + case <-ticker.C: + active, err := a.svc.SessionActive(r.Context(), rc.Session.TokenHash) + if err != nil && r.Context().Err() == nil { + log.Warn("event stream: session check failed", slog.Any("error", err)) + } + if !active { + return nil + } + if err := write([]byte(": keepalive\n\n")); err != nil { + return nil //nolint:nilerr // the client is gone + } + if n := sub.Dropped(); n > 0 { + log.Debug("event stream: slow client dropped events", slog.Int64("dropped", n)) + } + } + } +} diff --git a/server/internal/httpapi/harness_test.go b/server/internal/httpapi/harness_test.go index 919db85..4494090 100644 --- a/server/internal/httpapi/harness_test.go +++ b/server/internal/httpapi/harness_test.go @@ -23,6 +23,7 @@ import ( "git.beugsiud.de/beug.k/DDI/server/api" "git.beugsiud.de/beug.k/DDI/server/internal/agenthub" + "git.beugsiud.de/beug.k/DDI/server/internal/events" "git.beugsiud.de/beug.k/DDI/server/internal/httpserver" "git.beugsiud.de/beug.k/DDI/server/internal/license" "git.beugsiud.de/beug.k/DDI/server/internal/pki" @@ -97,6 +98,12 @@ func contractHandler(t *testing.T, next http.Handler) http.Handler { } r.Body = io.NopCloser(bytes.NewReader(body)) + if route.Operation.OperationID == "streamEvents" { + // A stream cannot be buffered: check status and content + // type as they are written (TestEventStream checks the rest). + next.ServeHTTP(&streamChecker{ResponseWriter: w, t: t, path: r.URL.Path}, r) + return + } rec := httptest.NewRecorder() next.ServeHTTP(rec, r) @@ -115,6 +122,38 @@ func contractHandler(t *testing.T, next http.Handler) http.Handler { }) } +// streamChecker validates the headers of a streaming response. +type streamChecker struct { + http.ResponseWriter + t *testing.T + path string + wrote bool +} + +func (s *streamChecker) WriteHeader(code int) { + if !s.wrote { + s.wrote = true + ct := s.Header().Get("Content-Type") + switch { + case code == http.StatusOK && ct != "text/event-stream": + s.t.Errorf("contract: stream %s has content type %q", s.path, ct) + case code != http.StatusOK && ct != "application/problem+json": + s.t.Errorf("contract: error %d of %s has content type %q", code, s.path, ct) + } + } + s.ResponseWriter.WriteHeader(code) +} + +func (s *streamChecker) Write(p []byte) (int, error) { + if !s.wrote { + s.WriteHeader(http.StatusOK) + } + return s.ResponseWriter.Write(p) +} + +// Unwrap lets http.ResponseController reach Flush and the deadlines. +func (s *streamChecker) Unwrap() http.ResponseWriter { return s.ResponseWriter } + // env is a running API backed by a fresh database. type env struct { t *testing.T @@ -125,6 +164,7 @@ type env struct { now *time.Time pki *pki.Authority hub *agenthub.Hub + bus *events.Bus } type stubProvider struct{} @@ -161,7 +201,10 @@ func (p storePersistence) EnsureInstalledAt(ctx context.Context, now time.Time) func newEnv(t *testing.T) *env { return newEnvWith(t, nil) } // newEnvWith allows tests to adjust the API options before start. -func newEnvWith(t *testing.T, adjust func(*Options)) *env { +func newEnvWith(t *testing.T, adjust func(*Options)) *env { return newEnvServer(t, adjust, nil) } + +// newEnvServer also allows tests to adjust the HTTP server (timeouts). +func newEnvServer(t *testing.T, adjust func(*Options), adjustServer func(*http.Server)) *env { t.Helper() st := storetest.New(t) keyData, _ := secrets.GenerateKeyFile("k1") @@ -182,10 +225,11 @@ func newEnvWith(t *testing.T, adjust func(*Options)) *env { t.Fatal(err) } hub := agenthub.New() + bus := events.New() svc := service.New(service.Options{Store: st, Keyring: kr, License: lic, ProductName: "DDI-Suite", Logger: log, Now: func() time.Time { return now }, PKI: authority, Hub: hub, AgentAPIAddress: "ddi.example.test:8444", - AgentServerNames: []string{"ddi.example.test"}}) - opts := Options{Service: svc, ProductName: "DDI-Suite", Version: "test", Logger: log} + AgentServerNames: []string{"ddi.example.test"}, Events: bus}) + opts := Options{Service: svc, ProductName: "DDI-Suite", Version: "test", Logger: log, EventKeepalive: 100 * time.Millisecond} srv := httptest.NewUnstartedServer(nil) if adjust != nil { opts.PublicURL = "http://" + srv.Listener.Addr().String() @@ -193,9 +237,14 @@ func newEnvWith(t *testing.T, adjust func(*Options)) *env { } a := New(opts) srv.Config.Handler = httpserver.RequestID(contractHandler(t, a.Handler())) + if adjustServer != nil { + adjustServer(srv.Config) + } srv.Start() t.Cleanup(srv.Close) - return &env{t: t, srv: srv, svc: svc, st: st, lic: lic, now: &now, pki: authority, hub: hub} + // Runs first: open event streams end, so that Close does not wait. + t.Cleanup(bus.Close) + return &env{t: t, srv: srv, svc: svc, st: st, lic: lic, now: &now, pki: authority, hub: hub, bus: bus} } // advance moves the shared clock (sessions, license). diff --git a/server/internal/httpapi/push_test.go b/server/internal/httpapi/push_test.go new file mode 100644 index 0000000..3146c8a --- /dev/null +++ b/server/internal/httpapi/push_test.go @@ -0,0 +1,582 @@ +// SPDX-License-Identifier: LicenseRef-DDI-Proprietary + +package httpapi + +import ( + "bufio" + "context" + "encoding/json" + "net/http" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" + + agentv1 "git.beugsiud.de/beug.k/DDI/proto/gen/ddi/agent/v1" + "git.beugsiud.de/beug.k/DDI/server/internal/agenthub" + "git.beugsiud.de/beug.k/DDI/server/internal/service" + "git.beugsiud.de/beug.k/DDI/server/internal/store" +) + +// officeNetwork creates a network with a name server entry. +func (c *client) officeNetwork() { + c.e.t.Helper() + c.mustStatus(201, "POST", "/api/v1/networks", map[string]any{ + "cidr": "192.0.2.0/24", "name": "Office", "gateway": "192.0.2.1", "domain": "office.test", + "dhcp_enabled": true, "pools": []map[string]string{{"start": "192.0.2.100", "end": "192.0.2.199"}}, + }) + c.mustStatus(201, "POST", "/api/v1/addresses", map[string]any{"ip": "192.0.2.53", "type": "static", "hostname": "ns1"}) +} + +func (c *client) addHost(ip, name string) { + c.e.t.Helper() + c.mustStatus(201, "POST", "/api/v1/addresses", map[string]any{"ip": ip, "type": "static", "hostname": name}) +} + +func (c *client) newRevision() service.RevisionDTO { + c.e.t.Helper() + var rev service.RevisionDTO + c.mustStatus(201, "POST", "/api/v1/revisions", nil).json(c.e.t, &rev) + return rev +} + +func (c *client) agent(id string) service.AgentDTO { + c.e.t.Helper() + var ag service.AgentDTO + c.mustStatus(200, "GET", "/api/v1/agents/"+id, nil).json(c.e.t, &ag) + return ag +} + +type deployResponse = listResponse[service.DeployTarget] + +func (c *client) deploy(n int64, agentIDs ...string) []service.DeployTarget { + c.e.t.Helper() + var body any + if len(agentIDs) > 0 { + body = map[string]any{"agent_ids": agentIDs} + } + var out deployResponse + c.mustStatus(200, "POST", "/api/v1/revisions/"+strconv.FormatInt(n, 10)+"/deploy", body).json(c.e.t, &out) + return out.Items +} + +// command returns the next command of conn or fails. +func command(t *testing.T, conn *agenthub.Conn) *agentv1.Command { + t.Helper() + select { + case cmd := <-conn.Commands(): + return cmd + case <-time.After(2 * time.Second): + t.Fatal("no command") + return nil + } +} + +func noCommand(t *testing.T, conn *agenthub.Conn) { + t.Helper() + select { + case cmd := <-conn.Commands(): + t.Errorf("unexpected command %v", cmd) + default: + } +} + +func (e *env) deployment(id int64) service.DeploymentDTO { + e.t.Helper() + page, err := e.svc.ListDeployments(context.Background(), service.DeploymentFilter{Limit: 500}) + if err != nil { + e.t.Fatal(err) + } + for _, d := range page.Items { + if d.ID == id { + return d + } + } + e.t.Fatalf("deployment %d not found", id) + return service.DeploymentDTO{} +} + +func TestAgentModesAndPush(t *testing.T) { + e := newEnv(t) + admin := e.adminClient() + viewer := e.userClient(admin, "vera", "viewer") + operator := e.userClient(admin, "otto", "operator") + ctx := context.Background() + admin.officeNetwork() + admin.configureDNS() + + auto := e.enroll(admin.createToken("dns").Secret, "dns") + manual := e.enroll(admin.createToken("dhcp").Secret, "dhcp") + if rev := operator.newRevision(); rev.Number != 1 { + t.Fatalf("revision %d", rev.Number) + } + + // Switching to manual keeps the agent on the latest revision and + // reconnects it. + conn := e.hub.Attach(manual, "dhcp", agenthub.ModeAutoPull) + viewer.mustStatus(403, "PATCH", "/api/v1/agents/"+manual, map[string]any{"mode": "manual"}) + operator.mustStatus(403, "PATCH", "/api/v1/agents/"+manual, map[string]any{"mode": "manual"}) + admin.mustStatus(422, "PATCH", "/api/v1/agents/"+manual, map[string]any{"mode": "sometimes"}, skipRequestValidation, "1") + admin.mustStatus(404, "PATCH", "/api/v1/agents/00000000-0000-4000-8000-000000000000", map[string]any{"mode": "manual"}) + var ag service.AgentDTO + admin.mustStatus(200, "PATCH", "/api/v1/agents/"+manual, map[string]any{"mode": "manual"}).json(t, &ag) + if ag.Mode != "manual" || ag.TargetRevision != 1 || ag.DesiredRevision != 1 || ag.Connected { + t.Errorf("manual agent %+v", ag) + } + if conn.Reason() != agenthub.ReasonModeChanged { + t.Errorf("stream not closed: %q", conn.Reason()) + } + // Unchanged mode: no audit entry, no reconnect. + mconn := e.hub.Attach(manual, "dhcp", agenthub.ModeManual) + admin.mustStatus(200, "PATCH", "/api/v1/agents/"+manual, map[string]any{"mode": "manual"}) + select { + case <-mconn.Done(): + t.Error("unchanged mode closed the stream") + default: + } + var audit auditPage + admin.mustStatus(200, "GET", "/api/v1/audit?action=agent.update", nil).json(t, &audit) + if len(audit.Items) != 1 { + t.Errorf("mode audit %+v", audit.Items) + } + + // A new revision goes to auto_pull agents only. + aconn := e.hub.Attach(auto, "dns", agenthub.ModeAutoPull) + admin.addHost("192.0.2.10", "www") + if rev := operator.newRevision(); rev.Number != 2 { + t.Fatalf("revision %d", rev.Number) + } + if cmd := command(t, aconn); cmd.GetApplyRevision().GetRevision() != 2 { + t.Errorf("auto_pull command %v", cmd) + } + noCommand(t, mconn) + if a := viewer.agent(auto); a.DesiredRevision != 2 || a.TargetRevision != 0 { + t.Errorf("auto_pull agent %+v", a) + } + if m := viewer.agent(manual); m.DesiredRevision != 1 { + t.Errorf("manual agent %+v", m) + } + + // Pushing an older revision to an auto_pull agent needs a rollback. + r := operator.mustStatus(422, "POST", "/api/v1/revisions/1/deploy", nil) + if p := r.problem(t); p.Code != "rollback_required" || p.Params["latest"] != "2" || len(p.Problems) != 1 || + p.Problems[0].Params["agent_id"] != auto { + t.Errorf("rollback required %+v", p) + } + viewer.mustStatus(403, "POST", "/api/v1/revisions/1/deploy", map[string]any{"agent_ids": []string{manual}}) + operator.mustStatus(404, "POST", "/api/v1/revisions/99/deploy", nil) + r = operator.mustStatus(422, "POST", "/api/v1/revisions/1/deploy", + map[string]any{"agent_ids": []string{"nonsense", "00000000-0000-4000-8000-000000000000"}}) + if p := r.problem(t); p.Code != "validation_failed" || len(p.Problems) != 2 || p.Problems[0].Code != "unknown_agent" || + p.Problems[0].Field != "agent_ids" { + t.Errorf("unknown agents %+v", p) + } + + // Push revision 1 to the manual agent: pending deployment, command + // with the deployment ID. + targets := operator.deploy(1, manual) + if len(targets) != 1 || targets[0].AgentID != manual || !targets[0].Sent || targets[0].DeploymentID == 0 || + targets[0].AgentHostname != "host-dhcp" { + t.Fatalf("targets %+v", targets) + } + first := targets[0].DeploymentID + cmd := command(t, mconn) + if cmd.GetApplyRevision().GetRevision() != 1 || cmd.GetApplyRevision().GetDeploymentId() != strconv.FormatInt(first, 10) { + t.Errorf("push command %v", cmd) + } + if d := e.deployment(first); d.Outcome != "pending" || d.FinishedAt != nil || d.RequestedByName != "otto" || d.Trigger != "push" { + t.Errorf("pending deployment %+v", d) + } + st, err := e.svc.AgentDesiredState(ctx, manual, nil) + if err != nil || st.DesiredRevision != 1 || st.PushDeploymentID != strconv.FormatInt(first, 10) { + t.Errorf("desired state %+v %v", st, err) + } + + // A newer push supersedes the open one and moves the target. + second := operator.deploy(2, manual)[0].DeploymentID + command(t, mconn) + if d := e.deployment(first); d.Outcome != "superseded" || d.FinishedAt == nil { + t.Errorf("superseded %+v", d) + } + if m := viewer.agent(manual); m.TargetRevision != 2 || m.DesiredRevision != 2 || m.LastDeployment == nil || + m.LastDeployment.ID != second { + t.Errorf("manual agent after push %+v", m) + } + + // The agent reports: the deployment is completed, polling no longer + // announces it, a resent report changes nothing. + started := e.now.Add(-time.Second).Truncate(time.Microsecond) + rep := service.DeploymentReport{CorrelationID: strconv.FormatInt(second, 10), Revision: 2, Trigger: "push", + Outcome: "succeeded", Message: "ok", ActiveRevision: 2, StartedAt: started, FinishedAt: *e.now} + for range 2 { + if err := e.svc.RecordDeployment(ctx, manual, rep); err != nil { + t.Fatal(err) + } + } + if d := e.deployment(second); d.Outcome != "succeeded" || d.FinishedAt == nil || d.Message != "ok" || !d.StartedAt.Equal(started) { + t.Errorf("completed %+v", d) + } + if st, _ := e.svc.AgentDesiredState(ctx, manual, nil); st.PushDeploymentID != "" || st.DesiredRevision != 2 { + t.Errorf("desired state after report %+v", st) + } + var deps service.DeploymentPage + viewer.mustStatus(200, "GET", "/api/v1/deployments?agent_id="+manual, nil).json(t, &deps) + if len(deps.Items) != 2 { + t.Errorf("deployments %+v", deps.Items) + } + if m := viewer.agent(manual); m.AppliedRevision != 2 || m.LastGoodRevision != 2 { + t.Errorf("agent state after report %+v", m) + } + // A report with an unknown push ID is stored like a pull report. + if err := e.svc.RecordDeployment(ctx, manual, service.DeploymentReport{CorrelationID: "999999", Revision: 2, + Trigger: "push", Outcome: "succeeded", ActiveRevision: 2, StartedAt: *e.now, FinishedAt: *e.now}); err != nil { + t.Fatal(err) + } + + // No agent goes back behind the revision it runs. + r = operator.mustStatus(422, "POST", "/api/v1/revisions/1/deploy", map[string]any{"agent_ids": []string{manual}}) + if p := r.problem(t); p.Code != "rollback_required" || p.Problems[0].Params["applied_revision"] != "2" { + t.Errorf("older than applied %+v", p) + } + + // Push to all agents: the latest revision is fine for both modes. + all := operator.deploy(2) + if len(all) != 2 { + t.Fatalf("push to all %+v", all) + } + if cmd := command(t, aconn); cmd.GetApplyRevision().GetDeploymentId() == "" { + t.Errorf("auto_pull push %v", cmd) + } + command(t, mconn) + + // Disconnected agents get the push later (sent false). + e.hub.Disconnect(auto, "test") + var autoPush int64 + for _, tg := range operator.deploy(2, auto) { + if tg.Sent { + t.Errorf("sent to a disconnected agent %+v", tg) + } + autoPush = tg.DeploymentID + } + set, err := e.svc.AgentHello(ctx, auto, service.HelloInput{AppliedRevision: 2, LastGoodRevision: 2}) + if err != nil || set.PushDeploymentID != strconv.FormatInt(autoPush, 10) || set.DesiredRevision != 2 { + t.Errorf("hello settings %+v %v", set, err) + } + + // A new revision supersedes the open push of the auto_pull agent. + admin.addHost("192.0.2.11", "mail") + operator.newRevision() + if d := e.deployment(autoPush); d.Outcome != "superseded" { + t.Errorf("auto_pull push after a new revision %+v", d) + } + + // Back to auto_pull: open pushes of older revisions are superseded. + open := operator.deploy(2, manual)[0].DeploymentID + admin.mustStatus(200, "PATCH", "/api/v1/agents/"+manual, map[string]any{"mode": "auto_pull"}).json(t, &ag) + if ag.DesiredRevision != 3 || ag.TargetRevision != 2 { + t.Errorf("auto_pull again %+v", ag) + } + if d := e.deployment(open); d.Outcome != "superseded" { + t.Errorf("open push after mode change %+v", d) + } + + // Revoked agents are rejected; read-only mode blocks pushes. + admin.mustStatus(200, "POST", "/api/v1/agents/"+manual+"/revoke", nil) + r = operator.mustStatus(422, "POST", "/api/v1/revisions/3/deploy", map[string]any{"agent_ids": []string{manual}}) + if p := r.problem(t); len(p.Problems) != 1 || p.Problems[0].Code != "agent_revoked" { + t.Errorf("revoked %+v", p) + } + if got := operator.deploy(3); len(got) != 1 || got[0].AgentID != auto { + t.Errorf("push to all skips revoked agents: %+v", got) + } + admin.mustStatus(200, "GET", "/api/v1/audit?action=revision.deploy", nil).json(t, &audit) + if len(audit.Items) != 6 { + t.Errorf("deploy audit %d", len(audit.Items)) + } + e.advance(31 * 24 * time.Hour) + operator.login("otto", "changed-password-1") + if p := operator.mustStatus(403, "POST", "/api/v1/revisions/3/deploy", nil).problem(t); p.Code != "license_read_only" { + t.Errorf("read-only %+v", p) + } +} + +func TestRollbackAndDiff(t *testing.T) { + e := newEnv(t) + admin := e.adminClient() + viewer := e.userClient(admin, "vera", "viewer") + operator := e.userClient(admin, "otto", "operator") + admin.officeNetwork() + admin.configureDNS() + auto := e.enroll(admin.createToken("dns").Secret, "dns") + conn := e.hub.Attach(auto, "dns", agenthub.ModeAutoPull) + + operator.newRevision() + command(t, conn) + admin.addHost("192.0.2.10", "www") + operator.newRevision() + command(t, conn) + + var d service.RevisionDiff + viewer.mustStatus(200, "GET", "/api/v1/revisions/2/diff", nil).json(t, &d) + if d.Number != 2 || d.Against != 1 || len(d.Files) == 0 { + t.Fatalf("diff %+v", d) + } + var zone *service.FileDiff + for i := range d.Files { + if d.Files[i].Status != "modified" || d.Files[i].Role != "dns" { + t.Errorf("file %+v", d.Files[i]) + } + if d.Files[i].Path == "zones/db.office.test" { + zone = &d.Files[i] + } + } + if zone == nil || !strings.Contains(zone.Diff, "192.0.2.10") || zone.Inserted < 3 || zone.Deleted < 2 || !strings.HasPrefix(zone.Diff, "@@ ") { + t.Errorf("zone diff %+v", zone) + } + d = service.RevisionDiff{} + viewer.mustStatus(200, "GET", "/api/v1/revisions/2/diff?against=0", nil).json(t, &d) + for _, f := range d.Files { + if f.Status != "added" || f.Deleted != 0 || f.Inserted == 0 { + t.Errorf("against nothing %+v", f) + } + } + if len(d.Files) < 2 || d.Files[len(d.Files)-1].Role != "dhcp" { + t.Errorf("files %+v", d.Files) + } + d = service.RevisionDiff{} + viewer.mustStatus(200, "GET", "/api/v1/revisions/1/diff?against=2", nil).json(t, &d) + if len(d.Files) == 0 || d.Files[0].Deleted == 0 { + t.Errorf("reverse diff %+v", d) + } + viewer.mustStatus(404, "GET", "/api/v1/revisions/2/diff?against=99", nil) + viewer.mustStatus(404, "GET", "/api/v1/revisions/99/diff", nil) + viewer.mustStatus(422, "GET", "/api/v1/revisions/2/diff?against=x", nil, skipRequestValidation, "1") + + // Rollback: a new revision with the content of revision 1 and a new + // SOA serial. + viewer.mustStatus(403, "POST", "/api/v1/revisions/1/rollback", nil) + operator.mustStatus(404, "POST", "/api/v1/revisions/99/rollback", nil) + var rev service.RevisionDTO + operator.mustStatus(201, "POST", "/api/v1/revisions/1/rollback", nil).json(t, &rev) + if rev.Number != 3 || rev.Comment != "Rollback to revision 1" || rev.CreatedByName != "otto" { + t.Errorf("rollback %+v", rev) + } + if cmd := command(t, conn); cmd.GetApplyRevision().GetRevision() != 3 { + t.Errorf("rollback not pushed: %v", cmd) + } + d = service.RevisionDiff{} + viewer.mustStatus(200, "GET", "/api/v1/revisions/3/diff?against=1", nil).json(t, &d) + if len(d.Files) == 0 { + t.Fatal("no serial change") + } + for _, f := range d.Files { + // Only the revision (header comment, SOA serial) differs. + if f.Role != "dns" || f.Status != "modified" || f.Inserted != f.Deleted || f.Inserted > 2 || + !strings.Contains(f.Diff, "+") || !strings.Contains(f.Diff, "(revision 3)") || strings.Contains(f.Diff, "192.0.2.10") { + t.Errorf("rollback differs from revision 1: %+v", f) + } + } + d = service.RevisionDiff{} + viewer.mustStatus(200, "GET", "/api/v1/revisions/3/diff", nil).json(t, &d) + if zone := findDiff(d.Files, "zones/db.office.test"); zone == nil || !strings.Contains(zone.Diff, "192.0.2.10") || zone.Deleted < 3 { + t.Errorf("rollback diff against 2: %+v", d.Files) + } + rev = service.RevisionDTO{} + operator.mustStatus(201, "POST", "/api/v1/revisions/3/rollback", map[string]any{"comment": " again "}).json(t, &rev) + if rev.Number != 4 || rev.Comment != "again" { + t.Errorf("second rollback %+v", rev) + } + command(t, conn) + var audit auditPage + admin.mustStatus(200, "GET", "/api/v1/audit?action=revision.rollback", nil).json(t, &audit) + if len(audit.Items) != 2 || audit.Items[1].ObjectID == nil || *audit.Items[1].ObjectID != "3" { + t.Errorf("rollback audit %+v", audit.Items) + } + // Normal creation still detects unchanged data: the data has www, + // revision 4 does not. + if rev := operator.newRevision(); rev.Number != 5 { + t.Errorf("revision after rollback %+v", rev) + } + + // Revisions without generator input (before M4) cannot be rolled back + // to. + if err := e.st.Q().InsertRevision(context.Background(), &store.Revision{Number: 6, CreatedAt: *e.now, CreatedByName: "old"}, nil); err != nil { + t.Fatal(err) + } + if p := operator.mustStatus(409, "POST", "/api/v1/revisions/6/rollback", nil).problem(t); p.Code != "rollback_unavailable" { + t.Errorf("no source %+v", p) + } +} + +func findDiff(files []service.FileDiff, path string) *service.FileDiff { + for i := range files { + if files[i].Path == path { + return &files[i] + } + } + return nil +} + +func TestPendingChanges(t *testing.T) { + e := newEnv(t) + admin := e.adminClient() + viewer := e.userClient(admin, "vera", "viewer") + admin.officeNetwork() + + var p service.PendingChanges + viewer.mustStatus(200, "GET", "/api/v1/config/pending", nil).json(t, &p) + if p.LatestRevision != 0 || !p.Changed || p.DNSError == nil || p.DNSError.Code != "dns_settings_missing" || p.DHCPError != nil || + len(p.Files) != 1 || p.Files[0].Status != "added" || p.Files[0].Role != "dhcp" || len(p.Changes) != 2 { + t.Fatalf("before the first revision %+v", p) + } + if p.Changes[0].Action != "address.create" || p.Changes[1].Action != "network.create" || p.Changes[0].IP != nil { + t.Errorf("changes %+v", p.Changes) + } + + admin.configureDNS() + admin.newRevision() + p = service.PendingChanges{} + viewer.mustStatus(200, "GET", "/api/v1/config/pending", nil).json(t, &p) + if p.LatestRevision != 1 || p.Changed || len(p.Files) != 0 || len(p.Changes) != 0 || p.DNSError != nil { + t.Fatalf("after the revision %+v", p) + } + + admin.addHost("192.0.2.10", "www") + admin.mustStatus(201, "POST", "/api/v1/users", map[string]any{"username": "x", "role": "viewer", "password": "initial-password-1"}) + p = service.PendingChanges{} + viewer.mustStatus(200, "GET", "/api/v1/config/pending", nil).json(t, &p) + if !p.Changed || len(p.Changes) != 1 || p.Changes[0].Action != "address.create" || findDiff(p.Files, "zones/db.office.test") == nil { + t.Fatalf("pending %+v", p) + } + for _, f := range p.Files { + if f.Status != "modified" || f.Role != "dns" { + t.Errorf("pending file %+v", f) + } + } + // The serial is not a change: the pending zone diff contains only the + // new host. + if z := findDiff(p.Files, "zones/db.office.test"); z.Deleted != 0 { + t.Errorf("zone diff %+v", z) + } +} + +// sse reads an event stream line by line. +type sse struct { + lines chan string + done chan struct{} + keepalives atomic.Int64 +} + +func (c *client) openEvents(ctx context.Context) (int, http.Header, *sse) { + c.e.t.Helper() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.e.srv.URL+"/api/v1/events", nil) + if err != nil { + c.e.t.Fatal(err) + } + res, err := c.http.Do(req) //nolint:bodyclose // closed by the reader goroutine + if err != nil { + c.e.t.Fatal(err) + } + s := &sse{lines: make(chan string, 1000), done: make(chan struct{})} + go func() { + defer close(s.done) + defer res.Body.Close() //nolint:errcheck // test + sc := bufio.NewScanner(res.Body) + for sc.Scan() { + if sc.Text() == ": keepalive" { + s.keepalives.Add(1) + } + s.lines <- sc.Text() + } + }() + return res.StatusCode, res.Header, s +} + +// event waits for the next event of type typ and returns its data. +func (s *sse) event(t *testing.T, typ string) string { + t.Helper() + timeout := time.After(5 * time.Second) + for { + select { + case l := <-s.lines: + if l != "event: "+typ { + continue + } + select { + case data := <-s.lines: + return strings.TrimPrefix(data, "data: ") + case <-timeout: + } + t.Fatalf("event %s without data", typ) + case <-timeout: + t.Fatalf("no %s event", typ) + } + } +} + +func TestEventStream(t *testing.T) { + // Short server timeouts: the stream must outlive them. + e := newEnvServer(t, nil, func(s *http.Server) { + s.ReadTimeout, s.WriteTimeout = 4*time.Second, 4*time.Second + }) + admin := e.adminClient() + e.client().mustStatus(401, "GET", "/api/v1/events", nil) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + status, header, stream := admin.openEvents(ctx) + if status != http.StatusOK || header.Get("Content-Type") != "text/event-stream" || header.Get("Cache-Control") != "no-store" { + t.Fatalf("stream response %d %v", status, header) + } + if l := <-stream.lines; l != "retry: 5000" { + t.Errorf("first line %q", l) + } + if e.bus.Subscribers() != 1 { + t.Errorf("subscribers %d", e.bus.Subscribers()) + } + + admin.officeNetwork() + if data := stream.event(t, "pending"); data != "{}" { + t.Errorf("pending data %q", data) + } + agentID := e.enroll(admin.createToken("dns").Secret, "dns") + if data := stream.event(t, "agent"); data != `{"id":"`+agentID+`"}` { + t.Errorf("agent data %q", data) + } + + // Past the server's write timeout the stream still delivers. + time.Sleep(5 * time.Second) + admin.configureDNS() + stream.event(t, "pending") + rev := admin.newRevision() + if data := stream.event(t, "revision"); data != `{"number":`+strconv.FormatInt(rev.Number, 10)+`}` { + t.Errorf("revision data %q", data) + } + var targets deployResponse + admin.mustStatus(200, "POST", "/api/v1/revisions/1/deploy", nil).json(t, &targets) + var dep struct { + ID int64 `json:"id"` + AgentID string `json:"agent_id"` + Revision int64 `json:"revision"` + } + if err := json.Unmarshal([]byte(stream.event(t, "deployment")), &dep); err != nil || dep.ID != targets.Items[0].DeploymentID || + dep.AgentID != agentID || dep.Revision != 1 { + t.Errorf("deployment event %+v %v", dep, err) + } + if n := stream.keepalives.Load(); n < 10 { + t.Errorf("keepalives %d", n) + } + + // Logging out ends the stream at the next session check. + admin.mustStatus(204, "POST", "/api/v1/auth/logout", nil) + select { + case <-stream.done: + case <-time.After(5 * time.Second): + t.Fatal("stream outlived the session") + } + time.Sleep(100 * time.Millisecond) + if n := e.bus.Subscribers(); n != 0 { + t.Errorf("subscribers after the end %d", n) + } +} diff --git a/server/internal/httpapi/routes.go b/server/internal/httpapi/routes.go index 4dad042..7d22f0a 100644 --- a/server/internal/httpapi/routes.go +++ b/server/internal/httpapi/routes.go @@ -12,7 +12,7 @@ import ( // permissions; TestEveryRouteIsProtected and the OpenAPI contract test check it. func (a *API) Routes() []Route { const ( - get, post, put, del = http.MethodGet, http.MethodPost, http.MethodPut, http.MethodDelete + get, post, put, patch, del = http.MethodGet, http.MethodPost, http.MethodPut, http.MethodPatch, http.MethodDelete ) perm := func(method, path string, p auth.Permission, write bool, h handlerFunc) Route { return Route{Method: method, Path: path, Access: permission, Perm: p, Write: write, Handler: h} @@ -58,10 +58,15 @@ func (a *API) Routes() []Route { // Generated configuration (read-only preview until agents arrive). perm(get, "/api/v1/config/preview", auth.PermIPAMRead, false, a.previewConfig), + perm(get, "/api/v1/config/pending", auth.PermIPAMRead, false, a.pendingChanges), + + // Live updates (Server-Sent Events, ADR 0022). + perm(get, "/api/v1/events", auth.PermIPAMRead, false, a.streamEvents), // Agents and enrollment (ADR 0021). perm(get, "/api/v1/agents", auth.PermIPAMRead, false, a.listAgents), perm(get, "/api/v1/agents/{id}", auth.PermIPAMRead, false, a.getAgent), + perm(patch, "/api/v1/agents/{id}", auth.PermSettingsAdm, true, a.updateAgent), // Locking agents out is a security function and stays available in // license read-only mode, like deleting enrollment tokens. perm(del, "/api/v1/agents/{id}", auth.PermSettingsAdm, false, a.deleteAgent), @@ -74,6 +79,9 @@ func (a *API) Routes() []Route { perm(get, "/api/v1/revisions", auth.PermIPAMRead, false, a.listRevisions), perm(post, "/api/v1/revisions", auth.PermDeploy, true, a.createRevision), perm(get, "/api/v1/revisions/{number}", auth.PermIPAMRead, false, a.getRevision), + perm(post, "/api/v1/revisions/{number}/deploy", auth.PermDeploy, true, a.deployRevision), + perm(post, "/api/v1/revisions/{number}/rollback", auth.PermDeploy, true, a.rollbackRevision), + perm(get, "/api/v1/revisions/{number}/diff", auth.PermIPAMRead, false, a.diffRevision), perm(get, "/api/v1/deployments", auth.PermIPAMRead, false, a.listDeployments), perm(get, "/api/v1/leases", auth.PermIPAMRead, false, a.listLeases), diff --git a/server/internal/service/agentplane.go b/server/internal/service/agentplane.go index e8e604e..aa28dc7 100644 --- a/server/internal/service/agentplane.go +++ b/server/internal/service/agentplane.go @@ -10,6 +10,8 @@ import ( "errors" "math" "net/netip" + "reflect" + "strconv" "strings" "time" @@ -129,6 +131,7 @@ func (s *Service) EnrollAgent(ctx context.Context, in EnrollInput) (*EnrollResul if err != nil { return nil, err } + s.publishAgent(res.AgentID) return &res, nil } @@ -192,6 +195,9 @@ func (s *Service) RenewAgentCertificate(ctx context.Context, agentID string, csr type AgentSettings struct { Mode string DesiredRevision uint64 + // PushDeploymentID is set while a push of DesiredRevision is open + // (decimal deployment ID, ADR 0022). + PushDeploymentID string } // HelloInput is the status an agent reports when it opens its stream. @@ -230,13 +236,46 @@ func healthJSON(h *AgentHealth) json.RawMessage { return raw } -// settings returns the agent's mode and desired revision. +// settings returns the agent's mode, desired revision and open push. A +// pending push of another revision is stale (the desired revision moved on +// without it) and is superseded. func (s *Service) settings(ctx context.Context, tx *store.Tx, a *store.Agent) (*AgentSettings, error) { - desired, err := latestRevisionNumber(ctx, tx) + latest, err := latestRevisionNumber(ctx, tx) if err != nil { return nil, translate(err) } - return &AgentSettings{Mode: a.Mode, DesiredRevision: uint64(desired)}, nil //nolint:gosec // non-negative + desired := desiredRevision(a, latest) + out := &AgentSettings{Mode: a.Mode, DesiredRevision: uint64(desired)} //nolint:gosec // non-negative + p, err := tx.PendingDeployment(ctx, a.ID) + if errors.Is(err, store.ErrNotFound) { + return out, nil + } + if err != nil { + return nil, translate(err) + } + if p.Revision == desired { + out.PushDeploymentID = strconv.FormatInt(p.ID, 10) + return out, nil + } + stale, err := tx.SupersedeDeployments(ctx, store.SupersedeFilter{AgentID: a.ID, KeepRevision: desired}, + s.now().UTC().Truncate(time.Microsecond)) + if err != nil { + return nil, translate(err) + } + for _, d := range stale { + s.publishDeployment(d) + } + return out, nil +} + +// pushDeploymentID parses the deployment ID of a server-created push +// (decimal); IDs of M3-style notifications (UUIDs) do not parse. +func pushDeploymentID(s string) (int64, bool) { + if s == "" || len(s) > 19 || strings.Trim(s, "0123456789") != "" { + return 0, false + } + id, err := strconv.ParseInt(s, 10, 64) + return id, err == nil && id > 0 } func agentStatusErr(err error) error { @@ -263,9 +302,24 @@ func (s *Service) AgentHello(ctx context.Context, agentID string, in HelloInput) if err != nil { return nil, agentStatusErr(err) } + s.publishAgent(agentID) return s.settings(ctx, q, a) } +// AgentDisconnected tells the UI that an agent's stream has ended. +func (s *Service) AgentDisconnected(agentID string) { s.publishAgent(agentID) } + +// statusChanged reports whether a heartbeat changed what the UI shows +// (revisions or health apart from the check time). +func statusChanged(before, after *store.Agent) bool { + if before.AppliedRevision != after.AppliedRevision || before.LastGoodRevision != after.LastGoodRevision { + return true + } + hb, ha := decodeHealth(before.Health), decodeHealth(after.Health) + hb.CheckedAt, ha.CheckedAt = nil, nil + return !reflect.DeepEqual(hb, ha) +} + // HeartbeatInput is a periodic status report. type HeartbeatInput struct { AppliedRevision uint64 @@ -275,11 +329,19 @@ type HeartbeatInput struct { // AgentHeartbeat records a heartbeat. func (s *Service) AgentHeartbeat(ctx context.Context, agentID string, in HeartbeatInput) error { - _, err := s.st.Q().UpdateAgentStatus(ctx, agentID, store.AgentStatus{AppliedRevision: revisionPtr(in.AppliedRevision), + q := s.st.Q() + before, err := q.GetAgent(ctx, agentID, false) + if err != nil { + return agentStatusErr(err) + } + after, err := q.UpdateAgentStatus(ctx, agentID, store.AgentStatus{AppliedRevision: revisionPtr(in.AppliedRevision), LastGoodRevision: revisionPtr(in.LastGoodRevision), Health: healthJSON(in.Health), SeenAt: s.now()}) if err != nil { return agentStatusErr(err) } + if statusChanged(before, after) { + s.publishAgent(agentID) + } return nil } @@ -353,11 +415,35 @@ func (s *Service) RecordDeployment(ctx context.Context, agentID string, in Deplo if in.Outcome == "succeeded" { st.LastGoodRevision = revisionPtr(in.Revision) } - return s.inTx(ctx, func(tx *store.Tx) error { + revision := int64(in.Revision) //nolint:gosec // at most MaxRevision (checked above) + message := cleanMessage(in.Message) + var recorded *store.Deployment + err := s.inTx(ctx, func(tx *store.Tx) error { + // A report with the ID of a server-created push completes that + // deployment (ADR 0022). + if id, ok := pushDeploymentID(in.CorrelationID); ok { + d, completed, err := tx.CompleteDeployment(ctx, id, agentID, revision, store.DeploymentResult{ + Outcome: in.Outcome, Message: message, StartedAt: in.StartedAt, FinishedAt: in.FinishedAt, + ActiveRevision: int64(in.ActiveRevision), //nolint:gosec // at most MaxRevision (checked above) + }) + switch { + case err == nil && !completed: + return nil // resent report: nothing changes + case err == nil: + recorded = d + if _, err := tx.UpdateAgentStatus(ctx, agentID, st); err != nil { + return agentStatusErr(err) + } + return nil + case !errors.Is(err, store.ErrNotFound): + return err + } + // Unknown ID: stored as a report like any other. + } + finished := in.FinishedAt d := &store.Deployment{ AgentID: agentID, CorrelationID: cleanText(in.CorrelationID, 100), Trigger: in.Trigger, Outcome: in.Outcome, - Message: cleanMessage(in.Message), StartedAt: in.StartedAt, FinishedAt: in.FinishedAt, - Revision: int64(in.Revision), //nolint:gosec // at most MaxRevision (checked above) + Message: message, StartedAt: in.StartedAt, FinishedAt: &finished, Revision: revision, ActiveRevision: int64(in.ActiveRevision), //nolint:gosec // at most MaxRevision (checked above) } inserted, err := tx.InsertDeployment(ctx, d) @@ -366,11 +452,20 @@ func (s *Service) RecordDeployment(ctx context.Context, agentID string, in Deplo // nothing; it must not roll back the agent's state either. return err } + recorded = d if _, err := tx.UpdateAgentStatus(ctx, agentID, st); err != nil { return agentStatusErr(err) } return nil }) + if err != nil { + return err + } + if recorded != nil { + s.publishDeployment(recorded) + s.publishAgent(agentID) + } + return nil } // cleanMessage keeps line breaks (checker output) but drops other control diff --git a/server/internal/service/agents.go b/server/internal/service/agents.go index c49b483..4c06fd8 100644 --- a/server/internal/service/agents.go +++ b/server/internal/service/agents.go @@ -76,24 +76,30 @@ func decodeHealth(raw []byte) AgentHealth { return h } -// DeploymentDTO is a deployment result reported by an agent. +// DeploymentDTO is a deployment: a result reported by an agent or a push +// requested by a user (pending until the agent reports). type DeploymentDTO struct { - ID int64 `json:"id"` - AgentID string `json:"agent_id"` - AgentHostname string `json:"agent_hostname"` - Revision int64 `json:"revision"` - Trigger string `json:"trigger"` - Outcome string `json:"outcome"` - Message string `json:"message"` - ActiveRevision int64 `json:"active_revision"` - StartedAt time.Time `json:"started_at"` - FinishedAt time.Time `json:"finished_at"` + ID int64 `json:"id"` + AgentID string `json:"agent_id"` + AgentHostname string `json:"agent_hostname"` + Revision int64 `json:"revision"` + Trigger string `json:"trigger"` + Outcome string `json:"outcome"` + Message string `json:"message"` + ActiveRevision int64 `json:"active_revision"` + StartedAt time.Time `json:"started_at"` + FinishedAt *time.Time `json:"finished_at,omitempty"` + RequestedByName string `json:"requested_by_name,omitempty"` } func deploymentDTO(d *store.Deployment) DeploymentDTO { - return DeploymentDTO{ID: d.ID, AgentID: d.AgentID, AgentHostname: d.AgentHostname, Revision: d.Revision, + out := DeploymentDTO{ID: d.ID, AgentID: d.AgentID, AgentHostname: d.AgentHostname, Revision: d.Revision, Trigger: d.Trigger, Outcome: d.Outcome, Message: d.Message, ActiveRevision: d.ActiveRevision, StartedAt: d.StartedAt, FinishedAt: d.FinishedAt} + if d.RequestedByName != nil { + out.RequestedByName = *d.RequestedByName + } + return out } // AgentDTO is an enrolled agent. @@ -112,16 +118,33 @@ type AgentDTO struct { AppliedRevision int64 `json:"applied_revision"` LastGoodRevision int64 `json:"last_good_revision"` DesiredRevision int64 `json:"desired_revision"` + TargetRevision int64 `json:"target_revision"` Health AgentHealth `json:"health"` RevokedAt *time.Time `json:"revoked_at,omitempty"` LastDeployment *DeploymentDTO `json:"last_deployment,omitempty"` } -func (s *Service) agentDTO(a *store.Agent, desired int64, last *store.Deployment) AgentDTO { +// desiredRevision is the revision an agent should run (ADR 0022): the +// latest in auto_pull mode, the last pushed one in manual mode, none when +// revoked. +func desiredRevision(a *store.Agent, latest int64) int64 { + switch { + case a.RevokedAt != nil: + return 0 + case a.Mode == agenthub.ModeManual: + return a.TargetRevision + default: + return latest + } +} + +// agentDTO converts a; latest is the newest revision number. +func (s *Service) agentDTO(a *store.Agent, latest int64, last *store.Deployment) AgentDTO { d := AgentDTO{ ID: a.ID, Role: a.Role, Hostname: a.Hostname, AgentVersion: a.AgentVersion, ServiceVersion: a.ServiceVersion, Mode: a.Mode, EnrolledAt: a.EnrolledAt, LastSeenAt: a.LastSeenAt, CertExpiresAt: a.CertNotAfter, - AppliedRevision: a.AppliedRevision, LastGoodRevision: a.LastGoodRevision, DesiredRevision: desired, + AppliedRevision: a.AppliedRevision, LastGoodRevision: a.LastGoodRevision, + DesiredRevision: desiredRevision(a, latest), TargetRevision: a.TargetRevision, Health: decodeHealth(a.Health), RevokedAt: a.RevokedAt, } if a.RemoteAddress != nil { @@ -134,14 +157,10 @@ func (s *Service) agentDTO(a *store.Agent, desired int64, last *store.Deployment dd := deploymentDTO(last) d.LastDeployment = &dd } - if a.RevokedAt != nil { - d.DesiredRevision = 0 - } return d } -// latestRevisionNumber returns the newest revision number, 0 if none. In -// M3 it is the desired revision of every agent (ADR 0021). +// latestRevisionNumber returns the newest revision number, 0 if none. func latestRevisionNumber(ctx context.Context, tx *store.Tx) (int64, error) { r, err := tx.LatestRevision(ctx) if errors.Is(err, store.ErrNotFound) { @@ -230,6 +249,7 @@ func (s *Service) RevokeAgent(ctx context.Context, a *Actor, id string) (AgentDT s.opts.Hub.Disconnect(id, agenthub.ReasonRevoked) out.Connected = false } + s.publishAgent(id) return out, nil } @@ -255,6 +275,7 @@ func (s *Service) DeleteAgent(ctx context.Context, a *Actor, id string) error { if s.opts.Hub != nil { s.opts.Hub.Disconnect(id, agenthub.ReasonDeleted) } + s.publishAgent(id) return nil } diff --git a/server/internal/service/configpreview.go b/server/internal/service/configpreview.go index a338b18..fca7c31 100644 --- a/server/internal/service/configpreview.go +++ b/server/internal/service/configpreview.go @@ -124,6 +124,12 @@ func (s *Service) generate(ctx context.Context, serial uint32) (*Generated, erro if err != nil { return nil, err } + return render(src), nil +} + +// render runs the generator on src (a pure function: equal input gives +// byte-identical output). +func render(src configgen.Source) *Generated { out := &Generated{Warnings: []configgen.Warning{}} if src.DNS.SOA.PrimaryNS == "" || src.DNS.SOA.Hostmaster == "" || len(src.DNS.NameServers) == 0 { @@ -145,5 +151,5 @@ func (s *Service) generate(ctx context.Context, serial uint32) (*Generated, erro } else { out.DHCP = arts } - return out, nil + return out } diff --git a/server/internal/service/deploy.go b/server/internal/service/deploy.go new file mode 100644 index 0000000..90a3cdb --- /dev/null +++ b/server/internal/service/deploy.go @@ -0,0 +1,633 @@ +// SPDX-License-Identifier: LicenseRef-DDI-Proprietary + +package service + +import ( + "cmp" + "context" + "encoding/json" + "errors" + "slices" + "strconv" + "strings" + "time" + "unicode/utf8" + + "git.beugsiud.de/beug.k/DDI/server/internal/agenthub" + "git.beugsiud.de/beug.k/DDI/server/internal/configgen" + "git.beugsiud.de/beug.k/DDI/server/internal/events" + "git.beugsiud.de/beug.k/DDI/server/internal/store" + "git.beugsiud.de/beug.k/DDI/server/internal/textdiff" +) + +// Revisions, pushes, rollbacks, diffs and pending changes (ADR 0021, 0022). + +// ErrRollbackUnavailable is returned for revisions without stored generator +// input (created before M4). +var ErrRollbackUnavailable = newError(KindConflict, "rollback_unavailable", + "the revision has no stored generator input and cannot be rolled back to") + +// diffContext is the number of context lines in revision diffs. +const diffContext = 3 + +// maxPendingChanges bounds the audit entries listed as pending changes. +const maxPendingChanges = 200 + +func validRevisionNumber(n int64) bool { return n >= 1 && n <= store.MaxRevision } + +// revisionAudit is the audit snapshot of a new revision. +type revisionAudit struct { + RevisionSummaryDTO + RollbackOf int64 `json:"rollback_of,omitempty"` +} + +// storeRevision stores the output g of src (rendered with src.Serial as +// revision number) and supersedes pending pushes of agents in auto_pull mode +// (the new revision replaces them). It runs in tx after LockRevisions. +func (s *Service) storeRevision(ctx context.Context, tx *store.Tx, a *Actor, src configgen.Source, g *Generated, + comment string, rollbackOf int64, +) (RevisionDTO, []*store.Deployment, error) { + warnings, err := json.Marshal(g.Warnings) + if err != nil { + return RevisionDTO{}, nil, err + } + source, err := json.Marshal(src) + if err != nil { + return RevisionDTO{}, nil, err + } + now := s.now().UTC().Truncate(time.Microsecond) + r := &store.Revision{Number: int64(src.Serial), CreatedAt: now, CreatedByName: a.Username, Comment: comment, + Warnings: warnings, Source: source} + if a.UserID != "" { + uid := a.UserID + r.CreatedBy = &uid + } + files := revisionFiles(g) + if err := tx.InsertRevision(ctx, r, files); err != nil { + return RevisionDTO{}, nil, err + } + superseded, err := tx.SupersedeDeployments(ctx, store.SupersedeFilter{AutoPullOnly: true, KeepRevision: r.Number}, now) + if err != nil { + return RevisionDTO{}, nil, err + } + summary := revisionAudit{RevisionSummaryDTO: RevisionSummaryDTO{Number: r.Number, CreatedAt: r.CreatedAt, + CreatedByName: r.CreatedByName, Comment: r.Comment, DNSFiles: len(g.DNS), DHCPFiles: len(g.DHCP), + Warnings: len(g.Warnings)}, RollbackOf: rollbackOf} + action := "revision.create" + if rollbackOf > 0 { + action = "revision.rollback" + } + num := strconv.FormatInt(r.Number, 10) + if err := s.audit(ctx, tx, a, action, "revision", num, "#"+num, nil, summary); err != nil { + return RevisionDTO{}, nil, err + } + return revisionDTO(r, files), superseded, nil +} + +// revisionCreated notifies agents and the UI after a new revision was +// committed. +func (s *Service) revisionCreated(a *Actor, rev *RevisionDTO, superseded []*store.Deployment) { + notified := 0 + if s.opts.Hub != nil { + notified = s.opts.Hub.NotifyRevision(uint64(rev.Number)) //nolint:gosec // positive + } + s.log.Info("revision created", "revision", rev.Number, "notified_agents", notified, + "superseded_pushes", len(superseded), "request_id", a.RequestID) + for _, d := range superseded { + s.publishDeployment(d) + } + s.publish(events.TypeRevision, events.Revision{Number: rev.Number}) + s.publish(events.TypePending, events.Pending{}) +} + +func checkComment(comment string) (string, error) { + comment = strings.TrimSpace(comment) + if utf8.RuneCountInString(comment) > 500 { + return "", invalidField("too_long", "comment") + } + return comment, nil +} + +// CreateRevision generates the configuration and stores it as a new +// revision (ADR 0021). If the output equals the latest revision, that one +// is returned with created false. Agents in auto_pull mode are notified. +func (s *Service) CreateRevision(ctx context.Context, a *Actor, comment string) (*RevisionDTO, bool, error) { + if err := s.requireWritable(); err != nil { + return nil, false, err + } + comment, err := checkComment(comment) + if err != nil { + return nil, false, err + } + var out RevisionDTO + var superseded []*store.Deployment + created := false + err = s.inTx(ctx, func(tx *store.Tx) error { + if err := tx.LockRevisions(ctx); err != nil { + return err + } + latest, err := tx.LatestRevision(ctx) + if err != nil && !errors.Is(err, store.ErrNotFound) { + return err + } + src, err := s.loadSource(ctx, 1) + if err != nil { + return err + } + next := int64(1) + if latest != nil { + // Generate with the latest number first: identical output + // means nothing changed since then. + src.Serial = uint32(latest.Number) //nolint:gosec // 1..2^32-1 by constraint + g := render(src) + if err := generationProblem(g); err != nil { + return err + } + stored, err := tx.RevisionFiles(ctx, latest.Number, "") + if err != nil { + return err + } + if sameFiles(revisionFiles(g), stored) { + out = revisionDTO(latest, stored) + return nil + } + if latest.Number >= store.MaxRevision { + return ErrRevisionLimit + } + next = latest.Number + 1 + } + src.Serial = uint32(next) //nolint:gosec // 1..2^32-1, checked above + g := render(src) + if err := generationProblem(g); err != nil { + return err + } + out, superseded, err = s.storeRevision(ctx, tx, a, src, g, comment, 0) + created = err == nil + return err + }) + if err != nil { + return nil, false, err + } + if created { + s.revisionCreated(a, &out, superseded) + } + return &out, created, nil +} + +// RollbackRevision creates a new revision with the content of revision n: +// n's generator input is rendered again with the next number, so that SOA +// serials keep increasing (ADR 0009, 0022). The result is stored even if it +// equals the latest revision. +func (s *Service) RollbackRevision(ctx context.Context, a *Actor, n int64, comment string) (*RevisionDTO, error) { + if err := s.requireWritable(); err != nil { + return nil, err + } + comment, err := checkComment(comment) + if err != nil { + return nil, err + } + if !validRevisionNumber(n) { + return nil, ErrNotFound + } + if comment == "" { + comment = "Rollback to revision " + strconv.FormatInt(n, 10) + } + var out RevisionDTO + var superseded []*store.Deployment + err = s.inTx(ctx, func(tx *store.Tx) error { + if err := tx.LockRevisions(ctx); err != nil { + return err + } + raw, err := tx.RevisionSource(ctx, n) + if err != nil { + return err + } + if raw == nil { + return ErrRollbackUnavailable + } + var src configgen.Source + if err := json.Unmarshal(raw, &src); err != nil { + e := *ErrRollbackUnavailable + e.err = err + return &e + } + latest, err := tx.LatestRevision(ctx) + if err != nil { + return err + } + if latest.Number >= store.MaxRevision { + return ErrRevisionLimit + } + src.Serial = uint32(latest.Number + 1) //nolint:gosec // checked above + g := render(src) + if err := generationProblem(g); err != nil { + return err + } + out, superseded, err = s.storeRevision(ctx, tx, a, src, g, comment, n) + return err + }) + if err != nil { + return nil, err + } + s.revisionCreated(a, &out, superseded) + return &out, nil +} + +// FileDiff is the change of one generated file. +type FileDiff struct { + Role string `json:"role"` + Path string `json:"path"` + Status string `json:"status"` // added, removed, modified + Diff string `json:"diff"` + Inserted int `json:"inserted"` + Deleted int `json:"deleted"` +} + +// RevisionDiff lists the files that differ between two revisions. +type RevisionDiff struct { + Number int64 `json:"number"` + Against int64 `json:"against"` + Files []FileDiff `json:"files"` +} + +func roleOrder(role string) int { + if role == RoleDNS { + return 0 + } + return 1 +} + +// diffFiles compares two file sets; unchanged files are left out. The +// result is ordered by role (DNS first) and path. +func diffFiles(older, newer []store.RevisionFile) []FileDiff { + type key struct{ role, path string } + oldBy := make(map[key]store.RevisionFile, len(older)) + for _, f := range older { + oldBy[key{f.Role, f.Path}] = f + } + newBy := make(map[key]store.RevisionFile, len(newer)) + keys := make([]key, 0, len(older)+len(newer)) + for _, f := range newer { + k := key{f.Role, f.Path} + newBy[k] = f + keys = append(keys, k) + } + for _, f := range older { + if k := (key{f.Role, f.Path}); !hasKey(newBy, k) { + keys = append(keys, k) + } + } + slices.SortFunc(keys, func(a, b key) int { + return cmp.Or(cmp.Compare(roleOrder(a.role), roleOrder(b.role)), strings.Compare(a.path, b.path)) + }) + keys = slices.Compact(keys) + + out := []FileDiff{} + for _, k := range keys { + o, inOld := oldBy[k] + n, inNew := newBy[k] + d := FileDiff{Role: k.role, Path: k.path} + switch { + case !inOld: + d.Status = "added" + case !inNew: + d.Status = "removed" + case string(o.Content) == string(n.Content) && o.Mode == n.Mode && o.CreateOnly == n.CreateOnly: + continue + default: + d.Status = "modified" + } + a, b := string(o.Content), string(n.Content) + d.Diff = textdiff.Unified(a, b, diffContext) + d.Inserted, d.Deleted = textdiff.Stats(textdiff.Lines(a, b)) + out = append(out, d) + } + return out +} + +func hasKey[K comparable, V any](m map[K]V, k K) bool { + _, ok := m[k] + return ok +} + +// DiffRevisions compares revision n with revision against (nil: n-1; 0: +// nothing, every file is added). +func (s *Service) DiffRevisions(ctx context.Context, n int64, against *int64) (*RevisionDiff, error) { + if !validRevisionNumber(n) { + return nil, ErrNotFound + } + m := n - 1 + if against != nil { + m = *against + } + if m < 0 || m > store.MaxRevision { + return nil, ErrNotFound + } + q := s.st.Q() + if _, err := q.GetRevision(ctx, n); err != nil { + return nil, translate(err) + } + newer, err := q.RevisionFiles(ctx, n, "") + if err != nil { + return nil, translate(err) + } + var older []store.RevisionFile + if m > 0 { + if _, err := q.GetRevision(ctx, m); err != nil { + return nil, translate(err) + } + if older, err = q.RevisionFiles(ctx, m, ""); err != nil { + return nil, translate(err) + } + } + return &RevisionDiff{Number: n, Against: m, Files: diffFiles(older, newer)}, nil +} + +// PendingChanges are the changes since the latest revision. +type PendingChanges struct { + LatestRevision int64 `json:"latest_revision"` + Changed bool `json:"changed"` + Files []FileDiff `json:"files"` + Changes []*store.AuditEntry `json:"changes"` + DNSError *GeneratorProblem `json:"dns_error,omitempty"` + DHCPError *GeneratorProblem `json:"dhcp_error,omitempty"` +} + +func withoutRole(files []store.RevisionFile, role string) []store.RevisionFile { + return slices.DeleteFunc(files, func(f store.RevisionFile) bool { return f.Role == role }) +} + +// PendingChanges generates the configuration from the current data with the +// latest revision's number (so the SOA serial does not count as a change) +// and compares it with that revision. Parts that cannot be generated are +// reported in DNSError/DHCPError and not compared. Changes lists the audit +// entries of generator data since the latest revision, newest first. +func (s *Service) PendingChanges(ctx context.Context) (*PendingChanges, error) { + q := s.st.Q() + latest, err := q.LatestRevision(ctx) + if err != nil && !errors.Is(err, store.ErrNotFound) { + return nil, translate(err) + } + serial := uint32(1) + out := &PendingChanges{} + var stored []store.RevisionFile + filter := store.ChangeFilter{ObjectTypes: configObjectTypes, Actions: configActions, Limit: maxPendingChanges} + if latest != nil { + serial = uint32(latest.Number) //nolint:gosec // 1..2^32-1 by constraint + out.LatestRevision = latest.Number + filter.AfterRevision, filter.Since = latest.Number, latest.CreatedAt + if stored, err = q.RevisionFiles(ctx, latest.Number, ""); err != nil { + return nil, translate(err) + } + } + src, err := s.loadSource(ctx, serial) + if err != nil { + return nil, err + } + g := render(src) + out.DNSError, out.DHCPError = g.DNSError, g.DHCPError + generated := revisionFiles(g) + if g.DNSError != nil { + stored = withoutRole(stored, RoleDNS) + } + if g.DHCPError != nil { + stored = withoutRole(stored, RoleDHCP) + } + out.Files = diffFiles(stored, generated) + out.Changed = len(out.Files) > 0 + + changes, err := q.ConfigChangesSince(ctx, filter) + if err != nil { + return nil, translate(err) + } + out.Changes = make([]*store.AuditEntry, len(changes)) + for i, c := range changes { + c.IP = nil // the source address of other users is for the audit log only + out.Changes[i] = c + } + return out, nil +} + +// DeployTarget is one agent a push was created for. +type DeployTarget struct { + AgentID string `json:"agent_id"` + AgentHostname string `json:"agent_hostname"` + DeploymentID int64 `json:"deployment_id"` + Sent bool `json:"sent"` +} + +// deployAudit is the audit snapshot of a push. +type deployAudit struct { + Revision int64 `json:"revision"` + AgentIDs []string `json:"agent_ids"` + DeploymentIDs []int64 `json:"deployment_ids"` +} + +// DeployRevision pushes revision n to the given agents (all agents that are +// not revoked if agentIDs is empty), ADR 0022: one pending deployment per +// agent, older pending pushes of these agents are superseded, agents in +// manual mode get n as target revision, connected agents receive an +// ApplyRevisionCommand at once. Revisions older than the latest cannot be +// pushed to agents in auto_pull mode (rollback_required). +func (s *Service) DeployRevision(ctx context.Context, a *Actor, n int64, agentIDs []string) ([]DeployTarget, error) { + if err := s.requireWritable(); err != nil { + return nil, err + } + if !validRevisionNumber(n) { + return nil, ErrNotFound + } + var created, superseded []*store.Deployment + err := s.inTx(ctx, func(tx *store.Tx) error { + // Serialize with revision creation: "latest" must not change + // while auto_pull agents are checked. + if err := tx.LockRevisions(ctx); err != nil { + return err + } + if _, err := tx.GetRevision(ctx, n); err != nil { + return err + } + latest, err := latestRevisionNumber(ctx, tx) + if err != nil { + return err + } + agents, err := s.deployTargets(ctx, tx, agentIDs) + if err != nil { + return err + } + var rollback []Problem + for _, ag := range agents { + // Agents in auto_pull mode always return to the latest + // revision, and no agent accepts a revision older than the + // one it runs (replay protection, ADR 0009); re-applying the + // running revision is allowed. + if (ag.Mode == agenthub.ModeAutoPull && n < latest) || ag.AppliedRevision > n { + rollback = append(rollback, Problem{Code: "rollback_required", Field: "agent_ids", + Params: map[string]string{"agent_id": ag.ID, "hostname": ag.Hostname, "mode": ag.Mode, + "applied_revision": strconv.FormatInt(ag.AppliedRevision, 10)}}) + } + } + if len(rollback) > 0 { + return &Error{Kind: KindInvalid, Code: "rollback_required", + Message: "agents cannot go back to an older revision by a push; use a rollback", + Params: map[string]string{"latest": strconv.FormatInt(latest, 10)}, + Problems: rollback} + } + now := s.now().UTC().Truncate(time.Microsecond) + requester := a.Username + audit := deployAudit{Revision: n, AgentIDs: []string{}, DeploymentIDs: []int64{}} + for _, ag := range agents { + sup, err := tx.SupersedeDeployments(ctx, store.SupersedeFilter{AgentID: ag.ID}, now) + if err != nil { + return err + } + superseded = append(superseded, sup...) + d := &store.Deployment{AgentID: ag.ID, AgentHostname: ag.Hostname, Revision: n, StartedAt: now, + RequestedByName: &requester} + if a.UserID != "" { + uid := a.UserID + d.RequestedBy = &uid + } + if err := tx.InsertPushDeployment(ctx, d); err != nil { + return err + } + if ag.Mode == agenthub.ModeManual && ag.TargetRevision != n { + if err := tx.SetAgentTarget(ctx, ag.ID, n); err != nil { + return err + } + } + created = append(created, d) + audit.AgentIDs = append(audit.AgentIDs, ag.ID) + audit.DeploymentIDs = append(audit.DeploymentIDs, d.ID) + } + num := strconv.FormatInt(n, 10) + return s.audit(ctx, tx, a, "revision.deploy", "revision", num, "#"+num, nil, audit) + }) + if err != nil { + return nil, err + } + out := make([]DeployTarget, 0, len(created)) + for _, d := range created { + t := DeployTarget{AgentID: d.AgentID, AgentHostname: d.AgentHostname, DeploymentID: d.ID} + if s.opts.Hub != nil { + t.Sent = s.opts.Hub.Send(d.AgentID, agenthub.ApplyCommand(uint64(n), strconv.FormatInt(d.ID, 10))) //nolint:gosec // positive + } + out = append(out, t) + } + for _, d := range superseded { + s.publishDeployment(d) + } + for _, d := range created { + s.publishDeployment(d) + s.publishAgent(d.AgentID) + } + s.log.Info("revision pushed", "revision", n, "agents", len(created), "request_id", a.RequestID) + return out, nil +} + +// deployTargets resolves and locks the agents of a push. +func (s *Service) deployTargets(ctx context.Context, tx *store.Tx, ids []string) ([]*store.Agent, error) { + if len(ids) == 0 { + all, err := tx.ListAgents(ctx) + if err != nil { + return nil, err + } + return slices.DeleteFunc(all, func(ag *store.Agent) bool { return ag.RevokedAt != nil }), nil + } + var out []*store.Agent + var problems []Problem + seen := map[string]bool{} + for i, id := range ids { + params := map[string]string{"agent_id": id, "index": strconv.Itoa(i)} + if seen[id] { + continue + } + seen[id] = true + if !ValidUUID(id) { + problems = append(problems, Problem{Code: "unknown_agent", Field: "agent_ids", Params: params}) + continue + } + ag, err := tx.GetAgent(ctx, id, true) + switch { + case errors.Is(err, store.ErrNotFound): + problems = append(problems, Problem{Code: "unknown_agent", Field: "agent_ids", Params: params}) + case err != nil: + return nil, err + case ag.RevokedAt != nil: + problems = append(problems, Problem{Code: "agent_revoked", Field: "agent_ids", Params: params}) + default: + out = append(out, ag) + } + } + if len(problems) > 0 { + return nil, invalid(problems...) + } + return out, nil +} + +// SetAgentMode changes the operating mode of an agent (ADR 0022). Switching +// to manual keeps the agent on the latest revision (it becomes the target); +// switching to auto_pull supersedes pending pushes of older revisions. A +// connected agent is disconnected so that it reconnects and receives the +// new mode in Welcome. +func (s *Service) SetAgentMode(ctx context.Context, a *Actor, id, mode string) (AgentDTO, error) { + if err := s.requireWritable(); err != nil { + return AgentDTO{}, err + } + if !ValidUUID(id) { + return AgentDTO{}, ErrNotFound + } + if mode != agenthub.ModeAutoPull && mode != agenthub.ModeManual { + return AgentDTO{}, invalidField("invalid_value", "mode") + } + var out AgentDTO + var superseded []*store.Deployment + changed := false + err := s.inTx(ctx, func(tx *store.Tx) error { + before, err := tx.GetAgent(ctx, id, true) + if err != nil { + return err + } + if before.Mode != mode { + latest, err := latestRevisionNumber(ctx, tx) + if err != nil { + return err + } + var target *int64 + if mode == agenthub.ModeManual { + target = &latest + } + if _, err := tx.SetAgentMode(ctx, id, mode, target); err != nil { + return err + } + if mode == agenthub.ModeAutoPull { + superseded, err = tx.SupersedeDeployments(ctx, store.SupersedeFilter{AgentID: id, KeepRevision: latest}, + s.now().UTC().Truncate(time.Microsecond)) + if err != nil { + return err + } + } + changed = true + } + if out, err = s.loadAgentDTO(ctx, tx, id); err != nil { + return err + } + if !changed { + return nil + } + return s.audit(ctx, tx, a, "agent.update", "agent", id, before.Hostname, s.agentDTO(before, 0, nil), out) + }) + if err != nil { + return AgentDTO{}, err + } + if changed { + if s.opts.Hub != nil { + s.opts.Hub.Disconnect(id, agenthub.ReasonModeChanged) + out.Connected = false + } + for _, d := range superseded { + s.publishDeployment(d) + } + s.publishAgent(id) + } + return out, nil +} diff --git a/server/internal/service/revisions.go b/server/internal/service/revisions.go index c918115..a2892f9 100644 --- a/server/internal/service/revisions.go +++ b/server/internal/service/revisions.go @@ -6,12 +6,8 @@ import ( "bytes" "context" "encoding/json" - "errors" "net/netip" - "strconv" - "strings" "time" - "unicode/utf8" "git.beugsiud.de/beug.k/DDI/server/internal/configgen" "git.beugsiud.de/beug.k/DDI/server/internal/store" @@ -122,89 +118,6 @@ func generationProblem(g *Generated) error { return nil } -// CreateRevision generates the configuration and stores it as a new -// revision (ADR 0021). If the output equals the latest revision, that one -// is returned with created false. Agents in auto_pull mode are notified. -func (s *Service) CreateRevision(ctx context.Context, a *Actor, comment string) (*RevisionDTO, bool, error) { - if err := s.requireWritable(); err != nil { - return nil, false, err - } - comment = strings.TrimSpace(comment) - if utf8.RuneCountInString(comment) > 500 { - return nil, false, invalidField("too_long", "comment") - } - var out RevisionDTO - created := false - err := s.inTx(ctx, func(tx *store.Tx) error { - if err := tx.LockRevisions(ctx); err != nil { - return err - } - latest, err := tx.LatestRevision(ctx) - if err != nil && !errors.Is(err, store.ErrNotFound) { - return err - } - next := int64(1) - if latest != nil { - // Generate with the latest number first: identical output - // means nothing changed since then. - g, err := s.generate(ctx, uint32(latest.Number)) //nolint:gosec // 1..2^32-1 by constraint - if err != nil { - return err - } - if err := generationProblem(g); err != nil { - return err - } - stored, err := tx.RevisionFiles(ctx, latest.Number, "") - if err != nil { - return err - } - if sameFiles(revisionFiles(g), stored) { - out = revisionDTO(latest, stored) - return nil - } - if latest.Number >= store.MaxRevision { - return ErrRevisionLimit - } - next = latest.Number + 1 - } - g, err := s.generate(ctx, uint32(next)) //nolint:gosec // 1..2^32-1, checked above - if err != nil { - return err - } - if err := generationProblem(g); err != nil { - return err - } - warnings, err := json.Marshal(g.Warnings) - if err != nil { - return err - } - r := &store.Revision{Number: next, CreatedAt: s.now().UTC().Truncate(time.Microsecond), CreatedByName: a.Username, - Comment: comment, Warnings: warnings} - if a.UserID != "" { - uid := a.UserID - r.CreatedBy = &uid - } - files := revisionFiles(g) - if err := tx.InsertRevision(ctx, r, files); err != nil { - return err - } - out = revisionDTO(r, files) - created = true - summary := RevisionSummaryDTO{Number: r.Number, CreatedAt: r.CreatedAt, CreatedByName: r.CreatedByName, - Comment: r.Comment, DNSFiles: len(g.DNS), DHCPFiles: len(g.DHCP), Warnings: len(g.Warnings)} - return s.audit(ctx, tx, a, "revision.create", "revision", strconv.FormatInt(r.Number, 10), - "#"+strconv.FormatInt(r.Number, 10), nil, summary) - }) - if err != nil { - return nil, false, err - } - if created && s.opts.Hub != nil { - n := s.opts.Hub.NotifyRevision(uint64(out.Number)) //nolint:gosec // positive - s.log.Info("revision created", "revision", out.Number, "notified_agents", n, "request_id", a.RequestID) - } - return &out, created, nil -} - // ListRevisions returns revisions newest first. before (0 = none) pages. func (s *Service) ListRevisions(ctx context.Context, limit int, before int64) (*RevisionPage, error) { if limit <= 0 || limit > MaxRevisionPage { diff --git a/server/internal/service/service.go b/server/internal/service/service.go index 73804f3..84c2306 100644 --- a/server/internal/service/service.go +++ b/server/internal/service/service.go @@ -7,10 +7,12 @@ import ( "encoding/json" "log/slog" "net/netip" + "slices" "time" "git.beugsiud.de/beug.k/DDI/server/internal/agenthub" "git.beugsiud.de/beug.k/DDI/server/internal/auth" + "git.beugsiud.de/beug.k/DDI/server/internal/events" "git.beugsiud.de/beug.k/DDI/server/internal/license" "git.beugsiud.de/beug.k/DDI/server/internal/pki" "git.beugsiud.de/beug.k/DDI/server/internal/secrets" @@ -52,6 +54,8 @@ type Options struct { AgentAPIAddress string // AgentServerNames are the SANs of the agent-port certificate. AgentServerNames []string + // Events receives change notifications for the live UI (optional). + Events *events.Bus } // Service implements the use cases. @@ -129,7 +133,36 @@ func (s *Service) audit(ctx context.Context, tx *store.Tx, a *Actor, action, obj return err } } - return tx.InsertAudit(ctx, e) + if err := tx.InsertAudit(ctx, e); err != nil { + return err + } + if affectsConfig(action, objType) { + tx.AfterCommit(events.TypePending, func() { s.publish(events.TypePending, events.Pending{}) }) + } + return nil +} + +// configObjectTypes and configActions identify audit entries of data the +// generator uses (pending changes, ADR 0022). +var ( + configObjectTypes = []string{"network", "address"} + configActions = []string{"settings.generator"} +) + +func affectsConfig(action, objType string) bool { + return slices.Contains(configObjectTypes, objType) || slices.Contains(configActions, action) +} + +// Events returns the event bus (nil if not configured). +func (s *Service) Events() *events.Bus { return s.opts.Events } + +// publish sends an event to the live UI. +func (s *Service) publish(typ string, data any) { s.opts.Events.Publish(typ, data) } + +func (s *Service) publishAgent(id string) { s.publish(events.TypeAgent, events.Agent{ID: id}) } + +func (s *Service) publishDeployment(d *store.Deployment) { + s.publish(events.TypeDeployment, events.Deployment{ID: d.ID, AgentID: d.AgentID, Revision: d.Revision}) } // inTx runs fn in a transaction and translates errors. diff --git a/server/internal/service/source_test.go b/server/internal/service/source_test.go new file mode 100644 index 0000000..0f93325 --- /dev/null +++ b/server/internal/service/source_test.go @@ -0,0 +1,167 @@ +// SPDX-License-Identifier: LicenseRef-DDI-Proprietary + +package service + +import ( + "bytes" + "context" + "encoding/json" + "net/netip" + "reflect" + "testing" + "time" + + "git.beugsiud.de/beug.k/DDI/server/internal/configgen" + "git.beugsiud.de/beug.k/DDI/server/internal/ipam" + "git.beugsiud.de/beug.k/DDI/server/internal/store/storetest" +) + +func addrPtr(s string) *netip.Addr { + a := netip.MustParseAddr(s) + return &a +} + +func intPtr(v int) *int { return &v } + +// testSource covers the types that must survive JSON: netip values (also +// zero values and pointers), maps, nil and empty slices, times. +func testSource(serial uint32) configgen.Source { + at := time.Date(2026, 10, 13, 12, 0, 0, 123456000, time.FixedZone("CEST", 2*3600)) + return configgen.Source{ + Serial: serial, + Networks: []ipam.Network{ + {ID: "n1", CIDR: netip.MustParsePrefix("192.0.2.0/24"), Name: "Office", Gateway: addrPtr("192.0.2.1"), + Domain: "office.test", SearchDomains: []string{"office.test", "example.test"}, DHCPEnabled: true, + LeaseTime: intPtr(7200), DNSServers: []netip.Addr{netip.MustParseAddr("192.0.2.53")}, + NTPServers: []netip.Addr{}, Pools: []ipam.Pool{{Start: netip.MustParseAddr("192.0.2.100"), End: netip.MustParseAddr("192.0.2.199")}}, + Version: 3, CreatedAt: at, UpdatedAt: at}, + {ID: "n2", CIDR: netip.MustParsePrefix("2001:db8:1::/64"), Name: "v6", Domain: "v6.test"}, + {ID: "n3", CIDR: netip.MustParsePrefix("198.51.100.0/25"), Name: "Lab", DHCPEnabled: true}, + }, + Addresses: []ipam.Address{ + {ID: "a1", NetworkID: "n1", IP: netip.MustParseAddr("192.0.2.53"), Type: ipam.TypeStatic, Hostname: "ns1", + FQDN: "ns1.office.test", Tags: []string{"dns"}, CreatedAt: at, UpdatedAt: at}, + {ID: "a2", NetworkID: "n1", IP: netip.MustParseAddr("192.0.2.10"), Type: ipam.TypeManualDHCP, + MAC: "00:11:22:33:44:55", Hostname: "printer", FQDN: "printer.office.test", LastSeen: &at}, + {ID: "a3", NetworkID: "n1", IP: netip.MustParseAddr("192.0.2.11"), Type: ipam.TypeManualDHCP, Hostname: "nomac"}, + {ID: "a4", NetworkID: "n2", IP: netip.MustParseAddr("2001:db8:1::10"), Type: ipam.TypeStatic, Hostname: "web", + FQDN: "web.v6.test"}, + {ID: "a5", NetworkID: "n1", IP: netip.MustParseAddr("192.0.2.12"), Type: ipam.TypeReserved}, + }, + SubnetIDs: map[string]uint32{"n1": 1, "n2": 2}, + DNS: configgen.DNSSettings{ + ZoneDir: "/var/lib/bind/ddi-suite", + Options: configgen.DNSOptions{Recursion: true, AllowRecursion: []netip.Prefix{netip.MustParsePrefix("192.0.2.0/24")}, Forwarders: []netip.Addr{netip.MustParseAddr("9.9.9.9")}}, + DefaultTTL: 3600, + SOA: configgen.SOA{PrimaryNS: "ns1.office.test", Hostmaster: "hostmaster.office.test", Refresh: 3600, Retry: 600, Expire: 604800, Minimum: 300}, + NameServers: []configgen.NameServer{{Name: "ns1.office.test", Addrs: []netip.Addr{netip.MustParseAddr("192.0.2.53")}}, + {Name: "ns2.example.net"}}, + }, + DHCP: configgen.DHCPSettings{Interfaces: []string{"eth0"}, DirectClients: true, ControlSocket: "/run/kea/kea4-ctrl-socket", + LeaseFile: "/var/lib/kea/kea-leases4.csv", ValidLifetime: 3600}, + } +} + +func sameOutput(t *testing.T, a, b *Generated) { + t.Helper() + if !sameFiles(revisionFiles(a), revisionFiles(b)) { + t.Error("rendered files differ") + } + if !reflect.DeepEqual(a.Warnings, b.Warnings) || !reflect.DeepEqual(a.DNSError, b.DNSError) || !reflect.DeepEqual(a.DHCPError, b.DHCPError) { + t.Errorf("warnings or errors differ: %+v / %+v", a, b) + } +} + +func TestSourceJSONRoundTrip(t *testing.T) { + src := testSource(7) + orig := render(src) + if orig.DNSError != nil || orig.DHCPError != nil || len(orig.DNS) == 0 || len(orig.DHCP) == 0 || len(orig.Warnings) == 0 { + t.Fatalf("test source does not exercise the generator: %+v", orig) + } + raw, err := json.Marshal(src) + if err != nil { + t.Fatal(err) + } + var back configgen.Source + if err := json.Unmarshal(raw, &back); err != nil { + t.Fatal(err) + } + sameOutput(t, orig, render(back)) + again, _ := json.Marshal(back) + if !bytes.Equal(raw, again) { + t.Errorf("JSON is not stable:\n%s\n%s", raw, again) + } + if back.Networks[0].Gateway == nil || *back.Networks[0].Gateway != *src.Networks[0].Gateway || + back.Networks[1].Gateway != nil || back.Networks[0].NTPServers == nil || back.Networks[1].DNSServers != nil || + back.SubnetIDs["n2"] != 2 || !back.Networks[0].CreatedAt.Equal(src.Networks[0].CreatedAt) { + t.Errorf("values changed: %+v", back.Networks) + } + // Another serial changes only the SOA serial. + back.Serial = 8 + if sameFiles(revisionFiles(orig), revisionFiles(render(back))) { + t.Error("serial did not change the zones") + } +} + +// TestStoredSourceReRendersIdentically: the source stored with a revision +// (through jsonb) renders the revision's files byte for byte. +func TestStoredSourceReRendersIdentically(t *testing.T) { + st := storetest.New(t) + ctx := context.Background() + svc := New(Options{Store: st}) + a := SystemActor + + g := DefaultGeneratorSettings() + g.DNS.PrimaryNS = "ns1.office.test" + g.DNS.Hostmaster = "hostmaster@office.test" + g.DNS.NameServers = []GeneratorNS{{Name: "ns1.office.test", Addresses: []string{"192.0.2.53"}}} + g.DNS.Recursion = true + g.DNS.AllowRecursion = []string{"192.0.2.0/24", "2001:db8::/32"} + g.DNS.Forwarders = []string{"9.9.9.9"} + if _, err := svc.SetGeneratorSettings(ctx, a, g); err != nil { + t.Fatal(err) + } + n4, err := svc.CreateNetwork(ctx, a, NetworkInput{CIDR: netip.MustParsePrefix("192.0.2.0/24"), Name: "Office", + Gateway: addrPtr("192.0.2.1"), Domain: "office.test", DHCPEnabled: true, LeaseTime: intPtr(7200), + DNSServers: []netip.Addr{netip.MustParseAddr("192.0.2.53")}, SearchDomains: []string{"office.test"}, + Pools: []PoolDTO{{Start: netip.MustParseAddr("192.0.2.100"), End: netip.MustParseAddr("192.0.2.199")}}}) + if err != nil { + t.Fatal(err) + } + n6, err := svc.CreateNetwork(ctx, a, NetworkInput{CIDR: netip.MustParsePrefix("2001:db8:1::/64"), Name: "v6", Domain: "v6.test"}) + if err != nil { + t.Fatal(err) + } + for _, in := range []AddressInput{ + {NetworkID: n4.ID, IP: netip.MustParseAddr("192.0.2.53"), Type: ipam.TypeStatic, Hostname: "ns1"}, + {NetworkID: n4.ID, IP: netip.MustParseAddr("192.0.2.10"), Type: ipam.TypeManualDHCP, MAC: "00:11:22:33:44:55", Hostname: "printer"}, + {NetworkID: n4.ID, IP: netip.MustParseAddr("192.0.2.150"), Type: ipam.TypeReserved}, + {NetworkID: n6.ID, IP: netip.MustParseAddr("2001:db8:1::10"), Type: ipam.TypeStatic, Hostname: "web"}, + } { + if _, err := svc.CreateAddress(ctx, a, in); err != nil { + t.Fatalf("address %v: %v", in.IP, err) + } + } + rev, created, err := svc.CreateRevision(ctx, a, "") + if err != nil || !created { + t.Fatalf("revision: %v %v", created, err) + } + raw, err := st.Q().RevisionSource(ctx, rev.Number) + if err != nil || raw == nil { + t.Fatalf("source %s %v", raw, err) + } + var src configgen.Source + if err := json.Unmarshal(raw, &src); err != nil { + t.Fatal(err) + } + if src.Serial != uint32(rev.Number) { //nolint:gosec // small + t.Errorf("stored serial %d", src.Serial) + } + stored, err := st.Q().RevisionFiles(ctx, rev.Number, "") + if err != nil { + t.Fatal(err) + } + if !sameFiles(revisionFiles(render(src)), stored) { + t.Error("the stored source does not render the stored files") + } +} diff --git a/server/internal/service/users.go b/server/internal/service/users.go index 4e0855b..07e89e9 100644 --- a/server/internal/service/users.go +++ b/server/internal/service/users.go @@ -151,6 +151,31 @@ func (s *Service) Authenticate(ctx context.Context, token string) (*Session, err return &Session{TokenHash: hash, CSRFToken: sess.CSRFToken, MFAPending: sess.MFAPending, User: userDTO(u), ExpiresAt: sess.ExpiresAt}, nil } +// SessionActive reports whether the session is still valid (not logged +// out, expired, idle or of a disabled user) without counting as activity. +// Long-lived requests (event stream) check it periodically. +func (s *Service) SessionActive(ctx context.Context, tokenHash []byte) (bool, error) { + now := s.now() + sess, err := s.st.Q().GetSession(ctx, tokenHash, now) + if errors.Is(err, store.ErrNotFound) { + return false, nil + } + if err != nil { + return false, translate(err) + } + if now.Sub(sess.LastSeenAt) > s.opts.SessionIdle { + return false, nil + } + u, err := s.st.Q().GetUser(ctx, sess.UserID) + if errors.Is(err, store.ErrNotFound) { + return false, nil + } + if err != nil { + return false, translate(err) + } + return !u.Disabled, nil +} + // Logout ends a session. func (s *Service) Logout(ctx context.Context, a *Actor) error { return s.inTx(ctx, func(tx *store.Tx) error { diff --git a/server/internal/store/agents.go b/server/internal/store/agents.go index 22945f4..99b4693 100644 --- a/server/internal/store/agents.go +++ b/server/internal/store/agents.go @@ -66,17 +66,20 @@ type Agent struct { Health json.RawMessage RevokedAt *time.Time LeaseSequence int64 + // TargetRevision is the last revision pushed explicitly (0 = none). + TargetRevision int64 } const agentColumns = `id, role, hostname, agent_version, service_version, mode, enrolled_at, last_seen_at, - remote_address, cert_serial, cert_not_after, applied_revision, last_good_revision, health, revoked_at, lease_sequence` + remote_address, cert_serial, cert_not_after, applied_revision, last_good_revision, health, revoked_at, lease_sequence, + target_revision` func scanAgent(row pgx.Row) (*Agent, error) { var a Agent var health []byte if err := row.Scan(&a.ID, &a.Role, &a.Hostname, &a.AgentVersion, &a.ServiceVersion, &a.Mode, &a.EnrolledAt, &a.LastSeenAt, &a.RemoteAddress, &a.CertSerial, &a.CertNotAfter, &a.AppliedRevision, &a.LastGoodRevision, - &health, &a.RevokedAt, &a.LeaseSequence); err != nil { + &health, &a.RevokedAt, &a.LeaseSequence, &a.TargetRevision); err != nil { return nil, mapErr(err) } a.Health = health @@ -134,6 +137,25 @@ func (t *Tx) RevokeAgent(ctx context.Context, id string, at time.Time) (*Agent, return scanAgent(t.q.QueryRow(ctx, `UPDATE agents SET revoked_at = coalesce(revoked_at, $2) WHERE id = $1 RETURNING `+agentColumns, id, at)) } +// SetAgentMode changes the operating mode; target (if not nil) becomes the +// target revision as well. +func (t *Tx) SetAgentMode(ctx context.Context, id, mode string, target *int64) (*Agent, error) { + return scanAgent(t.q.QueryRow(ctx, `UPDATE agents SET mode = $2::agent_mode, target_revision = coalesce($3, target_revision) + WHERE id = $1 RETURNING `+agentColumns, id, mode, target)) +} + +// SetAgentTarget sets the revision pushed explicitly to the agent. +func (t *Tx) SetAgentTarget(ctx context.Context, id string, revision int64) error { + tag, err := t.q.Exec(ctx, `UPDATE agents SET target_revision = $2 WHERE id = $1`, id, revision) + if err != nil { + return mapErr(err) + } + if tag.RowsAffected() == 0 { + return ErrNotFound + } + return nil +} + // DeleteAgent removes an agent with its deployments and leases. func (t *Tx) DeleteAgent(ctx context.Context, id string) error { tag, err := t.q.Exec(ctx, `DELETE FROM agents WHERE id = $1`, id) diff --git a/server/internal/store/agents_test.go b/server/internal/store/agents_test.go index da51e8a..e8a663e 100644 --- a/server/internal/store/agents_test.go +++ b/server/internal/store/agents_test.go @@ -193,17 +193,17 @@ func TestRevisionsAndDeployments(t *testing.T) { for i, outcome := range []string{"rejected", "succeeded"} { start := now.Add(time.Duration(i) * time.Minute) d := &store.Deployment{AgentID: agentA, Revision: 1, Trigger: "pull", Outcome: outcome, ActiveRevision: int64(i), - StartedAt: start, FinishedAt: start.Add(time.Second)} + StartedAt: start, FinishedAt: ptr(start.Add(time.Second))} if ok, err := st.Q().InsertDeployment(ctx, d); err != nil || !ok || d.ID == 0 { t.Fatalf("insert deployment: %v %v", ok, err) } } // A resent report is dropped without error. - dup := &store.Deployment{AgentID: agentA, Revision: 1, Trigger: "pull", Outcome: "rejected", StartedAt: now, FinishedAt: now.Add(time.Second)} + dup := &store.Deployment{AgentID: agentA, Revision: 1, Trigger: "pull", Outcome: "rejected", StartedAt: now, FinishedAt: ptr(now.Add(time.Second))} if ok, err := st.Q().InsertDeployment(ctx, dup); err != nil || ok { t.Errorf("duplicate report: %v %v", ok, err) } - bad := &store.Deployment{AgentID: agentA, Revision: 2, Trigger: "pull", Outcome: "succeeded", StartedAt: now, FinishedAt: now.Add(-time.Second)} + bad := &store.Deployment{AgentID: agentA, Revision: 2, Trigger: "pull", Outcome: "succeeded", StartedAt: now, FinishedAt: ptr(now.Add(-time.Second))} if _, err := st.Q().InsertDeployment(ctx, bad); !store.IsConstraint(err, "deployments_times") { t.Errorf("finished before started: %v", err) } diff --git a/server/internal/store/audit.go b/server/internal/store/audit.go index 813ec7a..4fab61b 100644 --- a/server/internal/store/audit.go +++ b/server/internal/store/audit.go @@ -7,8 +7,11 @@ import ( "encoding/json" "fmt" "net/netip" + "strconv" "strings" "time" + + "github.com/jackc/pgx/v5" ) // AuditEntry is a row of the audit log. @@ -107,6 +110,49 @@ func (t *Tx) ListAudit(ctx context.Context, f AuditFilter) ([]*AuditEntry, error if err != nil { return nil, mapErr(err) } + return scanAuditRows(rows) +} + +// ChangeFilter selects the audit entries that affect the generated +// configuration (ConfigChangesSince). +type ChangeFilter struct { + ObjectTypes []string + Actions []string + // AfterRevision is the revision whose creation (audit action + // revision.create or revision.rollback) starts the range; 0 = from the + // beginning. + AfterRevision int64 + // Since starts the range if the revision has no audit entry. + Since time.Time + Limit int +} + +// ConfigChangesSince returns matching entries newest first. The range is +// bounded by the ID of the revision's own audit entry, so the clocks of +// server and database do not matter. +func (t *Tx) ConfigChangesSince(ctx context.Context, f ChangeFilter) ([]*AuditEntry, error) { + limit := f.Limit + if limit <= 0 || limit > 10000 { + limit = 100 + } + rows, err := t.q.Query(ctx, ` + SELECT id, at, actor_id, actor_name, action, object_type, object_id, object_name, before, after, request_id, ip + FROM audit_log + WHERE (object_type = ANY($1) OR action = ANY($2)) + AND id > coalesce( + (SELECT max(id) FROM audit_log WHERE object_type = 'revision' AND object_id = $3 + AND action IN ('revision.create', 'revision.rollback')), + (SELECT max(id) FROM audit_log WHERE at < $4), + 0) + ORDER BY id DESC LIMIT `+strconv.Itoa(limit), + f.ObjectTypes, f.Actions, strconv.FormatInt(f.AfterRevision, 10), f.Since) + if err != nil { + return nil, mapErr(err) + } + return scanAuditRows(rows) +} + +func scanAuditRows(rows pgx.Rows) ([]*AuditEntry, error) { defer rows.Close() var out []*AuditEntry for rows.Next() { diff --git a/server/internal/store/deployments.go b/server/internal/store/deployments.go new file mode 100644 index 0000000..a9223d6 --- /dev/null +++ b/server/internal/store/deployments.go @@ -0,0 +1,235 @@ +// SPDX-License-Identifier: LicenseRef-DDI-Proprietary + +package store + +import ( + "context" + "errors" + "fmt" + "strings" + "time" + + "github.com/jackc/pgx/v5" +) + +// Deployment outcomes (enum deployment_outcome). +const ( + OutcomePending = "pending" + OutcomeSucceeded = "succeeded" + OutcomeRejected = "rejected" + OutcomeRolledBack = "rolled_back" + OutcomeFailed = "failed" + OutcomeSuperseded = "superseded" +) + +// Deployment is a row of deployments; AgentHostname comes from the agent. +// Rows created by a push from the server (ADR 0022) have RequestedByName +// set and stay pending until the agent reports. +type Deployment struct { + ID int64 + AgentID string + AgentHostname string + CorrelationID string + Revision int64 + Trigger string + Outcome string + Message string + ActiveRevision int64 + StartedAt time.Time + FinishedAt *time.Time + ReportedAt time.Time + RequestedBy *string + RequestedByName *string +} + +const deploymentColumns = `d.id, d.agent_id, a.hostname, d.correlation_id, d.revision, d.trigger_kind, d.outcome, d.message, + d.active_revision, d.started_at, d.finished_at, d.reported_at, d.requested_by, d.requested_by_name` + +func scanDeployment(row pgx.Row) (*Deployment, error) { + var d Deployment + if err := row.Scan(&d.ID, &d.AgentID, &d.AgentHostname, &d.CorrelationID, &d.Revision, &d.Trigger, &d.Outcome, + &d.Message, &d.ActiveRevision, &d.StartedAt, &d.FinishedAt, &d.ReportedAt, &d.RequestedBy, + &d.RequestedByName); err != nil { + return nil, mapErr(err) + } + return &d, nil +} + +func scanDeployments(rows pgx.Rows, err error) ([]*Deployment, error) { + if err != nil { + return nil, mapErr(err) + } + defer rows.Close() + var out []*Deployment + for rows.Next() { + d, err := scanDeployment(rows) + if err != nil { + return nil, err + } + out = append(out, d) + } + return out, mapErr(rows.Err()) +} + +// InsertDeployment stores a deployment report of an agent; ID and +// ReportedAt are set. A report with the same agent, revision and start time +// exists already if inserted is false (resent after an agent crash). +func (t *Tx) InsertDeployment(ctx context.Context, d *Deployment) (inserted bool, err error) { + err = mapErr(t.q.QueryRow(ctx, ` + INSERT INTO deployments (agent_id, correlation_id, revision, trigger_kind, outcome, message, active_revision, + started_at, finished_at) + VALUES ($1, $2, $3, $4::deployment_trigger, $5::deployment_outcome, $6, $7, $8, $9) + ON CONFLICT (agent_id, revision, started_at) WHERE requested_by_name IS NULL DO NOTHING + RETURNING id, reported_at`, + d.AgentID, d.CorrelationID, d.Revision, d.Trigger, d.Outcome, d.Message, d.ActiveRevision, + d.StartedAt, d.FinishedAt).Scan(&d.ID, &d.ReportedAt)) + if errors.Is(err, ErrNotFound) { + return false, nil + } + return err == nil, err +} + +// InsertPushDeployment stores a pending push deployment requested by a user +// (RequestedByName must be set). ID, ReportedAt and Outcome are set. +func (t *Tx) InsertPushDeployment(ctx context.Context, d *Deployment) error { + d.Trigger, d.Outcome, d.FinishedAt = "push", OutcomePending, nil + return mapErr(t.q.QueryRow(ctx, ` + INSERT INTO deployments (agent_id, revision, trigger_kind, outcome, started_at, requested_by, requested_by_name) + VALUES ($1, $2, 'push', 'pending', $3, $4, $5) + RETURNING id, reported_at`, + d.AgentID, d.Revision, d.StartedAt, d.RequestedBy, d.RequestedByName).Scan(&d.ID, &d.ReportedAt)) +} + +// PendingDeployment returns the newest pending deployment of an agent or +// ErrNotFound. +func (t *Tx) PendingDeployment(ctx context.Context, agentID string) (*Deployment, error) { + return scanDeployment(t.q.QueryRow(ctx, `SELECT `+deploymentColumns+` + FROM deployments d JOIN agents a ON a.id = d.agent_id + WHERE d.agent_id = $1 AND d.outcome = 'pending' ORDER BY d.id DESC LIMIT 1`, agentID)) +} + +// SupersedeFilter selects pending deployments to supersede. +type SupersedeFilter struct { + // AgentID restricts the update to one agent ("" = all agents). + AgentID string + // AutoPullOnly restricts it to agents in auto_pull mode. + AutoPullOnly bool + // KeepRevision keeps pending deployments of this revision (0 = none). + KeepRevision int64 +} + +// SupersedeDeployments marks matching pending deployments superseded at the +// given time and returns them. +func (t *Tx) SupersedeDeployments(ctx context.Context, f SupersedeFilter, at time.Time) ([]*Deployment, error) { + where := []string{"a.id = d.agent_id", "d.outcome = 'pending'"} + args := []any{at} + if f.AgentID != "" { + args = append(args, f.AgentID) + where = append(where, fmt.Sprintf("d.agent_id = $%d", len(args))) + } + if f.AutoPullOnly { + where = append(where, "a.mode = 'auto_pull'") + } + if f.KeepRevision > 0 { + args = append(args, f.KeepRevision) + where = append(where, fmt.Sprintf("d.revision <> $%d", len(args))) + } + return scanDeployments(t.q.Query(ctx, `UPDATE deployments d SET outcome = 'superseded', finished_at = greatest($1, d.started_at) + FROM agents a WHERE `+strings.Join(where, " AND ")+` RETURNING `+deploymentColumns, args...)) +} + +// DeploymentResult is what an agent reports for a push deployment. +type DeploymentResult struct { + Outcome string + Message string + ActiveRevision int64 + StartedAt time.Time + FinishedAt time.Time +} + +// CompleteDeployment stores the agent's result in the push deployment id of +// agentID for revision. A deployment that is pending or was superseded in +// the meantime is updated (completed true). One that has a result already +// is returned unchanged (a resent report). ErrNotFound means there is no +// such deployment of that agent and revision. +func (t *Tx) CompleteDeployment(ctx context.Context, id int64, agentID string, revision int64, r DeploymentResult) (d *Deployment, completed bool, err error) { + d, err = scanDeployment(t.q.QueryRow(ctx, ` + UPDATE deployments d SET outcome = $4::deployment_outcome, message = $5, active_revision = $6, started_at = $7, + finished_at = $8, reported_at = now() + FROM agents a + WHERE a.id = d.agent_id AND d.id = $1 AND d.agent_id = $2 AND d.revision = $3 AND d.requested_by_name IS NOT NULL + AND d.outcome IN ('pending', 'superseded') + RETURNING `+deploymentColumns, + id, agentID, revision, r.Outcome, r.Message, r.ActiveRevision, r.StartedAt, r.FinishedAt)) + if err == nil { + return d, true, nil + } + if !errors.Is(err, ErrNotFound) { + return nil, false, err + } + d, err = scanDeployment(t.q.QueryRow(ctx, `SELECT `+deploymentColumns+` + FROM deployments d JOIN agents a ON a.id = d.agent_id + WHERE d.id = $1 AND d.agent_id = $2 AND d.revision = $3 AND d.requested_by_name IS NOT NULL`, id, agentID, revision)) + if err != nil { + return nil, false, err + } + return d, false, nil +} + +// DeploymentFilter selects deployments. Zero values mean "no restriction". +type DeploymentFilter struct { + AgentID string + Revision int64 + BeforeID int64 + Limit int +} + +// ListDeployments returns deployments newest first. +func (t *Tx) ListDeployments(ctx context.Context, f DeploymentFilter) ([]*Deployment, error) { + var where []string + var args []any + add := func(cond string, v any) { + args = append(args, v) + where = append(where, fmt.Sprintf(cond, len(args))) + } + if f.AgentID != "" { + add("d.agent_id = $%d", f.AgentID) + } + if f.Revision > 0 { + add("d.revision = $%d", f.Revision) + } + if f.BeforeID > 0 { + add("d.id < $%d", f.BeforeID) + } + limit := f.Limit + if limit <= 0 { + limit = 100 + } + sql := `SELECT ` + deploymentColumns + ` FROM deployments d JOIN agents a ON a.id = d.agent_id` + if len(where) > 0 { + sql += " WHERE " + strings.Join(where, " AND ") + } + sql += fmt.Sprintf(" ORDER BY d.id DESC LIMIT %d", limit) + return scanDeployments(t.q.Query(ctx, sql, args...)) +} + +// LatestDeployments returns the newest deployment of every agent that has +// one, keyed by agent ID. +func (t *Tx) LatestDeployments(ctx context.Context) (map[string]*Deployment, error) { + list, err := scanDeployments(t.q.Query(ctx, `SELECT DISTINCT ON (d.agent_id) `+deploymentColumns+` + FROM deployments d JOIN agents a ON a.id = d.agent_id ORDER BY d.agent_id, d.id DESC`)) + if err != nil { + return nil, err + } + out := make(map[string]*Deployment, len(list)) + for _, d := range list { + out[d.AgentID] = d + } + return out, nil +} + +// LatestDeployment returns the newest deployment of an agent or ErrNotFound. +func (t *Tx) LatestDeployment(ctx context.Context, agentID string) (*Deployment, error) { + return scanDeployment(t.q.QueryRow(ctx, `SELECT `+deploymentColumns+` + FROM deployments d JOIN agents a ON a.id = d.agent_id WHERE d.agent_id = $1 ORDER BY d.id DESC LIMIT 1`, agentID)) +} diff --git a/server/internal/store/migrations/00004_push.sql b/server/internal/store/migrations/00004_push.sql new file mode 100644 index 0000000..98004d4 --- /dev/null +++ b/server/internal/store/migrations/00004_push.sql @@ -0,0 +1,65 @@ +-- SPDX-License-Identifier: LicenseRef-DDI-Proprietary +-- Pull and push (M4, ADR 0022): operating modes with a target revision, +-- pending push deployments created by the server, and the generator input of +-- every revision (for rollbacks with a new SOA serial). + +-- +goose Up + +-- Last revision pushed explicitly; the desired revision of agents in manual +-- mode (0 = none). +ALTER TABLE agents ADD COLUMN target_revision bigint NOT NULL DEFAULT 0; +ALTER TABLE agents ADD CONSTRAINT agents_target_revision CHECK (target_revision BETWEEN 0 AND 4294967295); + +-- Generator input (configgen.Source as JSON) the revision was rendered from; +-- NULL for revisions created before M4 (they cannot be rolled back to). +-- Adding a column does not fire the immutability trigger. +ALTER TABLE config_revisions ADD COLUMN source jsonb; +ALTER TABLE config_revisions ADD CONSTRAINT config_revisions_source_object + CHECK (source IS NULL OR jsonb_typeof(source) = 'object'); + +-- New outcomes: pending (pushed, no report yet) and superseded (replaced by a +-- newer push before the agent reported). A new enum type instead of ALTER +-- TYPE ... ADD VALUE, whose values cannot be used in the same transaction. +ALTER TYPE deployment_outcome RENAME TO deployment_outcome_m3; +CREATE TYPE deployment_outcome AS ENUM ('pending', 'succeeded', 'rejected', 'rolled_back', 'failed', 'superseded'); +ALTER TABLE deployments ALTER COLUMN outcome TYPE deployment_outcome USING outcome::text::deployment_outcome; +DROP TYPE deployment_outcome_m3; + +-- Pending deployments have no end yet. +ALTER TABLE deployments ALTER COLUMN finished_at DROP NOT NULL; +ALTER TABLE deployments ADD CONSTRAINT deployments_finished + CHECK ((outcome = 'pending') = (finished_at IS NULL)); + +-- Push deployments created by the server record who requested them. Rows +-- reported by agents (pulls, M3 pushes) have requested_by_name NULL. +ALTER TABLE deployments ADD COLUMN requested_by uuid REFERENCES users (id) ON DELETE SET NULL; +ALTER TABLE deployments ADD COLUMN requested_by_name text; +ALTER TABLE deployments ADD CONSTRAINT deployments_requested_by + CHECK (requested_by IS NULL OR requested_by_name IS NOT NULL); + +-- Duplicate protection applies to reports only: an agent may resend a report +-- after a crash, the copy is dropped. Server-created rows are matched by ID. +ALTER TABLE deployments DROP CONSTRAINT deployments_report_unique; +CREATE UNIQUE INDEX deployments_report_unique ON deployments (agent_id, revision, started_at) + WHERE requested_by_name IS NULL; +CREATE INDEX deployments_pending ON deployments (agent_id, id DESC) WHERE outcome = 'pending'; + +-- +goose Down +DROP INDEX deployments_pending; +DROP INDEX deployments_report_unique; +DELETE FROM deployments WHERE requested_by_name IS NOT NULL AND outcome IN ('pending', 'superseded'); +ALTER TABLE deployments ADD CONSTRAINT deployments_report_unique UNIQUE (agent_id, revision, started_at); +ALTER TABLE deployments DROP CONSTRAINT deployments_requested_by; +ALTER TABLE deployments DROP COLUMN requested_by_name; +ALTER TABLE deployments DROP COLUMN requested_by; +ALTER TABLE deployments DROP CONSTRAINT deployments_finished; +DELETE FROM deployments WHERE outcome IN ('pending', 'superseded'); +ALTER TABLE deployments ALTER COLUMN finished_at SET NOT NULL; +ALTER TYPE deployment_outcome RENAME TO deployment_outcome_m4; +CREATE TYPE deployment_outcome AS ENUM ('succeeded', 'rejected', 'rolled_back', 'failed'); +ALTER TABLE deployments ALTER COLUMN outcome TYPE deployment_outcome USING outcome::text::deployment_outcome; +DROP TYPE deployment_outcome_m4; +ALTER TABLE config_revisions DROP CONSTRAINT config_revisions_source_object; +ALTER TABLE config_revisions DROP COLUMN source; +ALTER TABLE agents DROP CONSTRAINT agents_target_revision; +ALTER TABLE agents DROP COLUMN target_revision; diff --git a/server/internal/store/push_test.go b/server/internal/store/push_test.go new file mode 100644 index 0000000..c84c683 --- /dev/null +++ b/server/internal/store/push_test.go @@ -0,0 +1,259 @@ +// SPDX-License-Identifier: LicenseRef-DDI-Proprietary + +package store_test + +import ( + "context" + "encoding/json" + "errors" + "testing" + "time" + + "git.beugsiud.de/beug.k/DDI/server/internal/store" + "git.beugsiud.de/beug.k/DDI/server/internal/store/storetest" +) + +func ptr[T any](v T) *T { return &v } + +func TestAgentModeAndTarget(t *testing.T) { + st := storetest.New(t) + ctx := context.Background() + a := newAgent(t, st, agentA, "dns", "ns1", "0a") + if a.TargetRevision != 0 || a.Mode != "auto_pull" { + t.Fatalf("new agent %+v", a) + } + got, err := st.Q().SetAgentMode(ctx, agentA, "manual", ptr(int64(4))) + if err != nil || got.Mode != "manual" || got.TargetRevision != 4 { + t.Fatalf("set mode %+v %v", got, err) + } + if got, err = st.Q().SetAgentMode(ctx, agentA, "auto_pull", nil); err != nil || got.TargetRevision != 4 { + t.Fatalf("mode without target %+v %v", got, err) + } + if err := st.Q().SetAgentTarget(ctx, agentA, 9); err != nil { + t.Fatal(err) + } + if got, _ := st.Q().GetAgent(ctx, agentA, false); got.TargetRevision != 9 { + t.Errorf("target %d", got.TargetRevision) + } + if err := st.Q().SetAgentTarget(ctx, agentA, 1<<32); !store.IsConstraint(err, "agents_target_revision") { + t.Errorf("target out of range: %v", err) + } + if err := st.Q().SetAgentTarget(ctx, agentB, 1); !errors.Is(err, store.ErrNotFound) { + t.Errorf("unknown agent: %v", err) + } + if _, err := st.Q().SetAgentMode(ctx, agentB, "manual", nil); !errors.Is(err, store.ErrNotFound) { + t.Errorf("mode of unknown agent: %v", err) + } +} + +func TestRevisionSource(t *testing.T) { + st := storetest.New(t) + ctx := context.Background() + now := time.Now().UTC().Truncate(time.Microsecond) + src := json.RawMessage(`{"Serial":1,"SubnetIDs":{"a":7},"Networks":null}`) + if err := st.Q().InsertRevision(ctx, &store.Revision{Number: 1, CreatedAt: now, CreatedByName: "admin", Source: src}, nil); err != nil { + t.Fatal(err) + } + if err := st.Q().InsertRevision(ctx, &store.Revision{Number: 2, CreatedAt: now, CreatedByName: "admin"}, nil); err != nil { + t.Fatal(err) + } + got, err := st.Q().RevisionSource(ctx, 1) + if err != nil { + t.Fatal(err) + } + var a, b map[string]any + if json.Unmarshal(got, &a) != nil || json.Unmarshal(src, &b) != nil || len(a) != 3 || a["SubnetIDs"].(map[string]any)["a"] != b["SubnetIDs"].(map[string]any)["a"] { + t.Errorf("source %s", got) + } + if got, err := st.Q().RevisionSource(ctx, 2); err != nil || got != nil { + t.Errorf("revision without source: %s %v", got, err) + } + if _, err := st.Q().RevisionSource(ctx, 3); !errors.Is(err, store.ErrNotFound) { + t.Errorf("unknown revision: %v", err) + } + if err := st.Q().InsertRevision(ctx, &store.Revision{Number: 3, CreatedAt: now, CreatedByName: "admin", + Source: json.RawMessage(`[1]`)}, nil); !store.IsConstraint(err, "config_revisions_source_object") { + t.Errorf("array source: %v", err) + } +} + +func TestPushDeployments(t *testing.T) { + st := storetest.New(t) + ctx := context.Background() + q := st.Q() + now := time.Now().UTC().Truncate(time.Microsecond) + newAgent(t, st, agentA, "dns", "ns1", "0a") + newAgent(t, st, agentB, "dhcp", "dhcp1", "0b") + if _, err := q.SetAgentMode(ctx, agentB, "manual", nil); err != nil { + t.Fatal(err) + } + + push := func(agent string, rev int64) *store.Deployment { + t.Helper() + d := &store.Deployment{AgentID: agent, Revision: rev, StartedAt: now, RequestedByName: ptr("admin")} + if err := q.InsertPushDeployment(ctx, d); err != nil { + t.Fatalf("push: %v", err) + } + if d.ID == 0 || d.Outcome != store.OutcomePending || d.Trigger != "push" { + t.Fatalf("pushed %+v", d) + } + return d + } + if _, err := q.PendingDeployment(ctx, agentA); !errors.Is(err, store.ErrNotFound) { + t.Fatalf("pending of none: %v", err) + } + // Same agent, revision and time: pushes are not subject to the + // duplicate protection of reports. + p1 := push(agentA, 1) + p2 := push(agentA, 1) + pb := push(agentB, 1) + got, err := q.PendingDeployment(ctx, agentA) + if err != nil || got.ID != p2.ID || got.FinishedAt != nil || got.RequestedByName == nil || *got.RequestedByName != "admin" || + got.AgentHostname != "ns1" { + t.Fatalf("pending %+v %v", got, err) + } + + // Superseding: per agent, keeping a revision, auto_pull only. + sup, err := q.SupersedeDeployments(ctx, store.SupersedeFilter{AgentID: agentA, KeepRevision: 1}, now) + if err != nil || len(sup) != 0 { + t.Fatalf("keep revision: %+v %v", sup, err) + } + sup, err = q.SupersedeDeployments(ctx, store.SupersedeFilter{AutoPullOnly: true}, now.Add(time.Second)) + if err != nil || len(sup) != 2 { + t.Fatalf("supersede auto_pull: %+v %v", sup, err) + } + for _, d := range sup { + if d.AgentID != agentA || d.Outcome != store.OutcomeSuperseded || d.FinishedAt == nil || !d.FinishedAt.Equal(now.Add(time.Second)) { + t.Errorf("superseded %+v", d) + } + } + if got, err := q.PendingDeployment(ctx, agentB); err != nil || got.ID != pb.ID { + t.Errorf("manual agent's push superseded: %+v %v", got, err) + } + + // Completing a superseded push records the real outcome; a resent + // report changes nothing. + res := store.DeploymentResult{Outcome: "succeeded", Message: "ok", ActiveRevision: 1, StartedAt: now.Add(time.Second), + FinishedAt: now.Add(2 * time.Second)} + d, done, err := q.CompleteDeployment(ctx, p1.ID, agentA, 1, res) + if err != nil || !done || d.Outcome != "succeeded" || d.Message != "ok" || d.FinishedAt == nil || !d.StartedAt.Equal(res.StartedAt) { + t.Fatalf("complete %+v %v %v", d, done, err) + } + res.Outcome = "failed" + if d, done, err := q.CompleteDeployment(ctx, p1.ID, agentA, 1, res); err != nil || done || d.Outcome != "succeeded" { + t.Errorf("resent %+v %v %v", d, done, err) + } + for _, c := range []struct { + id int64 + agent string + rev int64 + }{{p2.ID, agentB, 1}, {p2.ID, agentA, 2}, {p2.ID + 100, agentA, 1}} { + if _, _, err := q.CompleteDeployment(ctx, c.id, c.agent, c.rev, res); !errors.Is(err, store.ErrNotFound) { + t.Errorf("complete %+v: %v", c, err) + } + } + // Reports cannot be completed by ID. + rep := &store.Deployment{AgentID: agentA, Revision: 1, Trigger: "pull", Outcome: "succeeded", StartedAt: now, FinishedAt: ptr(now)} + if ok, err := q.InsertDeployment(ctx, rep); err != nil || !ok { + t.Fatal(err) + } + if ok, err := q.InsertDeployment(ctx, &store.Deployment{AgentID: agentA, Revision: 1, Trigger: "pull", Outcome: "failed", + StartedAt: now, FinishedAt: ptr(now)}); err != nil || ok { + t.Errorf("duplicate report: %v %v", ok, err) + } + if _, _, err := q.CompleteDeployment(ctx, rep.ID, agentA, 1, res); !errors.Is(err, store.ErrNotFound) { + t.Errorf("complete a report: %v", err) + } + // A pending outcome needs an open end and vice versa. + if _, err := q.InsertDeployment(ctx, &store.Deployment{AgentID: agentA, Revision: 3, Trigger: "pull", Outcome: "succeeded", + StartedAt: now}); !store.IsConstraint(err, "deployments_finished") { + t.Errorf("finished outcome without end: %v", err) + } + + list, err := q.ListDeployments(ctx, store.DeploymentFilter{AgentID: agentA}) + if err != nil || len(list) != 3 || list[0].ID != rep.ID || list[1].Outcome != store.OutcomeSuperseded { + t.Errorf("list %+v %v", list, err) + } + latest, _ := q.LatestDeployments(ctx) + if latest[agentB] == nil || latest[agentB].Outcome != store.OutcomePending { + t.Errorf("latest %+v", latest) + } +} + +func TestAfterCommit(t *testing.T) { + st := storetest.New(t) + ctx := context.Background() + var calls []string + err := st.InTx(ctx, func(tx *store.Tx) error { + tx.AfterCommit("k", func() { calls = append(calls, "k1") }) + tx.AfterCommit("k", func() { calls = append(calls, "k2") }) + tx.AfterCommit("", func() { calls = append(calls, "a") }) + if err := tx.Nested(ctx, func(n *store.Tx) error { + n.AfterCommit("", func() { calls = append(calls, "nested") }) + return nil + }); err != nil { + return err + } + if len(calls) != 0 { + t.Error("hook ran before commit") + } + return nil + }) + if err != nil || len(calls) != 3 || calls[0] != "k1" || calls[2] != "nested" { + t.Fatalf("calls %v %v", calls, err) + } + calls = nil + _ = st.InTx(ctx, func(tx *store.Tx) error { + tx.AfterCommit("", func() { calls = append(calls, "x") }) + return errors.New("rollback") + }) + if len(calls) != 0 { + t.Errorf("hook ran after rollback: %v", calls) + } + st.Q().AfterCommit("", func() { calls = append(calls, "now") }) + if len(calls) != 1 { + t.Errorf("hook without transaction: %v", calls) + } +} + +func TestConfigChangesSince(t *testing.T) { + st := storetest.New(t) + ctx := context.Background() + q := st.Q() + add := func(action, objType, objID string) { + t.Helper() + e := &store.AuditEntry{ActorName: "admin", Action: action, ObjectType: objType, ObjectName: action, ObjectID: &objID} + if err := q.InsertAudit(ctx, e); err != nil { + t.Fatal(err) + } + } + f := store.ChangeFilter{ObjectTypes: []string{"network", "address"}, Actions: []string{"settings.generator"}, Limit: 10} + add("network.create", "network", "n1") + add("user.create", "user", "u1") + add("settings.generator", "settings", "generator") + if got, err := q.ConfigChangesSince(ctx, f); err != nil || len(got) != 2 || got[0].Action != "settings.generator" { + t.Fatalf("no revision: %+v %v", got, err) + } + add("revision.create", "revision", "1") + add("address.create", "address", "a1") + add("revision.deploy", "revision", "1") + f.AfterRevision = 1 + if got, err := q.ConfigChangesSince(ctx, f); err != nil || len(got) != 1 || got[0].Action != "address.create" { + t.Fatalf("after revision 1: %+v %v", got, err) + } + add("revision.rollback", "revision", "2") + f.AfterRevision = 2 + if got, _ := q.ConfigChangesSince(ctx, f); len(got) != 0 { + t.Errorf("after rollback: %+v", got) + } + // Without an audit entry of the revision, the time bounds the range. + f.AfterRevision, f.Since = 9, time.Now().Add(time.Hour) + if got, _ := q.ConfigChangesSince(ctx, f); len(got) != 0 { + t.Errorf("since the future: %+v", got) + } + f.Limit = 1 + f.Since = time.Time{} + if got, _ := q.ConfigChangesSince(ctx, f); len(got) != 1 || got[0].Action != "address.create" { + t.Errorf("limit: %+v", got) + } +} diff --git a/server/internal/store/revisions.go b/server/internal/store/revisions.go index 0ecb50d..e41db47 100644 --- a/server/internal/store/revisions.go +++ b/server/internal/store/revisions.go @@ -5,9 +5,7 @@ package store import ( "context" "encoding/json" - "errors" "fmt" - "strings" "time" "github.com/jackc/pgx/v5" @@ -31,6 +29,10 @@ type Revision struct { DNSFiles int DHCPFiles int WarningCount int + // Source is the generator input (JSON object) the revision was rendered + // from. It is written by InsertRevision and read only by + // RevisionSource; nil for revisions created before M4. + Source json.RawMessage } // RevisionFile is one artifact of a revision. @@ -71,6 +73,19 @@ func (t *Tx) getRevision(ctx context.Context, cond string, args ...any) (*Revisi return &r, nil } +// RevisionSource returns the generator input of a revision: ErrNotFound for +// an unknown revision, nil for one created before M4. +func (t *Tx) RevisionSource(ctx context.Context, number int64) (json.RawMessage, error) { + var src []byte + if err := t.q.QueryRow(ctx, `SELECT source FROM config_revisions WHERE number = $1`, number).Scan(&src); err != nil { + return nil, mapErr(err) + } + if src == nil { + return nil, nil + } + return src, nil +} + // InsertRevision stores a revision with its files (in the given order per // role). CreatedAt is taken from r. func (t *Tx) InsertRevision(ctx context.Context, r *Revision, files []RevisionFile) error { @@ -78,8 +93,9 @@ func (t *Tx) InsertRevision(ctx context.Context, r *Revision, files []RevisionFi if len(warnings) == 0 { warnings = json.RawMessage("[]") } - if _, err := t.q.Exec(ctx, `INSERT INTO config_revisions (number, created_at, created_by, created_by_name, comment, warnings) - VALUES ($1, $2, $3, $4, $5, $6)`, r.Number, r.CreatedAt, r.CreatedBy, r.CreatedByName, r.Comment, []byte(warnings)); err != nil { + if _, err := t.q.Exec(ctx, `INSERT INTO config_revisions (number, created_at, created_by, created_by_name, comment, warnings, source) + VALUES ($1, $2, $3, $4, $5, $6, $7)`, r.Number, r.CreatedAt, r.CreatedBy, r.CreatedByName, r.Comment, []byte(warnings), + nullJSON(r.Source)); err != nil { return mapErr(err) } b := &pgx.Batch{} @@ -150,125 +166,3 @@ func (t *Tx) ListRevisions(ctx context.Context, limit int, before int64) ([]*Rev } return out, mapErr(rows.Err()) } - -// Deployment is a row of deployments; AgentHostname comes from the agent. -type Deployment struct { - ID int64 - AgentID string - AgentHostname string - CorrelationID string - Revision int64 - Trigger string - Outcome string - Message string - ActiveRevision int64 - StartedAt time.Time - FinishedAt time.Time - ReportedAt time.Time -} - -const deploymentColumns = `d.id, d.agent_id, a.hostname, d.correlation_id, d.revision, d.trigger_kind, d.outcome, d.message, - d.active_revision, d.started_at, d.finished_at, d.reported_at` - -func scanDeployment(row pgx.Row) (*Deployment, error) { - var d Deployment - if err := row.Scan(&d.ID, &d.AgentID, &d.AgentHostname, &d.CorrelationID, &d.Revision, &d.Trigger, &d.Outcome, - &d.Message, &d.ActiveRevision, &d.StartedAt, &d.FinishedAt, &d.ReportedAt); err != nil { - return nil, mapErr(err) - } - return &d, nil -} - -// InsertDeployment stores a deployment report; ID and ReportedAt are set. -// A report with the same agent, revision and start time exists already if -// inserted is false (resent after an agent crash). -func (t *Tx) InsertDeployment(ctx context.Context, d *Deployment) (inserted bool, err error) { - err = mapErr(t.q.QueryRow(ctx, ` - INSERT INTO deployments (agent_id, correlation_id, revision, trigger_kind, outcome, message, active_revision, - started_at, finished_at) - VALUES ($1, $2, $3, $4::deployment_trigger, $5::deployment_outcome, $6, $7, $8, $9) - ON CONFLICT ON CONSTRAINT deployments_report_unique DO NOTHING - RETURNING id, reported_at`, - d.AgentID, d.CorrelationID, d.Revision, d.Trigger, d.Outcome, d.Message, d.ActiveRevision, - d.StartedAt, d.FinishedAt).Scan(&d.ID, &d.ReportedAt)) - if errors.Is(err, ErrNotFound) { - return false, nil - } - return err == nil, err -} - -// DeploymentFilter selects deployments. Zero values mean "no restriction". -type DeploymentFilter struct { - AgentID string - Revision int64 - BeforeID int64 - Limit int -} - -// ListDeployments returns deployments newest first. -func (t *Tx) ListDeployments(ctx context.Context, f DeploymentFilter) ([]*Deployment, error) { - var where []string - var args []any - add := func(cond string, v any) { - args = append(args, v) - where = append(where, fmt.Sprintf(cond, len(args))) - } - if f.AgentID != "" { - add("d.agent_id = $%d", f.AgentID) - } - if f.Revision > 0 { - add("d.revision = $%d", f.Revision) - } - if f.BeforeID > 0 { - add("d.id < $%d", f.BeforeID) - } - limit := f.Limit - if limit <= 0 { - limit = 100 - } - sql := `SELECT ` + deploymentColumns + ` FROM deployments d JOIN agents a ON a.id = d.agent_id` - if len(where) > 0 { - sql += " WHERE " + strings.Join(where, " AND ") - } - sql += fmt.Sprintf(" ORDER BY d.id DESC LIMIT %d", limit) - rows, err := t.q.Query(ctx, sql, args...) - if err != nil { - return nil, mapErr(err) - } - defer rows.Close() - var out []*Deployment - for rows.Next() { - d, err := scanDeployment(rows) - if err != nil { - return nil, err - } - out = append(out, d) - } - return out, mapErr(rows.Err()) -} - -// LatestDeployments returns the newest deployment of every agent that has -// one, keyed by agent ID. -func (t *Tx) LatestDeployments(ctx context.Context) (map[string]*Deployment, error) { - rows, err := t.q.Query(ctx, `SELECT DISTINCT ON (d.agent_id) `+deploymentColumns+` - FROM deployments d JOIN agents a ON a.id = d.agent_id ORDER BY d.agent_id, d.id DESC`) - if err != nil { - return nil, mapErr(err) - } - defer rows.Close() - out := map[string]*Deployment{} - for rows.Next() { - d, err := scanDeployment(rows) - if err != nil { - return nil, err - } - out[d.AgentID] = d - } - return out, mapErr(rows.Err()) -} - -// LatestDeployment returns the newest deployment of an agent or ErrNotFound. -func (t *Tx) LatestDeployment(ctx context.Context, agentID string) (*Deployment, error) { - return scanDeployment(t.q.QueryRow(ctx, `SELECT `+deploymentColumns+` - FROM deployments d JOIN agents a ON a.id = d.agent_id WHERE d.agent_id = $1 ORDER BY d.id DESC LIMIT 1`, agentID)) -} diff --git a/server/internal/store/store.go b/server/internal/store/store.go index 1a8eb46..28e9298 100644 --- a/server/internal/store/store.go +++ b/server/internal/store/store.go @@ -148,14 +148,47 @@ func (t *Tx) execBatch(ctx context.Context, b *pgx.Batch) error { // Tx is a repository bound to a transaction. type Tx struct { q querier + // hooks run after a successful commit (nil without a transaction). + hooks *txHooks +} + +type txHooks struct { + fns []func() + keys map[string]bool } // Q returns repositories bound to the pool (no transaction). func (s *Store) Q() *Tx { return &Tx{q: s.pool} } -// InTx runs fn in a transaction that is committed if fn returns nil. +// InTx runs fn in a transaction that is committed if fn returns nil. The +// functions registered with AfterCommit run after the commit. func (s *Store) InTx(ctx context.Context, fn func(tx *Tx) error) error { - return pgx.BeginFunc(ctx, s.pool, func(t pgx.Tx) error { return fn(&Tx{q: t}) }) + h := &txHooks{keys: map[string]bool{}} + if err := pgx.BeginFunc(ctx, s.pool, func(t pgx.Tx) error { return fn(&Tx{q: t, hooks: h}) }); err != nil { + return err + } + for _, f := range h.fns { + f() + } + return nil +} + +// AfterCommit registers fn to run once the transaction has been committed, +// so that notifications never announce changes that are rolled back. A +// non-empty key registers fn only once per transaction (savepoints +// included). Without a transaction fn runs immediately. +func (t *Tx) AfterCommit(key string, fn func()) { + if t.hooks == nil { + fn() + return + } + if key != "" { + if t.hooks.keys[key] { + return + } + t.hooks.keys[key] = true + } + t.hooks.fns = append(t.hooks.fns, fn) } // Nested runs fn inside a savepoint when t is a transaction. If fn fails, @@ -166,5 +199,5 @@ func (t *Tx) Nested(ctx context.Context, fn func(tx *Tx) error) error { if !ok { return fn(t) } - return pgx.BeginFunc(ctx, outer, func(sp pgx.Tx) error { return fn(&Tx{q: sp}) }) + return pgx.BeginFunc(ctx, outer, func(sp pgx.Tx) error { return fn(&Tx{q: sp, hooks: t.hooks}) }) }