amadan
cli: carlos logs — the member's log read, with -f follow
API mode only (the box journal was never the member surface): filters map straight onto the endpoint, output is one line per record with stream tags and colored levels, -f polls the cursor at the serial's own 2s. E2E proves the #131 path end to end: login → merged timeline → wake reason, and a follow that catches a chunk shipped mid-poll. Usage + top-level help wired.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
4 files changed,
+376
−0
apimode.go+36 −0logs.go+177 −0logs_e2e_test.go+159 −0main.go+4 −0
diff --git a/apimode.go b/apimode.go| index ad4daf82e..6ae3887e6 100644 |
| --- a/apimode.go |
| +++ b/apimode.go |
| @@ -32,6 +32,7 @@ import ( |
| "time" |
| "github.com/carlosframework/platform/internal/appconfig" |
| + "github.com/carlosframework/platform/internal/applog" |
| "github.com/carlosframework/platform/internal/caps" |
| "github.com/carlosframework/platform/internal/ledger" |
| ) |
| @@ -1082,6 +1083,41 @@ func (s *apiSession) instanceList(ctx context.Context, a apiAcct, app string) ([ |
| return res.Instances, res.Truncated, nil |
| } |
| +// apiLogsResponse is the console's logs answer (internal/console/apilogs.go) |
| +// — records are applog's own wire shape, decoded as-is. |
| +type apiLogsResponse struct { |
| + Records []applog.Record `json:"records"` |
| + Truncated bool `json:"truncated"` |
| + Cursor string `json:"cursor"` |
| +} |
| + |
| +// logsGet reads one window (or one follow increment) of an app's merged log |
| +// timeline. query carries the filter params exactly as the endpoint speaks |
| +// them; apiPath may already hold a query (exact=sqid), hence the separator |
| +// dance. |
| +func (s *apiSession) logsGet(ctx context.Context, a apiAcct, app string, query url.Values) (apiLogsResponse, error) { |
| + path := apiPath(a, app, "/logs") |
| + if enc := query.Encode(); enc != "" { |
| + if strings.Contains(path, "?") { |
| + path += "&" + enc |
| + } else { |
| + path += "?" + enc |
| + } |
| + } |
| + var res apiLogsResponse |
| + status, body, err := s.getJSON(ctx, path) |
| + if err != nil { |
| + return res, err |
| + } |
| + if status != http.StatusOK { |
| + return res, apiErrorFromBody(status, body) |
| + } |
| + if err := json.Unmarshal(body, &res); err != nil { |
| + return res, fmt.Errorf("logs: unreadable response: %w", err) |
| + } |
| + return res, nil |
| +} |
| + |
| // instanceDelete retires one instance record. Idempotent server-side (the |
| // caller's goal is "not there"), so a 200 is the only success shape. |
| func (s *apiSession) instanceDelete(ctx context.Context, a apiAcct, app, host string) error { |
diff --git a/logs.go b/logs.go| new file mode 100644 |
| index 000000000..558f7fdb0 |
| --- /dev/null |
| +++ b/logs.go |
| @@ -0,0 +1,177 @@ |
| +// `carlos logs` — an app's merged log timeline with no box access (spec |
| +// 2026-08-10-carlos-logs-design.md §7, issues #131/#78): the app's own |
| +// output plus the platform's events about it (wakes, restarts, failures), |
| +// read through the console API. API mode ONLY, deliberately: the journal on |
| +// a box was never the member surface, and the box-local form is exactly the |
| +// gap this verb closes. |
| +package main |
| + |
| +import ( |
| + "context" |
| + "flag" |
| + "fmt" |
| + "io" |
| + "log" |
| + "net/url" |
| + "os" |
| + "sort" |
| + "strconv" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/applog" |
| +) |
| + |
| +const logsUsage = `usage: carlos logs -app <a> [flags] |
| + |
| +Shows the newest lines first-to-last: the app's own stdout/stderr ("app") |
| +merged with the platform's events about it ("platform" — wakes, restarts, |
| +failures with the reason). A site served from files alone has no app |
| +stream; platform events still appear. |
| + |
| +flags: |
| + -app <a> app whose logs to read (required) |
| + -since <t> window start: a duration back from now (15m, 1h, 24h) |
| + or an RFC3339 time (default 1h) |
| + -until <t> window end, same forms (default: now) |
| + -stream <s> only one stream: app | platform | edge | release |
| + -host <h> only one instance (its route host) |
| + -level <l> minimum level: debug | info | warn | error — only lines |
| + that CARRY a level; plain text never claims one |
| + -grep <re> only lines whose message matches this RE2 pattern |
| + -field k=v only structured lines with this field (repeatable) |
| + -n <n> at most this many lines (default 1000, max 5000) |
| + -f follow: poll for new lines every 2s until interrupted |
| + -account <a> account (sqid or name) the app belongs to (default "ops") |
| + -account-id <id> account sqid, matched as a sqid ONLY — supersedes -account |
| + -console <o> console to read through (default: the only logged-in one) |
| +` |
| + |
| +// logsFollowInterval is the -f poll cadence — the serial poll's own 2s, the |
| +// deployment's one convergence heartbeat. |
| +const logsFollowInterval = 2 * time.Second |
| + |
| +// logsMaxPolls bounds -f for the tests' benefit: 0 = forever (production). |
| +var logsMaxPolls = 0 |
| + |
| +func runLogs(args []string) { |
| + fs := flag.NewFlagSet("logs", flag.ExitOnError) |
| + app := fs.String("app", "", "app whose logs to read") |
| + since := fs.String("since", "1h", "window start: duration back from now, or RFC3339") |
| + until := fs.String("until", "", "window end (default: now)") |
| + stream := fs.String("stream", "", "only one stream: app | platform | edge | release") |
| + host := fs.String("host", "", "only one instance (route host)") |
| + level := fs.String("level", "", "minimum level: debug | info | warn | error") |
| + grep := fs.String("grep", "", "only messages matching this RE2 pattern") |
| + var fields multiFlag |
| + fs.Var(&fields, "field", "only structured lines with this k=v field (repeatable)") |
| + n := fs.Int("n", 0, "at most this many lines (default 1000, max 5000)") |
| + follow := fs.Bool("f", false, "follow: poll for new lines until interrupted") |
| + consoleFlag := fs.String("console", "", "console to read through (default: the only logged-in one)") |
| + acctRaw := fs.String("account", "ops", "account (sqid or name) the app belongs to") |
| + acctIDRaw := fs.String("account-id", "", "account sqid, matched as a sqid ONLY — supersedes -account") |
| + fs.Usage = func() { fmt.Fprint(os.Stderr, logsUsage) } |
| + fs.Parse(args) |
| + if *app == "" { |
| + log.Fatal(logsUsage) |
| + } |
| + |
| + q := url.Values{} |
| + for k, v := range map[string]string{ |
| + "since": *since, "until": *until, "stream": *stream, |
| + "host": *host, "level": *level, "grep": *grep, |
| + } { |
| + if v != "" { |
| + q.Set(k, v) |
| + } |
| + } |
| + for _, f := range fields { |
| + q.Add("field", f) |
| + } |
| + if *n > 0 { |
| + q.Set("limit", strconv.Itoa(*n)) |
| + } |
| + |
| + sess := requireAPISession("logs", *consoleFlag) |
| + acct := apiAccount(acctRaw, acctIDRaw) |
| + ctx := context.Background() |
| + res, err := sess.logsGet(ctx, acct, *app, q) |
| + if err != nil { |
| + log.Fatalf("logs: %v", err) |
| + } |
| + printLogRecords(stdout, res.Records) |
| + if len(res.Records) == 0 && !*follow { |
| + mutedf(stdout, "no log lines in this window — platform events appear on wakes and restarts; try -since 24h") |
| + } |
| + if res.Truncated { |
| + mutedf(stdout, "(showing the newest %d lines of this window — narrow it, or raise -n)", len(res.Records)) |
| + } |
| + if !*follow { |
| + return |
| + } |
| + |
| + // Follow: the same endpoint, strictly-after the cursor, forever. The |
| + // window params are dropped on purpose — a follow reads the PRESENT, |
| + // and the server treats a cursor request as unbounded-forward. |
| + fq := url.Values{} |
| + for _, k := range []string{"stream", "host", "level", "grep"} { |
| + if v := q.Get(k); v != "" { |
| + fq.Set(k, v) |
| + } |
| + } |
| + for _, f := range fields { |
| + fq.Add("field", f) |
| + } |
| + cursor := res.Cursor |
| + for polls := 0; logsMaxPolls == 0 || polls < logsMaxPolls; polls++ { |
| + time.Sleep(logsFollowInterval) |
| + if cursor != "" { |
| + fq.Set("cursor", cursor) |
| + } |
| + res, err := sess.logsGet(ctx, acct, *app, fq) |
| + if err != nil { |
| + // A follow rides through a blip; the next poll answers or the |
| + // member interrupts. Say it once per failure, quietly. |
| + mutedf(stdout, "(logs: %v — still following)", err) |
| + continue |
| + } |
| + printLogRecords(stdout, res.Records) |
| + if res.Cursor != "" { |
| + cursor = res.Cursor |
| + } |
| + } |
| +} |
| + |
| +// printLogRecords renders records one line each: UTC time, stream, host, |
| +// message (error lines red, warn yellow), then any structured fields dimmed. |
| +func printLogRecords(w io.Writer, recs []applog.Record) { |
| + c := colorFor(w) |
| + for _, r := range recs { |
| + msg := r.Msg |
| + switch r.Level { |
| + case applog.LevelError: |
| + msg = c.bad(msg) |
| + case applog.LevelWarn: |
| + msg = c.warn(msg) |
| + } |
| + line := c.muted(r.TS.UTC().Format("2006-01-02 15:04:05")) + " " + |
| + c.head("["+r.Stream+"]") + " " + c.muted(r.Host) + " " + msg |
| + keys := make([]string, 0, len(r.Fields)) |
| + for k := range r.Fields { |
| + keys = append(keys, k) |
| + } |
| + sort.Strings(keys) |
| + for _, k := range keys { |
| + line += " " + c.muted(k+"="+r.Fields[k]) |
| + } |
| + fmt.Fprintln(w, line) |
| + } |
| +} |
| + |
| +// multiFlag collects a repeatable string flag. |
| +type multiFlag []string |
| + |
| +func (m *multiFlag) String() string { return fmt.Sprint([]string(*m)) } |
| +func (m *multiFlag) Set(v string) error { |
| + *m = append(*m, v) |
| + return nil |
| +} |
diff --git a/logs_e2e_test.go b/logs_e2e_test.go| new file mode 100644 |
| index 000000000..04ed54406 |
| --- /dev/null |
| +++ b/logs_e2e_test.go |
| @@ -0,0 +1,159 @@ |
| +// End-to-end acceptance for `carlos logs` (#131/#78): a REAL console.Server |
| +// over a DirStore holding shipped log chunks, a device-flow login, then the |
| +// real runLogs driven in-process — the member's whole path from "my app is |
| +// 503ing" to reading the wake failure, with no box access anywhere. |
| +// TestE2EInstancesThroughConsole's shape. |
| +package main |
| + |
| +import ( |
| + "context" |
| + "net/http" |
| + "net/http/httptest" |
| + "path/filepath" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/applog" |
| + "github.com/carlosframework/platform/internal/console" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +func TestE2ELogsThroughConsole(t *testing.T) { |
| + useTempCredentials(t) |
| + clearDeploymentEnv(t) |
| + checkNoRegistry := requireNoRegistryTouch(t) |
| + |
| + tmp := t.TempDir() |
| + relStore := release.NewDirStore(filepath.Join(tmp, "store")) |
| + seed, pubB64, err := release.GenerateKey() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + priv, err := release.ParseSigningKey(seed, "e2e") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + pub, err := release.ParsePublicKey(pubB64, "e2e") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + cst, err := console.Open(filepath.Join(tmp, "console.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { cst.Close() }) |
| + |
| + var srv *console.Server |
| + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| + srv.ServeHTTP(w, r) |
| + })) |
| + t.Cleanup(ts.Close) |
| + srv, err = console.NewServer(console.Config{ |
| + Origin: ts.URL, |
| + Keymail: "http://keymail.invalid", // never dialled — see e2eDeviceLogin |
| + Registry: filepath.Join(tmp, "registry.db"), |
| + Releases: relStore, |
| + ReleasePub: pub, |
| + Key: priv, |
| + AppScheme: "http", |
| + AppsDomain: "oncarlos.test", |
| + }, cst) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + const addr = "paul@keymail.dev" |
| + if err := cst.AddMember(console.OpsAccountID, addr, console.RoleOwner, time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := cst.CreateAppIn(console.OpsAccountID, "titogo", addr, time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + token, expiresAt := e2eDeviceLogin(t, ts.URL, cst, addr, "e2e@laptop") |
| + credFile, err := loadCredentials() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + credFile.put(credential{Console: ts.URL, Address: addr, Token: token, ExpiresAt: expiresAt}) |
| + if err := credFile.save(); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + // A box shipped two chunks: routine output, then a wake failure — the |
| + // #131 case, as the shipping loop would have written it. |
| + now := time.Now().UTC() |
| + writeChunk := func(seq uint64, recs []applog.Record) { |
| + t.Helper() |
| + body, err := applog.EncodeChunk(recs) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + key := applog.ChunkKey(release.OpsAccount, "titogo", "box1", applog.HourOf(recs[0].TS), seq, len(body)) |
| + if _, err := relStore.Put(context.Background(), key, body, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + writeChunk(100, []applog.Record{ |
| + {TS: now.Add(-20 * time.Minute), Box: "box1", Host: "titogo.bab.oncarlos.test", |
| + Stream: applog.StreamApp, Msg: "listening on socket", Seq: 100, |
| + Fields: map[string]string{"request_id": "r1"}}, |
| + }) |
| + writeChunk(101, []applog.Record{ |
| + {TS: now.Add(-2 * time.Minute), Box: "box1", Host: "titogo.bab.oncarlos.test", |
| + Stream: applog.StreamPlatform, Level: applog.LevelError, Seq: 101, |
| + Msg: "wake failed: nothing is promoted for this instance's channel"}, |
| + }) |
| + |
| + // The plain read: both streams merged, oldest first, wake reason visible. |
| + buf := captureStdout(t) |
| + runLogs([]string{"-app", "titogo", "-account", "ops"}) |
| + out := buf.String() |
| + if !strings.Contains(out, "listening on socket") || !strings.Contains(out, "wake failed") { |
| + t.Fatalf("logs stdout missing the timeline:\n%s", out) |
| + } |
| + if strings.Index(out, "listening on socket") > strings.Index(out, "wake failed") { |
| + t.Fatalf("timeline out of order:\n%s", out) |
| + } |
| + if !strings.Contains(out, "[platform]") || !strings.Contains(out, "[app]") { |
| + t.Fatalf("stream tags missing:\n%s", out) |
| + } |
| + if !strings.Contains(out, "request_id=r1") { |
| + t.Fatalf("structured fields missing:\n%s", out) |
| + } |
| + |
| + // Filters reach the wire: -stream platform drops the app line. |
| + buf.Reset() |
| + runLogs([]string{"-app", "titogo", "-account", "ops", "-stream", "platform"}) |
| + if out := buf.String(); strings.Contains(out, "listening on socket") || !strings.Contains(out, "wake failed") { |
| + t.Fatalf("-stream platform: %q", out) |
| + } |
| + |
| + // -f follows: a line shipped between polls arrives; the poll cap is the |
| + // test's exit. |
| + logsMaxPolls = 2 |
| + t.Cleanup(func() { logsMaxPolls = 0 }) |
| + done := make(chan string, 1) |
| + go func() { |
| + fbuf := captureStdout(t) |
| + runLogs([]string{"-app", "titogo", "-account", "ops", "-f"}) |
| + done <- fbuf.String() |
| + }() |
| + time.Sleep(500 * time.Millisecond) // after the initial read, before poll 1 lands |
| + writeChunk(102, []applog.Record{ |
| + {TS: now.Add(-time.Second), Box: "box1", Host: "titogo.bab.oncarlos.test", |
| + Stream: applog.StreamApp, Msg: "fresh follow line", Seq: 102}, |
| + }) |
| + select { |
| + case out := <-done: |
| + if !strings.Contains(out, "fresh follow line") { |
| + t.Fatalf("follow missed the new line:\n%s", out) |
| + } |
| + case <-time.After(15 * time.Second): |
| + t.Fatal("follow did not finish its bounded polls") |
| + } |
| + |
| + // The whole flow ran through the console: nothing here may have touched |
| + // a local registry. |
| + checkNoRegistry() |
| +} |
diff --git a/main.go b/main.go| index 58b38f113..62106f5e4 100644 |
| --- a/main.go |
| +++ b/main.go |
| @@ -61,6 +61,8 @@ func main() { |
| runDeploy(args) |
| case "restart": |
| runRestart(args) |
| + case "logs": |
| + runLogs(args) |
| case "channels": |
| runChannels(args) |
| case "releases": |
| @@ -130,6 +132,8 @@ Your apps — ship and run software on a CARLOS deployment |
| env plain per-app config vars: set | unset | list |
| secrets sealed per-app secrets: set | unset | list | genkey |
| instances an app's instances: enable | create | list | delete |
| + logs an app's merged log timeline: its own output plus platform |
| + events (wakes, restarts, failures); -f follows live |
| ledger open ledgers: append | head | list | publish | blob | verify |
| vet check a release against the platform contract |