pipeline

Documentation

Overview

Package pipeline is Coremetry's ingest-time policy engine (v0.5.263). Operator-defined rules run BEFORE the sampler so dropped spans never touch the consumer's batch buffer, the sampler's tail buffer, or ClickHouse. Equivalent to Dynatrace OpenPipeline / Vector's transform pipeline for the drop case — implemented narrowly here so we don't grow a second-config-system surface inside Coremetry.

Scope:

• Signal: spans, logs, and metrics (v0.8.282 completed the logs + metrics application; the span-only MVP was v0.5.263). AcceptSpan / AcceptLog / AcceptMetric each scope to their own Signal so one shared catalog cleanly partitions rules.
• Rule kind: "drop", "enrich" (set resource attribute), and "sample" (probabilistic keep). Sample on metrics is a sharp tool — it estimates aggregates — but supported for symmetry.
• One Condition per rule (key = op + value). Multi-condition AND is a follow-up; one predicate covers the dominant "drop spans from service X" + "drop kind=internal" use cases.

HA story: rules live in system_settings (single JSON blob). Every replica loads at boot + on every PUT. No per-pod drift — same Redis-arbitrated config pattern as Tempo / Copilot.

Index

• type Condition
• type Engine
  • func New() *Engine
  • func (e *Engine) AcceptLog(l *chstore.Log) bool
  • func (e *Engine) AcceptMetric(m *chstore.MetricPoint) bool
  • func (e *Engine) AcceptSpan(sp *chstore.Span) bool
  • func (e *Engine) Delete(ctx context.Context, st store, id string) error
  • func (e *Engine) LoadPersisted(ctx context.Context, st store) error
  • func (e *Engine) LogStats()
  • func (e *Engine) Rules() []Rule
  • func (e *Engine) StartConfigRefresh(ctx context.Context, st store, interval time.Duration)
  • func (e *Engine) Upsert(ctx context.Context, st store, r Rule) (Rule, error)
• type Op
• type Rule
• type RuleKind
• type Signal

Types

type Condition

type Condition struct {
	Key   string `json:"key"`
	Op    Op     `json:"op"`
	Value string `json:"value"`
}

Condition is a single attribute predicate. Key supports the well-known span fields directly (service.name, name, kind, status_code) plus any attribute via the "attr." or "resource." prefix.

type Engine

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

Engine evaluates the active rule set against each incoming span. Methods are safe for concurrent use; the RWMutex protects the rules slice against the LoadRules / SaveRules admin path. AcceptSpan is the hot path — keep it allocation- free past the read lock.

func New

func New() *Engine

New returns an engine with no rules. Call LoadPersisted at boot to hydrate from system_settings.

func (*Engine) AcceptLog

func (e *Engine) AcceptLog(l *chstore.Log) bool

AcceptLog is the logs hot path — mirrors AcceptSpan exactly (v0.8.282). Called once per incoming log record before the consumer buffer. Returns false to drop; the caller bumps its logs-dropped-by-pipeline counter. Only rules whose Signal is SignalLogs are considered — spans/metrics rules are skipped so one shared rule catalog cleanly scopes per signal.

func (*Engine) AcceptMetric

func (e *Engine) AcceptMetric(m *chstore.MetricPoint) bool

AcceptMetric is the metrics hot path — mirrors AcceptSpan (v0.8.282). Returns false to drop the data point. Drop is the dominant use case (silence a noisy debug gauge / drop a whole instrument for cost); enrich adds resource context. Sample is supported for symmetry but is a sharp tool on metrics — probabilistically discarding points corrupts cumulative sums, counts, and histogram buckets, so the operator opts in per rule knowing the aggregate becomes an estimate. Only SignalMetrics rules are considered.

func (*Engine) AcceptSpan

func (e *Engine) AcceptSpan(sp *chstore.Span) bool

AcceptSpan is the hot path — called once per incoming span before sampling. Returns false to drop the span entirely; the caller bumps its dropped-by-pipeline counter and never touches the consumer buffer.

