feat(agent): add fly agent run, the base of the monitoring agent

agent_test.go

@@ -0,0 +1,156 @@
+package main
+
+// End-to-end tests of "fly agent run": the configuration errors, the lock and
+// a clean stop. The loop itself is tested in internal/agent with a fake clock.
+
+import (
+	"bytes"
+	"os"
+	"os/exec"
+	"strings"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+)
+
+const testToken = "flyagt_0123456789abcdefghijABCDEFGHIJ"
+
+// agentEnv is the environment of a valid agent with its own state directory.
+// PATH holds no docker command: the agent must not need Docker.
+func agentEnv(t *testing.T) []string {
+	t.Helper()
+
+	if os.Geteuid() == 0 {
+		t.Skip("fly refuses to run as root")
+	}
+
+	return []string{
+		"PATH=" + t.TempDir(),
+		"HOME=" + t.TempDir(),
+		"FLY_AGENT_URL=http://127.0.0.1:9",
+		"FLY_AGENT_TOKEN=" + testToken,
+		"FLY_AGENT_SERVER_ID=17",
+		"STATE_DIRECTORY=" + t.TempDir(),
+	}
+}
+
+// without returns environ without the variable key.
+func without(environ []string, key string) []string {
+	var out []string
+	for _, kv := range environ {
+		if !strings.HasPrefix(kv, key+"=") {
+			out = append(out, kv)
+		}
+	}
+	return out
+}
+
+// lockedBuffer is a bytes.Buffer that a child process and the test can use
+// at the same time.
+type lockedBuffer struct {
+	mu  sync.Mutex
+	buf bytes.Buffer
+}
+
+func (b *lockedBuffer) Write(p []byte) (int, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.Write(p)
+}
+
+func (b *lockedBuffer) String() string {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.String()
+}
+
+// startAgent starts "fly agent run" and waits until it logs that it started.
+func startAgent(t *testing.T, environ []string) (*exec.Cmd, *lockedBuffer) {
+	t.Helper()
+
+	cmd := exec.Command(flyBin, "agent", "run")
+	cmd.Env = environ
+	stderr := &lockedBuffer{}
+	cmd.Stderr = stderr
+	if err := cmd.Start(); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = cmd.Process.Kill() })
+
+	deadline := time.Now().Add(10 * time.Second)
+	for !strings.Contains(stderr.String(), "agent started") {
+		if time.Now().After(deadline) {
+			t.Fatalf("the agent did not start within 10s. stderr:\n%s", stderr)
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+
+	return cmd, stderr
+}
+
+func TestAgentConfigErrors(t *testing.T) {
+	tests := []struct {
+		name    string
+		environ func([]string) []string
+		want    string
+	}{
+		{"no token", func(e []string) []string { return without(e, "FLY_AGENT_TOKEN") }, "FLY_AGENT_TOKEN is not set"},
+		{"no state directory", func(e []string) []string { return without(e, "STATE_DIRECTORY") }, "STATE_DIRECTORY is not set"},
+		{"plain http", func(e []string) []string { return append(e, "FLY_AGENT_URL=http://example.com") }, "must be an https URL"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			res := runFly(t, t.TempDir(), tt.environ(agentEnv(t)), "agent", "run")
+
+			if res.code != 1 {
+				t.Errorf("exit code = %d, want 1", res.code)
+			}
+			if !strings.Contains(res.stderr, tt.want) {
+				t.Errorf("stderr = %q, want it to contain %q", res.stderr, tt.want)
+			}
+		})
+	}
+}
+
+func TestAgentRejectsArguments(t *testing.T) {
+	res := runFly(t, t.TempDir(), agentEnv(t), "agent", "run", "extra")
+	if res.code != 1 || !strings.Contains(res.stderr, "unknown command") && !strings.Contains(res.stderr, "accepts 0 arg") {
+		t.Errorf("fly agent run extra: exit %d, stderr %q; want a usage error", res.code, res.stderr)
+	}
+}
+
+func TestAgentRunsWithoutDockerAndStopsOnSIGTERM(t *testing.T) {
+	environ := agentEnv(t)
+	cmd, stderr := startAgent(t, environ)
+
+	// Only one agent can use the state directory.
+	second := runFly(t, t.TempDir(), environ, "agent", "run")
+	if second.code != 1 || !strings.Contains(second.stderr, "a different agent is running") {
+		t.Errorf("second agent: exit %d, stderr %q; want exit 1 and a lock error", second.code, second.stderr)
+	}
+
+	if err := cmd.Process.Signal(syscall.SIGTERM); err != nil {
+		t.Fatal(err)
+	}
+
+	done := make(chan error, 1)
+	go func() { done <- cmd.Wait() }()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Errorf("agent exit after SIGTERM: %v, want exit status 0. stderr:\n%s", err, stderr)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("the agent did not stop within 10s after SIGTERM")
+	}
+
+	out := stderr.String()
+	if !strings.Contains(out, "agent stopped") {
+		t.Errorf("stderr = %q, want an \"agent stopped\" line", out)
+	}
+	if strings.Contains(out, testToken) {
+		t.Error("the agent log contains the token")
+	}
+}

