multiline

README

Multiline

multiline is a small, dependency-free Go library that aggregates log output spanning several physical lines — such as panic and exception stack traces — back into a single logical entry.

Many log shippers treat each newline as a separate record, which scatters a single stack trace across many entries. multiline recognizes the start and continuation patterns of common stack traces and re-joins them, while passing ordinary single-line logs straight through untouched.

Supported formats

• Go (panic: / goroutine dumps)
• Java / JVM (also claims Node.js traces with an error-class headline like TypeError:, which share the at ... frame shape)
• Node.js (bare Error: headlines and V8 stack-trace markers)
• Python (including chained exceptions)
• .NET
• Ruby
• Rust (panics, with or without backtrace)
• PHP
• Kubernetes CRI partial lines (via the cri subpackage, see CRI partial lines)

Install

go get github.com/JohanLindvall/multiline

How it works

You create an Aggregator[T] with an emitter callback and feed it lines one at a time with Add. Lines are grouped by a key (typically a container or stream id) so interleaved streams stay separate. When a multi-line entry completes — or you call Flush, FlushBefore or Stop — the emitter receives an Entry:

Field	Meaning
Text	The entry text; aggregated source lines are joined by "\n" (empty with WithoutText).
Texts	The retained source lines, one element per line — a view borrowed until the emitter returns; copy to retain.
Key	The key the lines were added under.
Match	Name of the format that aggregated the entry ("go", "java", …), or "" for a line passed through as-is.
When	Time of the entry's first source line, as passed to AddAt (zero for lines fed via Add).
Data	The T value passed to Add with the entry's first source line.
Lines	Number of source lines the entry represents (including lines dropped by the size caps).
Truncated	Set when the size caps dropped or cut lines belonging to this entry.

T is a generic payload you attach to each line — a log timestamp, a file offset for checkpointing, or struct{} if you don't need one. An Aggregator is not safe for concurrent use.

package main

import (
	"context"
	"fmt"

	"github.com/JohanLindvall/multiline"
)

func main() {
	ml := multiline.New(func(_ context.Context, e multiline.Entry[any]) error {
		if e.Match != "" {
			fmt.Printf("[stacktrace %s]\n%s\n\n", e.Match, e.Text)
		} else {
			fmt.Printf("[plain] %s\n", e.Text)
		}
		return nil
	})

	ctx := context.Background()
	for _, line := range []string{
		"server started",
		"panic: runtime error: invalid memory address or nil pointer dereference",
		"",
		"goroutine 1 [running]:",
		"main.handler(0x0)",
		"\t/app/main.go:42 +0x1d",
		"shutting down",
	} {
		if err := ml.Add(ctx, "key", line, nil); err != nil {
			panic(err)
		}
	}
	if err := ml.Stop(ctx); err != nil {
		panic(err)
	}
}

The runnable version lives in examples/simple (go run ./examples/simple).

Methods

• New[T](emit, opts...) — create an aggregator. Defaults to the built-in matcher covering patterns.All; pass WithMatcher to change it.
• Add(ctx, key, line, data) — feed one line. An empty key bypasses aggregation and emits immediately.
• AddAt(ctx, key, line, when, data) — like Add with an explicit time, which FlushBefore compares against and Entry.When reports. Pass the log's own timestamp to make time-based flushing robust when replaying old logs.
• Flush(ctx, key) — emit the pending group for one key. Call it when a stream ends, e.g. when its container terminates.
• FlushBefore(ctx, t) — emit pending groups last touched before t.
• Stop(ctx) — flush everything (oldest first) and reset for reuse.
• Pending(key), Len(), Bytes() — cheap gauges for monitoring: whether a key has buffered lines, how many keys do, and the total buffered text bytes.

Buffering latency

A line that matches any start pattern is buffered until the next line for its key arrives, so the last entry of an idle stream stays pending. Every real deployment should flush stale groups periodically:

ticker := time.NewTicker(time.Second)
defer ticker.Stop()
for range ticker.C {
	if err := ml.FlushBefore(ctx, time.Now().Add(-5*time.Second)); err != nil {
		...
	}
}

Bounding memory

By default a group grows until its match completes. Three options bound the aggregator; entries that lost lines to a cap are flagged Truncated (0 means unlimited):

• WithMaxLines(n) — retain at most n lines per group; further lines are dropped while matching continues normally.
• WithMaxBytes(n) — retain at most n text bytes per group; the crossing line is cut on a UTF-8 rune boundary and later lines are dropped.
• WithMaxGroups(n) — track at most n keys with pending lines; beyond it the least recently touched group is flushed. This guards against key cardinality explosions.

