core

package module
v0.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 27, 2026 License: MIT Imports: 13 Imported by: 0

Documentation

Overview

Package core is crier's engine: the log record model, the ingestion pipeline, the buffer, and the exporter contract.

It has zero third-party runtime dependencies (NFR1) so that embedding crier as a library does not drag an exporter's SDK into the consumer's build. Exporters live in their own modules under exporters/.

Delivery semantics

Delivery is at-least-once, with no ordering guarantee across batches (ADR-0009). A record accepted by the pipeline may be exported more than once; callers must tolerate duplicates. Acceptance is not delivery: the receiver's 202 response means a record was admitted to the buffer, not that any backend has stored it.

Index

Examples

Constants

View Source 
const (
	DefaultBufferCapacity = 10_000
	DefaultBatchSize      = 512
	DefaultBatchWindow    = 5 * time.Second
)

In-memory buffer defaults (ADR-0002).

View Source 
const (
	// DefaultMaxDistinctValues is how many distinct values one attribute key
	// may carry within a window before the key is capped.
	DefaultMaxDistinctValues = 1000
	// DefaultMaxTrackedKeys bounds how many keys the guard tracks at once.
	DefaultMaxTrackedKeys = 256
	// DefaultCardinalityWindow is how long observations stay relevant.
	DefaultCardinalityWindow = 10 * time.Minute
	// DefaultCardinalityMark replaces a value once its key is capped.
	DefaultCardinalityMark = "…[high cardinality]"
)

Cardinality guard defaults (ADR-0010, FR12).

View Source 
const (
	// DefaultFailureThreshold is how many consecutive failures open a circuit.
	DefaultFailureThreshold = 5
	// DefaultCooldown is how long a circuit stays open before a probe.
	DefaultCooldown = 30 * time.Second
	// DefaultHalfOpenSuccesses is how many probes must succeed to close it.
	DefaultHalfOpenSuccesses = 1
)

Circuit breaker defaults.

View Source 
const (
	DefaultMaxAttributes  = 128
	DefaultMaxKeyBytes    = 256
	DefaultMaxValueBytes  = 8 * 1024
	DefaultMaxBodyBytes   = 64 * 1024
	DefaultTruncationMark = "…[truncated]"
	// DefaultUnsupportedMark replaces a value whose type cannot be bounded
	// cheaply. See Limits.Apply.
	DefaultUnsupportedMark = "…[unsupported value type]"
)

Default input limits (ADR-0010, FR12). Chosen to be generous for real applications and still bounded: the point is that no single record can cost unbounded memory, not that legitimate logs get clipped.

View Source 
const (
	// DefaultRetryAttempts is the total number of attempts, first included.
	DefaultRetryAttempts = 4
	// DefaultInitialBackoff is the base of the exponential backoff.
	DefaultInitialBackoff = 100 * time.Millisecond
	// DefaultMaxBackoff caps one backoff interval.
	DefaultMaxBackoff = 5 * time.Second
)

Retry defaults. Four attempts over roughly a second of backoff is enough to ride out a collector restart or a load-balancer reconvergence, and short enough that a batch does not sit on a dispatch worker while a destination is genuinely down — that case belongs to the circuit breaker, not to retry.

View Source 
const DefaultExportTimeout = 30 * time.Second

DefaultExportTimeout bounds one destination's Export call when FanOut is given no Timeout of its own.

View Source 
const DefaultExportWorkers = 4

DefaultExportWorkers is how many batches a Dispatcher keeps in flight when none is configured.

It is small on purpose. Workers bound concurrency, not memory: a batch in flight has already left the buffer, so more workers do not buy more capacity — they buy tolerance for destinations that are slow but working. Four is enough to keep a healthy destination busy while another rides out a backoff, and low enough that a total outage does not park a large slice of the buffer in goroutines that are all waiting on the same dead socket.

View Source 
const DefaultMetricLabelCap = 512

DefaultMetricLabelCap bounds how many distinct label values CountingMetrics will track for any one metric before folding the rest into OverflowLabel.

View Source 
const DefaultSkewThreshold = time.Minute

DefaultSkewThreshold is how far a source's asserted Timestamp may sit from the ObservedTimestamp before the deviation is reported. Ordinary scheduling and network delay produce small differences on every record; reporting those would bury the signal that actually matters — a source whose clock is wrong.

View Source 
const OverflowLabel = "<other>"

OverflowLabel replaces a label value once a metric has reached its label cap. Its presence in a snapshot means the real value was discarded, not that a source is literally named this.

View Source 
const RedactionMark = "[REDACTED]"

RedactionMark replaces every masked span and value. It is a fixed string so that the presence of redaction is greppable, and so the marker itself can never leak a hint about what it replaced (no length, no prefix, no hash).

View Source 
const SampleNothing = -1.0

SampleNothing is the SampleRate that discards every eligible record. It exists because a zero SampleRate means "unset", and a config that cannot express "none" pushes operators into approximating it with 0.0000001.

View Source 
const UnattributedSource = "<unattributed>"

UnattributedSource is the bucket for records that reach admission without an attested identity. It exists so such records are still accounted for rather than sharing an empty-string key with each other invisibly; in standalone mode authentication means it should stay empty, and a non-zero count there is worth investigating.

Variables

View Source 
var (
	// ErrBufferFull means the buffer as a whole is at capacity. Under the
	// default policy the receiver answers 503.
	ErrBufferFull = errors.New("buffer full")
	// ErrBufferClosed means the buffer is shutting down and will accept
	// nothing further. Records already inside are still drained.
	ErrBufferClosed = errors.New("buffer closed")
	// ErrSourceQuotaExhausted means this source's fair share is spent while
	// the buffer still has room (ADR-0011). Distinct from ErrBufferFull on
	// purpose: the operator response is entirely different.
	ErrSourceQuotaExhausted = errors.New("source quota exhausted")
)

Buffer errors. They are distinct because the receiver maps them to different responses and an operator reads them as different problems (ADR-0002, ADR-0011).

View Source 
var BodyPatterns = []string{

	`(?i)\b(?:bearer|basic|digest)\s+([A-Za-z0-9._~+/=-]{8,})`,

	`\b(eyJ[A-Za-z0-9_-]{4,}\.[A-Za-z0-9_-]{4,}\.[A-Za-z0-9_-]{4,})\b`,

	`(?i)\b(?:pass(?:word|wd|phrase)?|secret|token|api[_.-]?key|auth|credential)` +
		`\s*[=:]\s*"?([^"\s,;]{3,})"?`,

	`\b((?:AKIA|ASIA)[0-9A-Z]{16})\b`,

	`(?s)(-----BEGIN[A-Z ]*PRIVATE KEY-----.*?-----END[A-Z ]*PRIVATE KEY-----)`,
}

BodyPatterns are the value shapes masked inside free text by default.

Each captures the credential itself in group 1 where possible, so the surrounding context ("Authorization: Bearer ") survives and the line stays readable — a redacted log that no longer says what happened has traded one problem for another.

View Source 
var ErrCircuitOpen = errors.New("circuit open")

ErrCircuitOpen means the breaker rejected the batch without calling the destination, because recent calls failed and the cooldown has not elapsed.

It is a distinct sentinel because the caller's response differs: a batch refused by an open circuit never touched the network, so it costs nothing to have tried, and retrying it in a loop only spends the export deadline (ADR-0013).

View Source 
var ErrPermanent = errors.New("permanent export failure")

ErrPermanent marks a failure that retrying cannot fix: a malformed payload, a rejected credential, a 4xx other than 429. A retry decorator that sees it must give up immediately rather than spend its budget — and delay every batch queued behind it — on a batch the backend will never accept (ADR-0013).

Exporters signal it by wrapping: fmt.Errorf("%w: %v", ErrPermanent, err).

View Source 
var ErrRedactionFailed = errors.New("redaction failed")

ErrRedactionFailed reports that a record could not be redacted. The record is dropped, never exported (ADR-0014).

View Source 
var SensitiveKeySubstrings = []string{
	"pass",
	"secret",
	"token",
	"apikey", "api_key", "api-key", "api.key",
	"authoriz",
	"authentic",
	"credential",
	"privatekey", "private_key", "private-key", "private.key",
	"sessionid", "session_id", "session-id", "session.id",
	"cookie",
	"signature",
}

SensitiveKeySubstrings are the attribute-key fragments masked by default.

Substrings rather than regexes, matched case-insensitively, because the field is called "authorization" in one service and "auth_header" in the next — and because a substring scan is roughly an order of magnitude cheaper than a regex per attribute per record. The benchmark made that difference the dominant cost of the whole pipeline.

Separators are enumerated instead of matched with a character class for the same reason.

Functions

func IsPermanent 

func IsPermanent(err error) bool

IsPermanent reports whether err is a permanent failure.

Types

type BatchMutator 

type BatchMutator interface {
	Exporter

	// MutatesBatch reports whether Export modifies batch or its records.
	MutatesBatch() bool
}

BatchMutator is implemented by an exporter that writes to the batch it is given. FanOut hands such an exporter its own clone and leaves every other exporter sharing the original.

The default is therefore that a batch is read-only for the duration of Export, which is what makes concurrent dispatch safe without copying (ADR-0016). Cloning for everyone would put a deep copy of every batch on the hot path to defend against something almost no exporter does; cloning for nobody would make one badly behaved exporter corrupt its siblings' data in a way that shows up as a data race under load and nowhere else.

