amadan

carlos / platform Public

box side of carlos logs: exec/seed tees, wake events, ship loop

ExecBackend.Logs tees child+seed+sidecar output line-wise into the sink
(journald untouched; WaitDelay discipline extended to teed instance
children). Service.Logs publishes member-safe wake outcomes (#131's two
real cases pinned in tests), stop/exit pairing distinguishes crash from
lifecycle, Bounce records carlos restart. logshipLoop flushes per-app
chunks at 10s (2s after a platform event) and prunes its own box subtree
by age + key-embedded byte sizes. All nil-gated: a box without a bucket
changes nothing.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 644df4e5f0017aeb87e10f54bc20ea29365cb99c parent b16440c
11 files changed, +749 −27
  • docs/superpowers/specs/2026-08-10-carlos-logs-design.md +4 −1
  • internal/activator/activator.go +26 −0
  • internal/activator/backend_exec.go +59 −8
  • internal/activator/logsink.go +72 −0
  • internal/activator/logsink_test.go +122 −0
  • internal/applog/chunk.go +34 −4
  • internal/applog/chunk_test.go +13 −3
  • internal/applog/sink.go +35 −3
  • internal/host/agent.go +28 −8
  • internal/host/logship.go +214 −0
  • internal/host/logship_test.go +142 −0
