feat: add telemetry scheduler

.codecov.yml

@@ -13,3 +13,4 @@ coverage:
         threshold: 0.5%
 ignore:
   - "log_fallback.go"
+  - "internal/testutils"

client.go

@@ -15,6 +15,10 @@ import (
 
 	"github.com/getsentry/sentry-go/internal/debug"
 	"github.com/getsentry/sentry-go/internal/debuglog"
+	httpInternal "github.com/getsentry/sentry-go/internal/http"
+	"github.com/getsentry/sentry-go/internal/protocol"
+	"github.com/getsentry/sentry-go/internal/ratelimit"
+	"github.com/getsentry/sentry-go/internal/telemetry"
 )
 
 // The identifier of the SDK.
@@ -249,6 +253,8 @@ type ClientOptions struct {
 	//
 	// By default, this is empty and all status codes are traced.
 	TraceIgnoreStatusCodes [][]int
+	// EnableTelemetryBuffer enables the telemetry buffer layer for prioritized delivery of events.
+	EnableTelemetryBuffer bool
 }
 
 // Client is the underlying processor that is used by the main API and Hub
@@ -263,8 +269,10 @@ type Client struct {
 	sdkVersion      string
 	// Transport is read-only. Replacing the transport of an existing client is
 	// not supported, create a new client instead.
-	Transport   Transport
-	batchLogger *BatchLogger
+	Transport          Transport
+	batchLogger        *BatchLogger
+	telemetryBuffers   map[ratelimit.Category]*telemetry.Buffer[protocol.EnvelopeItemConvertible]
+	telemetryScheduler *telemetry.Scheduler
 }
 
 // NewClient creates and returns an instance of Client configured using
@@ -364,12 +372,15 @@ func NewClient(options ClientOptions) (*Client, error) {
 		sdkVersion:    SDKVersion,
 	}
 