cmd/agent.go

@@ -0,0 +1,43 @@
+package cmd
+
+import (
+	"log/slog"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/flywp/server-cli/internal/agent"
+	"github.com/spf13/cobra"
+)
+
+var agentCmd = &cobra.Command{
+	Use:   "agent",
+	Short: "Run the FlyWP monitoring agent",
+}
+
+// agentRunCmd does not need Docker: the agent must also report when Docker
+// is down.
+var agentRunCmd = &cobra.Command{
+	Use:   "run",
+	Short: "Run the monitoring agent until it is stopped",
+	Long: `Run the FlyWP monitoring agent until it is stopped. systemd starts this
+command (fly-agent.service). The agent reads FLY_AGENT_URL, FLY_AGENT_TOKEN,
+FLY_AGENT_SERVER_ID and STATE_DIRECTORY from the environment.`,
+	Args: cobra.NoArgs,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		cfg, err := agent.ConfigFromEnv(os.Getenv)
+		if err != nil {
+			return err
+		}
+
+		ctx, stop := signal.NotifyContext(cmd.Context(), syscall.SIGTERM, os.Interrupt)
+		defer stop()
+
+		return agent.Run(ctx, cfg, slog.New(slog.NewTextHandler(os.Stderr, nil)))
+	},
+}
+
+func init() {
+	agentCmd.AddCommand(agentRunCmd)
+	rootCmd.AddCommand(agentCmd)
+}

internal/agent/agent.go

@@ -0,0 +1,135 @@
+package agent
+
+import (
+	"context"
+	"errors"
+	"io/fs"
+	"log/slog"
+	"path/filepath"
+	"time"
+
+	"github.com/flywp/server-cli/internal/statefile"
+	"github.com/flywp/server-cli/internal/version"
+)
+
+// The report interval is the number of samples in one report. The control
+// plane sets it in each reply.
+const (
+	minReportInterval = 1
+	maxReportInterval = 10
+)
+
+// 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
+
+	// interval is the report interval, and pending is the number of samples
+	// since the last report.
+	interval int
+	pending  int
+
+	// last is the time of the last tick.
+	last time.Time
+}
+
+// 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 {
+	unlock, err := lock(cfg.StateDir)
+	if err != nil {
+		return err
+	}
+	defer unlock()
+
+	// 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)}
+	log.Info("agent started", "version", version.Version, "offset", cfg.Offset(), "report_interval", a.interval)
+	a.loop(ctx)
+	log.Info("agent stopped")
+
+	return nil
+}
+
+// loop calls tick at the offset second of each minute until ctx is done.
+func (a *agent) loop(ctx context.Context) {
+	for {
+		next := nextAfter(time.Now(), a.last, a.cfg.Offset())
+		timer := time.NewTimer(time.Until(next))
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return
+		case <-timer.C:
+			a.last = next
+			a.tick(ctx, next)
+		}
+	}
+}
+
+// nextAfter returns the next tick after now, and never a tick at or before
+// last. The timer runs on the monotonic clock, but the tick times come from
+// the wall clock: when the wall clock steps back, the timer fires before the
+// tick time, and without last the same tick would run two times.
+func nextAfter(now, last time.Time, offset time.Duration) time.Time {
+	next := nextTick(now, offset)
+	if !last.IsZero() && !next.After(last) {
+		next = nextTick(last, offset)
+	}
+
+	return next
+}
+
+// tick does the work of one minute: it takes a sample and, after each
+// interval samples, sends a report.
+func (a *agent) tick(ctx context.Context, now time.Time) {
+	a.log.Debug("tick", "at", now)
+
+	a.pending++
+	if a.pending >= a.interval {
+		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")
+}
+
+// nextTick returns the first time after now that is offset past a full minute.
+// It comes from the wall clock, so a slow tick moves to the next minute and
+// never runs two times in one minute.
+func nextTick(now time.Time, offset time.Duration) time.Time {
+	t := now.Truncate(time.Minute).Add(offset)
+	if !t.After(now) {
+		t = t.Add(time.Minute)
+	}
+
+	return t
+}
+
+// loadInterval returns the report interval of the last reply, or 1.
+func loadInterval(dir string, log *slog.Logger) int {
+	var s state
+	err := statefile.Read(filepath.Join(dir, "state.json"), &s)
+	switch {
+	case errors.Is(err, fs.ErrNotExist):
+		return minReportInterval
+	case err != nil:
+		log.Warn("ignoring the saved state", "error", err)
+		return minReportInterval
+	}
+
+	return clampInterval(s.ReportInterval)
+}
+
+func clampInterval(n int) int {
+	return min(max(n, minReportInterval), maxReportInterval)
+}