feat: add telemetry scheduler (#1107)

Author: giortzisg

.codecov.yml

@@ -16,3 +16,4 @@ coverage:
         informational: true
 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.
@@ -252,6 +256,8 @@ type ClientOptions struct {
 	// IMPORTANT: to not ignore any status codes, the option should be an empty slice and not nil. The nil option is
 	// used for defaulting to 404 ignores.
 	TraceIgnoreStatusCodes [][]int
+	// DisableTelemetryBuffer disables the telemetry buffer layer for prioritizing events and uses the old transport layer.
+	DisableTelemetryBuffer bool
 }
 
 // Client is the underlying processor that is used by the main API and Hub
@@ -266,8 +272,9 @@ 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
+	telemetryBuffer *telemetry.Buffer
 }
 
 // NewClient creates and returns an instance of Client configured using
@@ -371,12 +378,15 @@ func NewClient(options ClientOptions) (*Client, error) {
 		sdkVersion:    SDKVersion,
 	}
 
-	if options.EnableLogs {
+	client.setupTransport()
+
+	if !options.DisableTelemetryBuffer {
+		client.setupTelemetryBuffer()
+	} else if options.EnableLogs {
 		client.batchLogger = NewBatchLogger(&client)
 		client.batchLogger.Start()
 	}
 
-	client.setupTransport()
 	client.setupIntegrations()
 
 	return &client, nil
@@ -398,6 +408,52 @@ func (client *Client) setupTransport() {
 	client.Transport = transport
 }
 
+func (client *Client) setupTelemetryBuffer() {
+	if client.options.DisableTelemetryBuffer {
+		return
+	}
+
+	if client.dsn == nil {
+		debuglog.Println("Telemetry buffer disabled: no DSN configured")
+		return
+	}
+
+	// We currently disallow using custom Transport with the new Telemetry Buffer, 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")
+		if client.options.EnableLogs {
+			client.batchLogger = NewBatchLogger(client)
+			client.batchLogger.Start()
+		}
+		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}
+
+	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),
+	}
+
+	sdkInfo := &protocol.SdkInfo{
+		Name:    client.sdkIdentifier,
+		Version: client.sdkVersion,
+	}
+
+	client.telemetryBuffer = telemetry.NewBuffer(storage, transport, &client.dsn.Dsn, sdkInfo)
+}
+
 func (client *Client) setupIntegrations() {
 	integrations := []Integration{
 		new(contextifyFramesIntegration),
@@ -538,7 +594,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.telemetryBuffer != nil {
 		ctx, cancel := context.WithTimeout(context.Background(), timeout)
 		defer cancel()
 		return client.FlushWithContext(ctx)
@@ -562,6 +618,9 @@ func (client *Client) FlushWithContext(ctx context.Context) bool {
 	if client.batchLogger != nil {
 		client.batchLogger.Flush(ctx.Done())
 	}
+	if client.telemetryBuffer != nil {
+		return client.telemetryBuffer.FlushWithContext(ctx)
+	}
 	return client.Transport.FlushWithContext(ctx)
 }
 
@@ -570,6 +629,12 @@ 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.batchLogger != nil {
+		client.batchLogger.Shutdown()
+	}
 	client.Transport.Close()
 }
 
@@ -690,7 +755,13 @@ func (client *Client) processEvent(event *Event, hint *EventHint, scope EventMod
 		}
 	}
 
-	client.Transport.SendEvent(event)
+	if client.telemetryBuffer != nil {
+		if !client.telemetryBuffer.Add(event) {
+			debuglog.Println("Event dropped: telemetry buffer full or unavailable")
+		}
+	} else {
+		client.Transport.SendEvent(event)
+	}
 
 	return &event.EventID
 }

client_test.go

@@ -1,6 +1,7 @@
 package sentry
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
@@ -12,6 +13,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/getsentry/sentry-go/internal/debuglog"
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	pkgErrors "github.com/pkg/errors"
@@ -49,6 +51,7 @@ func setupClientTest() (*Client, *MockScope, *MockTransport) {
 	client, _ := NewClient(ClientOptions{
 		Dsn:       "http://[email protected]/1337",
 		Transport: transport,
+		// keep default buffers enabled
 		Integrations: func(_ []Integration) []Integration {
 			return []Integration{}
 		},
@@ -1032,3 +1035,43 @@ func TestClientSetsUpTransport(t *testing.T) {
 	client, _ = NewClient(ClientOptions{})
 	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)
+	defer debuglog.SetOutput(&bytes.Buffer{})
+
+	client, err := NewClient(ClientOptions{})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if client.telemetryBuffer != nil {
+		t.Fatal("expected telemetryBuffer to be nil when DSN is missing")
+	}
+
+	if _, ok := client.Transport.(*noopTransport); !ok {
+		t.Fatalf("expected noopTransport, got %T", client.Transport)
+	}
+}

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

@@ -493,6 +493,70 @@ func (e *Event) ToEnvelopeWithTime(dsn *protocol.Dsn, sentAt time.Time) (*protoc
 	return envelope, nil
 }
 
+// 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
+		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")
+	}
+
+	// TODO: all event types should be abstracted to implement EnvelopeItemConvertible and convert themselves.
+	var item *protocol.EnvelopeItem
+	switch e.Type {
+	case transactionType:
+		item = protocol.NewEnvelopeItem(protocol.EnvelopeItemTypeTransaction, eventBody)
+	case checkInType:
+		item = protocol.NewEnvelopeItem(protocol.EnvelopeItemTypeCheckIn, eventBody)
+	case logEvent.Type:
+		item = protocol.NewLogItem(len(e.Logs), eventBody)
+	default:
+		item = protocol.NewEnvelopeItem(protocol.EnvelopeItemTypeEvent, eventBody)
+	}
+
+	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
+}
+
+// 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,
 // to prevent accidentally storing T when we mean *T.
 // For example, the TraceContext must be stored as *TraceContext to pick up the
@@ -667,6 +731,69 @@ 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) {
+	type logJSON struct {
+		Timestamp  *float64                         `json:"timestamp,omitempty"`
+		TraceID    string                           `json:"trace_id,omitempty"`
+		Level      string                           `json:"level"`
+		Severity   int                              `json:"severity_number,omitempty"`
+		Body       string                           `json:"body,omitempty"`
+		Attributes map[string]protocol.LogAttribute `json:"attributes,omitempty"`
+	}
+
+	// Convert time.Time to seconds float if set
+	var ts *float64
+	if !l.Timestamp.IsZero() {
+		sec := float64(l.Timestamp.UnixNano()) / 1e9
+		ts = &sec
+	}
+
+	attrs := make(map[string]protocol.LogAttribute, len(l.Attributes))
+	for k, v := range l.Attributes {
+		attrs[k] = protocol.LogAttribute{Value: v.Value, Type: string(v.Type)}
+	}
+
+	logData, err := json.Marshal(logJSON{
+		Timestamp:  ts,
+		TraceID:    l.TraceID.String(),
+		Level:      string(l.Level),
+		Severity:   l.Severity,
+		Body:       l.Body,
+		Attributes: attrs,
+	})
+	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 (