ingestion

package
v0.2.2 Latest Latest
Warning

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

 Go to latest Published: Jun 7, 2026 License: MIT Imports: 21 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func HistogramPercentile 

func HistogramPercentile(bounds []float64, counts []uint64, p float64) float64

HistogramPercentile approximates the given percentile from explicit-bucket histogram data. bounds is an ascending list of upper bounds (length N); counts is the per-bucket counts (length N+1, the last bucket has no upper bound). Uses linear interpolation within the bucket that contains the percentile. Returns 0 if the histogram has no observations.

This is the exported entry point so the algorithm can be unit-tested without pmetric machinery.

Types

type AlwaysKeep added in v0.2.0 

type AlwaysKeep uint8

AlwaysKeep is a bitfield of the signal categories that bypass the token bucket. 

const (
	KeepErrors   AlwaysKeep = 1 << iota // status_code = 2
	KeepNPlusOne                        // trace has an n_plus_one issue
	KeepSlow                            // duration > 2× per-service p95
)

func ParseAlwaysKeep added in v0.2.0 

func ParseAlwaysKeep(s string) AlwaysKeep

ParseAlwaysKeep parses a comma-separated list of "error", "n_plus_one", "slow", "none".

type CacheMissStormDetector added in v0.2.0 

type CacheMissStormDetector struct{}

CacheMissStormDetector fires when ≥ 10 cache-miss spans cluster within a 50ms window.

func (*CacheMissStormDetector) Analyze added in v0.2.0 

func (*CacheMissStormDetector) Analyze(traceID, sessionID string, spans []*storage.Span, now int64) []*storage.TraceIssue

func (*CacheMissStormDetector) Kind added in v0.2.0 

func (*CacheMissStormDetector) Kind() string

type ChattyHTTPDetector added in v0.2.0 

type ChattyHTTPDetector struct{}

ChattyHTTPDetector fires when the same HTTP host is called ≥ 5 times in one trace.

func (*ChattyHTTPDetector) Analyze added in v0.2.0 

func (*ChattyHTTPDetector) Analyze(traceID, sessionID string, spans []*storage.Span, now int64) []*storage.TraceIssue

func (*ChattyHTTPDetector) Kind added in v0.2.0 

func (*ChattyHTTPDetector) Kind() string

type Counters added in v0.2.0 

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

Counters holds process-level drop counters. All fields are atomically updated. It implements api.DropCounterProvider via the four exported methods below.

func (*Counters) DroppedLogsTotal added in v0.2.0 

func (c *Counters) DroppedLogsTotal() int64

DroppedLogsTotal implements api.DropCounterProvider.

func (*Counters) DroppedMetricPointsTotal added in v0.2.0 

func (c *Counters) DroppedMetricPointsTotal() int64

DroppedMetricPointsTotal implements api.DropCounterProvider.

func (*Counters) DroppedSpansTotal added in v0.2.0 

func (c *Counters) DroppedSpansTotal() int64

DroppedSpansTotal implements api.DropCounterProvider.

func (*Counters) LastDropAt added in v0.2.0 

func (c *Counters) LastDropAt() int64

LastDropAt implements api.DropCounterProvider.

func (*Counters) Seed added in v0.2.0 

func (c *Counters) Seed(spans, logs, metrics int64)

Seed sets the initial values for the drop counters, loaded from persistent storage.

type Detector added in v0.2.0 

type Detector interface {
	Kind() string
	Analyze(traceID, sessionID string, spans []*storage.Span, now int64) []*storage.TraceIssue
}

Detector analyzes a trace's spans and returns any detected issues. Each implementation lives in its own detector_*.go file.

type ErrorChainDetector added in v0.2.0 

type ErrorChainDetector struct{}

ErrorChainDetector fires when a child span errored but its parent did not propagate the error (parent.status_code != 2 while child.status_code == 2).

func (*ErrorChainDetector) Analyze added in v0.2.0 

