pipelinetest

Documentation

Overview

Package pipelinetest provides primitives for testing OTLP pipelines: an in-process OTLP/HTTP sink that captures exported spans, and a subprocess-managed OpenTelemetry Collector. Tests generate signals with motel, push them through a real pipeline, and assert invariants on what the sink receives.

The package depends only on the OTLP protocol, not on any particular pipeline implementation. Anything that exports OTLP/HTTP can be the system under test; the OpenTelemetry Collector is the reference target.

Index

• Constants
• Variables
• func CheckConservation(sent *Sent, received []*tracepb.Span) error
• func CheckNoFabrication(sent *Sent, received []*tracepb.Span) error
• func CheckParentsKept(received []*tracepb.Span) error
• func CheckWholeTraces(sent *Sent, received []*tracepb.Span) error
• func CollectorBinary() (string, bool)
• func ReceivedKeys(received []*tracepb.Span) map[string]struct{}
• func SpanKey(traceID, spanID []byte) string
• func SupportsComponent(name string) bool
• func TracesConfig(defs string, names ...string) string
• type Collector
  • func Start(sink *Sink, config string) (*Collector, error)
  • func (c *Collector) Stop() error
• type Sent
  • func NewSent() *Sent
  • func (s *Sent) Add(traceID, spanID []byte)
  • func (s *Sent) Has(key string) bool
  • func (s *Sent) Len() int
• type Sink
  • func NewSink() *Sink
  • func (s *Sink) Close()
  • func (s *Sink) Count() int
  • func (s *Sink) Reset()
  • func (s *Sink) Spans() []*tracepb.Span
  • func (s *Sink) URL() string
  • func (s *Sink) WaitFor(want int, timeout time.Duration) bool
  • func (s *Sink) WaitSettled(idle, max time.Duration) []*tracepb.Span

Constants

const BinaryEnv = "MOTEL_COLLECTOR_BIN"

BinaryEnv names the environment variable that overrides the collector binary the harness runs. When unset, the harness looks for "otelcol" on PATH.

Variables

var ErrNoCollector = errors.New("no collector binary found")

ErrNoCollector is returned by Start when no collector binary can be found. Tests use errors.Is to skip when the harness cannot run.

Functions

func CheckConservation

func CheckConservation(sent *Sent, received []*tracepb.Span) error

CheckConservation reports any discrepancy between the identities sent and the identities received: a pass-through pipeline must deliver every sent span and fabricate none. It compares identity sets, not raw counts, so redelivery of a span (OTLP is at-least-once) is not a violation; only a missing or fabricated identity is. This is the conservation invariant.

func CheckNoFabrication

func CheckNoFabrication(sent *Sent, received []*tracepb.Span) error

CheckNoFabrication reports a received span whose identity was never sent: a sampler may drop spans but must never invent new ones. It is deliberately a subset check, not an exactly-once check — OTLP delivery is at-least-once, so a correct pipeline may redeliver a span on retry, and that is not a fabrication.

func CheckParentsKept

func CheckParentsKept(received []*tracepb.Span) error

CheckParentsKept reports an orphaned span: a received span whose parent span was dropped. Root spans (a zero parent ID) are exempt. This is the parent-child preservation invariant.

func CheckWholeTraces

func CheckWholeTraces(sent *Sent, received []*tracepb.Span) error

CheckWholeTraces reports a partially sampled trace: a trace with at least one received span but some sent span missing. This is the trace completeness invariant. Traces the pipeline dropped entirely are exempt.

func CollectorBinary

func CollectorBinary() (string, bool)

CollectorBinary resolves the collector binary path from BinaryEnv, falling back to "otelcol" on PATH. The boolean reports whether one was found.

func ReceivedKeys

func ReceivedKeys(received []*tracepb.Span) map[string]struct{}

ReceivedKeys reduces captured spans to their identity set.

func SpanKey

func SpanKey(traceID, spanID []byte) string

SpanKey is the hex-encoded (trace ID, span ID) identity of a span. It is the join key between the spans a test sent into a pipeline and the spans the Sink received, so invariant checks can compare the two sets.