-	if options.EnableLogs {
+	client.setupTransport()
+
+	if options.EnableTelemetryBuffer {
+		client.setupTelemetryBuffer()
+	} else if options.EnableLogs {
 		client.batchLogger = NewBatchLogger(&client)
 		client.batchLogger.Start()
 	}
 
-	client.setupTransport()
 	client.setupIntegrations()
 
 	return &client, nil
@@ -391,6 +402,37 @@ func (client *Client) setupTransport() {
 	client.Transport = transport
 }
 
+func (client *Client) setupTelemetryBuffer() {
+	if !client.options.EnableTelemetryBuffer {
+		return
+	}
+
+	if client.dsn == nil {
+		debuglog.Println("Telemetry buffer disabled: no DSN configured")
+		return
+	}
+
+	transport := httpInternal.NewAsyncTransport(httpInternal.TransportOptions{
+		Dsn:           client.options.Dsn,
+		HTTPClient:    client.options.HTTPClient,
+		HTTPTransport: client.options.HTTPTransport,
+		HTTPProxy:     client.options.HTTPProxy,
+		HTTPSProxy:    client.options.HTTPSProxy,
+		CaCerts:       client.options.CaCerts,
+	})
+	client.Transport = &internalAsyncTransportAdapter{transport: transport}
+
+	client.telemetryBuffers = map[ratelimit.Category]*telemetry.Buffer[protocol.EnvelopeItemConvertible]{
+		ratelimit.CategoryError:       telemetry.NewBuffer[protocol.EnvelopeItemConvertible](ratelimit.CategoryError, 100, telemetry.OverflowPolicyDropOldest, 1, 0),
+		ratelimit.CategoryTransaction: telemetry.NewBuffer[protocol.EnvelopeItemConvertible](ratelimit.CategoryTransaction, 1000, telemetry.OverflowPolicyDropOldest, 1, 0),
+		ratelimit.CategoryLog:         telemetry.NewBuffer[protocol.EnvelopeItemConvertible](ratelimit.CategoryLog, 100, telemetry.OverflowPolicyDropOldest, 100, 5*time.Second),
+		ratelimit.CategoryMonitor:     telemetry.NewBuffer[protocol.EnvelopeItemConvertible](ratelimit.CategoryMonitor, 100, telemetry.OverflowPolicyDropOldest, 1, 0),
+	}
+
+	client.telemetryScheduler = telemetry.NewScheduler(client.telemetryBuffers, transport, &client.dsn.Dsn)
+	client.telemetryScheduler.Start()
+}
+
 func (client *Client) setupIntegrations() {
 	integrations := []Integration{
 		new(contextifyFramesIntegration),
@@ -531,7 +573,7 @@ func (client *Client) RecoverWithContext(
 // the network synchronously, configure it to use the HTTPSyncTransport in the
 // call to Init.
 func (client *Client) Flush(timeout time.Duration) bool {
-	if client.batchLogger != nil {
+	if client.batchLogger != nil || client.telemetryScheduler != nil {
 		ctx, cancel := context.WithTimeout(context.Background(), timeout)
 		defer cancel()
 		return client.FlushWithContext(ctx)
@@ -555,6 +597,9 @@ func (client *Client) FlushWithContext(ctx context.Context) bool {
 	if client.batchLogger != nil {
 		client.batchLogger.Flush(ctx.Done())
 	}
+	if client.telemetryScheduler != nil {
+		client.telemetryScheduler.FlushWithContext(ctx)
+	}
 	return client.Transport.FlushWithContext(ctx)
 }
 
@@ -563,6 +608,9 @@ func (client *Client) FlushWithContext(ctx context.Context) bool {
 // Close should be called after Flush and before terminating the program
 // otherwise some events may be lost.
 func (client *Client) Close() {
+	if client.telemetryScheduler != nil {
+		client.telemetryScheduler.Stop(5 * time.Second)
+	}
 	client.Transport.Close()
 }
 
@@ -683,7 +731,19 @@ func (client *Client) processEvent(event *Event, hint *EventHint, scope EventMod
 		}
 	}
 
-	client.Transport.SendEvent(event)
+	if client.telemetryScheduler != nil {
+		category := event.toCategory()
+		if buffer, ok := client.telemetryBuffers[category]; ok {
+			buffer.Offer(event)
+			client.telemetryScheduler.Signal()
+		} else {
+			// fallback if we get an event type with unknown category. this shouldn't happen
+			debuglog.Printf("Unknown category for event type %s, sending directly", event.Type)
+			client.Transport.SendEvent(event)
+		}
+	} else {
+		client.Transport.SendEvent(event)
+	}
 
 	return &event.EventID
 }

interfaces.go

@@ -476,37 +476,8 @@ func (e *Event) SetException(exception error, maxErrorDepth int) {
 	}
 }
 
-// ToEnvelope converts the Event to a Sentry envelope.
-// This includes the event data and any attachments as separate envelope items.
-func (e *Event) ToEnvelope(dsn *protocol.Dsn) (*protocol.Envelope, error) {
-	return e.ToEnvelopeWithTime(dsn, time.Now())
-}
-
-// ToEnvelopeWithTime converts the Event to a Sentry envelope with a specific sentAt time.
-// This is primarily useful for testing with predictable timestamps.
-func (e *Event) ToEnvelopeWithTime(dsn *protocol.Dsn, sentAt time.Time) (*protocol.Envelope, error) {
-	// Create envelope header with trace context
-	trace := make(map[string]string)
-	if dsc := e.sdkMetaData.dsc; dsc.HasEntries() {
-		for k, v := range dsc.Entries {
-			trace[k] = v
-		}
-	}
-
-	header := &protocol.EnvelopeHeader{
-		EventID: string(e.EventID),
-		SentAt:  sentAt,
-		Trace:   trace,
-	}
-
-	if dsn != nil {
-		header.Dsn = dsn.String()
-	}
-
-	header.Sdk = &e.Sdk
-
-	envelope := protocol.NewEnvelope(header)
-
+// ToEnvelopeItem converts the Event to a Sentry envelope item.
+func (e *Event) ToEnvelopeItem() (*protocol.EnvelopeItem, error) {
 	eventBody, err := json.Marshal(e)
 	if err != nil {
 		// Try fallback: remove problematic fields and retry
@@ -527,25 +498,46 @@ func (e *Event) ToEnvelopeWithTime(dsn *protocol.Dsn, sentAt time.Time) (*protoc
 		DebugLogger.Printf("Event marshaling succeeded with fallback after removing problematic fields")
 	}
 
-	var mainItem *protocol.EnvelopeItem
+	// TODO: all event types should be abstracted to implement EnvelopeItemConvertible and convert themselves.
+	var item *protocol.EnvelopeItem
 	switch e.Type {
 	case transactionType:
-		mainItem = protocol.NewEnvelopeItem(protocol.EnvelopeItemTypeTransaction, eventBody)
+		item = protocol.NewEnvelopeItem(protocol.EnvelopeItemTypeTransaction, eventBody)
 	case checkInType:
-		mainItem = protocol.NewEnvelopeItem(protocol.EnvelopeItemTypeCheckIn, eventBody)
+		item = protocol.NewEnvelopeItem(protocol.EnvelopeItemTypeCheckIn, eventBody)
 	case logEvent.Type:
-		mainItem = protocol.NewLogItem(len(e.Logs), eventBody)
+		item = protocol.NewLogItem(len(e.Logs), eventBody)
 	default:
-		mainItem = protocol.NewEnvelopeItem(protocol.EnvelopeItemTypeEvent, eventBody)
+		item = protocol.NewEnvelopeItem(protocol.EnvelopeItemTypeEvent, eventBody)
 	}
 
-	envelope.AddItem(mainItem)
-	for _, attachment := range e.Attachments {
-		attachmentItem := protocol.NewAttachmentItem(attachment.Filename, attachment.ContentType, attachment.Payload)
-		envelope.AddItem(attachmentItem)
-	}
+	return item, nil
+}
+
+// GetCategory returns the rate limit category for this event.
+func (e *Event) GetCategory() ratelimit.Category {
+	return e.toCategory()
+}
+
+// GetEventID returns the event ID.
+func (e *Event) GetEventID() string {
+	return string(e.EventID)
+}
+
+// GetSdkInfo returns SDK information for the envelope header.
+func (e *Event) GetSdkInfo() *protocol.SdkInfo {
+	return &e.Sdk
+}
 
-	return envelope, nil
+// GetDynamicSamplingContext returns trace context for the envelope header.
+func (e *Event) GetDynamicSamplingContext() map[string]string {
+	trace := make(map[string]string)
+	if dsc := e.sdkMetaData.dsc; dsc.HasEntries() {
+		for k, v := range dsc.Entries {
+			trace[k] = v
+		}
+	}
+	return trace
 }
 
 // TODO: Event.Contexts map[string]interface{} => map[string]EventContext,

interfaces_test.go

@@ -590,7 +590,7 @@ func TestEvent_ToCategory(t *testing.T) {
 	}
 }
 
-func TestEvent_ToEnvelope(t *testing.T) {
+func TestEvent_CreateEnvelopeFromItems(t *testing.T) {
 	tests := []struct {
 		name      string
 		event     *Event
@@ -673,20 +673,25 @@ func TestEvent_ToEnvelope(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			envelope, err := tt.event.ToEnvelope(tt.dsn)
+			envelope, err := protocol.CreateEnvelopeFromItems([]protocol.EnvelopeItemConvertible{tt.event}, tt.dsn)
 
 			if (err != nil) != tt.wantError {
-				t.Errorf("ToEnvelope() error = %v, wantError %v", err, tt.wantError)
+				t.Errorf("CreateEnvelopeFromItems() error = %v, wantError %v", err, tt.wantError)
 				return
 			}
 
 			if err != nil {
-				return // Expected error, nothing more to check
+				return
+			}
+
+			for _, attachment := range tt.event.Attachments {
+				attachmentItem := protocol.NewAttachmentItem(attachment.Filename, attachment.ContentType, attachment.Payload)
+				envelope.AddItem(attachmentItem)
 			}
 
 			// Basic envelope validation
 			if envelope == nil {
-				t.Error("ToEnvelope() returned nil envelope")
+				t.Error("CreateEnvelopeFromItems() returned nil envelope")
 				return
 			}
 
@@ -699,8 +704,7 @@ func TestEvent_ToEnvelope(t *testing.T) {
 				t.Errorf("Expected EventID %s, got %s", tt.event.EventID, envelope.Header.EventID)
 			}
 
-			// Check that items were created
-			expectedItems := 1 // Main event item
+			expectedItems := 1
 			if tt.event.Attachments != nil {
 				expectedItems += len(tt.event.Attachments)
 			}
@@ -709,7 +713,6 @@ func TestEvent_ToEnvelope(t *testing.T) {
 				t.Errorf("Expected %d items, got %d", expectedItems, len(envelope.Items))
 			}
 
-			// Verify the envelope can be serialized
 			data, err := envelope.Serialize()
 			if err != nil {
 				t.Errorf("Failed to serialize envelope: %v", err)
@@ -722,37 +725,6 @@ func TestEvent_ToEnvelope(t *testing.T) {
 	}
 }
 
-func TestEvent_ToEnvelopeWithTime(t *testing.T) {
-	event := &Event{
-		EventID:   "12345678901234567890123456789012",
-		Message:   "test message",
-		Level:     LevelError,
-		Timestamp: time.Date(2023, 1, 1, 12, 0, 0, 0, time.UTC),
-	}
-
-	sentAt := time.Date(2023, 1, 1, 15, 0, 0, 0, time.UTC)
-	envelope, err := event.ToEnvelopeWithTime(nil, sentAt)
-
-	if err != nil {
-		t.Errorf("ToEnvelopeWithTime() error = %v", err)
-		return
-	}
-
-	if envelope == nil {
-		t.Error("ToEnvelopeWithTime() returned nil envelope")
-		return
-	}
-
-	if envelope.Header == nil {
-		t.Error("Envelope header is nil")
-		return
-	}
-
-	if !envelope.Header.SentAt.Equal(sentAt) {
-		t.Errorf("Expected SentAt %v, got %v", sentAt, envelope.Header.SentAt)
-	}
-}
-
 func TestEvent_ToEnvelope_FallbackOnMarshalError(t *testing.T) {
 	unmarshalableFunc := func() string { return "test" }
 
@@ -766,15 +738,15 @@ func TestEvent_ToEnvelope_FallbackOnMarshalError(t *testing.T) {
 		},
 	}
 
-	envelope, err := event.ToEnvelope(nil)
+	envelope, err := protocol.CreateEnvelopeFromItems([]protocol.EnvelopeItemConvertible{event}, nil)
 
 	if err != nil {
-		t.Errorf("ToEnvelope() should not error even with unmarshalable data, got: %v", err)
+		t.Errorf("CreateEnvelopeFromItems() should not error even with unmarshalable data, got: %v", err)
 		return
 	}
 
 	if envelope == nil {
-		t.Error("ToEnvelope() should not return a nil envelope")
+		t.Error("CreateEnvelopeFromItems() should not return a nil envelope")
 		return
 	}