Documentation
¶
Overview ¶
Package logbus fans one workload's output out to any number of live subscribers, with a bounded replay backlog for late joiners.
Every executor driver needs the same thing: a workload writes to a pipe, zero or more Stream consumers want the bytes, consumers come and go at arbitrary times, and none of them may ever back-pressure the workload. The logic is small but the concurrency is subtle enough that it was worth getting wrong only once — pkg/executor/localprocess shipped a "send on closed channel" panic that took down the whole UI server when a viewer closed a browser tab mid-run. This package is that logic, extracted so drivers added later (containers, remote agents) inherit the fix instead of re-deriving it.
Three invariants define the contract:
Emit never blocks. A subscriber whose buffer is full has chunks dropped, because blocking here would propagate back into the workload's own writes once its pipe filled — one stalled log viewer would stall the agent. Consumers detect loss via gaps in LogLine.Seq.
Send and close are serialised per subscriber. A plain non-blocking select cannot express this: when both a send and a close are ready Go picks at random, so the send can land on an already-closed channel.
Replay is captured under the same lock that registers the subscriber, so a chunk emitted concurrently with Subscribe is either in the backlog or delivered live — never both, never neither.
Index ¶
Constants ¶
const ( // DefaultSubscriberBuffer is how many chunks a single consumer may fall // behind before its chunks start being dropped. DefaultSubscriberBuffer = 512 // DefaultReplayBytes bounds the output retained for consumers that // subscribe after the workload started. It only has to cover the gap // between Start returning and Stream being called, so it is small. DefaultReplayBytes = 64 << 10 )
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Bus ¶
type Bus struct {
// contains filtered or unexported fields
}
Bus is the fan-out point for one workload's output. The zero value is not usable; call New.
func New ¶
func New(handleID string, stream executor.StreamName, opts Options) *Bus
New returns a Bus that stamps chunks with handleID and stream.
func (*Bus) Close ¶
func (b *Bus) Close()
Close releases every subscriber. Idempotent.
Drivers must call it only after the workload's terminal status has been recorded: a consumer that sees its channel close is entitled to read a terminal Status immediately, and executor.Run depends on that ordering.
func (*Bus) Emit ¶
Emit publishes one output chunk. Empty chunks are ignored. Emitting after Close is a no-op rather than a panic: a driver's output pump and its process reaper race by nature, and the pump losing that race must not take the process down.
func (*Bus) Redactor ¶ added in v0.0.2
Redactor returns the set installed at construction, or nil when the workload carries no credential.
It exists so a second consumer of the same workload's output — an interactive attach session (Task 20265) — can scrub with the identical set rather than rebuild one from a Spec it does not have. The bus is where the driver already recorded "these bytes must never appear in this handle's output", so it is the honest place to ask, and asking keeps the two paths from drifting: a credential added to the log filter is filtered in the terminal by construction.
The set is immutable after New, so handing it out shares no mutable state.
func (*Bus) Subscribe ¶
Subscribe returns a channel that first replays the bounded backlog, then delivers live chunks, then closes when the bus is closed.
Cancelling ctx unsubscribes this consumer only — the workload and every other consumer are unaffected. Subscribing to an already-closed bus is valid and yields the backlog followed by a close, so a caller that arrives after a short-lived workload finished still sees its output.
type Options ¶
type Options struct {
// SubscriberBuffer overrides DefaultSubscriberBuffer.
SubscriberBuffer int
// ReplayBytes overrides DefaultReplayBytes.
ReplayBytes int
// Now overrides the clock, for deterministic tests.
Now func() time.Time
// Redact removes the workload's own leased credentials from its output.
// Nil means the workload holds none, which is the common case and costs
// nothing.
//
// It belongs here, on the bus, rather than on each consumer: every path
// that output takes out of a driver — the live-log room, the replay
// backlog a late subscriber reads, the buffer executor.Run persists —
// starts at Emit. Filtering at any one consumer would leave the others.
Redact *redact.Set
}
Options tunes a Bus. Zero fields take the package defaults.
Source Files