func SupportsComponent

func SupportsComponent(name string) bool

SupportsComponent reports whether the collector binary advertises a component with the given name in its `components` output (for example "tail_sampling"), so tests can skip cleanly on builds that lack an optional component. It parses the output structurally and matches component names only, not module paths, and returns false when no binary is available or the subcommand fails. The component set is resolved once per binary path.

func TracesConfig

func TracesConfig(defs string, names ...string) string

TracesConfig renders a collector config for a single traces pipeline whose processor stage is defs and names. defs is the YAML body of the top-level `processors:` map, indented two spaces as it appears under that key (empty for a pass-through pipeline); names are the processors the pipeline references, in order. The result is a text/template Start renders with the connection details (OTLPHTTPPort, HealthPort, SinkURL), so it is suitable to pass straight to Start. Callers that vary only the processor stage share one skeleton instead of copying the whole config.

Types

type Collector

type Collector struct {
	// OTLPEndpoint is the host:port of the collector's OTLP/HTTP receiver,
	// ready for an otlptracehttp exporter.
	OTLPEndpoint string
	// contains filtered or unexported fields
}

Collector is a running OpenTelemetry Collector subprocess.

func Start

func Start(sink *Sink, config string) (*Collector, error)

Start launches a collector that forwards traces to sink and waits until it reports healthy. The config is a text/template; pass "" for a pass-through pipeline. Available template fields: OTLPHTTPPort, HealthPort, SinkURL.

The caller owns the returned Collector and must call Stop. If no collector binary is available, Start returns ErrNoCollector.

func (*Collector) Stop

func (c *Collector) Stop() error

Stop terminates the collector and waits for it to exit.

type Sent

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

Sent records the span identities a test pushed into a pipeline, grouped so the invariant checks can compare them against what the Sink received. The harness is agnostic to how a caller generates spans: translate each sent span into its raw trace and span ID bytes and call Add.

func NewSent

func NewSent() *Sent

NewSent returns an empty Sent ready for Add.

func (*Sent) Add

func (s *Sent) Add(traceID, spanID []byte)

Add records one sent span. Repeated identities are recorded once.

func (*Sent) Has

func (s *Sent) Has(key string) bool

Has reports whether key was recorded by Add.

func (*Sent) Len

func (s *Sent) Len() int

Len returns the number of distinct spans recorded.

type Sink

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

Sink is an in-process OTLP/HTTP trace receiver that records every span it receives. A pipeline under test exports to Sink.URL; tests assert on the captured spans.

func NewSink

func NewSink() *Sink

NewSink starts a Sink listening on an ephemeral loopback port. Call Close to stop it.

func (*Sink) Close

func (s *Sink) Close()

Close stops the sink's HTTP server.

func (*Sink) Count

func (s *Sink) Count() int

Count returns the number of spans received so far.

func (*Sink) Reset

func (s *Sink) Reset()

Reset discards all received spans, readying the sink for reuse.

func (*Sink) Spans

func (s *Sink) Spans() []*tracepb.Span

Spans returns a copy of all spans received so far.

func (*Sink) URL

func (s *Sink) URL() string

URL is the base endpoint for an OTLP/HTTP exporter. The exporter appends /v1/traces to it.

func (*Sink) WaitFor

func (s *Sink) WaitFor(want int, timeout time.Duration) bool

WaitFor blocks until the sink holds at least want spans or timeout elapses, and reports whether the count was reached. Use it when the pipeline conserves spans, so an exact expected count exists.

func (*Sink) WaitSettled

func (s *Sink) WaitSettled(idle, max time.Duration) []*tracepb.Span

WaitSettled blocks until no new span has arrived for idle, or until max elapses, then returns a snapshot of the spans received so far. It is the bounded "eventually received" assertion for lossy or transforming pipelines: a sampler drops spans, so there is no exact count to wait for, and a pipeline that buffers (for example tail sampling's decision wait) releases spans in bursts. Choose idle longer than the pipeline's largest internal delay so a quiet period means the pipeline has drained, not that it is still deciding.