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) 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
}
Options tunes a Bus. Zero fields take the package defaults.
Source Files