A decorator must forward this from the exporter it wraps, or a leaf's declaration becomes invisible the moment it is composed.

type BufferStore 

type BufferStore interface {
	// Enqueue admits one record. It returns ErrBufferFull, ErrBufferClosed,
	// or ctx.Err(); under DropPolicyDropOldest a successful Enqueue may have
	// evicted an older record, which is counted rather than reported.
	Enqueue(ctx context.Context, rec LogRecord) error

	// DequeueBatch returns the next batch, blocking until the batch is full,
	// the batch window expires, or ctx is done.
	//
	// After Close it returns whatever remains, one batch at a time, and then
	// ErrBufferClosed. That is what makes a bounded drain possible: the
	// consumer keeps calling until it sees ErrBufferClosed or runs out of
	// time (ADR-0015).
	DequeueBatch(ctx context.Context) ([]LogRecord, error)

	// Depth reports how many records are currently held.
	Depth() int

	// Close stops admission. It is idempotent and never discards records that
	// are already inside — draining them is the consumer's job.
	Close() error
}

BufferStore holds records between ingestion and export (FR3, ADR-0002).

It is an interface so a durable, WAL-backed implementation can replace the in-memory one without changing callers. None ships in the MVP; the seam exists so that adding one later is not a rewrite. It mirrors the Store split already validated in moat.

Implementations must be safe for concurrent use by many producers and many consumers.

type CardinalityGuard 

type CardinalityGuard struct {
	// MaxDistinctValues per key. Zero means DefaultMaxDistinctValues,
	// negative disables the guard.
	MaxDistinctValues int
	// MaxTrackedKeys bounds tracked keys. Zero means DefaultMaxTrackedKeys.
	MaxTrackedKeys int
	// Window is how long an observation counts for. Zero means
	// DefaultCardinalityWindow.
	Window time.Duration
	// Mark replaces a capped key's value. Zero means DefaultCardinalityMark.
	Mark string
	// Now supplies the clock. Nil means time.Now.
	Now func() time.Time
	// Metrics receives cap events. Nil discards.
	Metrics Metrics
	// contains filtered or unexported fields
}

CardinalityGuard replaces attribute values whose key has carried too many distinct values recently (ADR-0010).

The problem it solves is not primarily an attack. Request IDs, user IDs, and timestamps used as attribute values explode the series count of essentially every observability backend, degrading it and inflating its bill, and they get there through ordinary carelessness far more often than through malice.

Its own state is bounded

A guard that keeps a set of every value it has seen is the memory leak it exists to prevent. Three bounds apply:

  • MaxTrackedKeys caps how many keys are tracked. Past it, new keys are not tracked at all — they pass through unguarded rather than displacing a key already known to be a problem.
  • MaxDistinctValues caps the set per key. On reaching it the key is marked capped and its value set is released: once a key is capped, knowing which values it held buys nothing.
  • Window ages observations out, in two generations. A key that was noisy an hour ago and is quiet now recovers on its own.

The zero value is usable and safe for concurrent use.

func (*CardinalityGuard) Apply 

func (g *CardinalityGuard) Apply(rec *LogRecord)

Apply guards the record's attributes and resource attributes in place.

func (*CardinalityGuard) Observe 

func (g *CardinalityGuard) Observe(key, value string) bool

Observe records that key carried value and reports whether the key is now capped, meaning the caller should substitute a marker.

It is exported so a receiver can guard values it handles outside a LogRecord, and so the behaviour is directly testable.

func (*CardinalityGuard) TrackedKeys 

func (g *CardinalityGuard) TrackedKeys() int

TrackedKeys reports how many keys the guard currently holds state for. Exported for tests and for operators who want to see the guard's own bound being respected.

type CircuitBreaker 

type CircuitBreaker struct {
	// contains filtered or unexported fields
}

CircuitBreaker stops sending to a destination that keeps failing, so an unhealthy backend stops consuming dispatch workers that healthy ones need (FR6, ADR-0013).

It is composed per destination, innermost — FanOut(Retry(CircuitBreaker(e))). One breaker shared across destinations would let one broken backend silence the others, which is the coupling this whole layering exists to remove.

When every destination's breaker is open the pipeline is in the degraded state ADR-0015 surfaces through readiness; Open reports this breaker's part of that.

Safe for concurrent use.

func NewCircuitBreaker 

func NewCircuitBreaker(cfg CircuitBreakerConfig) (*CircuitBreaker, error)

NewCircuitBreaker validates cfg and returns the decorator.

func (*CircuitBreaker) Export 

func (cb *CircuitBreaker) Export(ctx context.Context, batch []LogRecord) error

Export sends batch unless the circuit is open.

func (*CircuitBreaker) MutatesBatch 

func (cb *CircuitBreaker) MutatesBatch() bool

MutatesBatch forwards the wrapped exporter's declaration (ADR-0016).

func (*CircuitBreaker) Name 

func (cb *CircuitBreaker) Name() string

Name returns the label this breaker's metrics carry.

func (*CircuitBreaker) Open 

func (cb *CircuitBreaker) Open() bool

Open reports whether the breaker is currently refusing calls. A half-open breaker is not open: it is admitting probes, so the destination is not yet written off.

func (*CircuitBreaker) Shutdown 

func (cb *CircuitBreaker) Shutdown(ctx context.Context) error

Shutdown releases the wrapped exporter.

func (*CircuitBreaker) State 

func (cb *CircuitBreaker) State() CircuitState

State reports the current state.

type CircuitBreakerConfig 

type CircuitBreakerConfig struct {
	// Name labels this destination's circuit metric. Required, and should
	// match the fan-out Destination name.
	Name string

	// Exporter is the destination being guarded.
	Exporter Exporter

	// FailureThreshold is how many consecutive failures open the circuit.
	// Zero means DefaultFailureThreshold.
	FailureThreshold int

	// Cooldown is how long the circuit stays open before admitting a probe.
	// Zero means DefaultCooldown.
	Cooldown time.Duration

	// HalfOpenSuccesses is how many consecutive probe successes close the
	// circuit again. Zero means DefaultHalfOpenSuccesses.
	HalfOpenSuccesses int

	// Metrics receives circuit transitions. Nil discards.
	Metrics Metrics

	// Now supplies the current time. Nil means time.Now; tests override it so
	// the cooldown can be asserted without waiting for it.
	Now func() time.Time
}

CircuitBreakerConfig configures a CircuitBreaker. Build one with NewCircuitBreaker, which validates eagerly (NFR4).

type CircuitReporter 

type CircuitReporter interface {
	// Name is the destination's metrics label.
	Name() string
	// Open reports whether calls are being refused right now.
	Open() bool
}

CircuitReporter is a destination that knows whether it is currently refusing calls. *CircuitBreaker implements it.

The Dispatcher uses it for one thing: when every destination is refusing, the pipeline is degraded, which readiness must reflect (ADR-0015).

type CircuitState 

type CircuitState int

CircuitState is a breaker's current state. 

const (
	// CircuitClosed passes every call through. The healthy state.
	CircuitClosed CircuitState = iota
	// CircuitOpen rejects every call with ErrCircuitOpen.
	CircuitOpen
	// CircuitHalfOpen lets one probe through to find out whether the
	// destination has recovered.
	CircuitHalfOpen
)

func (CircuitState) String 

func (s CircuitState) String() string

String implements fmt.Stringer.

type CountingMetrics 

type CountingMetrics struct {
	// LabelCap bounds distinct label values per metric. Zero means
	// DefaultMetricLabelCap. Set before first use.
	LabelCap int
	// contains filtered or unexported fields
}

CountingMetrics is an in-memory Metrics implementation, intended for tests and as the source for the standalone binary's metrics endpoint.

Label values are bounded (LabelCap): a metrics implementation that grows a map key per distinct client-supplied string is the same unbounded-cardinality leak the pipeline's own guard exists to prevent (ADR-0010), and IdentityDiscrepancy in particular takes a value straight from an untrusted caller. Past the cap, values collapse into OverflowLabel.

Safe for concurrent use.

func (*CountingMetrics) AttributeDropped 

func (m *CountingMetrics) AttributeDropped(key string)

AttributeDropped implements Metrics.

func (*CountingMetrics) AttributeTruncated 

func (m *CountingMetrics) AttributeTruncated(key string)

AttributeTruncated implements Metrics.

func (*CountingMetrics) BufferDepth 

func (m *CountingMetrics) BufferDepth(depth int)

BufferDepth implements Metrics.

func (*CountingMetrics) CardinalityCapped 

func (m *CountingMetrics) CardinalityCapped(key string)

CardinalityCapped implements Metrics.

func (*CountingMetrics) CircuitStateChanged 

func (m *CountingMetrics) CircuitStateChanged(exporter string, open bool)

CircuitStateChanged implements Metrics.

func (*CountingMetrics) ClockSkew 

func (m *CountingMetrics) ClockSkew(source string, deviation time.Duration)

ClockSkew implements Metrics.

func (*CountingMetrics) DeprecatedWireVersion 

func (m *CountingMetrics) DeprecatedWireVersion(version string)

DeprecatedWireVersion implements Metrics.

func (*CountingMetrics) ExportDegraded 