ml := multiline.New(emit,
	multiline.WithMaxLines(500),
	multiline.WithMaxBytes(64*1024),
	multiline.WithMaxGroups(10_000))

An emitter that writes lines to an io.Writer can consume Entry.Texts and pass WithoutText(), which skips joining aggregated lines into Entry.Text entirely — for a large capped trace, that saves a copy the size of the whole entry.

Custom formats

Matching is driven by declarative state machines in the patterns subpackage. The bundled definitions are exported (patterns.Go, patterns.Java, patterns.Python, patterns.DotNet, patterns.NodeJS, patterns.Ruby, patterns.Rust, patterns.PHP, collected in patterns.All), so you can compile a subset, or add your own set alongside them — its Name is what completed entries report as Match:

set := patterns.StateSet{Name: "tx", States: []patterns.State{
	{Name: patterns.StartState, Transitions: []patterns.Transition{
		{Pattern: `^BEGIN TX`, Next: "body"},
	}},
	{Name: "body", Transitions: []patterns.Transition{
		{Pattern: `^\s`, Next: "body"},
		{Pattern: `^(COMMIT|ROLLBACK)`, Next: "body"},
	}},
}}

matcher, err := patterns.Compile(append(patterns.All, set)...)
if err != nil {
	// invalid pattern, unknown state reference, ...
}
ml := multiline.New(emit, multiline.WithMatcher(matcher))

Notes:

• Every group begins at patterns.StartState; each set's transitions on it are merged into the shared start state, while all other state names are private to their set.
• A state is accepting unless NonTerminal is set. A group completes at the most recent line that landed in an accepting state; when it is flushed, those lines are emitted as one aggregated entry and any lines consumed after them are re-emitted individually. An aggregated entry always spans at least two source lines. Use NonTerminal for intermediate states that are not a valid stopping point.
• For full control you can implement the multiline.Matcher interface directly instead of compiling state sets.

A runnable example lives in examples/custom (go run ./examples/custom).

Kubernetes CRI partial lines

Container runtimes (containerd, CRI-O) write logs in the CRI format (<timestamp> <stream> P|F <content>) and split long application lines into P (partial) fragments closed by an F (full) line. Those fragments must be rejoined before stack-trace aggregation, and the fragments of one line are concatenated without a separator — so this is a separate stage, provided by the cri subpackage:

// Stack-trace aggregation over the rejoined lines...
traces := multiline.New(emitEntries)

// ...with CRI rejoining in front of it. Fragment runs are buffered per key
// and stream; rejoined lines reach the next stage with prefixes stripped,
// keyed "<key>/<stream>", stamped with their log timestamps.
logs := cri.New(traces.AddAt)

err := logs.Add(ctx, containerID, rawLine, data)

cri.New accepts the same WithMaxLines / WithMaxBytes / WithMaxGroups options to bound fragment buffering, and has its own Flush, FlushBefore, Stop (stop the upstream stage first) and Len / Bytes gauges. Lines that are not CRI-formatted pass through unmodified with a zero time, and cri.Parse is exported for callers that need the pieces. A tailer that already parses each line (to derive the key, or to route by stream) should feed the parse result to AddParsed instead of Add — the timestamp is then parsed exactly once per line on the whole path, roughly halving the per-line cost. Its (line, ok) parameters mirror Parse's results, so the non-CRI passthrough needs no separate call:

