feat: re-enable Telemetry Processor and simplify setup

client.go

@@ -294,7 +294,7 @@ type ClientOptions struct {
 type Client struct {
 	mu                    sync.RWMutex
 	options               ClientOptions
-	dsn                   *Dsn
+	dsn                   *protocol.Dsn
 	eventProcessors       []EventProcessor
 	integrations          []Integration
 	externalTraceResolver externalContextTraceResolver
@@ -395,10 +395,10 @@ func NewClient(options ClientOptions) (*Client, error) {
 		}
 	}
 
-	var dsn *Dsn
+	var dsn *protocol.Dsn
 	if options.Dsn != "" {
 		var err error
-		dsn, err = NewDsn(options.Dsn)
+		dsn, err = protocol.NewDsn(options.Dsn)
 		if err != nil {
 			return nil, err
 		}
@@ -412,32 +412,36 @@ func NewClient(options ClientOptions) (*Client, error) {
 		reportRecorder: report.NoopRecorder(),
 		reportProvider: report.NoopProvider(),
 	}
+	client.setupIntegrations()
 
 	if !options.DisableClientReports {
 		a := report.NewAggregator()
 		client.reportRecorder = a
 		client.reportProvider = a
 	}
 
-	client.setupTransport()
-
-	// noop Telemetry Buffers and Processor fow now
-	// if !options.DisableTelemetryBuffer {
-	// 	client.setupTelemetryProcessor()
-	// } else
-	if options.EnableLogs {
-		client.batchLogger = newLogBatchProcessor(&client)
-		client.batchLogger.Start()
-	}
+	// 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 !options.DisableTelemetryBuffer && client.options.Transport == nil {
+		client.setupTelemetryProcessor()
+	} else {
+		if client.options.Transport != nil {
+			debuglog.Println("Cannot enable Telemetry Processor with custom Transport: fallback to old transport")
+		}
+		client.setupTransport()
 
-	if !options.DisableMetrics {
-		client.batchMeter = newMetricBatchProcessor(&client)
-		client.batchMeter.Start()
+		if options.EnableLogs {
+			client.batchLogger = newLogBatchProcessor(&client)
+			client.batchLogger.Start()
+		}
+		if !options.DisableMetrics {
+			client.batchMeter = newMetricBatchProcessor(&client)
+			client.batchMeter.Start()
+		}
 	}
 	if options.OrgID != 0 && client.dsn != nil {
 		client.dsn.SetOrgID(options.OrgID)
 	}
-	client.setupIntegrations()
 
 	return &client, nil
 }
@@ -474,31 +478,16 @@ func (client *Client) setupTransport() {
 	client.Transport = transport
 }
 
-func (client *Client) setupTelemetryProcessor() { // nolint: unused
-	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 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 Processor/Buffers with custom Transport: fallback to old transport")
-		if client.options.EnableLogs {
-			client.batchLogger = newLogBatchProcessor(client)
-			client.batchLogger.Start()
-		}
-		if !client.options.DisableMetrics {
-			client.batchMeter = newMetricBatchProcessor(client)
-			client.batchMeter.Start()
-		}
-		return
+func (client *Client) setupTelemetryProcessor() {
+	sdkInfo := &protocol.SdkInfo{
+		Name:         client.GetSDKIdentifier(),
+		Version:      SDKVersion,
+		Integrations: client.listIntegrations(),
+		Packages: []SdkPackage{{
+			Name:    "sentry-go",
+			Version: SDKVersion,
+		}},
 	}
-
 	transport := httpInternal.NewAsyncTransport(httpInternal.TransportOptions{
 		Dsn:           client.options.Dsn,
 		HTTPClient:    client.options.HTTPClient,
@@ -508,6 +497,7 @@ func (client *Client) setupTelemetryProcessor() { // nolint: unused
 		CaCerts:       client.options.CaCerts,
 		Recorder:      client.reportRecorder,
 		Provider:      client.reportProvider,
+		SdkInfo:       sdkInfo,
 	})
 	client.Transport = &internalAsyncTransportAdapter{transport: transport}
 
@@ -519,12 +509,7 @@ func (client *Client) setupTelemetryProcessor() { // nolint: unused
 		ratelimit.CategoryTraceMetric: telemetry.NewRingBuffer[protocol.TelemetryItem](ratelimit.CategoryTraceMetric, 10*100, telemetry.OverflowPolicyDropOldest, 100, 5*time.Second, client.reportRecorder),
 	}
 
-	sdkInfo := &protocol.SdkInfo{
-		Name:    client.sdkIdentifier,
-		Version: client.sdkVersion,
-	}
-
-	client.telemetryProcessor = telemetry.NewProcessor(buffers, transport, &client.dsn.Dsn, sdkInfo, client.reportRecorder)
+	client.telemetryProcessor = telemetry.NewProcessor(buffers, transport, client.dsn, sdkInfo, client.reportRecorder)
 }
 
 func (client *Client) setupIntegrations() {

client_reports_test.go

@@ -7,22 +7,28 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"strings"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/getsentry/sentry-go/internal/ratelimit"
 	"github.com/getsentry/sentry-go/internal/testutils"
 	"github.com/getsentry/sentry-go/report"
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/stretchr/testify/require"
 )
 
 // TestClientReports_Integration tests that client reports are properly generated
 // and sent when events are dropped for various reasons.
 func TestClientReports_Integration(t *testing.T) {
 	var receivedBodies [][]byte
+	var mu sync.Mutex
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		body, _ := io.ReadAll(r.Body)
+		mu.Lock()
 		receivedBodies = append(receivedBodies, body)
+		mu.Unlock()
 		w.WriteHeader(http.StatusOK)
 		_, _ = w.Write([]byte(`{"id":"test-event-id"}`))
 	}))
@@ -73,21 +79,23 @@ func TestClientReports_Integration(t *testing.T) {
 	}
 
 	var got report.ClientReport
-	found := false
-	for _, b := range receivedBodies {
-		for _, line := range bytes.Split(b, []byte("\n")) {
-			if json.Unmarshal(line, &got) == nil && len(got.DiscardedEvents) > 0 {
-				found = true
-				break
+	require.Eventually(t, func() bool {
+		mu.Lock()
+		bodies := make([][]byte, len(receivedBodies))
+		copy(bodies, receivedBodies)
+		mu.Unlock()
+
+		for _, b := range bodies {
+			for _, line := range bytes.Split(b, []byte("\n")) {
+				var report report.ClientReport
+				if json.Unmarshal(line, &report) == nil && len(report.DiscardedEvents) > 0 {
+					got = report
+					return true
+				}
 			}
 		}
-		if found {
-			break
-		}
-	}
-	if !found {
-		t.Fatal("no client report found in envelope bodies")
-	}
+		return false
+	}, time.Second, 10*time.Millisecond, "no client report found in envelope bodies with: %v", got)
 
 	if got.Timestamp.IsZero() {
 		t.Error("client report missing timestamp")

client_test.go

@@ -14,6 +14,7 @@ import (
 	"time"
 
 	"github.com/getsentry/sentry-go/internal/debuglog"
+	internalHttp "github.com/getsentry/sentry-go/internal/http"
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	pkgErrors "github.com/pkg/errors"
@@ -1063,9 +1064,6 @@ func TestClientSetsUpTransport(t *testing.T) {
 		Transport: &MockTransport{},
 	})
 	require.IsType(t, &MockTransport{}, client.Transport)
-
-	client, _ = NewClient(ClientOptions{})
-	require.IsType(t, &noopTransport{}, client.Transport)
 }
 
 func TestClient_SetupTelemetryBuffer_NoDSN(t *testing.T) {
@@ -1078,13 +1076,10 @@ func TestClient_SetupTelemetryBuffer_NoDSN(t *testing.T) {
 		t.Fatalf("unexpected error: %v", err)
 	}
 
-	if client.telemetryProcessor != nil {
-		t.Fatal("expected telemetryProcessor to be nil when DSN is missing")
-	}
-
-	if _, ok := client.Transport.(*noopTransport); !ok {
-		t.Fatalf("expected noopTransport, got %T", client.Transport)
+	if client.telemetryProcessor == nil {
+		t.Fatal("expected telemetryProcessor to not be nil when DSN is missing")
 	}
+	require.IsType(t, &internalHttp.NoopTransport{}, client.Transport.(*internalAsyncTransportAdapter).transport)
 }
 
 type multiClientEnv struct {

dynamic_sampling_context_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"testing"
 
+	"github.com/getsentry/sentry-go/internal/protocol"
 	"github.com/getsentry/sentry-go/internal/testutils"
 )
 
@@ -192,7 +193,7 @@ func TestDynamicSamplingContextFromScope(t *testing.T) {
 				},
 			},
 			client: func() *Client {
-				dsn, _ := NewDsn("http://public@example.com/sentry/1")
+				dsn, _ := protocol.NewDsn("http://public@example.com/sentry/1")
 				return &Client{
 					options: ClientOptions{
 						Dsn:         dsn.String(),

internal/http/transport.go

@@ -42,6 +42,7 @@ type TransportOptions struct {
 	CaCerts       *x509.CertPool
 	Recorder      report.ClientReportRecorder
 	Provider      report.ClientReportProvider
+	SdkInfo       *protocol.SdkInfo
 }
 
 func getProxyConfig(options TransportOptions) func(*http.Request) (*url.URL, error) {
@@ -74,8 +75,11 @@ func getTLSConfig(options TransportOptions) *tls.Config {
 func getSentryRequestFromEnvelope(ctx context.Context, dsn *protocol.Dsn, envelope *protocol.Envelope) (r *http.Request, err error) {
 	defer func() {
 		if r != nil {
-			sdkName := envelope.Header.Sdk.Name
-			sdkVersion := envelope.Header.Sdk.Version
+			var sdkName, sdkVersion string
+			if envelope.Header.Sdk != nil {
+				sdkVersion = envelope.Header.Sdk.Version
+				sdkName = envelope.Header.Sdk.Name
+			}
 
 			r.Header.Set("User-Agent", fmt.Sprintf("%s/%s", sdkName, sdkVersion))
 			r.Header.Set("Content-Type", "application/x-sentry-envelope")
@@ -151,6 +155,7 @@ type SyncTransport struct {
 	transport http.RoundTripper
 	recorder  report.ClientReportRecorder
 	provider  report.ClientReportProvider
+	sdkInfo   *protocol.SdkInfo
 
 	mu     sync.Mutex
 	limits ratelimit.Map
@@ -173,13 +178,18 @@ func NewSyncTransport(options TransportOptions) protocol.TelemetryTransport {
 	if provider == nil {
 		provider = report.NoopProvider()
 	}
+	sdkInfo := &protocol.SdkInfo{}
+	if options.SdkInfo != nil {
+		sdkInfo = options.SdkInfo
+	}
 
 	transport := &SyncTransport{
 		Timeout:  defaultTimeout,
 		limits:   make(ratelimit.Map),
 		dsn:      dsn,
 		recorder: recorder,
 		provider: provider,
+		sdkInfo:  sdkInfo,
 	}
 
 	if options.HTTPTransport != nil {
@@ -288,6 +298,7 @@ type AsyncTransport struct {
 	transport http.RoundTripper
 	recorder  report.ClientReportRecorder
 	provider  report.ClientReportProvider
+	sdkInfo   *protocol.SdkInfo
 
 	queue chan *protocol.Envelope
 
@@ -321,6 +332,10 @@ func NewAsyncTransport(options TransportOptions) protocol.TelemetryTransport {
 	if provider == nil {
 		provider = report.NoopProvider()
 	}
+	sdkInfo := &protocol.SdkInfo{}
+	if options.SdkInfo != nil {
+		sdkInfo = options.SdkInfo
+	}
 
 	transport := &AsyncTransport{
 		QueueSize: defaultQueueSize,
@@ -330,6 +345,7 @@ func NewAsyncTransport(options TransportOptions) protocol.TelemetryTransport {
 		dsn:       dsn,
 		recorder:  recorder,
 		provider:  provider,
+		sdkInfo:   sdkInfo,
 	}
 
 	transport.queue = make(chan *protocol.Envelope, transport.QueueSize)
@@ -494,7 +510,8 @@ func (t *AsyncTransport) sendClientReport() {
 	}
 	header := &protocol.EnvelopeHeader{
 		SentAt: time.Now(),
-		Dsn:    t.dsn.String(),
+		Dsn:    t.dsn,
+		Sdk:    t.sdkInfo,
 	}
 	envelope := protocol.NewEnvelope(header)
 	envelope.AddItem(item)

internal/protocol/envelope.go

@@ -26,7 +26,7 @@ type EnvelopeHeader struct {
 	// Dsn can be used for self-authenticated envelopes.
 	// This means that the envelope has all the information necessary to be sent to sentry.
 	// In this case the full DSN must be stored in this key.
-	Dsn string `json:"dsn,omitempty"`
+	Dsn *Dsn `json:"dsn,omitempty"`
 
 	// Sdk carries the same payload as the sdk interface in the event payload but can be carried for all events.
 	// This means that SDK information can be carried for minidumps, session data and other submissions.

internal/telemetry/scheduler.go

@@ -237,10 +237,7 @@ func (s *Scheduler) processItems(buffer Buffer[protocol.TelemetryItem], category
 	switch category {
 	case ratelimit.CategoryLog:
 		logs := protocol.Logs(items)
-		header := &protocol.EnvelopeHeader{EventID: protocol.GenerateEventID(), SentAt: time.Now(), Sdk: s.sdkInfo}
-		if s.dsn != nil {
-			header.Dsn = s.dsn.String()
-		}
+		header := &protocol.EnvelopeHeader{EventID: protocol.GenerateEventID(), SentAt: time.Now(), Dsn: s.dsn, Sdk: s.sdkInfo}
 		envelope := protocol.NewEnvelope(header)
 		item, err := logs.ToEnvelopeItem()
 		if err != nil {
@@ -254,10 +251,7 @@ func (s *Scheduler) processItems(buffer Buffer[protocol.TelemetryItem], category
 		return
 	case ratelimit.CategoryTraceMetric:
 		metrics := protocol.Metrics(items)
-		header := &protocol.EnvelopeHeader{EventID: protocol.GenerateEventID(), SentAt: time.Now(), Sdk: s.sdkInfo}
-		if s.dsn != nil {
-			header.Dsn = s.dsn.String()
-		}
+		header := &protocol.EnvelopeHeader{EventID: protocol.GenerateEventID(), SentAt: time.Now(), Dsn: s.dsn, Sdk: s.sdkInfo}
 		envelope := protocol.NewEnvelope(header)
 		item, err := metrics.ToEnvelopeItem()
 		if err != nil {
@@ -287,15 +281,13 @@ func (s *Scheduler) sendItem(item protocol.EnvelopeItemConvertible) {
 	header := &protocol.EnvelopeHeader{
 		EventID: item.GetEventID(),
 		SentAt:  time.Now(),
+		Dsn:     s.dsn,
 		Trace:   item.GetDynamicSamplingContext(),
 		Sdk:     s.sdkInfo,
 	}
 	if header.EventID == "" {
 		header.EventID = protocol.GenerateEventID()
 	}
-	if s.dsn != nil {
-		header.Dsn = s.dsn.String()
-	}
 	envelope := protocol.NewEnvelope(header)
 	envItem, err := item.ToEnvelopeItem()
 	if err != nil {