func (m *CountingMetrics) ExportDegraded(degraded bool)

ExportDegraded implements Metrics.

func (*CountingMetrics) ExportLatency 

func (m *CountingMetrics) ExportLatency(exporter string, d time.Duration)

ExportLatency implements Metrics.

func (*CountingMetrics) ExportRetried 

func (m *CountingMetrics) ExportRetried(exporter string)

ExportRetried implements Metrics.

func (*CountingMetrics) IdentityDiscrepancy 

func (m *CountingMetrics) IdentityDiscrepancy(_, actual string)

IdentityDiscrepancy implements Metrics.

func (*CountingMetrics) RecordsDropped 

func (m *CountingMetrics) RecordsDropped(source string, reason DropReason, n int)

RecordsDropped implements Metrics.

func (*CountingMetrics) RecordsExported 

func (m *CountingMetrics) RecordsExported(exporter string, n int)

RecordsExported implements Metrics.

func (*CountingMetrics) RecordsFiltered 

func (m *CountingMetrics) RecordsFiltered(source string, n int)

RecordsFiltered implements Metrics.

func (*CountingMetrics) RecordsIngested 

func (m *CountingMetrics) RecordsIngested(source string, n int)

RecordsIngested implements Metrics.

func (*CountingMetrics) Snapshot 

func (m *CountingMetrics) Snapshot() Snapshot

Snapshot returns a copy of the current counters.

func (*CountingMetrics) TimestampMissing 

func (m *CountingMetrics) TimestampMissing(source string)

TimestampMissing implements Metrics.

type Crier 

type Crier struct {
	// contains filtered or unexported fields
}

Crier is the engine embedded in a host application (FR9): the same pipeline, buffer and export layer the daemon runs, with no receiver.

There is no receiver because in embedded mode there is nothing to receive from — the host calls Log directly, and owns the trust boundary itself.

Safe for concurrent use.

func New 

func New(opts Options) (*Crier, error)

New assembles the engine and starts exporting.

Example

ExampleNew shows the embedded engine: the same pipeline and export layer the daemon runs, with no receiver, because the host application calls it directly and owns the trust boundary itself. 

crier, err := New(Options{
	ServiceName:    "task-api",
	ServiceVersion: "1.4.0",
	Exporters:      map[string]Exporter{"primary": noopExporter{}},

	// Limits apply here exactly as they do to the HTTP receiver: a bug in
	// the host produces the same unbounded attribute map as a malicious
	// client (ADR-0010).
	Limits: Limits{MaxAttributes: 64},
})
if err != nil {
	panic(err)
}

if logErr := crier.Log(context.Background(), LogRecord{
	Severity:   SeverityError,
	Body:       "database unreachable",
	Attributes: map[string]any{"attempt": 3},
}); logErr != nil {
	panic(logErr)
}

// Loss at shutdown is permitted; silent loss is not, so the drain reports
// what it did (ADR-0015).
summary, err := crier.Shutdown(context.Background())
if err != nil {
	panic(err)
}
fmt.Println("clean drain:", summary.Clean())

Output:
clean drain: true

func (*Crier) Depth 

func (c *Crier) Depth() int

Depth reports how many records are buffered and not yet exported.

func (*Crier) Health 

func (c *Crier) Health() *Health

Health reports liveness and readiness, for a host that exposes its own health endpoints and wants crier's state in them.

func (*Crier) Log 

func (c *Crier) Log(ctx context.Context, rec LogRecord) error

Log runs one record through the pipeline and admits it.

It returns when the record is buffered, not when it is exported: export happens on the dispatcher's own workers, so a host application's latency never depends on whether a backend is healthy (ADR-0001, ADR-0009).

A filtered record returns nil. It was handled exactly as configured.

func (*Crier) LogBatch 

func (c *Crier) LogBatch(ctx context.Context, batch []LogRecord) (accepted int, err error)

LogBatch admits several records, returning how many were accepted and the first error.

It does not stop at the first failure: one record hitting a limit says nothing about the next.

func (*Crier) Shutdown 

func (c *Crier) Shutdown(ctx context.Context) (DrainSummary, error)

Shutdown stops accepting records, drains what is buffered within ctx, and releases the exporters.

The summary is returned rather than logged, because this package has no logger and should not acquire one: what the host does with the number of records lost is the host's decision. Ignoring it is also a decision, and one the return value at least makes visible (ADR-0015).

type Destination 

type Destination struct {
	// Name labels this destination's metrics. Required, and unique within a
	// fan-out.
	Name string

	// Exporter is the composed chain for this destination, retry and circuit
	// breaking included: FanOut(Retry(CircuitBreaker(e))), never the reverse
	// (ADR-0013).
	Exporter Exporter
}

Destination pairs an exporter with the label its counters carry.

The name is required rather than derived from the exporter's type, because two destinations are frequently the same type — two OTLP collectors, one per region — and counters that both report as "otlp" hide exactly the failure an operator is looking for.

type Dispatcher 

type Dispatcher struct {
	// contains filtered or unexported fields
}

Dispatcher drains the buffer and exports, on a bounded pool of workers (ADR-0016).

It is the only component that knows whether a batch reached anywhere at all, so it owns the two accounting duties on the export side: counting records that reached no destination (ADR-0015), and reporting the degraded state that readiness reflects.

func NewDispatcher 

func NewDispatcher(cfg DispatcherConfig) (*Dispatcher, error)

NewDispatcher validates cfg and returns the dispatcher. It does not start any workers; call Start.

func (*Dispatcher) Degraded 

func (d *Dispatcher) Degraded() bool

Degraded reports whether every destination is currently refusing calls — the state ADR-0015 requires readiness to reflect, so an orchestrator takes the instance out of service instead of feeding it records it cannot export.

It is false when any destination might still accept a batch, and false when no destination is guarded by a breaker at all: an unguarded destination has never told anyone it is unusable, and guessing that it is would take a healthy instance out of service.

Calling it is how the degraded metric stays current, so a readiness probe polling it is doing double duty.

func (*Dispatcher) Draining 

func (d *Dispatcher) Draining() bool

Draining reports whether Shutdown has begun.

Readiness reflects it: an instance that has stopped accepting records is not ready, whatever else is true of it (ADR-0015).

func (*Dispatcher) OpenCircuits 

func (d *Dispatcher) OpenCircuits() []string

OpenCircuits names the destinations currently refusing calls, so an operator reading a readiness failure is told which one to look at.

func (*Dispatcher) Shutdown 

func (d *Dispatcher) Shutdown(ctx context.Context) (DrainSummary, error)

Shutdown stops admission, drains what is buffered, and releases the exporters (FR10, ADR-0015).

ctx bounds the drain. Records still unexported when it expires are counted as DropShutdownTimeout and lost: bounding shutdown is a hard requirement in any orchestrated environment, so loss at shutdown is permitted — silent loss is not.

The context passed to Start must stay alive for the duration, or the workers stop before they have drained.

func (*Dispatcher) Start 

func (d *Dispatcher) Start(ctx context.Context)

Start launches the workers. It is safe to call once; later calls are no-ops.

ctx governs the workers' lifetime for cancellation, but it is not how you stop a Dispatcher: cancelling it abandons whatever is in the buffer. Shutdown drains first (ADR-0015).

type DispatcherConfig 

type DispatcherConfig struct {
	// Buffer is drained by the workers. Required.
	Buffer BufferStore

	// Exporter receives each batch — normally the *FanOut, with retry and
	// circuit breaking already composed inside it (ADR-0013). Required.
	Exporter Exporter

	// Workers is how many batches may be in flight at once. Zero means
	// DefaultExportWorkers.
	Workers int

	// Circuits are the breakers whose collective state decides whether the
	// pipeline is degraded. Nil discovers them by walking Exporter, which is
	// what makes readiness work without a second place to keep the same list
	// in step.
	//
	// Set it explicitly only for an exporter chain the walk cannot see
	// through — a custom decorator, or a breaker of your own.
	Circuits []CircuitReporter

	// Metrics receives export and drop counters. Nil discards.
	Metrics Metrics
}

DispatcherConfig configures a Dispatcher. Build one with NewDispatcher, which validates eagerly (NFR4).

type DrainSummary 

type DrainSummary struct {
	// Lost is how many records were still buffered when the deadline expired.
	Lost int
	// Duration is how long the drain took.
	Duration time.Duration
	// Destinations are the exporters the records would have gone to.
	Destinations []string
	// OpenCircuits are the destinations that were refusing calls when the
	// drain ended — the likeliest explanation for anything lost.
	OpenCircuits []string
}

DrainSummary is what a bounded shutdown actually achieved (FR10, ADR-0015).

Loss at shutdown is permitted — bounding shutdown time is a hard requirement in any orchestrated environment. Silent loss is not, which is why this type exists: the number has to be reportable, not merely counted.

func (DrainSummary) Clean 

func (s DrainSummary) Clean() bool

Clean reports whether the buffer emptied before the deadline.

func (DrainSummary) String 

func (s DrainSummary) String() string

String is the final summary line ADR-0015 requires before exit.

It names the count and the destinations rather than saying "some records were lost", because an operator reading it at 3am needs to know whether to look at crier or at the backend — and "to which exporters" is the half that answers that.

type DropKey 

type DropKey struct {
	Source string
	Reason DropReason
}