Rule evaluation walks the catalog in order. Multiple rules can match a single span:

• Drop short-circuits — first matching drop wins, span is gone.
• Sample probability-rolls against rule.Rate; failure to keep returns false (treated as dropped by the ingester). Continues to subsequent rules only when the keep roll succeeded.
• Enrich mutates the span's resource attributes in place and continues; a later drop / sample may still discard.

Hot-path discipline:

• read lock per call (uncontended in steady state)
• single math/rand/v2 call per sample-rule (lockless)
• no map lookups for well-known fields
• early-return on the first drop / drop-by-sample match

func (*Engine) Delete

func (e *Engine) Delete(ctx context.Context, st store, id string) error

Delete removes a rule by ID. Idempotent — unknown ID is a silent no-op so the audit log doesn't see double-fires when the admin clicks delete twice.

func (*Engine) LoadPersisted

func (e *Engine) LoadPersisted(ctx context.Context, st store) error

LoadPersisted hydrates the in-memory rule set from system_settings. Missing blob = empty rule set (engine accepts everything). Failure is non-fatal — caller logs and proceeds.

func (*Engine) LogStats

func (e *Engine) LogStats()

Log helper for the boot path — quick "loaded N rules" line without pulling the engine's lock through to main.go.

func (*Engine) Rules

func (e *Engine) Rules() []Rule

Rules returns a snapshot copy of the current rule set, sorted by name. Safe to mutate the returned slice.

func (*Engine) StartConfigRefresh

func (e *Engine) StartConfigRefresh(ctx context.Context, st store, interval time.Duration)

StartConfigRefresh — v0.5.324. Background poll keeps the pipeline rules in sync with the shared persisted blob across pods. interval ≤ 0 → 30s.

func (*Engine) Upsert

func (e *Engine) Upsert(ctx context.Context, st store, r Rule) (Rule, error)

Upsert creates or replaces a rule by ID + persists the new catalog. Returns the canonical Rule (with normalised fields) so the API handler can echo it back to the operator.

type Op

type Op string

Op enumerates the comparison operators a Condition supports. Kept narrow on purpose — full FilterExpr-grade ops belong on the query side, not the ingest side where the budget per span is microseconds.

const (
	OpEq         Op = "="
	OpNeq        Op = "!="
	OpContains   Op = "contains"
	OpStartsWith Op = "startsWith"
	OpEndsWith   Op = "endsWith"
)

type Rule

type Rule struct {
	ID      string    `json:"id"`
	Name    string    `json:"name"`
	Kind    RuleKind  `json:"kind"`
	Signal  Signal    `json:"signal"`
	Enabled bool      `json:"enabled"`
	When    Condition `json:"when"`

	// SetAttributes — enrich rules only (v0.5.270). When the
	// rule matches, every key/value pair is written to the
	// span's RESOURCE attributes (overrides if the key
	// already exists). Empty map = no-op.
	SetAttributes map[string]string `json:"setAttributes,omitempty"`

	// Rate — sample rules only (v0.5.270). Keep probability
	// in [0, 1]; 1.0 = keep everything (no-op), 0.0 = drop
	// everything (use a drop rule instead). Random keep
	// decision is local to this rule — the global head
	// sampler still runs afterwards and may further sample
	// the matching span out.
	Rate float64 `json:"rate,omitempty"`
}

Rule is one operator-defined pipeline policy.

type RuleKind

type RuleKind string

RuleKind enumerates the actions a matching rule performs.

const (
	KindDrop   RuleKind = "drop"   // drop the signal entirely
	KindEnrich RuleKind = "enrich" // add / override a resource attribute (v0.5.270)
	KindSample RuleKind = "sample" // probabilistic keep at Rate (v0.5.270)
)

type Signal

type Signal string

Signal scopes the rule to a single OTel signal type.

const (
	SignalSpans   Signal = "spans"
	SignalLogs    Signal = "logs"
	SignalMetrics Signal = "metrics"
)