refactor!: update Telemetry Processor logic (#1180)

Author: giortzisg

_examples/metrics/main.go

@@ -15,7 +15,7 @@ var meter sentry.Meter
 
 func main() {
 	err := sentry.Init(sentry.ClientOptions{
-		Dsn:              "https://53eb618df7695aaea90bcc816ff537e0@o447951.ingest.us.sentry.io/5774600",
+		Dsn:              "",
 		Debug:            true,
 		EnableTracing:    true,
 		TracesSampleRate: 1.0,

client.go

@@ -282,10 +282,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     *logBatchProcessor
-	batchMeter      *metricBatchProcessor
-	telemetryBuffer *telemetry.Buffer
+	Transport          Transport
+	batchLogger        *logBatchProcessor
+	batchMeter         *metricBatchProcessor
+	telemetryProcessor *telemetry.Processor
 }
 
 // NewClient creates and returns an instance of Client configured using
@@ -391,9 +391,9 @@ func NewClient(options ClientOptions) (*Client, error) {
 
 	client.setupTransport()
 
-	// noop Telemetry Buffers fow now
+	// noop Telemetry Buffers and Processor fow now
 	// if !options.DisableTelemetryBuffer {
-	// 	client.setupTelemetryBuffer()
+	// 	client.setupTelemetryProcessor()
 	// } else
 	if options.EnableLogs {
 		client.batchLogger = newLogBatchProcessor(&client)
@@ -426,7 +426,7 @@ func (client *Client) setupTransport() {
 	client.Transport = transport
 }
 
-func (client *Client) setupTelemetryBuffer() { // nolint: unused
+func (client *Client) setupTelemetryProcessor() { // nolint: unused
 	if client.options.DisableTelemetryBuffer {
 		return
 	}
@@ -436,10 +436,10 @@ func (client *Client) setupTelemetryBuffer() { // nolint: unused
 		return
 	}
 
-	// We currently disallow using custom Transport with the new Telemetry Buffer, due to the difference in transport signatures.
+	// We currently disallow using custom Transport with the new Telemetry Processor, due to the difference in transport signatures.
 	// The option should be enabled when the new Transport interface signature changes.
 	if client.options.Transport != nil {
-		debuglog.Println("Cannot enable Telemetry Buffer with custom Transport: fallback to old transport")
+		debuglog.Println("Cannot enable Telemetry Processor/Buffers with custom Transport: fallback to old transport")
 		if client.options.EnableLogs {
 			client.batchLogger = newLogBatchProcessor(client)
 			client.batchLogger.Start()
@@ -461,20 +461,20 @@ func (client *Client) setupTelemetryBuffer() { // nolint: unused
 	})
 	client.Transport = &internalAsyncTransportAdapter{transport: transport}
 
-	storage := map[ratelimit.Category]telemetry.Storage[protocol.EnvelopeItemConvertible]{
-		ratelimit.CategoryError:       telemetry.NewRingBuffer[protocol.EnvelopeItemConvertible](ratelimit.CategoryError, 100, telemetry.OverflowPolicyDropOldest, 1, 0),
-		ratelimit.CategoryTransaction: telemetry.NewRingBuffer[protocol.EnvelopeItemConvertible](ratelimit.CategoryTransaction, 1000, telemetry.OverflowPolicyDropOldest, 1, 0),
-		ratelimit.CategoryLog:         telemetry.NewRingBuffer[protocol.EnvelopeItemConvertible](ratelimit.CategoryLog, 10*100, telemetry.OverflowPolicyDropOldest, 100, 5*time.Second),
-		ratelimit.CategoryMonitor:     telemetry.NewRingBuffer[protocol.EnvelopeItemConvertible](ratelimit.CategoryMonitor, 100, telemetry.OverflowPolicyDropOldest, 1, 0),
-		ratelimit.CategoryTraceMetric: telemetry.NewRingBuffer[protocol.EnvelopeItemConvertible](ratelimit.CategoryTraceMetric, 10*100, telemetry.OverflowPolicyDropOldest, 100, 5*time.Second),
+	buffers := map[ratelimit.Category]telemetry.Buffer[protocol.TelemetryItem]{
+		ratelimit.CategoryError:       telemetry.NewRingBuffer[protocol.TelemetryItem](ratelimit.CategoryError, 100, telemetry.OverflowPolicyDropOldest, 1, 0),
+		ratelimit.CategoryTransaction: telemetry.NewRingBuffer[protocol.TelemetryItem](ratelimit.CategoryTransaction, 1000, telemetry.OverflowPolicyDropOldest, 1, 0),
+		ratelimit.CategoryLog:         telemetry.NewRingBuffer[protocol.TelemetryItem](ratelimit.CategoryLog, 10*100, telemetry.OverflowPolicyDropOldest, 100, 5*time.Second),
+		ratelimit.CategoryMonitor:     telemetry.NewRingBuffer[protocol.TelemetryItem](ratelimit.CategoryMonitor, 100, telemetry.OverflowPolicyDropOldest, 1, 0),
+		ratelimit.CategoryTraceMetric: telemetry.NewRingBuffer[protocol.TelemetryItem](ratelimit.CategoryTraceMetric, 10*100, telemetry.OverflowPolicyDropOldest, 100, 5*time.Second),
 	}
 
 	sdkInfo := &protocol.SdkInfo{
 		Name:    client.sdkIdentifier,
 		Version: client.sdkVersion,
 	}
 
-	client.telemetryBuffer = telemetry.NewBuffer(storage, transport, &client.dsn.Dsn, sdkInfo)
+	client.telemetryProcessor = telemetry.NewProcessor(buffers, transport, &client.dsn.Dsn, sdkInfo)
 }
 
 func (client *Client) setupIntegrations() {
@@ -547,7 +547,7 @@ func (client *Client) CaptureCheckIn(checkIn *CheckIn, monitorConfig *MonitorCon
 
 // CaptureEvent captures an event on the currently active client if any.
 //
-// The event must already be assembled. Typically code would instead use
+// The event must already be assembled. Typically, code would instead use
 // the utility methods like CaptureException. The return value is the
 // event ID. In case Sentry is disabled or event was dropped, the return value will be nil.
 func (client *Client) CaptureEvent(event *Event, hint *EventHint, scope EventModifier) *EventID {
@@ -567,8 +567,8 @@ func (client *Client) captureLog(log *Log, _ *Scope) bool {
 		}
 	}
 
-	if client.telemetryBuffer != nil {
-		if !client.telemetryBuffer.Add(log) {
+	if client.telemetryProcessor != nil {
+		if !client.telemetryProcessor.Add(log) {
 			debuglog.Print("Dropping log: telemetry buffer full or category missing")
 			return false
 		}
@@ -595,8 +595,8 @@ func (client *Client) captureMetric(metric *Metric, _ *Scope) bool {
 		}
 	}
 
-	if client.telemetryBuffer != nil {
-		if !client.telemetryBuffer.Add(metric) {
+	if client.telemetryProcessor != nil {
+		if !client.telemetryProcessor.Add(metric) {
 			debuglog.Printf("Dropping metric: telemetry buffer full or category missing")
 			return false
 		}
@@ -673,7 +673,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 || client.batchMeter != nil || client.telemetryBuffer != nil {
+	if client.batchLogger != nil || client.batchMeter != nil || client.telemetryProcessor != nil {
 		ctx, cancel := context.WithTimeout(context.Background(), timeout)
 		defer cancel()
 		return client.FlushWithContext(ctx)
@@ -700,8 +700,8 @@ func (client *Client) FlushWithContext(ctx context.Context) bool {
 	if client.batchMeter != nil {
 		client.batchMeter.Flush(ctx.Done())
 	}
-	if client.telemetryBuffer != nil {
-		return client.telemetryBuffer.FlushWithContext(ctx)
+	if client.telemetryProcessor != nil {
+		return client.telemetryProcessor.FlushWithContext(ctx)
 	}
 	return client.Transport.FlushWithContext(ctx)
 }
@@ -711,8 +711,8 @@ 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.telemetryBuffer != nil {
-		client.telemetryBuffer.Close(5 * time.Second)
+	if client.telemetryProcessor != nil {
+		client.telemetryProcessor.Close(5 * time.Second)
 	}
 	if client.batchLogger != nil {
 		client.batchLogger.Shutdown()
@@ -840,8 +840,8 @@ func (client *Client) processEvent(event *Event, hint *EventHint, scope EventMod
 		}
 	}
 
-	if client.telemetryBuffer != nil {
-		if !client.telemetryBuffer.Add(event) {
+	if client.telemetryProcessor != nil {
+		if !client.telemetryProcessor.Add(event) {
 			debuglog.Println("Event dropped: telemetry buffer full or unavailable")
 		}
 	} else {

client_test.go

@@ -1036,27 +1036,6 @@ func TestClientSetsUpTransport(t *testing.T) {
 	require.IsType(t, &noopTransport{}, client.Transport)
 }
 
-// func TestClient_SetupTelemetryBuffer_WithDSN(t *testing.T) {
-//	client, err := NewClient(ClientOptions{
-//		Dsn: "https://public@localhost/1",
-//	})
-//	if err != nil {
-//		t.Fatalf("unexpected error: %v", err)
-//	}
-//
-//	if client.telemetryBuffer == nil {
-//		t.Fatal("expected telemetryBuffer to be initialized")
-//	}
-//
-//	if _, ok := client.Transport.(*internalAsyncTransportAdapter); !ok {
-//		t.Fatalf("expected internalAsyncTransportAdapter, got %T", client.Transport)
-//	}
-//
-//	if !client.telemetryBuffer.Add(NewEvent()) {
-//		t.Fatal("expected Add to succeed with default buffers")
-//	}
-//}
-
 func TestClient_SetupTelemetryBuffer_NoDSN(t *testing.T) {
 	var buf bytes.Buffer
 	debuglog.SetOutput(&buf)
@@ -1067,8 +1046,8 @@ func TestClient_SetupTelemetryBuffer_NoDSN(t *testing.T) {
 		t.Fatalf("unexpected error: %v", err)
 	}
 
-	if client.telemetryBuffer != nil {
-		t.Fatal("expected telemetryBuffer to be nil when DSN is missing")
+	if client.telemetryProcessor != nil {
+		t.Fatal("expected telemetryProcessor to be nil when DSN is missing")
 	}
 
 	if _, ok := client.Transport.(*noopTransport); !ok {

interfaces.go

@@ -487,80 +487,6 @@ func (e *Event) SetException(exception error, maxErrorDepth int) {
 	e.Exception = exceptions
 }
 
-// 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)
-
-	eventBody, err := json.Marshal(e)
-	if err != nil {
-		// Try fallback: remove problematic fields and retry
-		e.Breadcrumbs = nil
-		e.Contexts = nil
-		e.Extra = map[string]interface{}{
-			"info": fmt.Sprintf("Could not encode original event as JSON. "+
-				"Succeeded by removing Breadcrumbs, Contexts and Extra. "+
-				"Please verify the data you attach to the scope. "+
-				"Error: %s", err),
-		}
-
-		eventBody, err = json.Marshal(e)
-		if err != nil {
-			return nil, fmt.Errorf("event could not be marshaled even with fallback: %w", err)
-		}
-
-		DebugLogger.Printf("Event marshaling succeeded with fallback after removing problematic fields")
-	}
-
-	var mainItem *protocol.EnvelopeItem
-	switch e.Type {
-	case transactionType:
-		mainItem = protocol.NewEnvelopeItem(protocol.EnvelopeItemTypeTransaction, eventBody)
-	case checkInType:
-		mainItem = protocol.NewEnvelopeItem(protocol.EnvelopeItemTypeCheckIn, eventBody)
-	case logEvent.Type:
-		mainItem = protocol.NewLogItem(len(e.Logs), eventBody)
-	case traceMetricEvent.Type:
-		mainItem = protocol.NewTraceMetricItem(len(e.Metrics), eventBody)
-	default:
-		mainItem = 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 envelope, nil
-}
-
 // ToEnvelopeItem converts the Event to a Sentry envelope item.
 func (e *Event) ToEnvelopeItem() (*protocol.EnvelopeItem, error) {
 	eventBody, err := json.Marshal(e)
@@ -831,21 +757,6 @@ type Log struct {
 	Attributes map[string]Attribute `json:"attributes,omitempty"`
 }
 
-// ToEnvelopeItem converts the Log to a Sentry envelope item for batching.
-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
@@ -930,20 +841,6 @@ type Metric struct {
 	Attributes map[string]Attribute `json:"attributes,omitempty"`
 }
 
-func (m *Metric) ToEnvelopeItem() (*protocol.EnvelopeItem, error) {
-	metricData, err := json.Marshal(m)
-	if err != nil {
-		return nil, err
-	}
-
-	return &protocol.EnvelopeItem{
-		Header: &protocol.EnvelopeItemHeader{
-			Type: protocol.EnvelopeItemTypeTraceMetric,
-		},
-		Payload: metricData,
-	}, nil
-}
-
 // MarshalJSON converts a Metric to JSON that skips SpanID and timestamp when zero.
 func (m *Metric) MarshalJSON() ([]byte, error) {
 	type metric Metric