DropKey identifies one (source, reason) drop bucket.

type DropPolicy 

type DropPolicy int

DropPolicy is what happens to a record when the buffer is full (ADR-0002). 

const (
	// DropPolicyReject returns ErrBufferFull. The default, because a silent
	// drop must never be the out-of-the-box behaviour — a caller opts into
	// losing data knowingly.
	DropPolicyReject DropPolicy = iota
	// DropPolicyBlock makes the sender wait for room.
	DropPolicyBlock
	// DropPolicyDropOldest evicts the oldest pending record to make room.
	DropPolicyDropOldest
)

func (DropPolicy) String 

func (p DropPolicy) String() string

String implements fmt.Stringer.

func (DropPolicy) Valid 

func (p DropPolicy) Valid() bool

Valid reports whether p is a defined policy.

type DropReason 

type DropReason string

DropReason says why a record did not survive to export. Every discard path in the pipeline names one.

The distinctions are not cosmetic. An operator seeing sustained DropBufferFull sizes the buffer up; seeing DropSourceQuota looks at one misbehaving source; seeing DropBackendUnavailable looks at the destination and leaves crier alone (ADR-0011, ADR-0015). Collapsing them into a single "dropped" counter would leave all three looking identical. 

const (
	// DropInvalid — the record failed validation and never entered the pipeline.
	DropInvalid DropReason = "invalid"
	// DropRedactionFailed — redaction errored, so the record was discarded
	// rather than exported unmasked (ADR-0014, fail-closed).
	DropRedactionFailed DropReason = "redaction_failed"
	// DropSourceQuota — this source's fair share was exhausted while the
	// buffer as a whole still had room (ADR-0011).
	DropSourceQuota DropReason = "source_quota"
	// DropBufferFull — the buffer was full under DropPolicyReject (ADR-0002).
	DropBufferFull DropReason = "buffer_full"
	// DropOldest — evicted to make room under DropPolicyDropOldest (ADR-0002).
	DropOldest DropReason = "drop_oldest"
	// DropBackendUnavailable — every exporter's circuit was open, so the
	// pipeline shed load rather than buffering without bound (ADR-0015).
	DropBackendUnavailable DropReason = "backend_unavailable"
	// DropShutdownTimeout — still unexported when the drain deadline expired.
	// Loss at shutdown is permitted; unaccounted loss is not (ADR-0015).
	DropShutdownTimeout DropReason = "shutdown_timeout"
)

Reasons a record can be dropped. This list is exhaustive by intent: adding a discard path means adding a reason here, which is the point.

type Exporter 

type Exporter interface {
	// Export delivers batch, honouring ctx for cancellation and deadlines.
	// Returning an error wrapping ErrPermanent tells the retry decorator not
	// to try again.
	Export(ctx context.Context, batch []LogRecord) error

	// Shutdown releases the exporter's resources. It is called once, after
	// the pipeline has drained, and must not block past ctx's deadline
	// (ADR-0015).
	Shutdown(ctx context.Context) error
}

Exporter sends a batch of records to one destination.

Implementations must be safe for concurrent use: fan-out dispatches to every exporter at once (ADR-0013).

Export returns nil only when the destination has accepted the whole batch. A partial success must be reported as an error, because the caller's only recovery is to re-send the batch — which is within the at-least-once contract (ADR-0009).

type FairShareBuffer 

type FairShareBuffer struct {
	// contains filtered or unexported fields
}

FairShareBuffer decorates a BufferStore with per-source admission, so a noisy source degrades itself rather than its neighbours (ADR-0011). That property is what makes one shared instance viable at all.

Identity

Admission and accounting key on rec.Resource.ServiceName. By the time a record reaches admission — step 8 of the stage order — that field has been overwritten from the authenticated principal at step 3 (ADR-0008), so a source cannot escape its own quota by renaming itself. Using it here is the reason that overwrite exists.

Quota state

In-process only; it does not survive a restart, and replicas do not share it. Distributed quota is out of scope for the MVP, the same line moat draws between memory and Redis stores.

func NewFairShareBuffer 

func NewFairShareBuffer(inner BufferStore, cfg FairShareConfig) (*FairShareBuffer, error)

NewFairShareBuffer wraps inner with per-source admission.

Reservations are validated eagerly (NFR4): reservations summing above the buffer's capacity is a configuration error that must fail at startup rather than silently under-deliver at runtime, where it looks like random loss.

func (*FairShareBuffer) Close 

func (f *FairShareBuffer) Close() error

Close implements BufferStore.

func (*FairShareBuffer) Depth 

func (f *FairShareBuffer) Depth() int

Depth implements BufferStore.

func (*FairShareBuffer) DequeueBatch 

func (f *FairShareBuffer) DequeueBatch(ctx context.Context) ([]LogRecord, error)

DequeueBatch implements BufferStore, returning each record's slot to its source's share.

func (*FairShareBuffer) Enqueue 

func (f *FairShareBuffer) Enqueue(ctx context.Context, rec LogRecord) error

Enqueue implements BufferStore, admitting the record only if its source's share allows it.

It returns ErrSourceQuotaExhausted — never ErrBufferFull — when the source is over its share while the buffer still has room. Collapsing the two would have operators resizing the buffer to fix a throttling problem, which does nothing.

func (*FairShareBuffer) SpareInUse 

func (f *FairShareBuffer) SpareInUse() int

SpareInUse reports how much unreserved capacity is currently taken.

func (*FairShareBuffer) UnlistedInUse 

func (f *FairShareBuffer) UnlistedInUse() int

UnlistedInUse reports how much of the shared unlisted pool is taken.

func (*FairShareBuffer) Usage 

func (f *FairShareBuffer) Usage(source string) int

Usage reports how many of a source's records are currently buffered.

type FairShareConfig 

type FairShareConfig struct {
	// Reservations is the guaranteed floor of buffer capacity per source, by
	// attested identity. A source at its floor can still use spare capacity;
	// it just loses that spare first when the buffer comes under pressure.
	Reservations map[string]int

	// UnlistedPool is capacity held back for sources with no explicit
	// reservation, shared among all of them.
	//
	// It is collective, not per-source, and that is a deliberate correction
	// rather than a shortcut: a per-source default cannot be guaranteed,
	// because the number of unlisted sources is unknowable at startup, so
	// granting each of them a floor admits more records than the buffer holds.
	// A collective pool keeps the bound exact — reserved + unlisted + spare is
	// always the capacity — at the cost of unlisted sources competing with
	// each other. Listing the sources that matter is the way to get a real
	// floor, which is the honest incentive.
	UnlistedPool int

	// Metrics receives quota rejections. Nil discards.
	Metrics Metrics
}

FairShareConfig configures per-source admission (ADR-0011).

type FanOut 

type FanOut struct {
	// contains filtered or unexported fields
}

FanOut sends every batch to every destination, concurrently, and joins the results (ADR-0013, ADR-0016).

It performs no retry of its own. Retry and circuit breaking are per destination, composed inside it: 

NewFanOut(FanOutConfig{Destinations: []Destination{
    {Name: "primary", Exporter: NewRetry(RetryConfig{Exporter: NewCircuitBreaker(...)})},
    {Name: "archive", Exporter: NewRetry(RetryConfig{Exporter: NewCircuitBreaker(...)})},
}})

Composed the other way round — a retry wrapping the fan-out — a failure at one destination re-sends the whole batch, so a healthy destination receives it once per attempt because an unrelated one is broken. That is audit finding A-1 and ADR-0013 exists to forbid it.

Safe for concurrent use.

Example

ExampleFanOut shows the composition ADR-0013 requires: retry and circuit breaking per destination, inside the fan-out. 

// One chain per destination. Innermost is the exporter, then its circuit
// breaker, then its retry — so each destination retries only its own
// batch.
build := func(name string, e Exporter) Destination {
	breaker, err := NewCircuitBreaker(CircuitBreakerConfig{Name: name, Exporter: e})
	if err != nil {
		panic(err)
	}
	retry, err := NewRetry(RetryConfig{Name: name, Exporter: breaker})
	if err != nil {
		panic(err)
	}
	return Destination{Name: name, Exporter: retry}
}

fanOut, err := NewFanOut(FanOutConfig{
	Destinations: []Destination{
		build("primary", noopExporter{}),
		build("archive", noopExporter{}),
	},
	Timeout: 30 * time.Second,
})
if err != nil {
	panic(err)
}

batch := []LogRecord{{Body: "hello", Severity: SeverityInfo}}
if err := fanOut.Export(context.Background(), batch); err != nil {
	var fe *FanOutError
	if errors.As(err, &fe) && fe.Partial() {
		// At least one destination has it. Re-sending to satisfy the
		// other would duplicate at the healthy one (ADR-0013).
		fmt.Println("partial:", err)
	}
}
fmt.Println("dispatched to", fanOut.Names())

Output:
dispatched to [primary archive]

func NewFanOut 

func NewFanOut(cfg FanOutConfig) (*FanOut, error)

NewFanOut validates cfg and returns the fan-out.

func (*FanOut) Destinations 

func (f *FanOut) Destinations() int

Destinations reports how many destinations the fan-out dispatches to.

func (*FanOut) Export 

func (f *FanOut) Export(ctx context.Context, batch []LogRecord) error

Export dispatches batch to every destination at once and waits for all of them.

