feat(agent): keep samples and events on disk until the control plane accepts them

agent_test.go

@@ -5,6 +5,9 @@ package main
 
 import (
 	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
 	"os"
 	"os/exec"
 	"strings"
@@ -89,6 +92,37 @@ func startAgent(t *testing.T, environ []string) (*exec.Cmd, *lockedBuffer) {
 	return cmd, stderr
 }
 
+func TestAgentSendsAgentStartedToTheControlPlane(t *testing.T) {
+	type request struct {
+		path, auth string
+		body       map[string][]map[string]any
+	}
+	got := make(chan request, 10)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		req := request{path: r.URL.Path, auth: r.Header.Get("Authorization")}
+		_ = json.NewDecoder(r.Body).Decode(&req.body)
+		got <- req
+		_, _ = w.Write([]byte(`{"accepted": 1}`))
+	}))
+	defer srv.Close()
+
+	environ := append(agentEnv(t), "FLY_AGENT_URL="+srv.URL)
+	cmd, stderr := startAgent(t, environ)
+	defer func() { _ = cmd.Process.Signal(syscall.SIGTERM); _ = cmd.Wait() }()
+
+	select {
+	case req := <-got:
+		if req.path != "/agent/v1/events" || req.auth != "Bearer "+testToken {
+			t.Errorf("request to %s with Authorization %q, want /agent/v1/events with the token", req.path, req.auth)
+		}
+		if events := req.body["events"]; len(events) != 1 || events[0]["name"] != "agent.started" {
+			t.Errorf("events = %v, want agent.started", events)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatalf("the control plane got no request within 10s. stderr:\n%s", stderr)
+	}
+}
+
 func TestAgentConfigErrors(t *testing.T) {
 	tests := []struct {
 		name    string

cmd/agent.go

@@ -33,7 +33,7 @@ FLY_AGENT_SERVER_ID and STATE_DIRECTORY from the environment.`,
 		ctx, stop := signal.NotifyContext(cmd.Context(), syscall.SIGTERM, os.Interrupt)
 		defer stop()
 
-		return agent.Run(ctx, cfg, slog.New(slog.NewTextHandler(os.Stderr, nil)))
+		return agent.Run(ctx, cfg, slog.New(slog.NewTextHandler(os.Stderr, nil)), nil)
 	},
 }
 

go.mod

@@ -6,6 +6,7 @@ toolchain go1.27.1
 
 require (
 	github.com/fatih/color v1.19.0
+	github.com/oklog/ulid/v2 v2.1.2
 	github.com/spf13/cobra v1.10.2
 	golang.org/x/mod v0.41.0
 	gopkg.in/yaml.v2 v2.4.0

go.sum

@@ -7,6 +7,9 @@ github.com/mattn/go-colorable v0.1.15 h1:+u9SLTRGnXv73cEsnsmoZBom+dMU88B2M0aDcWy
 github.com/mattn/go-colorable v0.1.15/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8=
 github.com/mattn/go-isatty v0.0.24 h1:tGZZoVgT/KiqK1c8ocVLeDS8BSWMRd47J3Lbz7vsReI=
 github.com/mattn/go-isatty v0.0.24/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI3HqDio6A=
+github.com/oklog/ulid/v2 v2.1.2 h1:IEclFb9JNvzYA6MW2SCxbLzcHTVsfqm3PrqGQJH5zec=
+github.com/oklog/ulid/v2 v2.1.2/go.mod h1:rcEKHmBBKfef9DhnvX7y1HZBYxjXb0cP5ExxNsTT1QQ=
+github.com/pborman/getopt v0.0.0-20170112200414-7148bc3a4c30/go.mod h1:85jBQOZwpVEaDAr341tbn15RS4fCAsIst0qp7i8ex1o=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/spf13/cobra v1.10.2 h1:DMTTonx5m65Ic0GOoRY2c16WCbHxOOw6xxezuLaBpcU=
 github.com/spf13/cobra v1.10.2/go.mod h1:7C1pvHqHw5A4vrJfjNwvOdzYu0Gml16OCs2GRiTUUS4=

internal/agent/agent.go

@@ -8,8 +8,10 @@ import (
 	"path/filepath"
 	"time"
 
+	"github.com/flywp/server-cli/internal/agent/wire"
 	"github.com/flywp/server-cli/internal/statefile"
 	"github.com/flywp/server-cli/internal/version"
+	"github.com/oklog/ulid/v2"
 )
 
 // The report interval is the number of samples in one report. The control
@@ -19,14 +21,31 @@ const (
 	maxReportInterval = 10
 )
 
+// ControlPlane is the part of the control plane that the agent uses.
+type ControlPlane interface {
+	PostMetrics(ctx context.Context, req *wire.MetricsRequest) (*wire.MetricsReply, error)
+	PostEvents(ctx context.Context, req *wire.EventsRequest) (*wire.EventsReply, error)
+}
+
+// Collector measures the server.
+type Collector interface {
+	// Sample measures the minute that ends at now.
+	Sample(now time.Time) (wire.Sample, error)
+	// Status describes the server now.
+	Status(ctx context.Context) wire.Status
+}
+
 // state is the part of the agent state that is not a queue.
 type state struct {
 	ReportInterval int `json:"report_interval"`
 }
 
 type agent struct {
-	cfg Config
-	log *slog.Logger
+	cfg       Config
+	log       *slog.Logger
+	cp        ControlPlane
+	collector Collector
+	outbox    *outbox
 
 	// interval is the report interval, and pending is the number of samples
 	// since the last report.
@@ -35,11 +54,21 @@ type agent struct {
 
 	// last is the time of the last tick.
 	last time.Time
+
+	// The waits of the events and the metrics requests after a failure, and
+	// whether the last report sent all samples.
+	eventsWait  backoff
+	metricsWait backoff
+	samplesSent bool
 }
 
 // Run runs the agent until ctx is done. Only one agent can run with the same
-// state directory.
-func Run(ctx context.Context, cfg Config, log *slog.Logger) error {
+// state directory. A nil collector takes no samples.
+func Run(ctx context.Context, cfg Config, log *slog.Logger, collector Collector) error {
+	return run(ctx, cfg, log, NewClient(cfg, nil), collector)
+}
+
+func run(ctx context.Context, cfg Config, log *slog.Logger, cp ControlPlane, collector Collector) error {
 	unlock, err := lock(cfg.StateDir)
 	if err != nil {
 		return err
@@ -49,8 +78,21 @@ func Run(ctx context.Context, cfg Config, log *slog.Logger) error {
 	// A crash during a write can leave a temporary file.
 	statefile.RemoveTemp(cfg.StateDir)
 
-	a := &agent{cfg: cfg, log: log, interval: loadInterval(cfg.StateDir, log)}
+	a := &agent{
+		cfg:       cfg,
+		log:       log,
+		cp:        cp,
+		collector: collector,
+		outbox:    loadOutbox(cfg.StateDir, log),
+		interval:  loadInterval(cfg.StateDir, log),
+	}
 	log.Info("agent started", "version", version.Version, "offset", cfg.Offset(), "report_interval", a.interval)
+
+	// Send the events at once, not at the next tick: after an update or a
+	// restart, they hold the result of the command.
+	a.addEvent(wire.EventAgentStarted, "", &wire.EventData{Version: version.Version})
+	a.send(ctx, false)
+
 	a.loop(ctx)
 	log.Info("agent stopped")
 
@@ -91,16 +133,54 @@ func nextAfter(now, last time.Time, offset time.Duration) time.Time {
 func (a *agent) tick(ctx context.Context, now time.Time) {
 	a.log.Debug("tick", "at", now)
 
+	if a.collector != nil {
+		s, err := a.collector.Sample(now)
+		if err != nil {
+			a.log.Warn("skipping the sample of this minute", "error", err)
+		} else {
+			s.RecordedAt = now
+			a.outbox.addSample(cleanSample(s))
+		}
+	}
+
 	a.pending++
-	if a.pending >= a.interval {
+	if a.pending < a.interval {
+		return
+	}
+
+	a.log.Debug("report")
+	a.send(ctx, true)
+
+	// Samples that could not go are tried again at the next tick, when their
+	// wait allows it, not only after the next full interval.
+	if a.samplesSent {
 		a.pending = 0
-		a.report(ctx)
 	}
 }
 
-// report sends the queued data to the control plane.
-func (a *agent) report(_ context.Context) {
-	a.log.Debug("report")
+// addEvent puts an event in the queue. Its ID is made now, so that each resend
+// has the same ID.
+func (a *agent) addEvent(name, commandID string, data *wire.EventData) {
+	a.outbox.addEvent(cleanEvent(wire.Event{
+		ID:        ulid.Make().String(),
+		Name:      name,
+		CommandID: commandID,
+		At:        time.Now().UTC(),
+		Data:      data,
+	}))
+}
+
+// setInterval applies the report interval of a reply.
+func (a *agent) setInterval(n int) {
+	if n < minReportInterval || n > maxReportInterval || n == a.interval {
+		return
+	}
+
+	a.log.Info("report interval changed", "from", a.interval, "to", n)
+	a.interval = n
+	if err := statefile.Write(filepath.Join(a.cfg.StateDir, "state.json"), state{ReportInterval: n}); err != nil {
+		a.log.Error("saving the report interval", "error", err)
+	}
 }
 
 // nextTick returns the first time after now that is offset past a full minute.

internal/agent/agent_test.go

@@ -30,6 +30,34 @@ func (r *recorder) Handle(_ context.Context, rec slog.Record) error {
 	return nil
 }
 
+// recordsOf returns the records with message msg.
+func (r *recorder) recordsOf(msg string) []slog.Record {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	var out []slog.Record
+	for _, rec := range r.records {
+		if rec.Message == msg {
+			out = append(out, rec)
+		}
+	}
+	return out
+}
+
+// attr returns the value of the attribute key of a record, as a string, or
+// nil.
+func attr(rec slog.Record, key string) any {
+	var v any
+	rec.Attrs(func(a slog.Attr) bool {
+		if a.Key == key {
+			v = a.Value.String()
+			return false
+		}
+		return true
+	})
+	return v
+}
+
 // times returns the times of the records with message msg.
 func (r *recorder) times(msg string) []time.Time {
 	r.mu.Lock()
@@ -95,7 +123,7 @@ func TestLoopTicksAtTheOffsetAndReportsEachInterval(t *testing.T) {
 		ctx, cancel := context.WithCancel(context.Background())
 		done := make(chan error)
 		go func() {
-			done <- Run(ctx, Config{ServerID: 17, StateDir: dir}, slog.New(rec))
+			done <- run(ctx, Config{ServerID: 17, StateDir: dir}, slog.New(rec), &fakeCP{}, nil)
 		}()
 
 		// The bubble starts at 2000-01-01 00:00:00 UTC. Let 4 minutes pass.
@@ -134,7 +162,7 @@ func TestRunRefusesASecondAgent(t *testing.T) {
 	}
 	defer unlock()
 
-	if err := Run(context.Background(), Config{StateDir: dir}, slog.New(&recorder{})); err == nil {
+	if err := run(context.Background(), Config{StateDir: dir}, slog.New(&recorder{}), &fakeCP{}, nil); err == nil {
 		t.Fatal("Run() = nil, want an error while a different agent holds the lock")
 	}
 }

internal/agent/clean.go

@@ -0,0 +1,88 @@
+package agent
+
+import (
+	"math"
+	"time"
+	"unicode/utf8"
+
+	"github.com/flywp/server-cli/internal/agent/wire"
+	"github.com/oklog/ulid/v2"
+)
+
+// The value ranges of the contract (section 4, "Value ranges", and section 6).
+// A value outside its range makes the control plane refuse the whole request
+// with a 400, and the agent then drops up to 240 samples. So the agent keeps
+// each value in its range before the value goes into a queue.
+const (
+	maxLoad          = 999999.99
+	maxVersionLen    = 32
+	maxStatusTextLen = 255
+	maxArchLen       = 16
+	maxEventNameLen  = 64
+	maxErrorLen      = 2000
+)
+
+func cleanSample(s wire.Sample) wire.Sample {
+	s.RecordedAt = s.RecordedAt.UTC().Truncate(time.Second)
+	s.CPUPercent = clamp(s.CPUPercent, 0, 100)
+	s.Load1 = clamp(s.Load1, 0, maxLoad)
+	for _, v := range []*uint64{
+		&s.MemoryUsedBytes, &s.MemoryTotalBytes, &s.SwapUsedBytes, &s.SwapTotalBytes,
+		&s.DiskUsedBytes, &s.DiskTotalBytes, &s.NetInBytes, &s.NetOutBytes,
+	} {
+		*v = clampInt(*v)
+	}
+	return s
+}
+
+func cleanStatus(s wire.Status) wire.Status {
+	s.OS = truncate(s.OS, maxStatusTextLen)
+	s.Kernel = truncate(s.Kernel, maxStatusTextLen)
+	s.Arch = truncate(s.Arch, maxArchLen)
+	s.UpdatesTotal = clampInt(s.UpdatesTotal)
+	s.UpdatesSecurity = clampInt(s.UpdatesSecurity)
+	s.UptimeSeconds = clampInt(s.UptimeSeconds)
+	return s
+}
+
+func cleanEvent(e wire.Event) wire.Event {
+	// A command id that is not a ULID makes a 400, and a 400 drops all the
+	// events of the request. Without the id, the event changes nothing.
+	if e.CommandID != "" {
+		if _, err := ulid.ParseStrict(e.CommandID); err != nil {
+			e.CommandID = ""
+		}
+	}
+	e.Name = truncate(e.Name, maxEventNameLen)
+	if e.Data != nil {
+		d := *e.Data
+		d.Version = truncate(d.Version, maxVersionLen)
+		d.Error = truncate(d.Error, maxErrorLen)
+		e.Data = &d
+	}
+	return e
+}
+
+// clampInt keeps v in the range of a signed 64-bit integer: the control plane
+// (PHP) cannot hold a larger integer, and fails the request with a 500. The
+// agent would then send the same sample again for 24 hours.
+func clampInt(v uint64) uint64 {
+	return min(v, math.MaxInt64)
+}
+
+// clamp keeps v between lo and hi. NaN becomes lo.
+func clamp(v, lo, hi float64) float64 {
+	if math.IsNaN(v) {
+		return lo
+	}
+	return min(max(v, lo), hi)
+}
+
+// truncate cuts s to at most n characters. The control plane counts
+// characters, not bytes.
+func truncate(s string, n int) string {
+	if utf8.RuneCountInString(s) <= n {
+		return s
+	}
+	return string([]rune(s)[:n])
+}