Update dependencies

vendor/github.com/getsentry/sentry-go/internal/telemetry/bucketed_buffer.go

@@ -0,0 +1,398 @@
+package telemetry
+
+import (
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/getsentry/sentry-go/internal/ratelimit"
+)
+
+const (
+	defaultBucketedCapacity = 100
+	perBucketItemLimit      = 100
+)
+
+type Bucket[T any] struct {
+	traceID       string
+	items         []T
+	createdAt     time.Time
+	lastUpdatedAt time.Time
+}
+
+// BucketedBuffer groups items by trace id, flushing per bucket.
+type BucketedBuffer[T any] struct {
+	mu sync.RWMutex
+
+	buckets    []*Bucket[T]
+	traceIndex map[string]int
+
+	head int
+	tail int
+
+	itemCapacity   int
+	bucketCapacity int
+
+	totalItems  int
+	bucketCount int
+
+	category       ratelimit.Category
+	priority       ratelimit.Priority
+	overflowPolicy OverflowPolicy
+	batchSize      int
+	timeout        time.Duration
+	lastFlushTime  time.Time
+
+	offered   int64
+	dropped   int64
+	onDropped func(item T, reason string)
+}
+
+func NewBucketedBuffer[T any](
+	category ratelimit.Category,
+	capacity int,
+	overflowPolicy OverflowPolicy,
+	batchSize int,
+	timeout time.Duration,
+) *BucketedBuffer[T] {
+	if capacity <= 0 {
+		capacity = defaultBucketedCapacity
+	}
+	if batchSize <= 0 {
+		batchSize = 1
+	}
+	if timeout < 0 {
+		timeout = 0
+	}
+
+	bucketCapacity := capacity / 10
+	if bucketCapacity < 10 {
+		bucketCapacity = 10
+	}
+
+	return &BucketedBuffer[T]{
+		buckets:        make([]*Bucket[T], bucketCapacity),
+		traceIndex:     make(map[string]int),
+		itemCapacity:   capacity,
+		bucketCapacity: bucketCapacity,
+		category:       category,
+		priority:       category.GetPriority(),
+		overflowPolicy: overflowPolicy,
+		batchSize:      batchSize,
+		timeout:        timeout,
+		lastFlushTime:  time.Now(),
+	}
+}
+
+func (b *BucketedBuffer[T]) Offer(item T) bool {
+	atomic.AddInt64(&b.offered, 1)
+
+	traceID := ""
+	if ta, ok := any(item).(TraceAware); ok {
+		if tid, hasTrace := ta.GetTraceID(); hasTrace {
+			traceID = tid
+		}
+	}
+
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.offerToBucket(item, traceID)
+}
+
+func (b *BucketedBuffer[T]) offerToBucket(item T, traceID string) bool {
+	if traceID != "" {
+		if idx, exists := b.traceIndex[traceID]; exists {
+			bucket := b.buckets[idx]
+			if len(bucket.items) >= perBucketItemLimit {
+				delete(b.traceIndex, traceID)
+			} else {
+				bucket.items = append(bucket.items, item)
+				bucket.lastUpdatedAt = time.Now()
+				b.totalItems++
+				return true
+			}
+		}
+	}
+
+	if b.totalItems >= b.itemCapacity {
+		return b.handleOverflow(item, traceID)
+	}
+	if b.bucketCount >= b.bucketCapacity {
+		return b.handleOverflow(item, traceID)
+	}
+
+	bucket := &Bucket[T]{
+		traceID:       traceID,
+		items:         []T{item},
+		createdAt:     time.Now(),
+		lastUpdatedAt: time.Now(),
+	}
+	b.buckets[b.tail] = bucket
+	if traceID != "" {
+		b.traceIndex[traceID] = b.tail
+	}
+	b.tail = (b.tail + 1) % b.bucketCapacity
+	b.bucketCount++
+	b.totalItems++
+	return true
+}
+
+func (b *BucketedBuffer[T]) handleOverflow(item T, traceID string) bool {
+	switch b.overflowPolicy {
+	case OverflowPolicyDropOldest:
+		oldestBucket := b.buckets[b.head]
+		if oldestBucket == nil {
+			atomic.AddInt64(&b.dropped, 1)
+			if b.onDropped != nil {
+				b.onDropped(item, "buffer_full_invalid_state")
+			}
+			return false
+		}
+		if oldestBucket.traceID != "" {
+			delete(b.traceIndex, oldestBucket.traceID)
+		}
+		droppedCount := len(oldestBucket.items)
+		atomic.AddInt64(&b.dropped, int64(droppedCount))
+		if b.onDropped != nil {
+			for _, di := range oldestBucket.items {
+				b.onDropped(di, "buffer_full_drop_oldest_bucket")
+			}
+		}
+		b.totalItems -= droppedCount
+		b.bucketCount--
+		b.head = (b.head + 1) % b.bucketCapacity
+		// add new bucket
+		bucket := &Bucket[T]{traceID: traceID, items: []T{item}, createdAt: time.Now(), lastUpdatedAt: time.Now()}
+		b.buckets[b.tail] = bucket
+		if traceID != "" {
+			b.traceIndex[traceID] = b.tail
+		}
+		b.tail = (b.tail + 1) % b.bucketCapacity
+		b.bucketCount++
+		b.totalItems++
+		return true
+	case OverflowPolicyDropNewest:
+		atomic.AddInt64(&b.dropped, 1)
+		if b.onDropped != nil {
+			b.onDropped(item, "buffer_full_drop_newest")
+		}
+		return false
+	default:
+		atomic.AddInt64(&b.dropped, 1)
+		if b.onDropped != nil {
+			b.onDropped(item, "unknown_overflow_policy")
+		}
+		return false
+	}
+}
+
+func (b *BucketedBuffer[T]) Poll() (T, bool) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	var zero T
+	if b.bucketCount == 0 {
+		return zero, false
+	}
+	bucket := b.buckets[b.head]
+	if bucket == nil || len(bucket.items) == 0 {
+		return zero, false
+	}
+	item := bucket.items[0]
+	bucket.items = bucket.items[1:]
+	b.totalItems--
+	if len(bucket.items) == 0 {
+		if bucket.traceID != "" {
+			delete(b.traceIndex, bucket.traceID)
+		}
+		b.buckets[b.head] = nil
+		b.head = (b.head + 1) % b.bucketCapacity
+		b.bucketCount--
+	}
+	return item, true
+}
+
+func (b *BucketedBuffer[T]) PollBatch(maxItems int) []T {
+	if maxItems <= 0 {
+		return nil
+	}
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if b.bucketCount == 0 {
+		return nil
+	}
+	res := make([]T, 0, maxItems)
+	for len(res) < maxItems && b.bucketCount > 0 {
+		bucket := b.buckets[b.head]
+		if bucket == nil {
+			break
+		}
+		n := maxItems - len(res)
+		if n > len(bucket.items) {
+			n = len(bucket.items)
+		}
+		res = append(res, bucket.items[:n]...)
+		bucket.items = bucket.items[n:]
+		b.totalItems -= n
+		if len(bucket.items) == 0 {
+			if bucket.traceID != "" {
+				delete(b.traceIndex, bucket.traceID)
+			}
+			b.buckets[b.head] = nil
+			b.head = (b.head + 1) % b.bucketCapacity
+			b.bucketCount--
+		}
+	}
+	return res
+}
+
+func (b *BucketedBuffer[T]) PollIfReady() []T {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if b.bucketCount == 0 {
+		return nil
+	}
+	ready := b.totalItems >= b.batchSize || (b.timeout > 0 && time.Since(b.lastFlushTime) >= b.timeout)
+	if !ready {
+		return nil
+	}
+	oldest := b.buckets[b.head]
+	if oldest == nil {
+		return nil
+	}
+	items := oldest.items
+	if oldest.traceID != "" {
+		delete(b.traceIndex, oldest.traceID)
+	}
+	b.buckets[b.head] = nil
+	b.head = (b.head + 1) % b.bucketCapacity
+	b.totalItems -= len(items)
+	b.bucketCount--
+	b.lastFlushTime = time.Now()
+	return items
+}
+
+func (b *BucketedBuffer[T]) Drain() []T {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if b.bucketCount == 0 {
+		return nil
+	}
+	res := make([]T, 0, b.totalItems)
+	for i := 0; i < b.bucketCount; i++ {
+		idx := (b.head + i) % b.bucketCapacity
+		bucket := b.buckets[idx]
+		if bucket != nil {
+			res = append(res, bucket.items...)
+			b.buckets[idx] = nil
+		}
+	}
+	b.traceIndex = make(map[string]int)
+	b.head = 0
+	b.tail = 0
+	b.totalItems = 0
+	b.bucketCount = 0
+	return res
+}
+
+func (b *BucketedBuffer[T]) Peek() (T, bool) {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+	var zero T
+	if b.bucketCount == 0 {
+		return zero, false
+	}
+	bucket := b.buckets[b.head]
+	if bucket == nil || len(bucket.items) == 0 {
+		return zero, false
+	}
+	return bucket.items[0], true
+}
+
+func (b *BucketedBuffer[T]) Size() int     { b.mu.RLock(); defer b.mu.RUnlock(); return b.totalItems }
+func (b *BucketedBuffer[T]) Capacity() int { b.mu.RLock(); defer b.mu.RUnlock(); return b.itemCapacity }
+func (b *BucketedBuffer[T]) Category() ratelimit.Category {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+	return b.category
+}
+func (b *BucketedBuffer[T]) Priority() ratelimit.Priority {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+	return b.priority
+}
+func (b *BucketedBuffer[T]) IsEmpty() bool {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+	return b.bucketCount == 0
+}
+func (b *BucketedBuffer[T]) IsFull() bool {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+	return b.totalItems >= b.itemCapacity
+}
+func (b *BucketedBuffer[T]) Utilization() float64 {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+	if b.itemCapacity == 0 {
+		return 0
+	}
+	return float64(b.totalItems) / float64(b.itemCapacity)
+}
+func (b *BucketedBuffer[T]) OfferedCount() int64  { return atomic.LoadInt64(&b.offered) }
+func (b *BucketedBuffer[T]) DroppedCount() int64  { return atomic.LoadInt64(&b.dropped) }
+func (b *BucketedBuffer[T]) AcceptedCount() int64 { return b.OfferedCount() - b.DroppedCount() }
+func (b *BucketedBuffer[T]) DropRate() float64 {
+	off := b.OfferedCount()
+	if off == 0 {
+		return 0
+	}
+	return float64(b.DroppedCount()) / float64(off)
+}
+
+func (b *BucketedBuffer[T]) GetMetrics() BufferMetrics {
+	b.mu.RLock()
+	size := b.totalItems
+	util := 0.0
+	if b.itemCapacity > 0 {
+		util = float64(b.totalItems) / float64(b.itemCapacity)
+	}
+	b.mu.RUnlock()
+	return BufferMetrics{Category: b.category, Priority: b.priority, Capacity: b.itemCapacity, Size: size, Utilization: util, OfferedCount: b.OfferedCount(), DroppedCount: b.DroppedCount(), AcceptedCount: b.AcceptedCount(), DropRate: b.DropRate(), LastUpdated: time.Now()}
+}
+
+func (b *BucketedBuffer[T]) SetDroppedCallback(callback func(item T, reason string)) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.onDropped = callback
+}
+func (b *BucketedBuffer[T]) Clear() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	for i := 0; i < b.bucketCapacity; i++ {
+		b.buckets[i] = nil
+	}
+	b.traceIndex = make(map[string]int)
+	b.head = 0
+	b.tail = 0
+	b.totalItems = 0
+	b.bucketCount = 0
+}
+func (b *BucketedBuffer[T]) IsReadyToFlush() bool {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+	if b.bucketCount == 0 {
+		return false
+	}
+	if b.totalItems >= b.batchSize {
+		return true
+	}
+	if b.timeout > 0 && time.Since(b.lastFlushTime) >= b.timeout {
+		return true
+	}
+	return false
+}
+func (b *BucketedBuffer[T]) MarkFlushed() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.lastFlushTime = time.Now()
+}