It returns nil only when every destination accepted the batch. Otherwise it returns a *FanOutError naming the ones that did not, which the caller inspects to tell a partial failure — the batch reached somewhere, so nothing is lost — from a total one, where the records are gone and must be counted (ADR-0015).

func (*FanOut) Names 

func (f *FanOut) Names() []string

Names returns the destination names, in configuration order.

func (*FanOut) Shutdown 

func (f *FanOut) Shutdown(ctx context.Context) error

Shutdown releases every destination's resources, concurrently, and joins the errors. One destination that hangs must not consume another's share of the shutdown deadline.

type FanOutConfig 

type FanOutConfig struct {
	// Destinations receive every batch. At least one is required.
	Destinations []Destination

	// Timeout bounds each destination's Export call. Zero means
	// DefaultExportTimeout.
	//
	// It is a ceiling on the whole composed chain, retry backoff included,
	// because that is the point: a destination that accepts a connection and
	// then never answers otherwise holds a dispatch worker forever, and no
	// circuit breaker helps — a call that never returns never reports the
	// failure a breaker would have counted (ADR-0016). Set it above the retry
	// budget, or retries are cut off by it.
	Timeout time.Duration

	// Metrics receives per-destination export counters. Nil discards.
	Metrics Metrics
}

FanOutConfig configures a FanOut. Build one with NewFanOut, which validates eagerly (NFR4).

type FanOutError 

type FanOutError struct {
	// Dispatched is how many destinations the batch went to.
	Dispatched int
	// Failures holds one error per destination that did not accept it.
	Failures map[string]error
}

FanOutError reports which destinations rejected a batch.

It names them rather than joining anonymous errors because the distinction between "one of three destinations is down" and "all three are down" is the difference between a warning and the degraded state that takes the instance out of service (ADR-0015), and a joined error string cannot be asked which one it is.

func (*FanOutError) AllFailed 

func (e *FanOutError) AllFailed() bool

AllFailed reports whether no destination accepted the batch — the batch is lost, and its records must be counted as dropped rather than exported.

func (*FanOutError) Error 

func (e *FanOutError) Error() string

Error implements error.

func (*FanOutError) Partial 

func (e *FanOutError) Partial() bool

Partial reports whether at least one destination accepted the batch. Nothing is lost in that case: delivery is at-least-once (ADR-0009), and re-sending to satisfy the failed destination would duplicate at the healthy one — which is the amplification ADR-0013 forbids.

func (*FanOutError) Unwrap 

func (e *FanOutError) Unwrap() []error

Unwrap exposes the per-destination errors, so errors.Is and errors.As reach through to sentinels such as ErrPermanent and ErrCircuitOpen.

type Filter 

type Filter struct {
	// MinSeverity drops records below this level. Zero (SeverityUnspecified)
	// keeps everything, which is also what a record with no severity gets.
	MinSeverity Severity

	// SampleRate is the fraction of eligible records kept, in [0,1]. Zero
	// means 1 — no sampling. Use SampleNothing to drop all eligible records.
	SampleRate float64

	// SampleFloor is the severity at or above which sampling never applies.
	// Zero means SeverityError.
	//
	// Sampling away errors defeats the purpose: the rare, important records
	// are exactly the ones a uniform sampler is most likely to discard, and
	// they are the reason anyone is looking at the logs.
	SampleFloor Severity

	// PerSource overrides the global settings by attested source identity
	// (ADR-0008) — never by anything the client asserts, or a noisy source
	// could exempt itself from its own sampling.
	PerSource map[string]SourceFilter

	// Rand returns a value in [0,1). Nil means math/rand/v2. Tests override it.
	Rand func() float64

	// Metrics receives filtered counts. Nil discards.
	Metrics Metrics
}

Filter is step 7 of the canonical stage order (ADR-0010, FR8): the severity threshold and sampler, applied before the buffer so a record that will never be exported never costs buffer memory.

Filtering is not dropping. A filtered record was never meant to leave, so it is counted through RecordsFiltered rather than RecordsDropped — folding the two together would make a correctly configured pipeline look lossy.

Per-exporter filtering remains available after dequeue as an additional narrowing, never as the only filter.

The zero value keeps everything. Safe for concurrent use.

func (*Filter) Keep 

func (f *Filter) Keep(rec *LogRecord, source string) bool

Keep reports whether rec should continue to the buffer, counting it as filtered if not.

source is the attested principal, used to resolve per-source overrides and to label the count.

func (*Filter) KeepBatch 

func (f *Filter) KeepBatch(batch []LogRecord, source string) []LogRecord

KeepBatch filters batch in place and returns the surviving prefix.

It reuses batch's backing array, so the caller must not keep the original slice header expecting the filtered records to still be there.

func (*Filter) Validate 

func (f *Filter) Validate() error

Validate reports configuration that cannot be right, so it fails at startup rather than silently discarding telemetry in production (NFR4).

type Health 

type Health struct {
	// contains filtered or unexported fields
}

Health answers the two questions an orchestrator asks (NFR5, ADR-0005).

It is deliberately not an HTTP handler: what "ready" means is engine behaviour and belongs where it can be tested without a server, and serving it is the daemon's job.

The zero value is not usable; build one with NewHealth.

func NewHealth 

func NewHealth(dispatcher *Dispatcher) (*Health, error)

NewHealth returns the health view of a running dispatcher.

func (*Health) Live 

func (h *Health) Live() bool

Live reports whether the process should stay alive.

It is true for as long as the process is running, including while degraded and while draining. Liveness that fails on a backend outage gets the instance killed and restarted into the same outage, losing whatever was buffered — which is a way of turning someone else's outage into data loss of our own.

func (*Health) Ready 

func (h *Health) Ready() (ready bool, reason string)

Ready reports whether this instance should receive traffic, and why not when it should not.

Not ready in two states, both from ADR-0015:

  • draining, because the buffer is closed and nothing further is accepted;
  • degraded, because every destination's circuit is open, so an accepted record has nowhere to go and would be counted as lost on arrival.

An operator seeing not-ready during a backend outage could read it as a crash loop, so the reason is returned rather than left to be inferred. The instance is still alive, still buffering what it already holds, and still probing the destinations behind their breakers.

type LatencyStat 

type LatencyStat struct {
	Count int64
	Total time.Duration
	Max   time.Duration
}

LatencyStat summarises observed durations without retaining every sample.

func (LatencyStat) Mean 

func (s LatencyStat) Mean() time.Duration

Mean returns the average observed duration, or zero if nothing was observed.

type Limits 

type Limits struct {
	// MaxAttributes caps entries per record, counting record and resource
	// attributes separately.
	MaxAttributes int
	// MaxKeyBytes caps an attribute key. An over-long key drops its attribute
	// rather than being shortened: truncating keys makes distinct fields
	// collide into one, which corrupts data instead of bounding it.
	MaxKeyBytes int
	// MaxValueBytes caps a string or []byte attribute value. Over-long values
	// are truncated with TruncationMark — losing one oversized field is better
	// than losing the event.
	MaxValueBytes int
	// MaxBodyBytes caps the log message.
	MaxBodyBytes int
	// TruncationMark is appended to anything shortened. Empty means
	// DefaultTruncationMark. It must be present: silently altered telemetry
	// that looks like source data is worse than obviously altered telemetry.
	TruncationMark string
	// UnsupportedMark replaces values of an unboundable type. Empty means
	// DefaultUnsupportedMark.
	UnsupportedMark string
	// Metrics receives every alteration. Nil discards, but nothing is ever
	// altered silently in the sense that matters — the marker is in the data.
	Metrics Metrics
}

Limits caps the size of a single record (ADR-0010, step 5).

They apply to embedded-library use as well as to the HTTP receiver. Embedded use is not exempt: a bug in the host application produces the same unbounded attribute map as a malicious client, and the process it takes down is the host's own.

The zero value applies the defaults above. Set a field negative to disable that one limit — an explicit, greppable choice rather than a zero that could mean either "unset" or "none".

func (Limits) Apply 

func (l Limits) Apply(rec *LogRecord)

Apply enforces the limits on rec in place.

It never rejects the record. Every limit here degrades the record rather than discarding it, because an event that arrives with one field clipped still carries the information someone will be looking for at 3am; an event that never arrives does not.

Attribute values must be strings, []byte, or scalars (bool, the integer and float kinds, time.Duration, time.Time). Anything else — a nested map, a slice, a struct — is replaced with UnsupportedMark and counted, because its size cannot be bounded without walking it, and walking an attacker-supplied structure on the hot path is the exhaustion vector this stage exists to close. Callers that need structure should flatten it into dotted keys, which is what the OTel semantic conventions do anyway.

type LogRecord 