func (*ErrorChainDetector) Analyze(traceID, sessionID string, spans []*storage.Span, now int64) []*storage.TraceIssue

func (*ErrorChainDetector) Kind added in v0.2.0 

func (*ErrorChainDetector) Kind() string

type LargePayloadDetector added in v0.2.0 

type LargePayloadDetector struct{}

LargePayloadDetector fires when http.response_content_length exceeds 1 MB.

func (*LargePayloadDetector) Analyze added in v0.2.0 

func (*LargePayloadDetector) Analyze(traceID, sessionID string, spans []*storage.Span, now int64) []*storage.TraceIssue

func (*LargePayloadDetector) Kind added in v0.2.0 

func (*LargePayloadDetector) Kind() string

type LintRule 

type LintRule struct {
	ID       string
	Severity string
	Check    func(s *storage.Span) (string, bool)
}

type N1Detector added in v0.2.0 

type N1Detector struct{}

N1Detector fires when ≥ 10 spans share a SQL fingerprint under the same parent.

func (*N1Detector) Analyze added in v0.2.0 

func (*N1Detector) Analyze(traceID, sessionID string, spans []*storage.Span, now int64) []*storage.TraceIssue

func (*N1Detector) Kind added in v0.2.0 

func (*N1Detector) Kind() string

type Pipeline 

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

func NewPipeline 

func NewPipeline(store *storage.DB, hub *ws.Hub) *Pipeline

func NewPipelineFull added in v0.2.0 

func NewPipelineFull(store *storage.DB, hub *ws.Hub, s *Sampler, lim *SourceLimiter) *Pipeline

func NewPipelineWithSampler added in v0.2.0 

func NewPipelineWithSampler(store *storage.DB, hub *ws.Hub, s *Sampler) *Pipeline

func (*Pipeline) DropCounters added in v0.2.0 

func (p *Pipeline) DropCounters() *Counters

DropCounters returns the sampler's drop counters for surfacing in /api/stats.

func (*Pipeline) IngestLogs 

func (p *Pipeline) IngestLogs(ctx context.Context, logs plog.Logs) error

func (*Pipeline) IngestMetrics 

func (p *Pipeline) IngestMetrics(ctx context.Context, md pmetric.Metrics) error

func (*Pipeline) IngestTraces 

func (p *Pipeline) IngestTraces(ctx context.Context, traces ptrace.Traces) error

func (*Pipeline) SetSelfService added in v0.2.0 

func (p *Pipeline) SetSelfService(name string)

SetSelfService configures the service.name treated as Spaniel's own self-telemetry; ingests from it are stored without generating new spans.

func (*Pipeline) Sources added in v0.2.0 

func (p *Pipeline) Sources() []storage.SourceStats

Sources returns a per-second stats snapshot for every source seen.

func (*Pipeline) Throughput added in v0.2.0 

func (p *Pipeline) Throughput() storage.Throughput

Throughput returns the current rolling per-second ingest rates.

type Sampler added in v0.2.0 

type Sampler struct {
	Counters Counters
	// contains filtered or unexported fields
}

Sampler applies tail-based sampling to incoming telemetry. When budget == 0 every span/log/metric is accepted (default off behaviour).

func NewSampler added in v0.2.0 

func NewSampler(budget int, keep AlwaysKeep) *Sampler

NewSampler creates a Sampler. budget=0 disables rate limiting.

func (*Sampler) DecideLog added in v0.2.0 

func (s *Sampler) DecideLog() bool

DecideLog always returns true when budget==0; otherwise consumes a token.

func (*Sampler) DecideMetric added in v0.2.0 

func (s *Sampler) DecideMetric() bool

DecideMetric always returns true when budget==0; otherwise consumes a token.

func (*Sampler) DecideSpan added in v0.2.0 

func (s *Sampler) DecideSpan(span *storage.Span) bool

DecideSpan returns true if the span should be stored, false if it should be dropped. Error spans and other always-keep signals bypass the budget.

