feat: add telemetry scheduler

.codecov.yml

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

batch_logger.go

@@ -127,5 +127,6 @@
 	event.Timestamp = time.Now()
 	event.Type = logEvent.Type
 	event.Logs = logs
-	l.client.CaptureEvent(event, nil, nil)
+	l.client.Transport.SendEvent(event)
+	//l.client.CaptureEvent(event, nil, nil)
 }

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,42 @@ 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),
+	}
+
+	sdkInfo := &protocol.SdkInfo{
+		Name:    client.sdkIdentifier,
+		Version: client.sdkVersion,
+	}
+
+	client.telemetryScheduler = telemetry.NewScheduler(client.telemetryBuffers, transport, &client.dsn.Dsn, sdkInfo)
+	client.telemetryScheduler.Start()
+}
+
 func (client *Client) setupIntegrations() {
 	integrations := []Integration{
 		new(contextifyFramesIntegration),
@@ -531,7 +578,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 +602,9 @@ func (client *Client) FlushWithContext(ctx context.Context) bool {
 	if client.batchLogger != nil {
 		client.batchLogger.Flush(ctx.Done())
 	}
+	if client.telemetryScheduler != nil {
+		return client.telemetryScheduler.FlushWithContext(ctx)
+	}
 	return client.Transport.FlushWithContext(ctx)
 }
 
@@ -563,6 +613,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()
 }
 

hub_test.go

@@ -9,6 +9,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/getsentry/sentry-go/internal/protocol"
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 )
@@ -177,9 +178,9 @@ func TestConfigureScope(t *testing.T) {
 }
 
 func TestLastEventID(t *testing.T) {
-	uuid := EventID(uuid())
-	hub := &Hub{lastEventID: uuid}
-	assertEqual(t, uuid, hub.LastEventID())
+	eventID := EventID(protocol.GenerateEventID())
+	hub := &Hub{lastEventID: eventID}
+	assertEqual(t, eventID, hub.LastEventID())
 }
 
 func TestLastEventIDUpdatesAfterCaptures(t *testing.T) {

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)
+}
 
-	return envelope, nil
+// GetSdkInfo returns SDK information for the envelope header.
+func (e *Event) GetSdkInfo() *protocol.SdkInfo {
+	return &e.Sdk
+}
+
+// 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,
@@ -722,6 +714,40 @@ type Log struct {
 	Attributes map[string]Attribute `json:"attributes,omitempty"`
 }
 
+// ToEnvelopeItem converts the Log to a Sentry envelope item.
+func (l *Log) ToEnvelopeItem() (*protocol.EnvelopeItem, error) {
+	logData, err := json.Marshal(l)
+	if err != nil {
+		return nil, err
+	}
+	return &protocol.EnvelopeItem{
+		Header: &protocol.EnvelopeItemHeader{
+			Type: protocol.EnvelopeItemTypeLog,
+		},
+		Payload: logData,
+	}, nil
+}
+
+// GetCategory returns the rate limit category for logs.
+func (l *Log) GetCategory() ratelimit.Category {
+	return ratelimit.CategoryLog
+}
+
+// GetEventID returns empty string (event ID set when batching).
+func (l *Log) GetEventID() string {
+	return ""
+}
+
+// GetSdkInfo returns nil (SDK info set when batching).
+func (l *Log) GetSdkInfo() *protocol.SdkInfo {
+	return nil
+}
+
+// GetDynamicSamplingContext returns nil (trace context set when batching).
+func (l *Log) GetDynamicSamplingContext() map[string]string {
+	return nil
+}
+
 type AttrType string
 
 const (