type LogRecord struct {
	// Timestamp is when the source claims the event happened. It is
	// source-asserted and therefore untrusted: it may be zero, wildly skewed,
	// or deliberately falsified. It is carried through and exported, but it is
	// never used for any decision crier makes (ADR-0009).
	Timestamp time.Time

	// ObservedTimestamp is when crier observed the record. It is assigned by
	// the pipeline, is always set, and is the authoritative time for ordering,
	// retention, and export (ADR-0009).
	ObservedTimestamp time.Time

	// Severity is the OTel severity number. Filtering and sampling compare
	// against it before the record reaches the buffer (ADR-0010).
	Severity Severity

	// SeverityText is the source's own label for the severity, preserved
	// verbatim because it is often more specific than the numeric mapping.
	SeverityText string

	// Body is the log message. Secrets leak here far more often than into
	// Attributes, so redaction covers it (ADR-0014, finding A-2).
	Body string

	// Attributes are the record's structured fields. Bounded in count, key
	// length, and value length, with a cardinality guard over values
	// (ADR-0010).
	Attributes map[string]any

	// Resource identifies the emitting service.
	Resource Resource

	// TraceID and SpanID correlate this record with a trace when the source
	// has one. Both are optional and never required for a record to be valid
	// (ADR-0004).
	TraceID string
	SpanID  string
}

LogRecord is crier's internal representation of a single log entry, aligned with the OpenTelemetry Logs data model (ADR-0004).

func (LogRecord) Clone 

func (rec LogRecord) Clone() LogRecord

Clone returns a deep copy of rec. Pipeline stages that mutate a record must clone it first when the original may be shared — most notably fan-out, where several exporters observe the same batch (ADR-0013).

func (LogRecord) EffectiveTime 

func (rec LogRecord) EffectiveTime() time.Time

EffectiveTime returns the timestamp to use for any decision or export.

It is ObservedTimestamp, always. The method exists so that call sites read as a deliberate choice rather than as an arbitrary pick between two timestamp fields (ADR-0009).

type MemoryBuffer 

type MemoryBuffer struct {
	// contains filtered or unexported fields
}

MemoryBuffer is the default BufferStore: a bounded ring buffer that batches by size or by time window, whichever comes first.

Safe for concurrent use by many producers and many consumers.

func NewMemoryBuffer 

func NewMemoryBuffer(cfg MemoryBufferConfig) (*MemoryBuffer, error)

NewMemoryBuffer builds a buffer from cfg, validating it eagerly (NFR4).

func (*MemoryBuffer) Capacity 

func (b *MemoryBuffer) Capacity() int

Capacity reports the configured bound.

func (*MemoryBuffer) Close 

func (b *MemoryBuffer) Close() error

Close implements BufferStore. It is idempotent and discards nothing: every record already inside stays available to DequeueBatch until drained.

func (*MemoryBuffer) Depth 

func (b *MemoryBuffer) Depth() int

Depth implements BufferStore.

func (*MemoryBuffer) DequeueBatch 

func (b *MemoryBuffer) DequeueBatch(ctx context.Context) ([]LogRecord, error)

DequeueBatch implements BufferStore.

func (*MemoryBuffer) Enqueue 

func (b *MemoryBuffer) Enqueue(ctx context.Context, rec LogRecord) error

Enqueue implements BufferStore.

source labels the drop counters. It is the attested principal, passed in rather than read from the record, because the record's own resource fields are not authoritative (ADR-0008).

func (*MemoryBuffer) EnqueueFrom 

func (b *MemoryBuffer) EnqueueFrom(ctx context.Context, rec LogRecord, source string) error

EnqueueFrom is Enqueue with an explicit source label for metrics.

type MemoryBufferConfig 

type MemoryBufferConfig struct {
	// Capacity is the maximum number of records held. Zero means
	// DefaultBufferCapacity. The bound is never relaxed under pressure — that
	// is the whole point of having one (ADR-0015).
	Capacity int
	// BatchSize is how many records a full batch holds. Zero means
	// DefaultBatchSize.
	BatchSize int
	// BatchWindow is how long the oldest pending record waits for company
	// before the batch goes out short. Zero means DefaultBatchWindow.
	BatchWindow time.Duration
	// Policy is what happens on a full buffer. Defaults to DropPolicyReject.
	Policy DropPolicy
	// Metrics receives depth and drop counts. Nil discards.
	Metrics Metrics
}

MemoryBufferConfig configures a MemoryBuffer.

type Metrics 

type Metrics interface {
	// RecordsIngested counts records accepted for processing, per source.
	RecordsIngested(source string, n int)

	// RecordsDropped counts records discarded, by source and reason. Source
	// may be empty where the record was rejected before identity was attested.
	RecordsDropped(source string, reason DropReason, n int)

	// RecordsFiltered counts records removed by the severity threshold or
	// sampler (ADR-0010, step 7). Filtered is not dropped: the record was
	// never meant to be exported, so it is not a loss.
	RecordsFiltered(source string, n int)

	// RecordsExported counts records a destination has accepted.
	RecordsExported(exporter string, n int)

	// ExportLatency observes how long one batch took, per exporter.
	ExportLatency(exporter string, d time.Duration)

	// ExportRetried counts retry attempts, per exporter (ADR-0013).
	ExportRetried(exporter string)

	// CircuitStateChanged reports an exporter's breaker opening or closing.
	// All-open is the degraded state surfaced through readiness (ADR-0015).
	CircuitStateChanged(exporter string, open bool)

	// ExportDegraded reports the pipeline entering or leaving the state where
	// no destination will accept anything (ADR-0015).
	//
	// It is separate from CircuitStateChanged because it is a different
	// question. One breaker opening is a destination to look at; every
	// breaker being open is an outage, and an operator should not have to
	// join the per-destination series to find that out.
	ExportDegraded(degraded bool)

	// BufferDepth reports current occupancy. A gauge, not a counter.
	BufferDepth(depth int)

	// AttributeTruncated counts values shortened to fit the length cap. The
	// record survives with a marker; losing one oversized field beats losing
	// the event (ADR-0010).
	AttributeTruncated(key string)

	// AttributeDropped counts attributes removed outright — the record was
	// over the attribute-count cap, or the key itself was longer than the key
	// cap. Kept apart from AttributeTruncated because losing a field is a
	// different event from shortening one (ADR-0010).
	AttributeDropped(key string)

	// CardinalityCapped counts values replaced because the key exceeded its
	// distinct-value threshold (ADR-0010).
	CardinalityCapped(key string)

	// IdentityDiscrepancy counts records whose client-asserted identity did
	// not match the authenticated principal. The record is still accepted,
	// attributed to the real principal (ADR-0008, finding D-2). A rising
	// count is either a misconfigured client or someone probing.
	IdentityDiscrepancy(claimed, actual string)

	// TimestampMissing counts records that asserted no Timestamp at all.
	//
	// Kept apart from ClockSkew deliberately: an absent timestamp is not a
	// clock that is wrong by the distance to the zero time, and folding it in
	// would swamp the skew statistic with a meaningless outlier. Both are
	// accepted, both are visible, neither is guessed at (ADR-0009).
	TimestampMissing(source string)

	// ClockSkew observes how far a source's asserted Timestamp sat from the
	// ObservedTimestamp. Skew never rejects a record (ADR-0009); it is
	// reported so a source with a broken clock is discoverable.
	ClockSkew(source string, deviation time.Duration)

	// DeprecatedWireVersion counts requests on a wire version scheduled for
	// removal, so the migration is driven by data (ADR-0012).
	DeprecatedWireVersion(version string)
}

Metrics is crier's self-observability seam (ADR-0005). It is an interface so that core carries no metrics-backend dependency (NFR1): the standalone binary can back it with Prometheus, an embedding application with whatever it already uses.

The methods are explicit rather than a generic Counter(name string) for one reason: "every drop path increments exactly one counter" is a property that can be reviewed when the events are named in the type, and cannot be when they are strings passed at call sites.

Implementations must be safe for concurrent use and must not block — every method sits on the hot path. Embed NopMetrics to implement only the subset you care about.

type NopMetrics 

type NopMetrics struct{}

NopMetrics implements Metrics as no-ops. Embed it to implement a subset: 

type bufferOnly struct {
    core.NopMetrics
    depth atomic.Int64
}
func (m *bufferOnly) BufferDepth(d int) { m.depth.Store(int64(d)) }

It is also the right default for an embedding application that has not wired metrics up yet — the pipeline must never require them to run.

func (NopMetrics) AttributeDropped 

func (NopMetrics) AttributeDropped(string)

AttributeDropped implements Metrics and counts nothing.

func (NopMetrics) AttributeTruncated 

func (NopMetrics) AttributeTruncated(string)

AttributeTruncated implements Metrics and counts nothing.

func (NopMetrics) BufferDepth 

func (NopMetrics) BufferDepth(int)

BufferDepth implements Metrics and records nothing.

func (NopMetrics) CardinalityCapped 

func (NopMetrics) CardinalityCapped(string)

CardinalityCapped implements Metrics and counts nothing.

func (NopMetrics) CircuitStateChanged 

func (NopMetrics) CircuitStateChanged(string, bool)

CircuitStateChanged implements Metrics and records nothing.

func (NopMetrics) ClockSkew 

func (NopMetrics) ClockSkew(string, time.Duration)

ClockSkew implements Metrics and observes nothing.

func (NopMetrics) DeprecatedWireVersion 

func (NopMetrics) DeprecatedWireVersion(string)

DeprecatedWireVersion implements Metrics and counts nothing.

func (NopMetrics) ExportDegraded 

func (NopMetrics) ExportDegraded(bool)

ExportDegraded implements Metrics and records nothing.

func (NopMetrics) ExportLatency 

func (NopMetrics) ExportLatency(string, time.Duration)

ExportLatency implements Metrics and observes nothing.

func (NopMetrics) ExportRetried 

