otlp

Documentation

Index

• func ConvertLogs(req *logscollpb.ExportLogsServiceRequest) []*chstore.Log
• func ConvertMetrics(req *metricscollpb.ExportMetricsServiceRequest) ([]*chstore.MetricPoint, []*chstore.ExemplarRow)
• func ConvertTraces(req *tracecollpb.ExportTraceServiceRequest) ([]*chstore.Span, []*chstore.SpanLinkRow)
• func HTTPHandler(ing *Ingester) http.Handler
• func SeriesFingerprint(metricName string, dpAttrs []*commonpb.KeyValue, ...) uint64
• type GRPCHandle
  • func StartGRPC(addr string, ing *Ingester) (*GRPCHandle, error)
  • func (h *GRPCHandle) Shutdown(grace time.Duration)
• type Ingester
  • func NewIngester(spans *consumer.Consumer[*chstore.Span], logs *consumer.Consumer[*chstore.Log], ...) *Ingester
  • func (ing *Ingester) ExemplarsDroppedCapped() uint64
  • func (ing *Ingester) ExemplarsDroppedNoTrace() uint64
  • func (ing *Ingester) ExemplarsIngested() uint64
  • func (ing *Ingester) LogsDroppedByPipeline() uint64
  • func (ing *Ingester) MetricsDroppedByPipeline() uint64
  • func (ing *Ingester) SetAutocomplete(a *acache.Store)
  • func (ing *Ingester) SetExemplarCap(n int)
  • func (ing *Ingester) SetExemplarPolicy(requireTraceContext bool)
  • func (ing *Ingester) SetExemplars(c *consumer.Consumer[*chstore.ExemplarRow])
  • func (ing *Ingester) SetPipeline(p *pipeline.Engine)
  • func (ing *Ingester) SetSpanLinks(c *consumer.Consumer[*chstore.SpanLinkRow])
  • func (ing *Ingester) SpanLinksDroppedInvalid() uint64
  • func (ing *Ingester) SpanLinksIngested() uint64
  • func (ing *Ingester) SpansDroppedByPipeline() uint64

Functions

func ConvertLogs

func ConvertLogs(req *logscollpb.ExportLogsServiceRequest) []*chstore.Log

func ConvertMetrics

func ConvertMetrics(req *metricscollpb.ExportMetricsServiceRequest) ([]*chstore.MetricPoint, []*chstore.ExemplarRow)

ConvertMetrics returns the metric points plus the OTLP exemplars extracted from their datapoints (v0.8.328 — previously dp.Exemplars was silently dropped, pivot-audit §2). Exemplar rows carry the SAME series fingerprint as their datapoint's metric row — the metric→trace pivot join key. The require-trace-context gate is NOT applied here (pure conversion, unit- testable); the Ingester's addExemplar applies it so HTTP + gRPC share one policy + one set of counters.

func ConvertTraces

func ConvertTraces(req *tracecollpb.ExportTraceServiceRequest) ([]*chstore.Span, []*chstore.SpanLinkRow)

ConvertTraces returns the span rows plus the span-link rows extracted from them (v0.8.329 — previously sp.Links was silently dropped, pivot-audit §2; mirrors how ConvertMetrics returns exemplars since v0.8.328). Link rows carry the OWNING span's identity (trace/span id, start time, service) — the forward key of the span_links table. The invalid-link gate (empty linked trace id) is NOT applied here: pure conversion, unit-testable; the Ingester's addSpanLink applies it so HTTP + gRPC share one gate + one set of counters (the v0.8.328 exemplar split, same reason).

func HTTPHandler

func HTTPHandler(ing *Ingester) http.Handler

HTTPHandler returns an http.Handler that accepts OTLP/HTTP (protobuf + JSON).

func SeriesFingerprint

func SeriesFingerprint(metricName string, dpAttrs []*commonpb.KeyValue, serviceName, serviceInstanceID string) uint64

