amadan

carlos / platform Public

host: arm operator paging at agent startup; park test keeps to its own contract (fix round 1)

Finding 1: runAgent builds the alert sinks from the sink env and arms
paging (armAgentAlerts) before armMovesPass and before any loop that runs
a moves pass. Arming no longer waits on the metrics store: metricsLoop is
handed the same instance and neither builds sinks nor arms paging. With
no sink configured it still logs once and pages nothing.
TestStuckMarkBeforeMetricsReachesTheSink drives a stuck mark through
PageOperator to a real webhook with the metrics loop never started and
with its store failing; TestAgentArmsPagingBeforeMovesConsumers pins the
order in runAgent.

Finding 2: the dual-receipt "findable by op" rule lives in the host test
only; the park test checks park.Local's contract, .journal.json naming
the op under the engine's layout.

Finding 3: pageStuck's comment calls delivery best effort: duplicates on
a replay, nothing on a crash between the mark and the send.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 0db2980ae159c2c563d86dc9b0d9e7588660c2eb parent d1e2726
7 files changed, +194 −50
  • internal/host/agent.go +8 −1
  • internal/host/agentalerts_test.go +161 −0
  • internal/host/alertsink.go +10 −0
  • internal/host/metrics_loop.go +5 −4
  • internal/host/movesretire_byop_test.go +2 −1
  • internal/host/movesrollback.go +4 −3
  • internal/park/local_linux_test.go +4 −41