func (NopMetrics) ExportRetried(string)

ExportRetried implements Metrics and counts nothing.

func (NopMetrics) IdentityDiscrepancy 

func (NopMetrics) IdentityDiscrepancy(string, string)

IdentityDiscrepancy implements Metrics and counts nothing.

func (NopMetrics) RecordsDropped 

func (NopMetrics) RecordsDropped(string, DropReason, int)

RecordsDropped implements Metrics and counts nothing.

func (NopMetrics) RecordsExported 

func (NopMetrics) RecordsExported(string, int)

RecordsExported implements Metrics and counts nothing.

func (NopMetrics) RecordsFiltered 

func (NopMetrics) RecordsFiltered(string, int)

RecordsFiltered implements Metrics and counts nothing.

func (NopMetrics) RecordsIngested 

func (NopMetrics) RecordsIngested(string, int)

RecordsIngested implements Metrics and counts nothing.

func (NopMetrics) TimestampMissing 

func (NopMetrics) TimestampMissing(string)

TimestampMissing implements Metrics and counts nothing.

type Normalizer 

type Normalizer struct {
	// Now supplies the observation time. Nil means time.Now. Tests override it;
	// production should not.
	Now func() time.Time

	// SkewThreshold is the deviation past which skew is reported. Zero means
	// DefaultSkewThreshold. Negative disables reporting.
	SkewThreshold time.Duration

	// Metrics receives skew and missing-timestamp observations. Nil discards.
	Metrics Metrics
}

Normalizer is step 4 of the canonical stage order (ADR-0010): it stamps the authoritative timestamp and reports what the source claimed.

It never rejects a record. A missing or absurd Timestamp is a broken source, not a hostile one, and dropping its logs would destroy exactly the evidence needed to notice the breakage (ADR-0009).

The zero value is usable: it uses time.Now, DefaultSkewThreshold, and discards metrics.

func (*Normalizer) Normalize 

func (n *Normalizer) Normalize(rec *LogRecord, source string)

Normalize assigns rec.ObservedTimestamp and reports on rec.Timestamp.

source is the attested principal (ADR-0008), used only to label the observations — never taken from the record, which is why it is a parameter rather than read from rec.Resource.

ObservedTimestamp is assigned once. A record that already carries one has passed through a pipeline stage before, and re-stamping it would move the authoritative time forward every hop.

func (*Normalizer) NormalizeBatch 

func (n *Normalizer) NormalizeBatch(batch []LogRecord, source string)

NormalizeBatch applies Normalize to every record, sharing one observation time across the batch so records that arrived together are stamped together.

type Options 

type Options struct {
	// ServiceName identifies the host application. It becomes the records'
	// resource identity and the key for fair share, filtering and metrics.
	//
	// Required. In embedded mode there is no authenticated principal to derive
	// it from — the host application is the trust boundary (FR11) — so it is
	// asserted here once rather than per record.
	ServiceName string

	// ServiceVersion is optional, and worth setting: it is what makes a
	// deployment distinguishable in the backend.
	ServiceVersion string

	// Exporters are the destinations, by name. At least one is required.
	//
	// Each is wrapped in its own circuit breaker and its own retry, inside the
	// fan-out — the composition ADR-0013 requires. Assembling it here rather
	// than asking the host to is the point: built the other way round, one
	// failing destination re-sends the batch to every healthy one, and that
	// mistake is invisible until a backend goes down.
	Exporters map[string]Exporter

	// Capacity bounds the buffer. Zero means DefaultBufferCapacity.
	Capacity int
	// BatchSize and BatchWindow control how records are grouped for export.
	// Zero means the buffer's defaults.
	BatchSize   int
	BatchWindow time.Duration
	// DropPolicy is what happens to a record when the buffer is full. The
	// zero value rejects, which is the default a caller has to opt out of.
	DropPolicy DropPolicy

	// Limits caps record size (ADR-0010). The zero value applies defaults.
	//
	// They apply here exactly as they do to the HTTP receiver, and for the
	// same reason: a bug in the host application produces the same unbounded
	// attribute map as a malicious client, and the buffer cannot tell them
	// apart.
	Limits Limits

	// Cardinality guards attribute value cardinality. Nil disables the guard.
	Cardinality *CardinalityGuard

	// Redactor masks sensitive data. Nil disables redaction, which is the
	// explicit, auditable choice ADR-0014 requires an operator to make.
	Redactor *Redactor

	// Filter applies the severity threshold and sampler. Nil keeps everything.
	Filter *Filter

	// Workers bounds how many batches are exported at once. Zero means
	// DefaultExportWorkers.
	Workers int

	// ExportTimeout bounds one destination's export, retry backoff included.
	// Zero means DefaultExportTimeout.
	ExportTimeout time.Duration

	// RetryAttempts bounds attempts per destination, the first included. Zero
	// means DefaultRetryAttempts; one disables retrying without removing the
	// decorator.
	RetryAttempts int

	// FailureThreshold is how many consecutive failures open a destination's
	// circuit. Zero means DefaultFailureThreshold.
	FailureThreshold int

	// Cooldown is how long an open circuit waits before probing. Zero means
	// DefaultCooldown.
	Cooldown time.Duration

	// Metrics receives every counter. Nil discards, which is the right default
	// for a host that has not wired metrics up yet — the engine must never
	// require them to run.
	Metrics Metrics
}

Options configures an embedded engine. Build one with New, which validates eagerly (NFR4).

Everything has a working default except Exporters and ServiceName, because an engine with no destination and no identity is not a thing anyone meant to build.

type Pipeline 

type Pipeline struct {
	// contains filtered or unexported fields
}

Pipeline applies the canonical stage order to every record, whatever receiver it arrived through (ADR-0010).

  1. normalize -> 5. record limits -> 6. redact -> 7. filter/sample -> 8. admit

Steps 1 to 3 — transport limits, parsing, and identity attestation — belong to the receiver, because they are about the request rather than the record. The attested identity arrives here as Admit's source argument.

The order is a contract, not an implementation detail. Everything cheap and everything reductive happens before the buffer, so bounded memory is only ever spent on records that will actually be exported. Only step 8 touches the BufferStore.

Safe for concurrent use.

func NewPipeline 

func NewPipeline(cfg PipelineConfig) (*Pipeline, error)

NewPipeline validates cfg and assembles the stages.

func (*Pipeline) Admit 

func (p *Pipeline) Admit(ctx context.Context, rec LogRecord, source string) error

Admit runs rec through every stage and, if it survives, enqueues it.

source is the attested principal (ADR-0008). It overwrites the record's identity fields before any stage that keys on identity, so a client cannot influence its own quota, its own sampling, or how its records are attributed. Pass an empty source only where there is genuinely no authenticated principal — embedded-library use, where the host application is the boundary.

A filtered record returns nil. Filtering is not failure: the record was handled exactly as configured, and a receiver that answers 202 for admitted records should answer 202 here too (ADR-0009 — 202 means admitted, not delivered, and a record deliberately discarded by policy was still accepted).

Errors are the ones a caller must act on: ErrRedactionFailed (dropped fail-closed), ErrSourceQuotaExhausted (this source is throttled), ErrBufferFull (capacity pressure), ErrBufferClosed (shutting down).

func (*Pipeline) AdmitBatch 

func (p *Pipeline) AdmitBatch(ctx context.Context, batch []LogRecord, source string) (admitted int, firstErr error)

AdmitBatch runs every record through the pipeline, returning how many were enqueued and the first error encountered.

It does not stop at the first failure. A batch is a transport detail: one record hitting a source quota says nothing about the next, and abandoning the rest would turn one throttled record into a whole request lost.

func (*Pipeline) Buffer 

func (p *Pipeline) Buffer() BufferStore

Buffer exposes the underlying store, so a consumer can dequeue and a shutdown can drain (ADR-0015).

type PipelineConfig 

type PipelineConfig struct {
	// Buffer is where admitted records land. Required.
	Buffer BufferStore

	// Normalizer stamps ObservedTimestamp (step 4). Nil uses a default one.
	Normalizer *Normalizer

	// Limits caps record size (step 5).
	Limits Limits

	// Cardinality guards attribute value cardinality (step 5). Nil disables
	// the guard; the size limits still apply.
	//
	// It runs after the size limits, per ADR-0010's ordering, which has a
	// consequence worth knowing: truncation collapses long values that shared
	// a prefix into one, so the guard sees the truncated form and reports
	// lower cardinality than the source actually emitted. That is the right
	// trade — the backend also only ever sees the truncated form, and it is
	// the backend's series count the guard exists to protect.
	Cardinality *CardinalityGuard

	// Redactor masks sensitive data (step 6).
	//
	// Nil disables redaction entirely, which is the explicit, auditable
	// choice ADR-0014 requires an operator to make. It is not the same as a
	// redactor that fails open — there is no such thing.
	Redactor *Redactor

	// Filter applies the severity threshold and sampler (step 7). Nil keeps
	// everything.
	Filter *Filter

	// Metrics receives pipeline counters, and is wired into any stage above
	// that does not carry its own.
	//
	// Filling in a stage's nil Metrics means mutating a struct the caller
	// supplied, which is worth stating plainly. The alternative is worse: an
	// operator sets Metrics once on the pipeline, every stage keeps its own
	// nil, and the counters that matter most silently stay at zero. A project
	// whose first rule is "no silent drops" cannot ship that default.
	Metrics Metrics
}

