add buffer wrapper

Author: giortzisg

client.go

@@ -269,10 +269,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
-	telemetryBuffers   map[ratelimit.Category]telemetry.BufferInterface[protocol.EnvelopeItemConvertible]
-	telemetryScheduler *telemetry.Scheduler
+	Transport       Transport
+	batchLogger     *BatchLogger
+	telemetryBuffer *telemetry.Buffer
 }
 
 // NewClient creates and returns an instance of Client configured using
@@ -422,20 +421,19 @@ func (client *Client) setupTelemetryBuffer() {
 	})
 	client.Transport = &internalAsyncTransportAdapter{transport: transport}
 
-	client.telemetryBuffers = map[ratelimit.Category]telemetry.BufferInterface[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),
+	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, 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.telemetryScheduler = telemetry.NewScheduler(client.telemetryBuffers, transport, &client.dsn.Dsn, sdkInfo)
-	client.telemetryScheduler.Start()
+	client.telemetryBuffer = telemetry.NewBuffer(storage, transport, &client.dsn.Dsn, sdkInfo)
 }
 
 func (client *Client) setupIntegrations() {
@@ -578,7 +576,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.telemetryScheduler != nil {
+	if client.batchLogger != nil || client.telemetryBuffer != nil {
 		ctx, cancel := context.WithTimeout(context.Background(), timeout)
 		defer cancel()
 		return client.FlushWithContext(ctx)
@@ -602,8 +600,8 @@ 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)
+	if client.telemetryBuffer != nil {
+		return client.telemetryBuffer.FlushWithContext(ctx)
 	}
 	return client.Transport.FlushWithContext(ctx)
 }
@@ -613,8 +611,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.telemetryScheduler != nil {
-		client.telemetryScheduler.Stop(5 * time.Second)
+	if client.telemetryBuffer != nil {
+		client.telemetryBuffer.Close(5 * time.Second)
 	}
 	client.Transport.Close()
 }
@@ -736,7 +734,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
 }

interfaces.go

@@ -731,34 +731,47 @@ type Log struct {
 	Attributes map[string]Attribute `json:"attributes,omitempty"`
 }
 
-// ToEnvelopeItem converts the Log to a Sentry envelope item.
+// 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
+	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
 	}
-	return &protocol.EnvelopeItem{
-		Header: &protocol.EnvelopeItemHeader{
-			Type: protocol.EnvelopeItemTypeLog,
-		},
-		Payload: logData,
-	}, nil
-}
 
-// ToLogPayload converts the Log to a protocol.LogItem for batching.
-func (l *Log) ToLogPayload() protocol.LogItem {
 	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)}
 	}
-	return protocol.LogItem{
-		Timestamp:  l.Timestamp,
+
+	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.

internal/protocol/log_batch.go

@@ -2,7 +2,6 @@ package protocol
 
 import (
 	"encoding/json"
-	"time"
 
 	"github.com/getsentry/sentry-go/internal/ratelimit"
 )
@@ -13,57 +12,28 @@ type LogAttribute struct {
 	Type  string `json:"type"`
 }
 
-// LogItem represents the serialized shape of a single log record inside a batched
-// log envelope item. Keep in sync with sentry.Log fields that are meant to be emitted.
-type LogItem struct {
-	Timestamp  time.Time               `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]LogAttribute `json:"attributes,omitempty"`
-}
-
-// LogPayloader is implemented by items that can convert to a LogItem for batching.
-type LogPayloader interface {
-	ToLogPayload() LogItem
-}
+// Logs is a container for multiple log items which knows how to convert
+// itself into a single batched log envelope item.
+type Logs []EnvelopeItemConvertible
 
-// MarshalJSON encodes timestamp as seconds since epoch per Sentry logs spec.
-func (lp LogItem) MarshalJSON() ([]byte, error) {
-	// Convert time.Time to seconds float if set
-	var ts *float64
-	if !lp.Timestamp.IsZero() {
-		sec := float64(lp.Timestamp.UnixNano()) / 1e9
-		ts = &sec
+func (ls Logs) ToEnvelopeItem() (*EnvelopeItem, error) {
+	// Convert each log to its JSON representation
+	items := make([]json.RawMessage, 0, len(ls))
+	for _, log := range ls {
+		envItem, err := log.ToEnvelopeItem()
+		if err != nil {
+			continue
+		}
+		items = append(items, envItem.Payload)
 	}
 
-	out := 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]LogAttribute `json:"attributes,omitempty"`
-	}{
-		Timestamp:  ts,
-		TraceID:    lp.TraceID,
-		Level:      lp.Level,
-		Severity:   lp.Severity,
-		Body:       lp.Body,
-		Attributes: lp.Attributes,
+	if len(items) == 0 {
+		return nil, nil
 	}
-	return json.Marshal(out)
-}
-
-// Logs is a container for multiple LogItem items which knows how to convert
-// itself into a single batched log envelope item.
-type Logs []LogItem
 
-func (ls Logs) ToEnvelopeItem() (*EnvelopeItem, error) {
 	wrapper := struct {
-		Items []LogItem `json:"items"`
-	}{Items: ls}
+		Items []json.RawMessage `json:"items"`
+	}{Items: items}
 
 	payload, err := json.Marshal(wrapper)
 	if err != nil {