telemetry

package
v1.0.0-alpha.1 Latest Latest
Warning

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

 Go to latest Published: Oct 19, 2026 License: MIT Imports: 11 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrQueueFull = errors.New("transport queue full")

ErrQueueFull reports that a transport rejected an envelope because its queue is full.

Functions

This section is empty.

Types

type Bucket added in v0.38.0 

type Bucket[T any] struct {
	// contains filtered or unexported fields
}

type BucketedBuffer added in v0.38.0 

type BucketedBuffer[T any] struct {
	// contains filtered or unexported fields
}

BucketedBuffer groups items by trace id, flushing per bucket.

func NewBucketedBuffer added in v0.38.0 

func NewBucketedBuffer[T any](
	category ratelimit.Category,
	capacity int,
	overflowPolicy OverflowPolicy,
	batchSize int,
	timeout time.Duration,
	recorder report.ClientReportRecorder,
) *BucketedBuffer[T]

func (*BucketedBuffer[T]) AcceptedCount added in v0.38.0 

func (b *BucketedBuffer[T]) AcceptedCount() int64

func (*BucketedBuffer[T]) Capacity added in v0.38.0 

func (b *BucketedBuffer[T]) Capacity() int

func (*BucketedBuffer[T]) Category added in v0.38.0 

func (b *BucketedBuffer[T]) Category() ratelimit.Category

func (*BucketedBuffer[T]) Clear added in v0.38.0 

func (b *BucketedBuffer[T]) Clear()

func (*BucketedBuffer[T]) Drain added in v0.38.0 

func (b *BucketedBuffer[T]) Drain() []T

func (*BucketedBuffer[T]) DropRate added in v0.38.0 

func (b *BucketedBuffer[T]) DropRate() float64

func (*BucketedBuffer[T]) DroppedCount added in v0.38.0 

func (b *BucketedBuffer[T]) DroppedCount() int64

func (*BucketedBuffer[T]) GetMetrics added in v0.38.0 

func (b *BucketedBuffer[T]) GetMetrics() BufferMetrics

func (*BucketedBuffer[T]) IsEmpty added in v0.38.0 

func (b *BucketedBuffer[T]) IsEmpty() bool

func (*BucketedBuffer[T]) IsFull added in v0.38.0 

func (b *BucketedBuffer[T]) IsFull() bool

func (*BucketedBuffer[T]) IsReadyToFlush added in v0.38.0 

func (b *BucketedBuffer[T]) IsReadyToFlush() bool

func (*BucketedBuffer[T]) MarkFlushed added in v0.38.0 

func (b *BucketedBuffer[T]) MarkFlushed()

func (*BucketedBuffer[T]) Offer added in v0.38.0 

func (b *BucketedBuffer[T]) Offer(item T) bool

func (*BucketedBuffer[T]) OfferedCount added in v0.38.0 

func (b *BucketedBuffer[T]) OfferedCount() int64

func (*BucketedBuffer[T]) Peek added in v0.38.0 

func (b *BucketedBuffer[T]) Peek() (T, bool)

func (*BucketedBuffer[T]) Poll added in v0.38.0 

func (b *BucketedBuffer[T]) Poll() (T, bool)

func (*BucketedBuffer[T]) PollBatch added in v0.38.0 

func (b *BucketedBuffer[T]) PollBatch(maxItems int) []T

func (*BucketedBuffer[T]) PollIfReady added in v0.38.0 

func (b *BucketedBuffer[T]) PollIfReady() []T

func (*BucketedBuffer[T]) Priority added in v0.38.0 

func (b *BucketedBuffer[T]) Priority() ratelimit.Priority

func (*BucketedBuffer[T]) SetDroppedCallback added in v0.38.0 

func (b *BucketedBuffer[T]) SetDroppedCallback(callback func(item T, reason string))

func (*BucketedBuffer[T]) Size added in v0.38.0 

func (b *BucketedBuffer[T]) Size() int

func (*BucketedBuffer[T]) Utilization added in v0.38.0 

func (b *BucketedBuffer[T]) Utilization() float64

type Buffer 

type Buffer[T any] interface {
	// Core operations
	Offer(item T) bool
	Poll() (T, bool)
	PollBatch(maxItems int) []T
	PollIfReady() []T
	Drain() []T
	Peek() (T, bool)

	// State queries
	Size() int
	Capacity() int
	IsEmpty() bool
	IsFull() bool
	Utilization() float64

	// Flush management
	IsReadyToFlush() bool
	MarkFlushed()

	// Category/Priority
	Category() ratelimit.Category
	Priority() ratelimit.Priority

	// Metrics
	OfferedCount() int64
	DroppedCount() int64
	AcceptedCount() int64
	DropRate() float64
	GetMetrics() BufferMetrics

	// Configuration
	SetDroppedCallback(callback func(item T, reason string))
	Clear()
}

