feat: rotation-aware log streaming engine + sdk bindings#755
Open
toksdotdev wants to merge 9 commits into
Open
feat: rotation-aware log streaming engine + sdk bindings#755toksdotdev wants to merge 9 commits into
toksdotdev wants to merge 9 commits into
Conversation
toksdotdev
force-pushed
the
feat/log-stream
branch
from
8531656 to
8668e79
Compare
- added `microsandbox::logs` module owning `log_stream` (returns a `Stream<LogEntry>`) and `read_logs` (sorted snapshot) over `exec.log` plus rotated siblings, `runtime.log`, and `kernel.log`. - `log_stream` supports cursor-based per-source resume and an optional `follow` mode driven by filesystem notify events with a poll fallback. - opaque `LogCursor` snapshots positions across active sources at emit time; pass it to `LogStreamStart::From` to resume. - introduced `MicrosandboxError::MissedRotation` for consumers that fall behind the on-disk retention window and `InvalidCursor` for malformed resume tokens. - removed the prior polling `sandbox/logs.rs`; callers (CLI, rust example, python/node/go sdks) now reach the api via `microsandbox::logs::*` and await the now-async `read_logs`. - added integration tests that drive the public api against synthetic on-disk log layouts via an `MSB_HOME` tempdir, plus engine-level unit tests for the parser, cursor, and multiplexer.
- replaced the bespoke poll loop in `msb logs` with `microsandbox::logs`: snapshots go through `read_logs` (chronologically sorted) and `--follow` consumes a `log_stream` with `start: Since(now)` after the snapshot completes. - dropped the local rotation tracking, JSON Lines reader, text-log timestamp parser, and the 200ms `tokio::time::sleep` poll; `--follow` now wakes on filesystem change notifications via the engine. - `--follow` surfaces `MicrosandboxError::MissedRotation` with a hint to restart, replacing the prior silent gap when retention outpaced the consumer. - new bridge helpers `mask_to_engine_sources` and `engine_entry_to_cli` translate the cli source flags into engine sources and re-encode non-utf8 bodies as base64 so `--json` keeps round-trip fidelity.
…sources onto SourceMask
- renamed `Multiplexer` to `LogEngine` across `crates/microsandbox/lib/logs/{mod,stream}.rs` and the engine module's docstring; the name is more descriptive of what the type does and matches the way the rest of the codebase talks about subsystem engines.
- moved the cli's free `mask_to_engine_sources(mask)` into a `SourceMask::to_engine_sources(self)` method so the conversion lives next to the data it operates on; updated the single caller and the unit test.
- exposed `Sandbox.logStream(opts)` and `SandboxHandle.logStream(opts)`. both return a `LogStream`: an `AsyncIterable<LogEntry>` (`for await...of` friendly) with a `recv()` method and `Symbol.asyncDispose` for cleanup, mirroring `MetricsStream`. - added `LogStreamOptions` with `sources`, `since`, `fromCursor`, `until`, and `follow`. `since` and `fromCursor` are mutually exclusive; passing both rejects at the napi boundary. - added an opaque `cursor: string` field on every `LogEntry` (snapshot and streaming paths). pass it back via `fromCursor` to resume strictly after that entry. - native side: new `JsLogStream` class wraps a `tokio::mpsc::Receiver<Result<LogEntry>>` fed by a background task that drains the engine stream; shared helper `spawn_log_stream` is reused by both `Sandbox` and `SandboxHandle`.
- exposed `Sandbox.log_stream(...)` and `SandboxHandle.log_stream(...)`. both return a `LogStream`: an async iterator (`async for entry in stream`) of `LogEntry` values that calls back into the engine via `pyo3_async_runtimes::tokio::future_into_py`, mirroring `MetricsStream`. - added keyword args `sources`, `since_ms`, `from_cursor`, `until_ms`, `follow`. `since_ms` and `from_cursor` are mutually exclusive; passing both raises `ValueError` at the binding boundary. - added an opaque `cursor: str` field on every `LogEntry` (snapshot and streaming paths). pass it back via `from_cursor` to resume strictly after that entry. - exported `LogEntry` and `LogStream` from the top-level `microsandbox` package — `LogEntry` was previously available only via the private `_microsandbox` module.
- exposed `Sandbox.LogStream(ctx, opts)` and `SandboxHandle.LogStream(ctx, opts)`. both return a `*LogStreamHandle` with `Recv(ctx)` (returns the next entry, or `nil, nil` when the stream ends) and `Close()`, mirroring `MetricsStreamHandle`. - added `LogStreamOptions` with `Sources`, `Since`, `FromCursor`, `Until`, `Follow`. `Since` and `FromCursor` are mutually exclusive; passing both rejects at the ffi boundary. - added a `Cursor` field on `LogEntry` (snapshot + streaming paths). pass it back via `FromCursor` to resume strictly after that entry. - native side: new ffi entry points `msb_sandbox_log_stream`, `msb_sandbox_handle_log_stream`, `msb_log_recv`, `msb_log_close`. background forwarder drains the engine stream into an unbounded `mpsc` channel; the go side blocks on `recv()` per call.
annotated `RotationConfig::{path, find_index, queue_all_rotated}`, `LogEngine::{new, snapshot_cursor, step, build_watcher, into_stream}`, and `Reader::{open, read_chunk, handle_rotation, replace_live}` plus the helper `StepResult` enum and `ReaderState` struct. one-to-three-sentence comments describing intent and side effects.
new `tests/log_stream_e2e.rs` boots an alpine sandbox under `#[msb_test]` and exercises the relay-tap → exec.log → consumer path end-to-end: - `logs_captures_exec_stdout_from_running_sandbox`: snapshot via `Sandbox::logs` after an exec; asserts the marker arrives with `source = Stdout` and a real `session_id`. - `log_stream_follow_catches_live_writes`: opens `log_stream(follow: true, start: Since(now))` first, then execs; asserts the marker arrives on the live stream within 5s (validating the notify-driven wake path against a real producer). - `log_stream_resume_from_cursor_excludes_replayed_entries`: execs A, captures the cursor of A's entry, execs B, opens a fresh stream from that cursor; asserts B is yielded but A is not replayed. each scenario uses a unique sandbox name and cleans up on success. runs only via `cargo nextest run -p microsandbox --test log_stream_e2e --run-ignored=only` (msb_test marks them ignored by default).
toksdotdev
force-pushed
the
feat/log-stream
branch
from
8668e79 to
cc3329e
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
summary
microsandbox::logsengine:log_stream(Stream<LogEntry>with cursor-based resume and notify-driven follow) plusread_logs(sorted snapshot, drains the same engine), coveringexec.logand rotated siblings,runtime.log, andkernel.log. introducedMicrosandboxError::MissedRotationfor consumers that fall behind retention, andInvalidCursorfor bad resume tokens.msb logsandmsb logs -fnow drive the engine instead of a bespoke poll loop. snapshot usesread_logs(chronologically sorted);--followuseslog_streamwithstart: Since(now)after the snapshot completes and surfacesMissedRotationwith a hint to restart.logStream/log_stream/LogStreamwrappers on bothSandboxandSandboxHandle. all three mirror the existing metrics-stream pattern (async iterator on node/python,Recv/Closehandle on go). everyLogEntrycarries an opaquecursorfield (snapshot + stream) for resume.since/fromCursorare mutually exclusive; passing both rejects at the binding boundary.MultiplexertoLogEngineand folded the cli'smask_to_engine_sourcesinto aSourceMask.to_engine_sources()method.test plan
cargo test -p microsandbox --lib logs::(34 unit tests across cursor / parser / stream / types)cargo test -p microsandbox --test log_stream(8 integration tests driving the public api through synthetic on-disk layouts viaMSB_HOMEtempdir)cargo test -p microsandbox-cli --lib commands::logs::(13 unit tests)cargo check --workspace+cargo clippy --workspacecleannpm run build:native && npm run build:ts && npm run typecheck && npm run test:unit(64 unit tests)maturin developthenpytest tests/go test ./sdk/go/...(4 new bridge tests aroundCursorfield +LogStreamOptions)MSB_HOME=$tmpdir msb logs <name>with hand-laidexec.logexercises the snapshot path; appending entries with future timestamps validates-fpicks them up via fs notify