vendor/github.com/getsentry/sentry-go/internal/telemetry/buffer.go

@@ -0,0 +1,378 @@
+package telemetry
+
+import (
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/getsentry/sentry-go/internal/ratelimit"
+)
+
+const defaultCapacity = 100
+
+// RingBuffer is a thread-safe ring buffer with overflow policies.
+type RingBuffer[T any] struct {
+	mu       sync.RWMutex
+	items    []T
+	head     int
+	tail     int
+	size     int
+	capacity int
+
+	category       ratelimit.Category
+	priority       ratelimit.Priority
+	overflowPolicy OverflowPolicy
+
+	batchSize     int
+	timeout       time.Duration
+	lastFlushTime time.Time
+
+	offered   int64
+	dropped   int64
+	onDropped func(item T, reason string)
+}
+
+func NewRingBuffer[T any](category ratelimit.Category, capacity int, overflowPolicy OverflowPolicy, batchSize int, timeout time.Duration) *RingBuffer[T] {
+	if capacity <= 0 {
+		capacity = defaultCapacity
+	}
+
+	if batchSize <= 0 {
+		batchSize = 1
+	}
+
+	if timeout < 0 {
+		timeout = 0
+	}
+
+	return &RingBuffer[T]{
+		items:          make([]T, capacity),
+		capacity:       capacity,
+		category:       category,
+		priority:       category.GetPriority(),
+		overflowPolicy: overflowPolicy,
+		batchSize:      batchSize,
+		timeout:        timeout,
+		lastFlushTime:  time.Now(),
+	}
+}
+
+func (b *RingBuffer[T]) SetDroppedCallback(callback func(item T, reason string)) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.onDropped = callback
+}
+
+func (b *RingBuffer[T]) Offer(item T) bool {
+	atomic.AddInt64(&b.offered, 1)
+
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if b.size < b.capacity {
+		b.items[b.tail] = item
+		b.tail = (b.tail + 1) % b.capacity
+		b.size++
+		return true
+	}
+
+	switch b.overflowPolicy {
+	case OverflowPolicyDropOldest:
+		oldItem := b.items[b.head]
+		b.items[b.head] = item
+		b.head = (b.head + 1) % b.capacity
+		b.tail = (b.tail + 1) % b.capacity
+
+		atomic.AddInt64(&b.dropped, 1)
+		if b.onDropped != nil {
+			b.onDropped(oldItem, "buffer_full_drop_oldest")
+		}
+		return true
+
+	case OverflowPolicyDropNewest:
+		atomic.AddInt64(&b.dropped, 1)
+		if b.onDropped != nil {
+			b.onDropped(item, "buffer_full_drop_newest")
+		}
+		return false
+
+	default:
+		atomic.AddInt64(&b.dropped, 1)
+		if b.onDropped != nil {
+			b.onDropped(item, "unknown_overflow_policy")
+		}
+		return false
+	}
+}
+
+func (b *RingBuffer[T]) Poll() (T, bool) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	var zero T
+	if b.size == 0 {
+		return zero, false
+	}
+
+	item := b.items[b.head]
+	b.items[b.head] = zero
+	b.head = (b.head + 1) % b.capacity
+	b.size--
+
+	return item, true
+}
+
+func (b *RingBuffer[T]) PollBatch(maxItems int) []T {
+	if maxItems <= 0 {
+		return nil
+	}
+
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if b.size == 0 {
+		return nil
+	}
+
+	itemCount := maxItems
+	if itemCount > b.size {
+		itemCount = b.size
+	}
+
+	result := make([]T, itemCount)
+	var zero T
+
+	for i := 0; i < itemCount; i++ {
+		result[i] = b.items[b.head]
+		b.items[b.head] = zero
+		b.head = (b.head + 1) % b.capacity
+		b.size--
+	}
+
+	return result
+}
+
+func (b *RingBuffer[T]) Drain() []T {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if b.size == 0 {
+		return nil
+	}
+
+	result := make([]T, b.size)
+	index := 0
+	var zero T
+
+	for i := 0; i < b.size; i++ {
+		pos := (b.head + i) % b.capacity
+		result[index] = b.items[pos]
+		b.items[pos] = zero
+		index++
+	}
+
+	b.head = 0
+	b.tail = 0
+	b.size = 0
+
+	return result
+}
+
+func (b *RingBuffer[T]) Peek() (T, bool) {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+
+	var zero T
+	if b.size == 0 {
+		return zero, false
+	}
+
+	return b.items[b.head], true
+}
+
+func (b *RingBuffer[T]) Size() int {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+	return b.size
+}
+
+func (b *RingBuffer[T]) Capacity() int {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+	return b.capacity
+}
+
+func (b *RingBuffer[T]) Category() ratelimit.Category {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+	return b.category
+}
+
+func (b *RingBuffer[T]) Priority() ratelimit.Priority {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+	return b.priority
+}
+
+func (b *RingBuffer[T]) IsEmpty() bool {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+	return b.size == 0
+}
+
+func (b *RingBuffer[T]) IsFull() bool {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+	return b.size == b.capacity
+}
+
+func (b *RingBuffer[T]) Utilization() float64 {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+	return float64(b.size) / float64(b.capacity)
+}
+
+func (b *RingBuffer[T]) OfferedCount() int64 {
+	return atomic.LoadInt64(&b.offered)
+}
+
+func (b *RingBuffer[T]) DroppedCount() int64 {
+	return atomic.LoadInt64(&b.dropped)
+}
+
+func (b *RingBuffer[T]) AcceptedCount() int64 {
+	return b.OfferedCount() - b.DroppedCount()
+}
+
+func (b *RingBuffer[T]) DropRate() float64 {
+	offered := b.OfferedCount()
+	if offered == 0 {
+		return 0.0
+	}
+	return float64(b.DroppedCount()) / float64(offered)
+}
+
+func (b *RingBuffer[T]) Clear() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	var zero T
+	for i := 0; i < b.capacity; i++ {
+		b.items[i] = zero
+	}
+
+	b.head = 0
+	b.tail = 0
+	b.size = 0
+}
+
+func (b *RingBuffer[T]) GetMetrics() BufferMetrics {
+	b.mu.RLock()
+	size := b.size
+	util := float64(b.size) / float64(b.capacity)
+	b.mu.RUnlock()
+
+	return BufferMetrics{
+		Category:      b.category,
+		Priority:      b.priority,
+		Capacity:      b.capacity,
+		Size:          size,
+		Utilization:   util,
+		OfferedCount:  b.OfferedCount(),
+		DroppedCount:  b.DroppedCount(),
+		AcceptedCount: b.AcceptedCount(),
+		DropRate:      b.DropRate(),
+		LastUpdated:   time.Now(),
+	}
+}
+
+func (b *RingBuffer[T]) IsReadyToFlush() bool {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+
+	if b.size == 0 {
+		return false
+	}
+
+	if b.size >= b.batchSize {
+		return true
+	}
+
+	if b.timeout > 0 && time.Since(b.lastFlushTime) >= b.timeout {
+		return true
+	}
+
+	return false
+}
+
+func (b *RingBuffer[T]) MarkFlushed() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.lastFlushTime = time.Now()
+}
+
+func (b *RingBuffer[T]) PollIfReady() []T {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if b.size == 0 {
+		return nil
+	}
+
+	ready := b.size >= b.batchSize ||
+		(b.timeout > 0 && time.Since(b.lastFlushTime) >= b.timeout)
+
+	if !ready {
+		return nil
+	}
+
+	itemCount := b.batchSize
+	if itemCount > b.size {
+		itemCount = b.size
+	}
+
+	result := make([]T, itemCount)
+	var zero T
+
+	for i := 0; i < itemCount; i++ {
+		result[i] = b.items[b.head]
+		b.items[b.head] = zero
+		b.head = (b.head + 1) % b.capacity
+		b.size--
+	}
+
+	b.lastFlushTime = time.Now()
+	return result
+}
+
+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"`
+}
+
+// OverflowPolicy defines how the ring buffer handles overflow.
+type OverflowPolicy int
+
+const (
+	OverflowPolicyDropOldest OverflowPolicy = iota
+	OverflowPolicyDropNewest
+)
+
+func (op OverflowPolicy) String() string {
+	switch op {
+	case OverflowPolicyDropOldest:
+		return "drop_oldest"
+	case OverflowPolicyDropNewest:
+		return "drop_newest"
+	default:
+		return "unknown"
+	}
+}