Buffer defines the common interface for all buffer implementations.

type BufferMetrics 

type BufferMetrics struct {
	Category      ratelimit.Category `json:"category"`
	Priority      ratelimit.Priority `json:"priority"`
	Capacity      int                `json:"capacity"`
	Size          int                `json:"size"`
	Utilization   float64            `json:"utilization"`
	OfferedCount  int64              `json:"offered_count"`
	DroppedCount  int64              `json:"dropped_count"`
	AcceptedCount int64              `json:"accepted_count"`
	DropRate      float64            `json:"drop_rate"`
	LastUpdated   time.Time          `json:"last_updated"`
}

type EnvelopeConvertible 

type EnvelopeConvertible interface {
	// GetCategory returns the rate limit category for this item.
	GetCategory() ratelimit.Category

	// GetEventID returns the event ID for this item.
	GetEventID() string

	// GetSdkInfo returns SDK information for the envelope header.
	GetSdkInfo() *protocol.SdkInfo

	// GetDynamicSamplingContext returns trace context for the envelope header.
	GetDynamicSamplingContext() map[string]string

	// ToEnvelope converts the item to a Sentry envelope using the provided header.
	ToEnvelope(*protocol.EnvelopeHeader) (*protocol.Envelope, error)
}

EnvelopeConvertible represents items that can convert themselves to envelopes.

type Item 

type Item interface {
	// GetCategory returns the rate limit category for this item.
	GetCategory() ratelimit.Category

	// MakeSerializationSafe prevents serialization races, by serializing user mutable data on the foreground.
	// Should be used before passing telemetry to the processor.
	MakeSerializationSafe()
}

Item represents any telemetry data that can be stored in buffers.

type ItemContainer 

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

func NewItemContainer 

func NewItemContainer(category ratelimit.Category, items []Item) ItemContainer

NewItemContainer constructs a batched envelope producer from buffered telemetry items.

func (ItemContainer) GetCategory 

func (b ItemContainer) GetCategory() ratelimit.Category

func (ItemContainer) GetDynamicSamplingContext 

func (ItemContainer) GetDynamicSamplingContext() map[string]string

func (ItemContainer) GetEventID 

func (ItemContainer) GetEventID() string

func (ItemContainer) GetSdkInfo 

func (ItemContainer) GetSdkInfo() *protocol.SdkInfo

func (ItemContainer) ToEnvelope 

func (b ItemContainer) ToEnvelope(header *protocol.EnvelopeHeader) (*protocol.Envelope, error)

func (ItemContainer) ToEnvelopeItem 

func (b ItemContainer) ToEnvelopeItem() (*protocol.EnvelopeItem, error)

type OverflowPolicy 

type OverflowPolicy int

OverflowPolicy defines how the ring buffer handles overflow. 

const (
	OverflowPolicyDropOldest OverflowPolicy = iota
	OverflowPolicyDropNewest
)

func (OverflowPolicy) String 

func (op OverflowPolicy) String() string

type Processor added in v0.42.0 

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

Processor is the top-level object that wraps the scheduler and buffers.

func NewProcessor added in v0.42.0 

func NewProcessor(
	buffers map[ratelimit.Category]Buffer[Item],
	transport Transport,
	dsn *protocol.Dsn,
	sdkInfo func() *protocol.SdkInfo,
	recorder report.ClientReportRecorder,
	provider report.ClientReportProvider,
) *Processor

NewProcessor creates a new Processor with the given configuration.

func (*Processor) Add added in v0.42.0 

func (b *Processor) Add(ctx context.Context, item Item) bool

Add buffers item if its category has a buffer and otherwise sends it inline.

The processor should call MakeSerializationSafe to eliminate any race on user mutable fields, since the serialization happens on a background goroutine.

func (*Processor) Close added in v0.42.0 

func (b *Processor) Close(timeout time.Duration)

Close stops the buffer, flushes remaining data, and releases resources.

func (*Processor) FlushWithContext added in v0.42.0 

func (b *Processor) FlushWithContext(ctx context.Context) bool

FlushWithContext flushes with a custom context for cancellation.

type RingBuffer added in v0.38.0 

type RingBuffer[T any] struct {
	// contains filtered or unexported fields
}

RingBuffer is a thread-safe ring buffer with overflow policies.

func NewRingBuffer added in v0.38.0 

func NewRingBuffer[T any](category ratelimit.Category, capacity int, overflowPolicy OverflowPolicy, batchSize int, timeout time.Duration, recorder report.ClientReportRecorder) *RingBuffer[T]