func (*Sampler) DecideTrace added in v0.2.0 

func (s *Sampler) DecideTrace(spans []*storage.Span, hasN1 bool, serviceP95 map[string]int64) bool

DecideTrace makes a keep/drop decision for an entire buffered trace. It checks always-keep signals first; if none fire it falls back to the token bucket, consuming n tokens (one per span).

func (*Sampler) NeedsBuffer added in v0.2.0 

func (s *Sampler) NeedsBuffer() bool

NeedsBuffer reports whether the sampler requires a per-trace buffer before the store/drop decision can be made (i.e., N+1 or Slow signals are enabled and rate limiting is active). When budget==0 every span is accepted regardless of signal, so buffering would only add latency with no benefit.

type SerialPromiseDetector added in v0.2.0 

type SerialPromiseDetector struct{}

SerialPromiseDetector fires when ≥ 3 sequential outbound calls under one parent could be parallelised (total child time ≈ parent duration).

func (*SerialPromiseDetector) Analyze added in v0.2.0 

func (*SerialPromiseDetector) Analyze(traceID, sessionID string, spans []*storage.Span, now int64) []*storage.TraceIssue

func (*SerialPromiseDetector) Kind added in v0.2.0 

func (*SerialPromiseDetector) Kind() string

type SlowDBDetector added in v0.2.0 

type SlowDBDetector struct{}

SlowDBDetector fires when a single DB span exceeds 250 ms.

func (*SlowDBDetector) Analyze added in v0.2.0 

func (*SlowDBDetector) Analyze(traceID, sessionID string, spans []*storage.Span, now int64) []*storage.TraceIssue

func (*SlowDBDetector) Kind added in v0.2.0 

func (*SlowDBDetector) Kind() string

type SourceLimiter added in v0.2.0 

type SourceLimiter struct {
	RPS   float64 // fill rate per source; 0 = unlimited
	Burst int     // bucket capacity; defaults to RPS*5 when 0
	// contains filtered or unexported fields
}

SourceLimiter manages per-source token buckets and rolling stats. RPS==0 disables rate limiting (all spans accepted) but still tracks stats.

func NewSourceLimiter added in v0.2.0 

func NewSourceLimiter(rps float64, burst int) *SourceLimiter

NewSourceLimiter creates a limiter. rps==0 means track-only (no limiting).

func (*SourceLimiter) Allow added in v0.2.0 

func (sl *SourceLimiter) Allow(service string, isError bool, bytes int64) bool

Allow checks whether a span from the given service should be accepted. It always records the outcome in the rolling window. bytes is an estimate of the span payload size (used for bytes/s stats).

func (*SourceLimiter) Snapshot added in v0.2.0 

func (sl *SourceLimiter) Snapshot() []storage.SourceStats

Snapshot returns a per-second stats row for every source seen so far.

type SynchronousIODetector added in v0.2.0 

type SynchronousIODetector struct{}

SynchronousIODetector fires when a server-kind GET handler executes ≥ 3 DB spans inline (blocking I/O on a latency-sensitive path).

func (*SynchronousIODetector) Analyze added in v0.2.0 

func (*SynchronousIODetector) Analyze(traceID, sessionID string, spans []*storage.Span, now int64) []*storage.TraceIssue

func (*SynchronousIODetector) Kind added in v0.2.0 

func (*SynchronousIODetector) Kind() string

type TracingGapDetector added in v0.2.0 

type TracingGapDetector struct{}

TracingGapDetector fires when a parent span has > 50ms of duration not covered by any child span (unexplained latency).

func (*TracingGapDetector) Analyze added in v0.2.0 

func (*TracingGapDetector) Analyze(traceID, sessionID string, spans []*storage.Span, now int64) []*storage.TraceIssue

func (*TracingGapDetector) Kind added in v0.2.0 

func (*TracingGapDetector) Kind() string

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.