vendor/github.com/getsentry/sentry-go/internal/telemetry/buffer_wrapper.go

@@ -0,0 +1,49 @@
+package telemetry
+
+import (
+	"context"
+	"time"
+
+	"github.com/getsentry/sentry-go/internal/protocol"
+	"github.com/getsentry/sentry-go/internal/ratelimit"
+)
+
+// Processor is the top-level object that wraps the scheduler and buffers.
+type Processor struct {
+	scheduler *Scheduler
+}
+
+// NewProcessor creates a new Processor with the given configuration.
+func NewProcessor(
+	buffers map[ratelimit.Category]Buffer[protocol.TelemetryItem],
+	transport protocol.TelemetryTransport,
+	dsn *protocol.Dsn,
+	sdkInfo *protocol.SdkInfo,
+) *Processor {
+	scheduler := NewScheduler(buffers, transport, dsn, sdkInfo)
+	scheduler.Start()
+
+	return &Processor{
+		scheduler: scheduler,
+	}
+}
+
+// Add adds a TelemetryItem to the appropriate buffer based on its category.
+func (b *Processor) Add(item protocol.TelemetryItem) bool {
+	return b.scheduler.Add(item)
+}
+
+// Flush forces all buffers to flush within the given timeout.
+func (b *Processor) Flush(timeout time.Duration) bool {
+	return b.scheduler.Flush(timeout)
+}
+
+// FlushWithContext flushes with a custom context for cancellation.
+func (b *Processor) FlushWithContext(ctx context.Context) bool {
+	return b.scheduler.FlushWithContext(ctx)
+}
+
+// Close stops the buffer, flushes remaining data, and releases resources.
+func (b *Processor) Close(timeout time.Duration) {
+	b.scheduler.Stop(timeout)
+}