SeriesFingerprint returns the stable identity of one metric series: (metric name, datapoint attribute set, service.name + service.instance.id). Resource identity is deliberately limited to those two keys (pivot-audit open question #3, approved): per-instance series pivot per-instance; service-level rollups use the metric+service fallback read path.

Types

type GRPCHandle

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

GRPCHandle is the shutdown handle StartGRPC returns (v0.8.336, HA audit H1). Opaque on purpose: main owns WHEN to stop accepting, the otlp package owns HOW (graceful-with-bound), and main never imports google.golang.org/grpc.

func StartGRPC

func StartGRPC(addr string, ing *Ingester) (*GRPCHandle, error)

StartGRPC listens, registers the OTLP services and serves in a background goroutine, returning a handle main GracefulStops during shutdown (v0.8.336, HA audit H1 — the server used to outlive the consumers: post-SIGTERM Exports were ACKed into channels nobody drained, and the abrupt connection cut on process exit is the client-side trigger for the otelcol zero-addresses wedge).

func (*GRPCHandle) Shutdown

func (h *GRPCHandle) Shutdown(grace time.Duration)

Shutdown drains gracefully — GOAWAY to collectors, in-flight Exports finish — but never longer than `grace`: GracefulStop can hang on a stuck stream, and a shutdown that outlives terminationGracePeriod gets SIGKILLed mid-drain, losing more than the hard Stop would.

type Ingester

type Ingester struct {
	Spans   *consumer.Consumer[*chstore.Span]
	Logs    *consumer.Consumer[*chstore.Log]
	Metrics *consumer.Consumer[*chstore.MetricPoint]
	// Exemplars (v0.8.328) — OTLP metric exemplars extracted alongside the
	// metric points (cross-signal pivot). Its flusher batches into
	// chstore.InsertExemplars with the same consumer machinery / flush
	// cadence as Metrics. nil-safe: addExemplar no-ops the enqueue (counters
	// still tick) on pods where main didn't wire it.
	Exemplars *consumer.Consumer[*chstore.ExemplarRow]
	// SpanLinks (v0.8.329) — OTel span links extracted alongside the span
	// rows (cross-signal pivot Phase 1b). Its flusher batches into
	// chstore.InsertSpanLinks with the same consumer machinery / flush
	// cadence as Spans. nil-safe: addSpanLink no-ops the enqueue (counters
	// still tick) on pods where main didn't wire it.
	SpanLinks *consumer.Consumer[*chstore.SpanLinkRow]
	// contains filtered or unexported fields
}

Ingester wires the three OTLP consumers together. Every received span is stored — in-binary head/tail sampling was removed (v0.8.73); sampling, when wanted, happens at the collector (the sample-to-Coremetry / 100%-to-Tempo split). An optional pipeline drop/enrich engine still runs first.

func NewIngester

func NewIngester(
	spans *consumer.Consumer[*chstore.Span],
	logs *consumer.Consumer[*chstore.Log],
	metrics *consumer.Consumer[*chstore.MetricPoint],
) *Ingester

func (*Ingester) ExemplarsDroppedCapped

func (ing *Ingester) ExemplarsDroppedCapped() uint64

func (*Ingester) ExemplarsDroppedNoTrace

func (ing *Ingester) ExemplarsDroppedNoTrace() uint64

func (*Ingester) ExemplarsIngested

func (ing *Ingester) ExemplarsIngested() uint64

ExemplarsIngested / ExemplarsDroppedNoTrace are the two v0.8.328 exemplar ingest totals surfaced on /admin/stats (SystemStats.Exemplars), following the pipeline-counter accessor pattern above.

func (*Ingester) LogsDroppedByPipeline

func (ing *Ingester) LogsDroppedByPipeline() uint64

LogsDroppedByPipeline / MetricsDroppedByPipeline mirror the span accessor (v0.8.282). Surfaced on /admin/stats so the operator sees how many logs / metric points a pipeline rule discarded before the consumer buffer. Distinct from the queue-full / write-failed loss counters — pipeline drops are INTENTIONAL, not data loss.

func (*Ingester) MetricsDroppedByPipeline

func (ing *Ingester) MetricsDroppedByPipeline() uint64

func (*Ingester) SetAutocomplete

func (ing *Ingester) SetAutocomplete(a *acache.Store)

SetAutocomplete wires the Redis autocomplete cache so every accepted span also populates the service/operation/attribute picker facets. Called from main(); nil keeps the old behaviour (no autocomplete cache).

func (*Ingester) SetExemplarCap

func (ing *Ingester) SetExemplarCap(n int)

SetExemplarCap arms the per-series×minute ingest cap (v0.8.433, exemplar audit Faz C). n <= 0 keeps the unlimited default. Called at boot before traffic like SetExemplarPolicy — not safe to flip live.

func (*Ingester) SetExemplarPolicy

func (ing *Ingester) SetExemplarPolicy(requireTraceContext bool)

SetExemplarPolicy applies config exemplars.require_trace_context (default true). require=false stores trace-less exemplars too — useful when the operator wants the value/attr context even without a click-through target.

func (*Ingester) SetExemplars

func (ing *Ingester) SetExemplars(c *consumer.Consumer[*chstore.ExemplarRow])

SetExemplars wires the exemplar consumer (v0.8.328). Called from main(); nil keeps addExemplar counting but not enqueueing (api-only pods).

func (*Ingester) SetPipeline

func (ing *Ingester) SetPipeline(p *pipeline.Engine)

SetPipeline wires the ingest-time policy engine. Always called from main(); nil keeps the old behaviour (every span flows through unchanged).

func (*Ingester) SetSpanLinks

func (ing *Ingester) SetSpanLinks(c *consumer.Consumer[*chstore.SpanLinkRow])

SetSpanLinks wires the span-link consumer (v0.8.329). Called from main(); nil keeps addSpanLink counting but not enqueueing (api-only pods).

func (*Ingester) SpanLinksDroppedInvalid

func (ing *Ingester) SpanLinksDroppedInvalid() uint64

func (*Ingester) SpanLinksIngested

func (ing *Ingester) SpanLinksIngested() uint64

SpanLinksIngested / SpanLinksDroppedInvalid are the two v0.8.329 span-link ingest totals surfaced on /admin/stats (SystemStats.SpanLinks), following the exemplar-counter accessor pattern above.

func (*Ingester) SpansDroppedByPipeline

func (ing *Ingester) SpansDroppedByPipeline() uint64

SpansDroppedByPipeline is a monotonic counter of spans dropped by a pipeline rule since boot. Surfaced on /api/health alongside sampler-dropped + buffer-dropped counters so the operator can see the policy engine's effect at a glance.