func (*RingBuffer[T]) AcceptedCount added in v0.38.0 

func (b *RingBuffer[T]) AcceptedCount() int64

func (*RingBuffer[T]) Capacity added in v0.38.0 

func (b *RingBuffer[T]) Capacity() int

func (*RingBuffer[T]) Category added in v0.38.0 

func (b *RingBuffer[T]) Category() ratelimit.Category

func (*RingBuffer[T]) Clear added in v0.38.0 

func (b *RingBuffer[T]) Clear()

func (*RingBuffer[T]) Drain added in v0.38.0 

func (b *RingBuffer[T]) Drain() []T

func (*RingBuffer[T]) DropRate added in v0.38.0 

func (b *RingBuffer[T]) DropRate() float64

func (*RingBuffer[T]) DroppedCount added in v0.38.0 

func (b *RingBuffer[T]) DroppedCount() int64

func (*RingBuffer[T]) GetMetrics added in v0.38.0 

func (b *RingBuffer[T]) GetMetrics() BufferMetrics

func (*RingBuffer[T]) IsEmpty added in v0.38.0 

func (b *RingBuffer[T]) IsEmpty() bool

func (*RingBuffer[T]) IsFull added in v0.38.0 

func (b *RingBuffer[T]) IsFull() bool

func (*RingBuffer[T]) IsReadyToFlush added in v0.38.0 

func (b *RingBuffer[T]) IsReadyToFlush() bool

func (*RingBuffer[T]) MarkFlushed added in v0.38.0 

func (b *RingBuffer[T]) MarkFlushed()

func (*RingBuffer[T]) Offer added in v0.38.0 

func (b *RingBuffer[T]) Offer(item T) bool

func (*RingBuffer[T]) OfferedCount added in v0.38.0 

func (b *RingBuffer[T]) OfferedCount() int64

func (*RingBuffer[T]) Peek added in v0.38.0 

func (b *RingBuffer[T]) Peek() (T, bool)

func (*RingBuffer[T]) Poll added in v0.38.0 

func (b *RingBuffer[T]) Poll() (T, bool)

func (*RingBuffer[T]) PollBatch added in v0.38.0 

func (b *RingBuffer[T]) PollBatch(maxItems int) []T

func (*RingBuffer[T]) PollIfReady added in v0.38.0 

func (b *RingBuffer[T]) PollIfReady() []T

func (*RingBuffer[T]) Priority added in v0.38.0 

func (b *RingBuffer[T]) Priority() ratelimit.Priority

func (*RingBuffer[T]) SetDroppedCallback added in v0.38.0 

func (b *RingBuffer[T]) SetDroppedCallback(callback func(item T, reason string))

func (*RingBuffer[T]) Size added in v0.38.0 

func (b *RingBuffer[T]) Size() int

func (*RingBuffer[T]) Utilization added in v0.38.0 

func (b *RingBuffer[T]) Utilization() float64

type Scheduler added in v0.38.0 

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

Scheduler implements a weighted round-robin scheduler for processing buffered events.

func NewScheduler added in v0.38.0 

func NewScheduler(
	buffers map[ratelimit.Category]Buffer[Item],
	transport Transport,
	dsn *protocol.Dsn,
	sdkInfo func() *protocol.SdkInfo,
	recorder report.ClientReportRecorder,
	provider report.ClientReportProvider,
) *Scheduler

func (*Scheduler) Add added in v0.38.0 

func (s *Scheduler) Add(item Item) bool

func (*Scheduler) FlushWithContext added in v0.38.0 

func (s *Scheduler) FlushWithContext(ctx context.Context) bool

func (*Scheduler) Signal added in v0.38.0 

func (s *Scheduler) Signal()

func (*Scheduler) Start added in v0.38.0 

func (s *Scheduler) Start()

func (*Scheduler) Stop added in v0.38.0 

func (s *Scheduler) Stop(timeout time.Duration)

type TraceAware added in v0.38.0 

type TraceAware interface {
	GetTraceID() (string, bool)
}

TraceAware is implemented by items that can expose a trace ID. BucketedBuffer uses this to group items by trace.

type Transport 

type Transport interface {
	// SendEnvelope sends an envelope to Sentry. Returns an error only if
	// the envelope is rejected.
	SendEnvelope(ctx context.Context, envelope *protocol.Envelope) error

	// FlushWithContext waits for all pending envelopes to be sent.
	FlushWithContext(ctx context.Context) bool

	// Close shuts down the transport gracefully
	Close()
}

Transport represents the envelope-first transport interface. This interface is designed for the telemetry buffer system and provides sends with backpressure signals.

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.