PipelineConfig assembles the processing stages. Build a Pipeline with NewPipeline, which validates the configuration eagerly (NFR4).

type RedactionConfig 

type RedactionConfig struct {
	// KeySubstrings are case-insensitive fragments matched against attribute
	// keys; a match masks the whole value. Nil means SensitiveKeySubstrings.
	// Empty and non-nil means no substring rules.
	KeySubstrings []string

	// KeyPatterns are regexes matched against attribute keys, for what a
	// substring cannot express. They are checked after KeySubstrings and cost
	// considerably more per attribute, so prefer a substring where one will do.
	KeyPatterns []string
	// BodyPatterns are regexes matched against Body and against string
	// attribute values. Capture group 1, when present, is what gets masked;
	// otherwise the whole match is. Nil means BodyPatterns.
	BodyPatterns []string
	// SkipBody turns off body scanning. It does not turn off key rules.
	//
	// This is the honest lever for the cost described in ADR-0014: body
	// redaction is the most expensive stage in the pipeline. It is not a
	// fail-open switch — attribute redaction still applies, and a record that
	// fails redaction is still dropped.
	SkipBody bool
	// Metrics receives drop counts on failure. Nil discards.
	Metrics Metrics
}

RedactionConfig describes what to mask. Compile it into a Redactor before use — a config is not usable directly, so an invalid rule cannot reach the hot path.

type Redactor 

type Redactor struct {
	// contains filtered or unexported fields
}

Redactor masks sensitive data in a record (ADR-0006, ADR-0014). Build one with NewRedactor; the zero value is not usable.

Safe for concurrent use: compiled patterns are immutable and regexp.Regexp is safe for concurrent use by design.

func NewRedactor 

func NewRedactor(cfg RedactionConfig) (*Redactor, error)

NewRedactor compiles cfg.

Compilation is eager and failure is fatal to startup (NFR4, ADR-0014): a config typo becomes a deployment failure rather than a service that runs while silently leaking. The error names the offending pattern, since a regex rejected without saying which one is a support ticket.

func (*Redactor) Redact 

func (r *Redactor) Redact(rec *LogRecord, source string) error

Redact masks rec in place, covering record attributes, resource attributes, and Body (ADR-0014, findings A-2 and A-3).

It returns an error wrapping ErrRedactionFailed if the record could not be fully processed. A caller receiving that error must drop the record. There is no fail-open mode and there will not be one: a security control that degrades into permitting the thing it guards against is not a control. An operator who wants unredacted export disables redaction explicitly, which is an auditable choice rather than a silent degradation.

Best-effort on Body

Pattern matching over free text cannot be complete. A secret with no recognisable shape, interpolated into a message, will survive. This is why structured attributes are preferred: key-based redaction is reliable in a way body scanning cannot be.

func (*Redactor) RedactString 

func (r *Redactor) RedactString(s string) string

RedactString masks every configured pattern in s, replacing the captured credential — not the whole match — where a rule captures one.

Exported so a receiver can mask text it handles outside a LogRecord, and so the behaviour is directly testable.

type Resource 

type Resource struct {
	// ServiceName is semconv "service.name". Authoritative, server-derived.
	ServiceName string
	// ServiceVersion is semconv "service.version".
	ServiceVersion string
	// Attributes carries any further resource-level attributes. These are
	// descriptive, not identifying, and survive identity attestation.
	Attributes map[string]any
}

Resource identifies the origin of a log record, using OpenTelemetry semantic-convention attribute names (ADR-0004).

The identity fields are not client-asserted. In standalone mode the receiver overwrites them from the authenticated principal and counts the discrepancy (ADR-0008); descriptive attributes supplied by the client are preserved.

func (Resource) Clone 

func (r Resource) Clone() Resource

Clone returns a deep copy of r, safe to mutate independently.

type Retry 

type Retry struct {
	// contains filtered or unexported fields
}

Retry re-sends a failed batch to one destination, with bounded attempts and exponential backoff (ADR-0013, FR6).

It is composed *inside* the fan-out — FanOut(Retry(CircuitBreaker(e))) — so it only ever re-sends its own destination's batch. Wrapped the other way round, one destination's failure re-sends the batch to every healthy destination as well, once per attempt. That is duplicate amplification, audit finding A-1.

Safe for concurrent use.

func NewRetry 

func NewRetry(cfg RetryConfig) (*Retry, error)

NewRetry validates cfg and returns the decorator.

func (*Retry) Export 

func (r *Retry) Export(ctx context.Context, batch []LogRecord) error

Export sends batch, retrying only failures that retrying could fix.

It gives up immediately on:

  • a permanent failure (ErrPermanent) — a malformed payload or a rejected credential is not going to be accepted on the third attempt, and the budget spent on it delays every batch queued behind this one;
  • an open circuit (ErrCircuitOpen) — the breaker below has already decided this destination is unhealthy, and asking it again in a loop turns a fail-fast into a stall that spends the whole export deadline;
  • a cancelled or expired context — nobody is waiting for the answer.

func (*Retry) MutatesBatch 

func (r *Retry) MutatesBatch() bool

MutatesBatch forwards the wrapped exporter's declaration, so a leaf that writes to its batch is not hidden behind the decorator chain (ADR-0016).

func (*Retry) Shutdown 

func (r *Retry) Shutdown(ctx context.Context) error

Shutdown releases the wrapped exporter.

type RetryConfig 

type RetryConfig struct {
	// Name labels this destination's retry counter. Required, and should
	// match the fan-out Destination name, or the retry count and the export
	// count for one destination land under different labels.
	Name string

	// Exporter is what gets retried — the circuit breaker, which wraps the
	// real exporter (ADR-0013).
	Exporter Exporter

	// MaxAttempts bounds attempts, the first included. Zero means
	// DefaultRetryAttempts. One disables retrying without removing the
	// decorator, which is the honest way to turn it off.
	MaxAttempts int

	// InitialBackoff is the first interval; it doubles per attempt. Zero means
	// DefaultInitialBackoff.
	InitialBackoff time.Duration

	// MaxBackoff caps one interval. Zero means DefaultMaxBackoff.
	MaxBackoff time.Duration

	// Metrics receives retry counts. Nil discards.
	Metrics Metrics

	// Rand returns a value in [0,1), used for jitter. Nil means math/rand/v2.
	Rand func() float64
}

RetryConfig configures a Retry. Build one with NewRetry, which validates eagerly (NFR4).

type RetryHint 

type RetryHint interface {
	error

	// RetryAfter is how long the destination asked us to wait.
	RetryAfter() time.Duration
}

RetryHint is implemented by an error that knows how long the caller should wait before trying again — an HTTP 429 or 503 carrying Retry-After (ADR-0017).

Retry waits for the longer of its own backoff and the hint. Ignoring a destination that has just told us how long it needs is how a rate limit turns into an outage: the sender keeps arriving early, keeps being refused, and spends its whole budget doing it.

The hint is bounded by the export deadline, not obeyed unconditionally (ADR-0016). A destination asking for an hour gets the deadline, and the batch fails rather than parking a dispatch worker for an hour.

type Severity 

type Severity int

Severity is the OpenTelemetry log severity number. The numeric values are fixed by the OTel Logs data model and are relied upon for range comparisons, so a threshold filter can be expressed as a single >= check. 

const (
	SeverityUnspecified Severity = 0
	SeverityTrace       Severity = 1
	SeverityDebug       Severity = 5
	SeverityInfo        Severity = 9
	SeverityWarn        Severity = 13
	SeverityError       Severity = 17
	SeverityFatal       Severity = 21
)

Severity levels, aligned with the OpenTelemetry Logs data model.

func (Severity) String 

func (s Severity) String() string

String returns the short name of the severity range s falls into.

func (Severity) Valid 

func (s Severity) Valid() bool

Valid reports whether s is within the range defined by the OTel data model.

type Snapshot 

type Snapshot struct {
	Ingested              map[string]int64
	Dropped               map[DropKey]int64
	Filtered              map[string]int64
	Exported              map[string]int64
	Retries               map[string]int64
	OpenCircuits          map[string]bool
	Degraded              bool
	DegradedTransitions   int64
	ExportLatency         map[string]LatencyStat
	BufferDepth           int
	AttributesTruncated   map[string]int64
	AttributesDropped     map[string]int64
	CardinalityCapped     map[string]int64
	IdentityDiscrepancies map[string]int64
	TimestampMissing      map[string]int64
	ClockSkew             map[string]LatencyStat
	DeprecatedWireVersion map[string]int64
}

Snapshot is a point-in-time copy of CountingMetrics. Maps are copies, so a caller may read them while the pipeline keeps running.

func (Snapshot) DroppedBy 

func (s Snapshot) DroppedBy(reason DropReason) int64

DroppedBy returns the count for one reason across all sources.

func (Snapshot) TotalDropped 

func (s Snapshot) TotalDropped() int64

TotalDropped sums every drop bucket.

type SourceFilter 

type SourceFilter struct {
	MinSeverity *Severity
	SampleRate  *float64
}

SourceFilter overrides Filter's thresholds for one source. A nil field means "inherit"; this is why they are pointers rather than plain values, where a zero would be indistinguishable from an unset override.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.