diff --git a/docs/superpowers/specs/2026-08-10-carlos-logs-design.md b/docs/superpowers/specs/2026-08-10-carlos-logs-design.md
index 4223853d7..86299f55f 100644
--- a/docs/superpowers/specs/2026-08-10-carlos-logs-design.md
+++ b/docs/superpowers/specs/2026-08-10-carlos-logs-design.md
@@ -209,4 +209,7 @@ the drop is recorded in-stream (synthetic record) and visible in the CLI.
- Unit-tenant journald ingestion (stage 2).
- Edge request lines (stage 2) and the release join (stage 3).
- Full-text index/search over all history; alerting on log content.
-- Console web UI tab (API + CLI first; the tab reads the same endpoint).
+
+(The console web UI Logs tab on the app page IS stage 1 — Paul, 2026-08-10:
+a member should see logs in the console, not only through the CLI. It
+renders the same merged timeline server-side with stream/since filters.)
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 9014bfe37..9a07ae907 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -363,6 +363,12 @@ type Service struct {
now func() time.Time // a seam for the unit tests' clock
liveVer func(string) (string, error) // live/<host> symlink target; a seam for tests
+ // Logs, when non-nil, receives member-visible platform events (wake
+ // outcomes, restart bounces) for `carlos logs` — set before Start, like
+ // the backend's own Logs field. Nil (every test, a box without the
+ // shipping loop) emits nothing.
+ Logs EventSink
+
// mesh seam: Eleven kept the multi-box placement layer here (mesh *meshSvc):
// requests for hosts whose HRW owner is another live box were proxied
// there, and the due-time sweeper only woke hosts this box owned. Nil
@@ -573,6 +579,19 @@ func (a *Service) wake(e *actEntry) {
defer cancel()
w, err := a.doWake(ctx, e.route)
+ // The member-facing record of this wake (spec §5): composed here, where
+ // the route and the whole error are in hand — the journal line below the
+ // backends already write stays exactly as it was.
+ if err != nil {
+ emitEvent(a.Logs, e.route, "error", memberSafeWakeReason(err))
+ } else {
+ msg := "instance woke"
+ if w.spawnVersion != "" {
+ msg += " (version " + w.spawnVersion + ")"
+ }
+ emitEvent(a.Logs, e.route, "info", msg)
+ }
+
a.mu.Lock()
if err != nil {
e.err = err
@@ -737,6 +756,7 @@ func (a *Service) Bounce(host string) bool {
if e == nil || e.state != actAwake {
return false
}
+ emitEvent(a.Logs, e.route, "info", "instance restarting (carlos restart)")
a.beginStopLocked(e)
return true
}
@@ -746,6 +766,12 @@ func (a *Service) Bounce(host string) bool {
// the entry. The local DB stays in place — it's the warm cache for the next
// wake; S3 is the durable truth under the lease.
func (a *Service) stop(e *actEntry) {
+ // The deliberate-stop marker for the member's timeline: hibernation,
+ // convergence cycling, bounce, shutdown all pass through here, so an
+ // "exited" line WITHOUT this line before it reads as what it is — a
+ // crash.
+ emitEvent(a.Logs, e.route, "info", "instance stopping")
+
// Co-stop, sidecar FIRST — before the next-due probe and the instance's
// own SIGTERM. Two reasons: the sidecar must never outlive its instance
// (an orphan's reconnect loop would be a standing wake source — the
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 57775d560..22c8d30ce 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -20,6 +20,7 @@ import (
"syscall"
"time"
+ "github.com/carlosframework/platform/internal/applog"
"github.com/carlosframework/platform/internal/cgroup"
"github.com/carlosframework/platform/internal/edge"
"github.com/carlosframework/platform/internal/lease"
@@ -62,6 +63,40 @@ type ExecBackend struct {
// metrics sampler can account it (exec-child accounting spec). Nil — no
// delegation: old unit file, tests, darwin — spawns exactly as before.
Cgroups *cgroup.Manager
+ // Logs, when non-nil, gets each instance and seed child's stdout/stderr
+ // teed in line-wise for `carlos logs` (spec §5) — journald keeps every
+ // byte it got before. Nil (tests, a box without the shipping loop)
+ // spawns exactly as before: fd passthrough, no pipes.
+ Logs EventSink
+}
+
+// execLogLineMax caps one teed log line (spec §5): longer is truncated with
+// a marker, never split into phantom lines.
+const execLogLineMax = 8 << 10
+
+// execDrainGrace bounds how long a dead child's held-open output pipes can
+// delay its Wait. Teeing turns the child's stdout/stderr into PIPES (with a
+// plain *os.File the fd passed straight through), and os/exec's Wait then
+// waits for the read side to drain — which a grandchild holding the write
+// end keeps open forever. Same discipline, same reason as seedDrainGrace;
+// without it, an exited instance would look alive to the activator for as
+// long as some stray grandchild lived, and Stop would hang on p.done.
+var execDrainGrace = 5 * time.Second
+
+// teeStd wires cmd's stdout/stderr: straight to the agent's own streams
+// (journald under systemd) exactly as before, PLUS — only when b.Logs is
+// set — a line-splitting tee into the log sink tagged as rt's app output.
+func (b *ExecBackend) teeStd(cmd *exec.Cmd, rt registry.Route) {
+ if b.Logs == nil {
+ cmd.Stdout = os.Stdout
+ cmd.Stderr = os.Stderr
+ return
+ }
+ acct, app := release.RouteAccount(rt), rt.App
+ emit := func(line string) { b.Logs.App(acct, app, rt.Host, line) }
+ cmd.Stdout = io.MultiWriter(os.Stdout, applog.NewLineWriter(execLogLineMax, emit))
+ cmd.Stderr = io.MultiWriter(os.Stderr, applog.NewLineWriter(execLogLineMax, emit))
+ cmd.WaitDelay = execDrainGrace
}
var _ Backend = (*ExecBackend)(nil)
@@ -238,9 +273,17 @@ func (b *ExecBackend) Seed(ctx context.Context, rt registry.Route) error {
// Like the instance child, the seed logs through the agent's own streams
// (journald under systemd) — and its stderr is ALSO teed into a bounded
// buffer, because the wake error this returns is the whole diagnosis an
- // operator gets for a host that refuses to come up.
+ // operator gets for a host that refuses to come up. With Logs set the
+ // seed's output ships to `carlos logs` too: it is app output, and on a
+ // failed first wake it is the diagnosis the MEMBER gets.
cmd.Stdout = os.Stdout
cmd.Stderr = io.MultiWriter(os.Stderr, &stderr)
+ if b.Logs != nil {
+ acct, app := release.RouteAccount(rt), rt.App
+ emit := func(line string) { b.Logs.App(acct, app, host, line) }
+ cmd.Stdout = io.MultiWriter(os.Stdout, applog.NewLineWriter(execLogLineMax, emit))
+ cmd.Stderr = io.MultiWriter(os.Stderr, &stderr, applog.NewLineWriter(execLogLineMax, emit))
+ }
// WaitDelay is what makes seedTimeout real. Because Stderr is a plain
// io.Writer (the tee), os/exec pipes it and Wait waits for that pipe to
// close — and a seed that spawned its own grandchild holds the write end
@@ -474,9 +517,10 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
cmd.Env = childEnv(childVars)
// The child logs through the agent's own stderr/stdout (journald
// under systemd); there's no per-instance unit to own a log stream
- // anymore.
- cmd.Stdout = os.Stdout
- cmd.Stderr = os.Stderr
+ // anymore. With Logs set the same bytes are ALSO teed line-wise
+ // into the log sink (`carlos logs`); teeStd owns the WaitDelay
+ // that piping entails.
+ b.teeStd(cmd, rt)
return cmd
})
if err != nil {
@@ -505,6 +549,14 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
}
go func() {
p.err = cmd.Wait()
+ // The member-facing record of the exit: a crash shows an "exited"
+ // line with no "stopping" line before it — that pairing, not the
+ // exit code alone, is what tells a member crash from lifecycle.
+ if p.err != nil {
+ emitEvent(b.Logs, rt, "warn", "instance exited: "+redactPaths(p.err.Error()))
+ } else {
+ emitEvent(b.Logs, rt, "info", "instance exited (exit 0)")
+ }
// Last exiter wins: rmdir refuses while the sidecar (same host
// cgroup) is still alive, and that failure is tolerated.
b.Cgroups.CleanupHost(host)
@@ -668,10 +720,9 @@ func (b *ExecBackend) StartSidecar(_ context.Context, rt registry.Route) (Proc,
cmd := exec.Command(b.liveLink(rt.Host), "sidecar", "run")
cmd.Env = childEnv(fileEnv)
// Like the instance child: logs through the agent's own streams
- // (journald under systemd) — no per-host unit owns a log stream
- // anymore.
- cmd.Stdout = os.Stdout
- cmd.Stderr = os.Stderr
+ // (journald under systemd), teed into the log sink when armed — a
+ // sidecar's output is the app's output.
+ b.teeStd(cmd, rt)
return cmd
})
if err != nil {
diff --git a/internal/activator/logsink.go b/internal/activator/logsink.go
new file mode 100644
index 000000000..e8fa84242
--- /dev/null
+++ b/internal/activator/logsink.go
@@ -0,0 +1,72 @@
+// The activator's half of `carlos logs` (spec 2026-08-10-carlos-logs-design
+// §5): the seam the agent injects its applog sink through, and the
+// member-safe rendering of wake failures. The reason a wake failed is the
+// whole diagnosis a member gets for a 503 (#131), and it is composed HERE,
+// at the emit site — never relayed from the journal, and never carrying box
+// filesystem layout.
+package activator
+
+import (
+ "regexp"
+ "strings"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// EventSink is where the activator publishes member-visible log lines.
+// applog.Sink satisfies it; the interface lives here so this package
+// depends on the contract, not the buffer. Nil disables everything —
+// tests and non-shipping deployments change nothing.
+type EventSink interface {
+ // App records one line of the app's own output.
+ App(acct, app, host, line string)
+ // Event records one platform event with member-safe text.
+ Event(acct, app, host, level, msg string)
+}
+
+// emitEvent publishes one platform event for rt's app, tolerating a nil
+// sink. level is an applog.Level* value ("info"/"warn"/"error" — string
+// here to keep the dependency one-way).
+func emitEvent(sink EventSink, rt registry.Route, level, msg string) {
+ if sink == nil {
+ return
+ }
+ sink.Event(release.RouteAccount(rt), rt.App, rt.Host, level, msg)
+}
+
+// absPathToken matches an absolute-path token: box filesystem layout,
+// which a member-safe message must not carry (#131). Replacement keeps the
+// base name — "…/lpay.bde.oncarlos.com.sock" still tells the member WHICH
+// socket without telling them where the box keeps it.
+var absPathToken = regexp.MustCompile(`(^|[\s"(=])(/[\w@.\-/]+)`)
+
+// memberSafeWakeReason renders a wake error for the member-facing platform
+// stream. Known failure shapes get a diagnosis in the member's terms; the
+// rest pass through with absolute paths reduced to base names. The error
+// text itself originates in our own code (backends, materialize, lease) —
+// this is a rendering pass, not a secrecy boundary.
+func memberSafeWakeReason(err error) string {
+ s := err.Error()
+ switch {
+ case strings.Contains(s, "address already in use"):
+ // The stale-socket wake trap (#129): every later wake fails until
+ // the leftover socket goes away. Say that, not the bind syscall.
+ return "wake failed: instance socket already in use — a previous run's socket was left behind (this wake will keep failing until the platform clears it)"
+ case strings.Contains(s, "no adoptable head"):
+ return "wake failed: nothing is promoted for this instance's channel (or the head is still bake-held) — promote a build, or check `carlos channels`"
+ }
+ return "wake failed: " + redactPaths(s)
+}
+
+// redactPaths reduces absolute-path tokens to their base names.
+func redactPaths(s string) string {
+ return absPathToken.ReplaceAllStringFunc(s, func(m string) string {
+ i := strings.Index(m, "/")
+ prefix, path := m[:i], m[i:]
+ if j := strings.LastIndexByte(path, '/'); j >= 0 && j+1 < len(path) {
+ return prefix + path[j+1:]
+ }
+ return prefix + path
+ })
+}
diff --git a/internal/activator/logsink_test.go b/internal/activator/logsink_test.go
new file mode 100644
index 000000000..c119cee1e
--- /dev/null
+++ b/internal/activator/logsink_test.go
@@ -0,0 +1,122 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "strings"
+ "testing"
+ "time"
+)
+
+// TestMemberSafeWakeReason pins the two real 2026-08-09 cases from #131 —
+// each needs a different response from the member, and neither may carry
+// box filesystem layout.
+func TestMemberSafeWakeReason(t *testing.T) {
+ cases := []struct {
+ name string
+ err error
+ want []string // substrings that must appear
+ deny []string // substrings that must NOT appear
+ }{
+ {
+ "stale socket (#129)",
+ errors.New(`start lpay.bde.oncarlos.com: listen unix /data/exec/lpay.bde.oncarlos.com/inst.sock: bind: address already in use`),
+ []string{"socket already in use", "previous run"},
+ []string{"/data", "bind:"},
+ },
+ {
+ "nothing promoted (#130)",
+ errors.New(`materialize lpay.bde.oncarlos.com: no adoptable head (nothing promoted for its channel, or the head is bake-held)`),
+ []string{"nothing is promoted", "carlos channels"},
+ nil,
+ },
+ {
+ "unknown errors keep meaning, lose layout",
+ errors.New(`seed h: fork/exec /opt/carlos/versions/lpay/1.2.3/lpay: permission denied`),
+ []string{"wake failed", "permission denied", "lpay"},
+ []string{"/opt/carlos"},
+ },
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ got := memberSafeWakeReason(tc.err)
+ for _, w := range tc.want {
+ if !strings.Contains(got, w) {
+ t.Errorf("missing %q in %q", w, got)
+ }
+ }
+ for _, d := range tc.deny {
+ if strings.Contains(got, d) {
+ t.Errorf("leaked %q in %q", d, got)
+ }
+ }
+ })
+ }
+}
+
+// eventRec and fakeSink are the package's EventSink fixture.
+type eventRec struct{ acct, app, host, level, msg string }
+
+type fakeSink struct {
+ apps []eventRec
+ events []eventRec
+}
+
+func (f *fakeSink) App(acct, app, host, line string) {
+ f.apps = append(f.apps, eventRec{acct, app, host, "", line})
+}
+func (f *fakeSink) Event(acct, app, host, level, msg string) {
+ f.events = append(f.events, eventRec{acct, app, host, level, msg})
+}
+
+// TestWakeOutcomesReachTheSink: the activator's wake path publishes exactly
+// one member-facing event per outcome — the #131 fact, at its source. The
+// route carries app+account context so the sink can key the record.
+func TestWakeOutcomesReachTheSink(t *testing.T) {
+ // Failure: Start refuses, the event is the member-safe rendering.
+ a, b, _, _ := testActivator(t, time.Hour)
+ sink := &fakeSink{}
+ a.Logs = sink
+ b.startErr = errors.New("listen unix /data/exec/h/inst.sock: bind: address already in use")
+ rt := testRoute("lpay.bde.oncarlos.com")
+ rt.App = "lpay"
+ rt.Slug = ""
+ if _, err := a.beginRequest(context.Background(), rt); err == nil {
+ t.Fatal("wake should have failed")
+ }
+ var wakeFails []eventRec
+ for _, e := range sink.events {
+ if strings.Contains(e.msg, "wake failed") {
+ wakeFails = append(wakeFails, e)
+ }
+ }
+ if len(wakeFails) != 1 {
+ t.Fatalf("wake-failure events: %d (%v)", len(wakeFails), sink.events)
+ }
+ ev := wakeFails[0]
+ if ev.level != "error" || ev.host != "lpay.bde.oncarlos.com" || ev.app != "lpay" {
+ t.Fatalf("event mis-tagged: %+v", ev)
+ }
+ if strings.Contains(ev.msg, "/data") || !strings.Contains(ev.msg, "socket already in use") {
+ t.Fatalf("event not member-safe: %q", ev.msg)
+ }
+
+ // Success: a woke event, tagged the same way.
+ a2, _, _, _ := testActivator(t, time.Hour)
+ sink2 := &fakeSink{}
+ a2.Logs = sink2
+ rel, err := a2.beginRequest(context.Background(), rt)
+ if err != nil {
+ t.Fatalf("wake failed: %v", err)
+ }
+ rel()
+ var woke bool
+ for _, e := range sink2.events {
+ if strings.Contains(e.msg, "instance woke") && e.level == "info" {
+ woke = true
+ }
+ }
+ if !woke {
+ t.Fatalf("no woke event: %+v", sink2.events)
+ }
+}
diff --git a/internal/applog/chunk.go b/internal/applog/chunk.go
index ad53ffffe..6603f8ab0 100644
--- a/internal/applog/chunk.go
+++ b/internal/applog/chunk.go
@@ -7,6 +7,8 @@ import (
"encoding/json"
"fmt"
"io"
+ "strconv"
+ "strings"
"time"
"github.com/carlosframework/platform/internal/release"
@@ -46,10 +48,38 @@ func HourPrefix(acct, app, box, hour string) string {
}
// ChunkKey names one flushed chunk: seq is the first record's per-box
-// sequence number, zero-padded so lexical listing order is numeric order.
-// Chunks are written once and never appended or rewritten (spec §3).
-func ChunkKey(acct, app, box, hour string, seq uint64) string {
- return fmt.Sprintf("%s%012d.jsonl.gz", HourPrefix(acct, app, box, hour), seq)
+// sequence number, zero-padded (16 digits — seq is milliseconds-based, see
+// NewSink) so lexical listing order is numeric order, and size is the
+// object's own byte length. Size lives IN the key on purpose:
+// release.Store's List returns keys only, and retention's byte cap needs
+// sizes — embedding them costs a suffix, extending the Store interface
+// costs every implementation. Chunks are written once and never appended
+// or rewritten (spec §3).
+func ChunkKey(acct, app, box, hour string, seq uint64, size int) string {
+ return fmt.Sprintf("%s%016d-%d.jsonl.gz", HourPrefix(acct, app, box, hour), seq, size)
+}
+
+// ChunkKeySize recovers the byte size a ChunkKey carries, false for any key
+// not shaped like one (retention treats unparsable as size-unknown = 0 —
+// age still prunes it, the byte cap just cannot count it).
+func ChunkKeySize(key string) (int64, bool) {
+ base := key
+ if i := strings.LastIndexByte(base, '/'); i >= 0 {
+ base = base[i+1:]
+ }
+ base, ok := strings.CutSuffix(base, ".jsonl.gz")
+ if !ok {
+ return 0, false
+ }
+ _, sizeStr, ok := strings.Cut(base, "-")
+ if !ok {
+ return 0, false
+ }
+ n, err := strconv.ParseInt(sizeStr, 10, 64)
+ if err != nil || n < 0 {
+ return 0, false
+ }
+ return n, true
}
// EncodeChunk renders records as gzipped JSONL, one record per line.
diff --git a/internal/applog/chunk_test.go b/internal/applog/chunk_test.go
index 137b6a6f5..e5069f123 100644
--- a/internal/applog/chunk_test.go
+++ b/internal/applog/chunk_test.go
@@ -48,9 +48,19 @@ func TestKeysAndHourFormat(t *testing.T) {
if got := HourPrefix("bde", "lpay", "box1", "2026-08-10-21"); got != "apps/bde/lpay/logs/box1/2026-08-10-21/" {
t.Fatalf("HourPrefix = %q", got)
}
- // Zero-padded so lexical order is numeric order.
- if got := ChunkKey("bde", "lpay", "box1", "2026-08-10-21", 41822); got != "apps/bde/lpay/logs/box1/2026-08-10-21/000000041822.jsonl.gz" {
- t.Fatalf("ChunkKey = %q", got)
+ // Zero-padded so lexical order is numeric order; the byte size rides in
+ // the key so retention can count without a Store interface change.
+ key := ChunkKey("bde", "lpay", "box1", "2026-08-10-21", 41822, 8123)
+ if key != "apps/bde/lpay/logs/box1/2026-08-10-21/0000000000041822-8123.jsonl.gz" {
+ t.Fatalf("ChunkKey = %q", key)
+ }
+ if n, ok := ChunkKeySize(key); !ok || n != 8123 {
+ t.Fatalf("ChunkKeySize = %d, %v", n, ok)
+ }
+ for _, bad := range []string{"x/no-size.jsonl.gz", "x/whatever.json", "0000000000041822-8123.jsonl.gz.bak"} {
+ if _, ok := ChunkKeySize(bad); ok {
+ t.Errorf("ChunkKeySize(%q) parsed", bad)
+ }
}
}
diff --git a/internal/applog/sink.go b/internal/applog/sink.go
index 0f7490796..309b46d8a 100644
--- a/internal/applog/sink.go
+++ b/internal/applog/sink.go
@@ -29,7 +29,9 @@ type Sink struct {
mu sync.Mutex
apps map[AppKey]*appBuf
- seq uint64 // per box process, shared across apps — the follow cursor's spine
+ seq uint64 // monotonic across apps AND restarts — the follow cursor's spine
+
+ notify chan struct{} // poked (non-blocking) on Event; the ship loop's fast-flush signal
}
type appBuf struct {
@@ -41,7 +43,13 @@ type appBuf struct {
}
// NewSink returns a Sink stamping records with box and timestamps from now
-// (a seam so tests control the clock).
+// (a seam so tests control the clock). The sequence counter starts at the
+// construction instant's UnixMilli rather than zero: chunk keys embed the
+// first record's seq, and per-box seq order is a promise the cursor relies
+// on — a restart that re-based to zero would write keys that sort BEFORE
+// last boot's and break exact per-box ordering. Milliseconds always outrun
+// the admitted line rate (the rate cap is ~17 KiB/s), so a rebase lands
+// above everything the previous process numbered.
func NewSink(box string, now func() time.Time) *Sink {
return &Sink{
MaxBytesPerApp: 4 << 20,
@@ -50,7 +58,27 @@ func NewSink(box string, now func() time.Time) *Sink {
box: box,
now: now,
apps: make(map[AppKey]*appBuf),
+ seq: uint64(now().UnixMilli()),
+ notify: make(chan struct{}, 1),
+ }
+}
+
+// Wake is the ship loop's fast-flush signal: it receives after an Event has
+// been buffered, so a wake failure reaches the bucket in seconds rather
+// than at the next tick (spec §5). Level-triggered and lossy by design —
+// one pending poke summarizes any number of events.
+func (s *Sink) Wake() <-chan struct{} { return s.notify }
+
+// Known lists every app this sink has ever buffered for, in no particular
+// order — the ship loop's retention worklist.
+func (s *Sink) Known() []AppKey {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ out := make([]AppKey, 0, len(s.apps))
+ for k := range s.apps {
+ out = append(out, k)
}
+ return out
}
// App buffers one line of app output for (acct, app), parsing structured
@@ -84,13 +112,17 @@ func (s *Sink) App(acct, app, host, line string) {
// a bug, not a budget).
func (s *Sink) Event(acct, app, host, level, msg string) {
s.mu.Lock()
- defer s.mu.Unlock()
s.append(s.buf(AppKey{acct, app}), Record{
Host: host,
Stream: StreamPlatform,
Level: level,
Msg: msg,
})
+ s.mu.Unlock()
+ select { // non-blocking: a pending poke already says everything
+ case s.notify <- struct{}{}:
+ default:
+ }
}
// Drain returns everything buffered, keyed by app, and resets the buffers
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 94a784712..c9051bec2 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -91,6 +91,7 @@ import (
"github.com/carlosframework/platform/internal/activator"
"github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/applog"
"github.com/carlosframework/platform/internal/caps"
"github.com/carlosframework/platform/internal/cgroup"
"github.com/carlosframework/platform/internal/domains"
@@ -139,6 +140,15 @@ func RunAgent(args []string) {
// per-instance readings.
cgm := cgroup.Setup(log.Printf)
+ // Log shipping (`carlos logs`): one sink for every producer in this
+ // process, armed only when a bucket exists to ship to — a nil sink
+ // leaves every producer exactly as before (no tees, no pipes, no
+ // events). The box identity is the one metrics publishes under.
+ var logSink *applog.Sink
+ if resolveStatusBucket(os.Getenv) != "" {
+ logSink = applog.NewSink(status.Label(readMachineID()), time.Now)
+ }
+
cfg, err := activator.ConfigFromEnv()
if err != nil {
log.Fatalf("agent: %v", err)
@@ -150,17 +160,24 @@ func RunAgent(args []string) {
// the split backend dispatches per route on Backing. Note the arming
// gate is still the hibernate bucket env even though unit tenants
// never touch the bucket — one opt-in switch for the whole activator.
+ eb := &activator.ExecBackend{
+ Cfg: *cfg,
+ OptDir: release.OptDir(),
+ DataDir: dd,
+ Cache: cache,
+ Materialize: materializeHook(reg, dd),
+ Cgroups: cgm,
+ }
+ if logSink != nil { // a typed-nil in the interface would NOT be nil
+ eb.Logs = logSink
+ }
svc = activator.New(*cfg, &activator.SplitBackend{
- Exec: &activator.ExecBackend{
- Cfg: *cfg,
- OptDir: release.OptDir(),
- DataDir: dd,
- Cache: cache,
- Materialize: materializeHook(reg, dd),
- Cgroups: cgm,
- },
+ Exec: eb,
Unit: &activator.UnitBackend{OptDir: release.OptDir()},
}, reg) // the registry satisfies activator.DueStore directly
+ if logSink != nil {
+ svc.Logs = logSink
+ }
svc.Start()
act = svc
log.Printf("agent: hibernation activator armed (bucket %s, idle %s)", cfg.Bucket, cfg.Idle)
@@ -168,6 +185,9 @@ func RunAgent(args []string) {
go adoptLoop(reg, svc)
go fastConvergeLoop(reg, svc)
+ if logSink != nil {
+ go logshipLoop(logSink, status.Label(readMachineID()))
+ }
go statusLoop(reg)
go domainsLoop(reg, peekAcmeCache(args))
go instancesLoop(reg)
diff --git a/internal/host/logship.go b/internal/host/logship.go
new file mode 100644
index 000000000..6861c0a28
--- /dev/null
+++ b/internal/host/logship.go
@@ -0,0 +1,214 @@
+// The log shipping loop — the box-side pump of `carlos logs` (spec
+// 2026-08-10-carlos-logs-design.md §5). Producers (the activator's exec
+// tees and platform events) buffer into one applog.Sink; every 10s — or
+// within ~2s of a platform event, which is what lets `carlos deploy`'s
+// timeout window see the wake error — this loop drains it and writes one
+// immutable gzipped chunk per app into the deployment bucket. On each hour
+// boundary it prunes ITS OWN box subtree per app: hours beyond the age
+// floor, then oldest-first beyond the byte cap (sizes ride in the chunk
+// keys — applog.ChunkKeySize — so no Store interface change). The
+// sampler/flush split mirrors metricsLoop for the same reason: the loop
+// shares a process with the box's traffic, so it recovers panics and
+// disables itself without a bucket rather than taking the box down.
+package host
+
+import (
+ "context"
+ "log"
+ "os"
+ "time"
+
+ "github.com/carlosframework/platform/internal/applog"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+const (
+ logshipTick = 10 * time.Second
+ logshipDebounce = 2 * time.Second // event-poke settle: batch a wake's burst into one PUT
+ logshipMaxAge = 7 * 24 * time.Hour
+ logshipMaxBytes = 50 << 20 // per app per box
+ // logshipMaxPending caps records held across failed flushes, per app.
+ // The bucket being unwritable must not grow the agent's heap forever
+ // (samplerMaxBuf's reasoning); oldest go first, silently — a box-wide
+ // bucket outage is loudly visible elsewhere, and the gap in seq numbers
+ // records that something is missing.
+ logshipMaxPending = 4096
+)
+
+func logshipLoop(sink *applog.Sink, box string) {
+ // Same rule as metricsLoop: this loop shares a process with the agent
+ // serving traffic — log shipping stopping is a bad day, the box
+ // stopping is an outage.
+ defer func() {
+ if r := recover(); r != nil {
+ log.Printf("agent: logship loop panic, log shipping stopped for this process: %v", r)
+ }
+ }()
+
+ ctx := context.Background()
+ bucket := resolveStatusBucket(os.Getenv)
+ if bucket == "" {
+ log.Printf("agent: logship loop disabled: neither CARLOS_STATUS_BUCKET nor CARLOS_DEPLOYMENT_BUCKET is set")
+ return
+ }
+ s, err := release.NewS3Store(ctx, bucket)
+ if err != nil {
+ log.Printf("agent: logship loop disabled: %v", err)
+ return
+ }
+ log.Printf("agent: logship loop armed (10s ticks, event fast-flush), publishing to bucket %s", bucket)
+
+ pending := make(map[applog.AppKey][]applog.Record)
+ pruned := "" // last hour a prune ran for, "" = not yet
+ ticker := time.NewTicker(logshipTick)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ticker.C:
+ case <-sink.Wake():
+ time.Sleep(logshipDebounce) // let the event's burst finish
+ }
+ logshipGather(sink, pending)
+ logshipFlush(ctx, s, pending)
+
+ if hour := applog.HourOf(time.Now()); hour != pruned {
+ // Prune covers every app this box is CURRENTLY holding or has
+ // just flushed; an app that stops logging entirely ages out via
+ // other boxes' passes or the app's own deletion sweep. Sub-cap
+ // apps pay one ListDirs each per hour.
+ for _, k := range logshipKnownApps(pending, sink) {
+ logshipPrune(ctx, s, box, k, time.Now())
+ }
+ pruned = hour
+ }
+ }
+}
+
+// logshipGather drains the sink into pending, enforcing the per-app cap —
+// oldest first, so what survives a long bucket outage is the freshest tail.
+func logshipGather(sink *applog.Sink, pending map[applog.AppKey][]applog.Record) {
+ for k, recs := range sink.Drain() {
+ merged := append(pending[k], recs...)
+ if over := len(merged) - logshipMaxPending; over > 0 {
+ merged = merged[over:]
+ }
+ pending[k] = merged
+ }
+}
+
+// logshipFlush writes one chunk per pending app — create-only, keyed by the
+// batch's first seq, so a retried batch can never overwrite a delivered one
+// — and forgets what landed. A failed app keeps its records for the next
+// round; other apps' flushes proceed regardless.
+func logshipFlush(ctx context.Context, s release.Store, pending map[applog.AppKey][]applog.Record) {
+ for k, recs := range pending {
+ if len(recs) == 0 {
+ delete(pending, k)
+ continue
+ }
+ body, err := applog.EncodeChunk(recs)
+ if err != nil { // marshal of our own records — a bug, not a retry case
+ log.Printf("agent: logship encode %s/%s: %v (dropping %d records)", k.Acct, k.App, err, len(recs))
+ delete(pending, k)
+ continue
+ }
+ key := applog.ChunkKey(k.Acct, k.App, recs[0].Box, applog.HourOf(recs[0].TS), recs[0].Seq, len(body))
+ if _, err := s.Put(ctx, key, body, ""); err != nil {
+ log.Printf("agent: logship flush %s/%s: %v (%d records held)", k.Acct, k.App, err, len(recs))
+ continue
+ }
+ delete(pending, k)
+ }
+}
+
+// logshipKnownApps is the union of apps with pending records and apps the
+// sink has ever buffered for — the prune worklist.
+func logshipKnownApps(pending map[applog.AppKey][]applog.Record, sink *applog.Sink) []applog.AppKey {
+ seen := make(map[applog.AppKey]bool)
+ var out []applog.AppKey
+ for k := range pending {
+ if !seen[k] {
+ seen[k] = true
+ out = append(out, k)
+ }
+ }
+ for _, k := range sink.Known() {
+ if !seen[k] {
+ seen[k] = true
+ out = append(out, k)
+ }
+ }
+ return out
+}
+
+// logshipPrune enforces retention for ONE app on THIS box's subtree only —
+// each box owns what it wrote (spec §5). Hour granularity: an hour goes
+// when it is older than the age floor, or (oldest first) while the
+// subtree's key-embedded sizes total over the byte cap. Errors are logged
+// and left for the next hourly pass — retention is idempotent.
+func logshipPrune(ctx context.Context, s release.Store, box string, k applog.AppKey, now time.Time) {
+ prefix := applog.BoxPrefix(k.Acct, k.App, box)
+ hourDirs, err := s.ListDirs(ctx, prefix)
+ if err != nil {
+ log.Printf("agent: logship prune %s: %v", prefix, err)
+ return
+ }
+ // ListDirs returns full prefixes ending in "/", sorted is not promised —
+ // the hour names sort chronologically, so order them ourselves via the
+ // keys' lexical shape. Collect per-hour keys+sizes as we go.
+ type hourEnt struct {
+ dir string
+ keys []string
+ bytes int64
+ }
+ var hours []hourEnt
+ var total int64
+ for _, dir := range hourDirs {
+ keys, err := s.List(ctx, dir)
+ if err != nil {
+ log.Printf("agent: logship prune %s: %v", dir, err)
+ return
+ }
+ var b int64
+ for _, key := range keys {
+ if n, ok := applog.ChunkKeySize(key); ok {
+ b += n
+ }
+ }
+ hours = append(hours, hourEnt{dir: dir, keys: keys, bytes: b})
+ total += b
+ }
+ // hourDirs from both implementations come lexically sorted (S3 lists in
+ // key order; DirStore sorts), and hour names are zero-padded UTC — so
+ // hours[0] is oldest.
+ cutoff := now.Add(-logshipMaxAge)
+ for _, h := range hours {
+ hourName := hourNameOf(h.dir)
+ drop := applog.HourBefore(hourName, cutoff) || total > logshipMaxBytes
+ if !drop {
+ break // hours are oldest-first; nothing further is droppable
+ }
+ for _, key := range h.keys {
+ if err := s.Delete(ctx, key); err != nil {
+ log.Printf("agent: logship prune %s: %v", key, err)
+ return
+ }
+ }
+ total -= h.bytes
+ }
+}
+
+// hourNameOf extracts the hour directory name from its full prefix
+// (".../logs/<box>/<hour>/" → "<hour>").
+func hourNameOf(dir string) string {
+ trimmed := dir
+ if n := len(trimmed); n > 0 && trimmed[n-1] == '/' {
+ trimmed = trimmed[:n-1]
+ }
+ for i := len(trimmed) - 1; i >= 0; i-- {
+ if trimmed[i] == '/' {
+ return trimmed[i+1:]
+ }
+ }
+ return trimmed
+}
diff --git a/internal/host/logship_test.go b/internal/host/logship_test.go
new file mode 100644
index 000000000..38e8419b0
--- /dev/null
+++ b/internal/host/logship_test.go
@@ -0,0 +1,142 @@
+package host
+
+import (
+ "context"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/applog"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func logshipFixture(t *testing.T) (release.Store, *applog.Sink) {
+ t.Helper()
+ s := release.NewDirStore(t.TempDir())
+ now := time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC)
+ return s, applog.NewSink("box1", func() time.Time { return now })
+}
+
+// TestLogshipFlushWritesOneChunkPerApp: a drain with two apps lands as two
+// immutable chunks, each under its own app prefix, decodable and complete.
+func TestLogshipFlushWritesOneChunkPerApp(t *testing.T) {
+ s, sink := logshipFixture(t)
+ sink.App("bde", "lpay", "lpay.bde.oncarlos.com", "hello")
+ sink.Event("bab", "kass", "kass.example.com", applog.LevelInfo, "instance woke")
+
+ pending := make(map[applog.AppKey][]applog.Record)
+ logshipGather(sink, pending)
+ logshipFlush(context.Background(), s, pending)
+ if len(pending) != 0 {
+ t.Fatalf("flushed apps still pending: %v", pending)
+ }
+
+ for _, app := range []applog.AppKey{{Acct: "bde", App: "lpay"}, {Acct: "bab", App: "kass"}} {
+ keys, err := s.List(context.Background(), applog.Prefix(app.Acct, app.App))
+ if err != nil || len(keys) != 1 {
+ t.Fatalf("chunks for %v: %v (%v)", app, keys, err)
+ }
+ body, _, err := s.Get(context.Background(), keys[0])
+ if err != nil {
+ t.Fatal(err)
+ }
+ recs, err := applog.DecodeChunk(body)
+ if err != nil || len(recs) != 1 {
+ t.Fatalf("chunk decode for %v: %d recs, %v", app, len(recs), err)
+ }
+ // The key carries the real byte size — retention counts with it.
+ if n, ok := applog.ChunkKeySize(keys[0]); !ok || n != int64(len(body)) {
+ t.Fatalf("key size %d/%v, body %d: %s", n, ok, len(body), keys[0])
+ }
+ }
+}
+
+// TestLogshipFlushKeepsRecordsWhenTheBucketRefuses: a failed Put must hold
+// the app's records for the next round, not lose them — and the pending cap
+// bounds that holding.
+func TestLogshipFlushKeepsRecordsWhenTheBucketRefuses(t *testing.T) {
+ _, sink := logshipFixture(t)
+ sink.App("bde", "lpay", "h", "precious")
+ pending := make(map[applog.AppKey][]applog.Record)
+ logshipGather(sink, pending)
+ logshipFlush(context.Background(), failingStore{}, pending)
+ k := applog.AppKey{Acct: "bde", App: "lpay"}
+ if len(pending[k]) != 1 || pending[k][0].Msg != "precious" {
+ t.Fatalf("records lost on failed flush: %v", pending)
+ }
+ // Cap: a long outage trims oldest, keeps the freshest tail.
+ for i := 0; i < logshipMaxPending+10; i++ {
+ sink.App("bde", "lpay", "h", "line")
+ }
+ logshipGather(sink, pending)
+ if len(pending[k]) != logshipMaxPending {
+ t.Fatalf("pending cap not enforced: %d", len(pending[k]))
+ }
+ if pending[k][0].Msg == "precious" {
+ t.Fatal("cap kept oldest instead of freshest")
+ }
+}
+
+// TestLogshipPrune: hours past the age floor go; then oldest-first until
+// the key-embedded sizes fit the byte cap; the box only touches its own
+// subtree.
+func TestLogshipPrune(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+ now := time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC)
+ k := applog.AppKey{Acct: "bde", App: "lpay"}
+
+ put := func(box, hour string, seq uint64, size int) string {
+ key := applog.ChunkKey(k.Acct, k.App, box, hour, seq, size)
+ if _, err := s.Put(ctx, key, make([]byte, size), ""); err != nil {
+ t.Fatal(err)
+ }
+ return key
+ }
+ oldKey := put("box1", "2026-08-01-10", 1, 10) // over the 7d floor
+ bigKey := put("box1", "2026-08-10-10", 2, 60<<20) // pushes total over the byte cap
+ liveKey := put("box1", "2026-08-10-11", 3, 10)
+ otherBox := put("box2", "2026-08-01-10", 4, 10) // NOT ours to prune
+
+ logshipPrune(ctx, s, "box1", k, now)
+
+ for key, want := range map[string]bool{oldKey: false, bigKey: false, liveKey: true, otherBox: true} {
+ _, _, err := s.Get(ctx, key)
+ if got := err == nil; got != want {
+ t.Errorf("%s survived=%v, want %v", key, got, want)
+ }
+ }
+}
+
+// failingStore refuses writes — the bucket-outage fixture.
+type failingStore struct{ release.Store }
+
+func (failingStore) Put(context.Context, string, []byte, string) (string, error) {
+ return "", context.DeadlineExceeded
+}
+
+// TestHourNameOf pins the prefix→name extraction.
+func TestHourNameOf(t *testing.T) {
+ if got := hourNameOf("apps/bde/lpay/logs/box1/2026-08-10-11/"); got != "2026-08-10-11" {
+ t.Fatalf("hourNameOf = %q", got)
+ }
+}
+
+// TestLogshipPruneKeepsRecentUnderCap: nothing under both limits is touched.
+func TestLogshipPruneKeepsRecentUnderCap(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+ now := time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC)
+ k := applog.AppKey{Acct: "bde", App: "lpay"}
+ key := applog.ChunkKey(k.Acct, k.App, "box1", "2026-08-10-11", 1, 10)
+ if _, err := s.Put(ctx, key, make([]byte, 10), ""); err != nil {
+ t.Fatal(err)
+ }
+ logshipPrune(ctx, s, "box1", k, now)
+ if _, _, err := s.Get(ctx, key); err != nil {
+ t.Fatalf("live chunk pruned: %v", err)
+ }
+ if !strings.HasPrefix(key, "apps/bde/lpay/logs/") {
+ t.Fatalf("key layout drifted: %s", key)
+ }
+}