diff --git a/internal/host/agent.go b/internal/host/agent.go
index aa9ddafa5..39258dedb 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -246,6 +246,13 @@ func runAgent(args []string, buildClaims ClaimFactory) {
cgm = cgroup.Setup(log.Printf)
}
+ // Alert sinks and the operator paging hook (moves spec §13), built from
+ // the sink env alone and armed here, before armMovesPass and before any
+ // loop that runs a moves pass starts, so a stuck mark never finds paging
+ // unarmed. They do not wait on the metrics store: the metrics loop is
+ // handed this same instance, and a box whose store fails still pages.
+ alerts := armAgentAlerts(agentCtx, os.Getenv, &processAlertCounters, newSNSClient)
+
// Log shipping (`carlos logs`): one sink for every producer in this
// process, armed only when there is somewhere to ship to — a nil sink
// leaves every producer exactly as before (no tees, no pipes, no
@@ -596,7 +603,7 @@ func runAgent(args []string, buildClaims ClaimFactory) {
go steeringLoop()
// The sweep reads the registry only through this closure — the stamped
// bespoke units to sample beside the carlos- ones, re-read each tick.
- go metricsLoop(cgm.Root(), func() []string { return stampedUnitNames(reg) })
+ go metricsLoop(cgm.Root(), func() []string { return stampedUnitNames(reg) }, alerts)
go configSyncLoop(reg, svc, dd, claims.Writes)
go configGenLoop(reg, svc, dd, claims.Writes)
go socketAdvisoryLoop(reg, svc)
diff --git a/internal/host/agentalerts_test.go b/internal/host/agentalerts_test.go
new file mode 100644
index 000000000..149f3a49b
--- /dev/null
+++ b/internal/host/agentalerts_test.go
@@ -0,0 +1,161 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "go/ast"
+ "go/parser"
+ "go/token"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "path/filepath"
+ "sync"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// TestStuckMarkBeforeMetricsReachesTheSink (fix round 1, finding 1): the
+// agent arms paging at startup from the sink env alone (armAgentAlerts), so
+// a stuck mark reaches a configured webhook whether or not the metrics loop
+// has started, and when its store fails to initialise. The revert's stuck
+// mark runs with the production hook, PageOperator.
+func TestStuckMarkBeforeMetricsReachesTheSink(t *testing.T) {
+ for _, metrics := range []string{"never started", "store init failed"} {
+ t.Run(metrics, func(t *testing.T) {
+ var mu sync.Mutex
+ var bodies [][]byte
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ b, _ := io.ReadAll(r.Body)
+ mu.Lock()
+ bodies = append(bodies, b)
+ mu.Unlock()
+ }))
+ defer srv.Close()
+ env := map[string]string{"CARLOS_ALERT_WEBHOOK": srv.URL}
+ sinks := armAgentAlerts(context.Background(), func(k string) string { return env[k] }, &alertCounters{}, (&fakeClientFactory{}).build)
+ t.Cleanup(func() { armPaging(nil) })
+
+ if metrics == "store init failed" {
+ // No bucket, and a fleet name that does not validate:
+ // openStatusStore fails and the loop returns at once.
+ t.Setenv("CARLOS_STATUS_BUCKET", "")
+ t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "")
+ t.Setenv("CARLOS_FLEET", "not a fleet!")
+ if _, _, err := openStatusStore(context.Background(), os.Getenv); err == nil {
+ t.Fatal("fixture: the metrics store opened")
+ }
+ metricsLoop(t.TempDir(), nil, sinks)
+ }
+
+ h := newRevertHarness(t)
+ h.fromStopped(t)
+ h.arm(t)
+ h.cfg.Page = PageOperator
+ if err := os.WriteFile(filepath.Join(h.cfg.DataDir, p1Host, "upload.txt"), []byte("changed while stopped"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ h.pass(t)
+ if m := h.move(t); m.State != provision.MoveStuck {
+ t.Fatalf("not stuck: %+v", m)
+ }
+ mu.Lock()
+ defer mu.Unlock()
+ if len(bodies) != 1 {
+ t.Fatalf("webhook received %d posts, want the one page", len(bodies))
+ }
+ var p struct{ Kind, Level, Subject, Body string }
+ if err := json.Unmarshal(bodies[0], &p); err != nil {
+ t.Fatal(err)
+ }
+ if p.Kind != "page" || p.Level != "incident" || p.Subject != "carlos move stuck: "+p1Host || p.Body != stuckPageBody("ie-box", "source", provision.StuckBaselineMismatch) {
+ t.Fatalf("the page: %+v", p)
+ }
+ })
+ }
+}
+
+// TestAgentArmsPagingBeforeMovesConsumers (fix round 1, finding 1): in
+// runAgent the alert sinks are built and paging armed once, before
+// armMovesPass and before any loop that runs a moves pass starts; the
+// metrics loop is handed that same instance and neither builds sinks nor
+// arms paging itself.
+func TestAgentArmsPagingBeforeMovesConsumers(t *testing.T) {
+ fset := token.NewFileSet()
+ parse := func(name string) *ast.File {
+ f, err := parser.ParseFile(fset, name, nil, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return f
+ }
+ body := func(f *ast.File, name string) *ast.BlockStmt {
+ for _, d := range f.Decls {
+ if fd, ok := d.(*ast.FuncDecl); ok && fd.Recv == nil && fd.Name.Name == name {
+ return fd.Body
+ }
+ }
+ t.Fatalf("no func %s", name)
+ return nil
+ }
+ callee := func(c *ast.CallExpr) string {
+ if id, ok := c.Fun.(*ast.Ident); ok {
+ return id.Name
+ }
+ return ""
+ }
+ // first call position of each name in runAgent, and how many times.
+ at := map[string]token.Pos{}
+ count := map[string]int{}
+ var armed string // the variable armAgentAlerts' result is assigned to
+ var metricsArg ast.Expr
+ ast.Inspect(body(parse("agent.go"), "runAgent"), func(n ast.Node) bool {
+ switch n := n.(type) {
+ case *ast.AssignStmt:
+ if len(n.Rhs) == 1 && len(n.Lhs) == 1 {
+ if c, ok := n.Rhs[0].(*ast.CallExpr); ok && callee(c) == "armAgentAlerts" {
+ if id, ok := n.Lhs[0].(*ast.Ident); ok {
+ armed = id.Name
+ }
+ }
+ }
+ case *ast.CallExpr:
+ name := callee(n)
+ if name == "" {
+ return true
+ }
+ count[name]++
+ if _, ok := at[name]; !ok {
+ at[name] = n.Pos()
+ }
+ if name == "metricsLoop" && len(n.Args) == 3 {
+ metricsArg = n.Args[2]
+ }
+ }
+ return true
+ })
+ if count["armAgentAlerts"] != 1 || armed == "" {
+ t.Fatalf("runAgent calls armAgentAlerts %d times, assigned to %q; want once, kept", count["armAgentAlerts"], armed)
+ }
+ for _, later := range []string{"armMovesPass", "adoptLoop", "fastConvergeLoop", "metricsLoop"} {
+ if at[later] == token.NoPos {
+ t.Fatalf("runAgent no longer calls %s; update this test", later)
+ }
+ if at["armAgentAlerts"] > at[later] {
+ t.Errorf("paging is armed after %s starts (%s before %s)", later, fset.Position(at[later]), fset.Position(at["armAgentAlerts"]))
+ }
+ }
+ if id, ok := metricsArg.(*ast.Ident); !ok || id.Name != armed {
+ t.Errorf("metricsLoop is not handed the armed sinks %q", armed)
+ }
+ ast.Inspect(body(parse("metrics_loop.go"), "metricsLoop"), func(n ast.Node) bool {
+ if c, ok := n.(*ast.CallExpr); ok {
+ if name := callee(c); name == "buildAlertSinks" || name == "armPaging" || name == "armAgentAlerts" {
+ t.Errorf("metricsLoop calls %s; it must reuse the armed sinks", name)
+ }
+ }
+ return true
+ })
+}
diff --git a/internal/host/alertsink.go b/internal/host/alertsink.go
index ea204da0e..2a1509634 100644
--- a/internal/host/alertsink.go
+++ b/internal/host/alertsink.go
@@ -128,6 +128,16 @@ func PageOperator(ctx context.Context, level, subject, body string) error {
return alerting.Page(ctx, level, subject, body)
}
+// armAgentAlerts builds the agent's alert sinks from the env and arms
+// paging on them (armPaging). runAgent calls it once at startup, before
+// anything that can run a moves pass, and hands the same instance to the
+// metrics loop, so arming never waits on the metrics store.
+func armAgentAlerts(ctx context.Context, getenv func(string) string, counts *alertCounters, newClient snsClientFactory) *alertSinks {
+ sinks := buildAlertSinks(ctx, getenv, counts, newClient)
+ armPaging(sinks)
+ return sinks
+}
+
// armPaging makes a the sinks PageOperator pages on. nil means none.
func armPaging(a *alertSinks) {
if a == nil {
diff --git a/internal/host/metrics_loop.go b/internal/host/metrics_loop.go
index 254b74868..2e0dad037 100644
--- a/internal/host/metrics_loop.go
+++ b/internal/host/metrics_loop.go
@@ -745,7 +745,11 @@ func BoxLabel() string { return metricsBoxLabel(os.Getenv, readMachineID()) }
// re-read once per tick so a freshly stamped route is picked up within a
// minute. It is a func, not the registry handle, so this loop keeps its "no
// state but a bucket and /proc" shape; nil = none.
-func metricsLoop(execRoot string, stampedUnits func() []string) {
+//
+// sinks are the agent's alert sinks, built and armed for paging by runAgent
+// at startup (armAgentAlerts); the loop delivers its alerts on them and
+// never builds or arms its own. nil delivers nowhere.
+func metricsLoop(execRoot string, stampedUnits func() []string, sinks *alertSinks) {
// This loop is always on and shares a process with the agent that serves
// this box's traffic, so a panic in it — an unreadable cgroup shape, a
// nil no one expected — must not take the box down with it. Monitoring
@@ -789,9 +793,6 @@ func metricsLoop(execRoot string, stampedUnits func() []string) {
st := loadAlertState(statePath, box)
rules := metrics.RulesFromEnv(os.Getenv)
logRejectedAlertOverrides(os.Getenv)
- sinks := buildAlertSinks(ctx, os.Getenv, &processAlertCounters, newSNSClient)
- // The moves code's operator pages go out on the same sinks.
- armPaging(sinks)
log.Printf("agent: metrics loop armed (60s ticks, 5min flush), publishing to %s", target)
diff --git a/internal/host/movesretire_byop_test.go b/internal/host/movesretire_byop_test.go
index 56c03028c..f054c87b7 100644
--- a/internal/host/movesretire_byop_test.go
+++ b/internal/host/movesretire_byop_test.go
@@ -16,7 +16,8 @@ import (
// (Op) and the moves-owned rename's moves-retirement.json (op), each in the
// event directory .parked/<incarnation>/<host>/<retirement>/. It returns the
// retirement ids whose receipt names op, and those with no readable receipt.
-// internal/park's retirementsOf is the same rule.
+// This is the one copy of the rule; internal/park's test checks only that
+// park.Local's journal carries Op.
func retirementsByOp(t *testing.T, dataDir, incarnation, host, op string) (found, unreadable []string) {
t.Helper()
dir := filepath.Join(dataDir, ".parked", incarnation, host)
diff --git a/internal/host/movesrollback.go b/internal/host/movesrollback.go
index 91413190a..e35a74372 100644
--- a/internal/host/movesrollback.go
+++ b/internal/host/movesrollback.go
@@ -141,9 +141,10 @@ func (w *movesWorker) logEvent(ctx context.Context, rec provision.Record, event
// §13: stuck is an incident; K11). role is this box's part in the move,
// "source" or "target", and reason the recorded stuck sentence. The page
// names the host, the move, the box and the reason, and nothing else: no
-// path, no error text, no secret. A replay that lands the same mark again
-// pages again. Unwired (nil Page), nothing pages; a failed page is logged
-// and never stops the arm.
+// path, no error text, no secret. Delivery is best effort: a replay that
+// lands the same mark again pages again, and a crash between the mark and
+// the send, or a send that fails (logged, never retried), pages nothing.
+// Unwired (nil Page), nothing pages; a failed page never stops the arm.
func (w *movesWorker) pageStuck(ctx context.Context, rec provision.Record, role, reason string) {
f := w.cfg.Page
if f == nil {
diff --git a/internal/park/local_linux_test.go b/internal/park/local_linux_test.go
index 106b2db64..fcf56deb4 100644
--- a/internal/park/local_linux_test.go
+++ b/internal/park/local_linux_test.go
@@ -679,45 +679,14 @@ func TestLocalUntouchedBoxAcknowledgement(t *testing.T) {
}
}
-// retirementsOf is how F3 finds a host's local retirements by op, one rule
-// for both receipts a retirement can carry (K20): park.Local's .journal.json
-// (Op) and the moves-owned rename's moves-retirement.json (op), each in the
-// event directory .parked/<incarnation>/<host>/<retirement>/. It returns
-// the retirement ids whose receipt names op.
-func retirementsOf(t *testing.T, dataDir, incarnation, host, op string) []string {
- t.Helper()
- dir := filepath.Join(dataDir, ".parked", incarnation, host)
- es, err := os.ReadDir(dir)
- if err != nil {
- t.Fatal(err)
- }
- var out []string
- for _, e := range es {
- var engine struct{ Op string }
- var moves struct {
- Op string `json:"op"`
- }
- named := ""
- if b, err := os.ReadFile(filepath.Join(dir, e.Name(), ".journal.json")); err == nil && json.Unmarshal(b, &engine) == nil {
- named = engine.Op
- } else if b, err := os.ReadFile(filepath.Join(dir, e.Name(), "moves-retirement.json")); err == nil && json.Unmarshal(b, &moves) == nil {
- named = moves.Op
- }
- if named == op {
- out = append(out, e.Name())
- }
- }
- return out
-}
-
// TestLocalRetirementUnderAMoveOpIsFindableByOp (moves spec §6.4, D6, K20):
// a park-local whose op is a move's id, under each move kind that queues
// one, lands under the engine's own layout,
// .parked/<incarnation>/<host>/<retirement>/artifacts/, and its journal names
-// the move's op, so F3 selects a move's local retirements by op as it does
-// the remote ones by manifest. The moves-owned rename's receipt is the
-// other half; internal/host's TestMoveLocalRetirementsAreFindableByOp
-// covers it with the same rule.
+// the move's op. That is park.Local's own contract; how F3 selects a
+// move's local retirements by op across both receipts (this journal and
+// the moves-owned rename's) is internal/host's
+// TestMoveLocalRetirementsAreFindableByOp.
func TestLocalRetirementUnderAMoveOpIsFindableByOp(t *testing.T) {
const move = "mv_abcdefghijklmnopqrstuvwxyz"
for _, op := range []hostclaim.Op{
@@ -755,12 +724,6 @@ func TestLocalRetirementUnderAMoveOpIsFindableByOp(t *testing.T) {
if j.Op != move || j.Host != f.host || j.Incarnation != f.captured.Incarnation || j.Retirement != "retirement" || j.Obligation != "local" || j.Mode != mode {
t.Fatalf("the journal does not name the move: %+v", j)
}
- if got := retirementsOf(t, f.l.DataDir, f.captured.Incarnation, f.host, move); len(got) != 1 || got[0] != "retirement" {
- t.Fatalf("retirements of %s: %v", move, got)
- }
- if got := retirementsOf(t, f.l.DataDir, f.captured.Incarnation, f.host, "delete"); len(got) != 0 {
- t.Fatalf("another op found the move's retirement: %v", got)
- }
})
}
}