l, ok := cri.Parse(raw)
// ...derive key from l...
err := logs.AddParsed(ctx, key, raw, l, ok, data)
``` A runnable pipeline lives in
[examples/cri](examples/cri/main.go) (`go run ./examples/cri`). Docker's
json-file driver is a different format and needs JSON unwrapping instead.

## License

MIT — see [LICENSE](LICENSE).

Documentation

Overview

Package multiline aggregates log output spanning several physical lines — such as panic and exception stack traces — back into a single logical entry. Lines are fed one at a time to an Aggregator, grouped per key, and completed entries are handed to an Emitter callback.

The bundled matcher recognizes Go, Java (and Node.js), Python, .NET, Ruby, Rust and PHP stack traces; custom formats are declared in the patterns subpackage and selected with WithMatcher.

Index

• type Aggregator
  • func New[T any](emit Emitter[T], opts ...Option) *Aggregator[T]
  • func (a *Aggregator[T]) Add(ctx context.Context, key, line string, data T) error
  • func (a *Aggregator[T]) AddAt(ctx context.Context, key, line string, when time.Time, data T) error
  • func (a *Aggregator[T]) Bytes() int
  • func (a *Aggregator[T]) Flush(ctx context.Context, key string) error
  • func (a *Aggregator[T]) FlushBefore(ctx context.Context, t time.Time) error
  • func (a *Aggregator[T]) Len() int
  • func (a *Aggregator[T]) Pending(key string) bool
  • func (a *Aggregator[T]) Stop(ctx context.Context) error
• type Emitter
• type Entry
• type Matcher
• type Option
  • func WithClock(now func() time.Time) Option
  • func WithMatcher(matcher Matcher) Option
  • func WithMaxBytes(n int) Option
  • func WithMaxGroups(n int) Option
  • func WithMaxLines(n int) Option
  • func WithoutText() Option

Examples

• New

Types

type Aggregator

type Aggregator[T any] struct {
	// contains filtered or unexported fields
}

Aggregator joins log entries that span several lines into a single entry. Lines are grouped per key (see Aggregator.Add); when a group completes, the joined lines are passed to the emitter. Grouping is driven by a Matcher. An Aggregator is not safe for concurrent use.

func New

func New[T any](emit Emitter[T], opts ...Option) *Aggregator[T]

New creates an aggregator that hands completed entries to emit. By default it recognizes the stack-trace formats in patterns.All; pass WithMatcher to change that.

Example

package main

import (
	"context"
	"fmt"

	"github.com/JohanLindvall/multiline"
)

func main() {
	ml := multiline.New(func(_ context.Context, e multiline.Entry[struct{}]) error {
		fmt.Printf("match=%q lines=%d text=%q\n", e.Match, e.Lines, e.Text)
		return nil
	})

	ctx := context.Background()
	for _, line := range []string{
		"GET /healthz 200",
		"java.lang.NullPointerException: boom",
		"\tat com.example.Foo.bar(Foo.java:12)",
		"\tat com.example.Main.main(Main.java:5)",
		"shutting down",
	} {
		if err := ml.Add(ctx, "container-1", line, struct{}{}); err != nil {
			panic(err)
		}
	}
	if err := ml.Stop(ctx); err != nil {
		panic(err)
	}

}

Output:
match="" lines=1 text="GET /healthz 200"
match="java" lines=3 text="java.lang.NullPointerException: boom\n\tat com.example.Foo.bar(Foo.java:12)\n\tat com.example.Main.main(Main.java:5)"
match="" lines=1 text="shutting down"

func (*Aggregator[T]) Add

func (a *Aggregator[T]) Add(ctx context.Context, key, line string, data T) error

Add feeds a single line into the aggregator. The key groups related lines and is typically a container or stream id; an empty key bypasses aggregation and emits the line immediately. data rides along with the line and is handed back through the emitter. Add returns the first error produced by the emitter, if any. Entries fed via Add carry a zero Entry.When; staleness for Aggregator.FlushBefore is tracked with the aggregator clock only when a line is actually buffered, keeping the pass-through path free of clock reads.

func (*Aggregator[T]) AddAt

func (a *Aggregator[T]) AddAt(ctx context.Context, key, line string, when time.Time, data T) error

AddAt is Aggregator.Add with an explicit time for the line, which Aggregator.FlushBefore compares against and Entry.When reports — pass the log's own timestamp to make time-based flushing robust when replaying old logs. Times are assumed to be non-decreasing across calls. A zero when is allowed (Add uses one): staleness falls back to the aggregator clock and Entry.When stays zero.

func (*Aggregator[T]) Bytes

func (a *Aggregator[T]) Bytes() int

Bytes returns the total text bytes currently buffered across all groups — a cheap gauge for memory monitoring.

func (*Aggregator[T]) Flush

func (a *Aggregator[T]) Flush(ctx context.Context, key string) error

Flush emits the pending group for key, if any. Use it when a stream ends, for example when its container terminates.

func (*Aggregator[T]) FlushBefore

func (a *Aggregator[T]) FlushBefore(ctx context.Context, t time.Time) error

FlushBefore emits every pending group last touched before t, freeing groups that have gone stale. Call it periodically, e.g. from a ticker:

ml.FlushBefore(ctx, time.Now().Add(-5*time.Second))

Groups are kept in last-touched order, so flushing stops at the first group touched at or after t. It returns the first error produced while emitting.

func (*Aggregator[T]) Len

func (a *Aggregator[T]) Len() int

Len returns the number of keys with buffered lines.

func (*Aggregator[T]) Pending

func (a *Aggregator[T]) Pending(key string) bool

Pending reports whether key has buffered lines.

func (*Aggregator[T]) Stop

func (a *Aggregator[T]) Stop(ctx context.Context) error

Stop flushes every pending group, oldest first, leaving the aggregator empty and reusable. It returns the first error produced while emitting; groups emitted before the error are not re-delivered by a retry.

type Emitter

type Emitter[T any] func(ctx context.Context, entry Entry[T]) error

Emitter receives completed entries. Returning an error aborts the Add or flush call that produced the entry; lines already buffered in the same group are not re-delivered.

type Entry

type Entry[T any] struct {
	// Text is the entry text; for an aggregated entry the source lines are
	// joined by "\n". It is left empty when [WithoutText] is configured.
	Text string
	// Texts is the entry's retained source lines, one element per line. It is
	// a view borrowed from internal buffers, valid only until the emitter
	// returns — copy it (e.g. slices.Clone) to retain. Writing the elements
	// to an io.Writer avoids Text's joined allocation entirely (see
	// [WithoutText]).
	Texts []string
	// Key is the key the entry's lines were added under. It allows chaining
	// aggregation stages: an emitter can feed another Aggregator keyed by
	// entry.Key (see examples/cri).
	Key string
	// Match names the format that aggregated this entry (a patterns.StateSet
	// name such as "go" or "java"). It is "" when the line passed through
	// as-is.
	Match string
	// When is the time of the entry's first source line, as passed to AddAt.
	// Lines fed without a time (Add, or AddAt with a zero when) carry a zero
	// When.
	When time.Time
	// Data is the value passed to Add for the entry's first source line.
	Data T
	// Lines is the number of source lines the entry represents. It counts
	// lines dropped by WithMaxLines/WithMaxBytes, so it can exceed the number
	// of lines in Text.
	Lines int
	// Truncated is set when lines belonging to this entry were dropped or cut
	// by WithMaxLines/WithMaxBytes.
	Truncated bool
}

Entry is one completed log entry handed to the Emitter.

type Matcher

type Matcher interface {
	// Step applies line to the active states and returns the new active set,
	// plus the index of an accepting state the line landed in (-1 if none).
	// An empty next means line does not continue any active state. Step must
	// not retain or modify the active slice; the aggregator retains the
	// returned slice until the group's next line, so implementations must
	// return slices they will never mutate (shared immutable slices are
	// fine).
	Step(line string, active []int) (next []int, accepted int)
	// Format returns the format name reported as [Entry].Match for a group
	// that completed in the state at index.
	Format(index int) string
}

Matcher decides how successive lines are grouped. Implementations track matcher state as opaque int indices, where index 0 is the start state a new group begins from. The built-in implementation is patterns.StateMachine; implementations must be immutable or otherwise safe for the (single-threaded) use an Aggregator makes of them.

type Option

type Option func(*config)

Option configures an Aggregator at construction time.

func WithClock

func WithClock(now func() time.Time) Option

WithClock replaces time.Now as the source of the arrival times that Aggregator.Add stamps groups with (used by FlushBefore). Prefer Aggregator.AddAt to supply per-line times, e.g. log timestamps.

func WithMatcher

func WithMatcher(matcher Matcher) Option

WithMatcher selects a custom Matcher (typically a patterns.StateMachine built via patterns.Compile) instead of the built-in one.

func WithMaxBytes

func WithMaxBytes(n int) Option

WithMaxBytes caps the total text bytes retained in a single group. The line that crosses the limit is cut on a UTF-8 rune boundary, subsequent lines are dropped, and the resulting entry is flagged Truncated. A value <= 0 means unlimited.

func WithMaxGroups

func WithMaxGroups(n int) Option

WithMaxGroups caps the number of keys with pending lines. Adding a line for a new key beyond the cap flushes the least recently touched group first. A value <= 0 means unlimited. This guards against unbounded key cardinality (note that Go maps keep their high-water bucket memory, so the cap also bounds that); time-based flushing is Aggregator.FlushBefore.

func WithMaxLines

func WithMaxLines(n int) Option

WithMaxLines caps the number of lines retained in a single group. Further lines are dropped while matching continues normally, and the resulting entry is flagged Truncated. A value <= 0 means unlimited. This guards against an unterminated match growing without bound.

func WithoutText

func WithoutText() Option

WithoutText skips building Entry.Text (it is left empty), for emitters that consume Entry.Texts instead. This avoids joining an aggregated entry's lines into one string — for a large capped trace, a copy the size of the whole entry.