vendor/github.com/getsentry/sentry-go/internal/telemetry/scheduler.go

@@ -0,0 +1,301 @@
+package telemetry
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/getsentry/sentry-go/internal/debuglog"
+	"github.com/getsentry/sentry-go/internal/protocol"
+	"github.com/getsentry/sentry-go/internal/ratelimit"
+)
+
+// Scheduler implements a weighted round-robin scheduler for processing buffered events.
+type Scheduler struct {
+	buffers   map[ratelimit.Category]Buffer[protocol.TelemetryItem]
+	transport protocol.TelemetryTransport
+	dsn       *protocol.Dsn
+	sdkInfo   *protocol.SdkInfo
+
+	currentCycle []ratelimit.Priority
+	cyclePos     int
+
+	ctx          context.Context
+	cancel       context.CancelFunc
+	processingWg sync.WaitGroup
+
+	mu         sync.Mutex
+	cond       *sync.Cond
+	startOnce  sync.Once
+	finishOnce sync.Once
+}
+
+func NewScheduler(
+	buffers map[ratelimit.Category]Buffer[protocol.TelemetryItem],
+	transport protocol.TelemetryTransport,
+	dsn *protocol.Dsn,
+	sdkInfo *protocol.SdkInfo,
+) *Scheduler {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	priorityWeights := map[ratelimit.Priority]int{
+		ratelimit.PriorityCritical: 5,
+		ratelimit.PriorityHigh:     4,
+		ratelimit.PriorityMedium:   3,
+		ratelimit.PriorityLow:      2,
+		ratelimit.PriorityLowest:   1,
+	}
+
+	var currentCycle []ratelimit.Priority
+	for priority, weight := range priorityWeights {
+		hasBuffers := false
+		for _, buffer := range buffers {
+			if buffer.Priority() == priority {
+				hasBuffers = true
+				break
+			}
+		}
+
+		if hasBuffers {
+			for i := 0; i < weight; i++ {
+				currentCycle = append(currentCycle, priority)
+			}
+		}
+	}
+
+	s := &Scheduler{
+		buffers:      buffers,
+		transport:    transport,
+		dsn:          dsn,
+		sdkInfo:      sdkInfo,
+		currentCycle: currentCycle,
+		ctx:          ctx,
+		cancel:       cancel,
+	}
+	s.cond = sync.NewCond(&s.mu)
+
+	return s
+}
+
+func (s *Scheduler) Start() {
+	s.startOnce.Do(func() {
+		s.processingWg.Add(1)
+		go s.run()
+	})
+}
+
+func (s *Scheduler) Stop(timeout time.Duration) {
+	s.finishOnce.Do(func() {
+		s.Flush(timeout)
+
+		s.cancel()
+		s.cond.Broadcast()
+
+		done := make(chan struct{})
+		go func() {
+			defer close(done)
+			s.processingWg.Wait()
+		}()
+
+		select {
+		case <-done:
+		case <-time.After(timeout):
+			debuglog.Printf("scheduler stop timed out after %v", timeout)
+		}
+	})
+}
+
+func (s *Scheduler) Signal() {
+	s.cond.Signal()
+}
+
+func (s *Scheduler) Add(item protocol.TelemetryItem) bool {
+	category := item.GetCategory()
+	buffer, exists := s.buffers[category]
+	if !exists {
+		return false
+	}
+
+	accepted := buffer.Offer(item)
+	if accepted {
+		s.Signal()
+	}
+	return accepted
+}
+
+func (s *Scheduler) Flush(timeout time.Duration) bool {
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	return s.FlushWithContext(ctx)
+}
+
+func (s *Scheduler) FlushWithContext(ctx context.Context) bool {
+	s.flushBuffers()
+	return s.transport.FlushWithContext(ctx)
+}
+
+func (s *Scheduler) run() {
+	defer s.processingWg.Done()
+
+	go func() {
+		ticker := time.NewTicker(100 * time.Millisecond)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ticker.C:
+				s.cond.Broadcast()
+			case <-s.ctx.Done():
+				return
+			}
+		}
+	}()
+
+	for {
+		s.mu.Lock()
+
+		for !s.hasWork() && s.ctx.Err() == nil {
+			s.cond.Wait()
+		}
+
+		if s.ctx.Err() != nil {
+			s.mu.Unlock()
+			return
+		}
+
+		s.mu.Unlock()
+		s.processNextBatch()
+	}
+}
+
+func (s *Scheduler) hasWork() bool {
+	for _, buffer := range s.buffers {
+		if buffer.IsReadyToFlush() {
+			return true
+		}
+	}
+	return false
+}
+
+func (s *Scheduler) processNextBatch() {
+	if len(s.currentCycle) == 0 {
+		return
+	}
+
+	priority := s.currentCycle[s.cyclePos]
+	s.cyclePos = (s.cyclePos + 1) % len(s.currentCycle)
+
+	var bufferToProcess Buffer[protocol.TelemetryItem]
+	var categoryToProcess ratelimit.Category
+	for category, buffer := range s.buffers {
+		if buffer.Priority() == priority && buffer.IsReadyToFlush() {
+			bufferToProcess = buffer
+			categoryToProcess = category
+			break
+		}
+	}
+
+	if bufferToProcess != nil {
+		s.processItems(bufferToProcess, categoryToProcess, false)
+	}
+}
+
+func (s *Scheduler) processItems(buffer Buffer[protocol.TelemetryItem], category ratelimit.Category, force bool) {
+	var items []protocol.TelemetryItem
+
+	if force {
+		items = buffer.Drain()
+	} else {
+		items = buffer.PollIfReady()
+	}
+
+	// drop the current batch if rate-limited or if transport is full
+	if len(items) == 0 || s.isRateLimited(category) || !s.transport.HasCapacity() {
+		return
+	}
+
+	switch category {
+	case ratelimit.CategoryLog:
+		logs := protocol.Logs(items)
+		header := &protocol.EnvelopeHeader{EventID: protocol.GenerateEventID(), SentAt: time.Now(), Sdk: s.sdkInfo}
+		if s.dsn != nil {
+			header.Dsn = s.dsn.String()
+		}
+		envelope := protocol.NewEnvelope(header)
+		item, err := logs.ToEnvelopeItem()
+		if err != nil {
+			debuglog.Printf("error creating log batch envelope item: %v", err)
+			return
+		}
+		envelope.AddItem(item)
+		if err := s.transport.SendEnvelope(envelope); err != nil {
+			debuglog.Printf("error sending envelope: %v", err)
+		}
+		return
+	case ratelimit.CategoryTraceMetric:
+		metrics := protocol.Metrics(items)
+		header := &protocol.EnvelopeHeader{EventID: protocol.GenerateEventID(), SentAt: time.Now(), Sdk: s.sdkInfo}
+		if s.dsn != nil {
+			header.Dsn = s.dsn.String()
+		}
+		envelope := protocol.NewEnvelope(header)
+		item, err := metrics.ToEnvelopeItem()
+		if err != nil {
+			debuglog.Printf("error creating trace metric batch envelope item: %v", err)
+			return
+		}
+		envelope.AddItem(item)
+		if err := s.transport.SendEnvelope(envelope); err != nil {
+			debuglog.Printf("error sending envelope: %v", err)
+		}
+		return
+	default:
+		// if the buffers are properly configured, buffer.PollIfReady should return a single item for every category
+		// other than logs. We still iterate over the items just in case, because we don't want to send broken envelopes.
+		for _, it := range items {
+			convertible, ok := it.(protocol.EnvelopeItemConvertible)
+			if !ok {
+				debuglog.Printf("item does not implement EnvelopeItemConvertible: %T", it)
+				continue
+			}
+			s.sendItem(convertible)
+		}
+	}
+}
+
+func (s *Scheduler) sendItem(item protocol.EnvelopeItemConvertible) {
+	header := &protocol.EnvelopeHeader{
+		EventID: item.GetEventID(),
+		SentAt:  time.Now(),
+		Trace:   item.GetDynamicSamplingContext(),
+		Sdk:     s.sdkInfo,
+	}
+	if header.EventID == "" {
+		header.EventID = protocol.GenerateEventID()
+	}
+	if s.dsn != nil {
+		header.Dsn = s.dsn.String()
+	}
+	envelope := protocol.NewEnvelope(header)
+	envItem, err := item.ToEnvelopeItem()
+	if err != nil {
+		debuglog.Printf("error while converting to envelope item: %v", err)
+		return
+	}
+	envelope.AddItem(envItem)
+	if err := s.transport.SendEnvelope(envelope); err != nil {
+		debuglog.Printf("error sending envelope: %v", err)
+	}
+}
+
+func (s *Scheduler) flushBuffers() {
+	for category, buffer := range s.buffers {
+		if !buffer.IsEmpty() {
+			s.processItems(buffer, category, true)
+		}
+	}
+}
+
+func (s *Scheduler) isRateLimited(category ratelimit.Category) bool {
+	return s.transport.IsRateLimited(category)
+}

vendor/github.com/getsentry/sentry-go/internal/telemetry/storage.go

@@ -0,0 +1,42 @@
+package telemetry
+
+import (
+	"github.com/getsentry/sentry-go/internal/ratelimit"
+)
+
+// Buffer defines the common interface for all buffer implementations.
+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()
+}

vendor/github.com/getsentry/sentry-go/internal/telemetry/trace_aware.go

@@ -0,0 +1,7 @@
+package telemetry
+
+// TraceAware is implemented by items that can expose a trace ID.
+// BucketedBuffer uses this to group items by trace.
+type TraceAware interface {
+	GetTraceID() (string, bool)
+}