diff --git a/internal/telemetry/interfaces.go b/internal/telemetry/interfaces.go index fe85f3277..2a6540d36 100644 --- a/internal/telemetry/interfaces.go +++ b/internal/telemetry/interfaces.go @@ -2,6 +2,7 @@ package telemetry import ( "context" + "errors" "time" "github.com/getsentry/sentry-go/internal/ratelimit" @@ -36,6 +37,9 @@ type EnvelopeConvertible interface { ToEnvelope(*protocol.EnvelopeHeader) (*protocol.Envelope, error) } +// ErrQueueFull reports that a transport rejected an envelope because its queue is full. +var ErrQueueFull = errors.New("transport queue full") + // Transport represents the envelope-first transport interface. // This interface is designed for the telemetry buffer system and provides // non-blocking sends with backpressure signals. @@ -44,12 +48,6 @@ type Transport interface { // backpressure error if the queue is full. SendEnvelope(envelope *protocol.Envelope) error - // HasCapacity reports whether the transport has capacity to accept at least one more envelope. - HasCapacity() bool - - // IsRateLimited checks if a specific category is currently rate limited - IsRateLimited(category ratelimit.Category) bool - // Flush waits for all pending envelopes to be sent, with timeout Flush(timeout time.Duration) bool diff --git a/internal/telemetry/scheduler.go b/internal/telemetry/scheduler.go index 3d7b4898e..d1fdf0f47 100644 --- a/internal/telemetry/scheduler.go +++ b/internal/telemetry/scheduler.go @@ -2,6 +2,7 @@ package telemetry import ( "context" + "errors" "sync" "time" @@ -242,19 +243,6 @@ func (s *Scheduler) processItems(buffer Buffer[Item], category ratelimit.Categor return } - if s.isRateLimited(category) { - for _, item := range items { - s.recorder.RecordItem(report.ReasonRateLimitBackoff, item) - } - return - } - if !s.transport.HasCapacity() { - for _, item := range items { - s.recorder.RecordItem(report.ReasonQueueOverflow, item) - } - return - } - for _, item := range s.envelopeConvertibles(category, items) { s.sendItem(item) } @@ -301,6 +289,11 @@ func (s *Scheduler) sendItem(item EnvelopeConvertible) { } if err := s.transport.SendEnvelope(envelope); err != nil { debuglog.Printf("error sending envelope: %v", err) + reason := report.ReasonSendError + if errors.Is(err, ErrQueueFull) { + reason = report.ReasonQueueOverflow + } + s.recorder.RecordForEnvelope(reason, envelope) } } @@ -322,7 +315,3 @@ func (s *Scheduler) flushBuffers(ctx context.Context) bool { } return ctx.Err() == nil } - -func (s *Scheduler) isRateLimited(category ratelimit.Category) bool { - return s.transport.IsRateLimited(category) -} diff --git a/internal/telemetry/scheduler_test.go b/internal/telemetry/scheduler_test.go index 591bfde3b..554613b85 100644 --- a/internal/telemetry/scheduler_test.go +++ b/internal/telemetry/scheduler_test.go @@ -311,39 +311,6 @@ func TestTelemetrySchedulerFlushAlreadyCanceled(t *testing.T) { require.True(t, scheduler.Flush(testutils.FlushTimeout())) } -func TestTelemetrySchedulerRateLimiting(t *testing.T) { - transport := &testutils.MockTelemetryTransport{} - dsn := &protocol.Dsn{} - - buffer := NewRingBuffer[Item](ratelimit.CategoryError, 10, OverflowPolicyDropOldest, 1, 0, nil) - buffers := map[ratelimit.Category]Buffer[Item]{ - ratelimit.CategoryError: buffer, - } - // no log buffer used in simplified scheduler tests - sdkInfo := &protocol.SdkInfo{Name: "test-sdk", Version: "1.0.0"} - - scheduler := NewScheduler(buffers, transport, dsn, func() *protocol.SdkInfo { return sdkInfo }, nil) - - transport.SetRateLimited("error", true) - - scheduler.Start() - defer scheduler.Stop(100 * time.Millisecond) - - item := &testTelemetryItem{id: 1, data: "test"} - buffer.Offer(item) - scheduler.Signal() - - time.Sleep(200 * time.Millisecond) - - if transport.GetSendCount() > 0 { - t.Errorf("Expected 0 items to be processed due to rate limiting, got %d", transport.GetSendCount()) - } - - if transport.GetRateLimitedCalls() == 0 { - t.Error("Expected rate limit check to be called") - } -} - func TestTelemetrySchedulerStartStop(t *testing.T) { transport := &testutils.MockTelemetryTransport{} dsn := &protocol.Dsn{} @@ -405,43 +372,68 @@ func TestTelemetrySchedulerContextCancellation(t *testing.T) { } } -func TestTelemetrySchedulerRecordsFullDiscardCountsOnEnvelopeError(t *testing.T) { - transport := &testutils.MockTelemetryTransport{} - dsn := &protocol.Dsn{} - recorder := reportpkg.NewAggregator() +type rejectingTransport struct { + testutils.MockTelemetryTransport + err error +} - buffer := NewRingBuffer[Item](ratelimit.CategoryTransaction, 10, OverflowPolicyDropOldest, 1, 0, nil) - buffers := map[ratelimit.Category]Buffer[Item]{ - ratelimit.CategoryTransaction: buffer, +func (t *rejectingTransport) SendEnvelope(*protocol.Envelope) error { return t.err } + +type transactionTelemetryItem struct{ testTelemetryItem } + +func (t *transactionTelemetryItem) ToEnvelope(header *protocol.EnvelopeHeader) (*protocol.Envelope, error) { + return protocol.NewEnvelope(header, protocol.NewTransactionItem(3, []byte(`{}`))), nil +} + +func TestTelemetrySchedulerRecordsFullDiscardCountsOnEnvelopeError(t *testing.T) { + transaction := testTelemetryItem{data: "tx", category: ratelimit.CategoryTransaction} + tests := []struct { + name string + item Item + err error + reason reportpkg.DiscardReason + }{ + {"conversion error", &failingTransactionTelemetryItem{testTelemetryItem: transaction, spanCount: 3}, nil, reportpkg.ReasonInternalError}, + {"queue full", &transactionTelemetryItem{transaction}, ErrQueueFull, reportpkg.ReasonQueueOverflow}, + {"send error", &transactionTelemetryItem{transaction}, errors.New("send failed"), reportpkg.ReasonSendError}, } - sdkInfo := &protocol.SdkInfo{Name: "test-sdk", Version: "1.0.0"} + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + transport := &rejectingTransport{err: tt.err} + dsn := &protocol.Dsn{} + recorder := reportpkg.NewAggregator() - scheduler := NewScheduler(buffers, transport, dsn, func() *protocol.SdkInfo { return sdkInfo }, recorder) + buffer := NewRingBuffer[Item](ratelimit.CategoryTransaction, 10, OverflowPolicyDropOldest, 1, 0, nil) + buffers := map[ratelimit.Category]Buffer[Item]{ + ratelimit.CategoryTransaction: buffer, + } + sdkInfo := &protocol.SdkInfo{Name: "test-sdk", Version: "1.0.0"} - buffer.Offer(&failingTransactionTelemetryItem{ - testTelemetryItem: testTelemetryItem{data: "tx", category: ratelimit.CategoryTransaction}, - spanCount: 3, - }) + scheduler := NewScheduler(buffers, transport, dsn, func() *protocol.SdkInfo { return sdkInfo }, recorder) - scheduler.Flush(time.Second) + buffer.Offer(tt.item) - clientReport := recorder.TakeReport() - if clientReport == nil { - t.Fatal("expected client report") - } + scheduler.Flush(time.Second) - outcomes := map[ratelimit.Category]int64{} - for _, discarded := range clientReport.DiscardedEvents { - if discarded.Reason != reportpkg.ReasonInternalError { - t.Fatalf("unexpected reason: %s", discarded.Reason) - } - outcomes[discarded.Category] += discarded.Quantity - } + clientReport := recorder.TakeReport() + if clientReport == nil { + t.Fatal("expected client report") + } - if outcomes[ratelimit.CategoryTransaction] != 1 { - t.Fatalf("expected one discarded transaction, got %d", outcomes[ratelimit.CategoryTransaction]) - } - if outcomes[ratelimit.CategorySpan] != 3 { - t.Fatalf("expected discarded span count to be recorded, got %d", outcomes[ratelimit.CategorySpan]) + outcomes := map[ratelimit.Category]int64{} + for _, discarded := range clientReport.DiscardedEvents { + if discarded.Reason != tt.reason { + t.Fatalf("unexpected reason: %s", discarded.Reason) + } + outcomes[discarded.Category] += discarded.Quantity + } + + if outcomes[ratelimit.CategoryTransaction] != 1 { + t.Fatalf("expected one discarded transaction, got %d", outcomes[ratelimit.CategoryTransaction]) + } + if outcomes[ratelimit.CategorySpan] != 3 { + t.Fatalf("expected discarded span count to be recorded, got %d", outcomes[ratelimit.CategorySpan]) + } + }) } } diff --git a/internal/testutils/mocks.go b/internal/testutils/mocks.go index 204c79f70..9539322e4 100644 --- a/internal/testutils/mocks.go +++ b/internal/testutils/mocks.go @@ -6,18 +6,14 @@ import ( "sync/atomic" "time" - "github.com/getsentry/sentry-go/internal/ratelimit" "github.com/getsentry/sentry-go/protocol" ) type MockTelemetryTransport struct { - sentEnvelopes []*protocol.Envelope - rateLimited map[string]bool - sendError error - mu sync.Mutex - sendCount int64 - rateLimitedCalls int64 - capacity int + sentEnvelopes []*protocol.Envelope + sendError error + mu sync.Mutex + sendCount int64 } func (m *MockTelemetryTransport) SendEnvelope(envelope *protocol.Envelope) error { @@ -33,26 +29,6 @@ func (m *MockTelemetryTransport) SendEnvelope(envelope *protocol.Envelope) error return nil } -func (m *MockTelemetryTransport) IsRateLimited(category ratelimit.Category) bool { - atomic.AddInt64(&m.rateLimitedCalls, 1) - m.mu.Lock() - defer m.mu.Unlock() - - if m.rateLimited == nil { - return false - } - return m.rateLimited[string(category)] -} - -func (m *MockTelemetryTransport) HasCapacity() bool { - m.mu.Lock() - defer m.mu.Unlock() - if m.capacity == 0 { - return true - } - return int(m.sendCount) < m.capacity -} - func (m *MockTelemetryTransport) Flush(_ time.Duration) bool { return true } @@ -76,29 +52,13 @@ func (m *MockTelemetryTransport) GetSentEnvelopes() []*protocol.Envelope { return result } -func (m *MockTelemetryTransport) SetRateLimited(category string, limited bool) { - m.mu.Lock() - defer m.mu.Unlock() - if m.rateLimited == nil { - m.rateLimited = make(map[string]bool) - } - m.rateLimited[category] = limited -} - func (m *MockTelemetryTransport) GetSendCount() int64 { return atomic.LoadInt64(&m.sendCount) } -func (m *MockTelemetryTransport) GetRateLimitedCalls() int64 { - return atomic.LoadInt64(&m.rateLimitedCalls) -} - func (m *MockTelemetryTransport) Reset() { m.mu.Lock() defer m.mu.Unlock() m.sentEnvelopes = nil - m.rateLimited = nil atomic.StoreInt64(&m.sendCount, 0) - atomic.StoreInt64(&m.rateLimitedCalls, 0) - m.capacity = 0 } diff --git a/report/aggregator.go b/report/aggregator.go index 99c284feb..b5d281c02 100644 --- a/report/aggregator.go +++ b/report/aggregator.go @@ -1,6 +1,7 @@ package report import ( + "encoding/json" "sync" "sync/atomic" "time" @@ -115,10 +116,21 @@ func (a *Aggregator) RecordForEnvelope(reason DiscardReason, envelope *protocol. a.Record(reason, ratelimit.CategoryLogByte, int64(*item.Header.Length)) } case protocol.EnvelopeItemTypeTraceMetric: - a.RecordOne(reason, ratelimit.CategoryTraceMetric) + if item.Header.ItemCount != nil { + a.Record(reason, ratelimit.CategoryTraceMetric, int64(*item.Header.ItemCount)) + } case protocol.EnvelopeItemTypeCheckIn: a.RecordOne(reason, ratelimit.CategoryMonitor) - case protocol.EnvelopeItemTypeAttachment, protocol.EnvelopeItemTypeClientReport: + case protocol.EnvelopeItemTypeClientReport: + var clientReport ClientReport + if err := json.Unmarshal(item.Payload, &clientReport); err != nil { + debuglog.Printf("failed to decode client report: %v", err) + continue + } + for _, event := range clientReport.DiscardedEvents { + a.Record(event.Reason, event.Category, event.Quantity) + } + case protocol.EnvelopeItemTypeAttachment: // Skip — not reportable categories } } diff --git a/transport.go b/transport.go index 1eddf1205..cea2825d1 100644 --- a/transport.go +++ b/transport.go @@ -29,9 +29,9 @@ const ( ) var ( - ErrTransportQueueFull = errors.New("transport queue full") + ErrTransportQueueFull = telemetry.ErrQueueFull ErrTransportClosed = errors.New("transport is closed") - ErrEmptyEnvelope = errors.New("empty envelope provided") + ErrInvalidEnvelope = errors.New("invalid envelope: missing header or items") ) type TransportOptions struct { @@ -129,7 +129,9 @@ func categoryFromEnvelope(envelope *protocol.Envelope) ratelimit.Category { return ratelimit.CategoryMonitor case protocol.EnvelopeItemTypeLog: return ratelimit.CategoryLog - case protocol.EnvelopeItemTypeAttachment: + case protocol.EnvelopeItemTypeTraceMetric: + return ratelimit.CategoryTraceMetric + case protocol.EnvelopeItemTypeAttachment, protocol.EnvelopeItemTypeClientReport: continue default: return ratelimit.CategoryAll @@ -139,18 +141,20 @@ func categoryFromEnvelope(envelope *protocol.Envelope) ratelimit.Category { return ratelimit.CategoryAll } -// SyncTransport is a blocking implementation of Transport. -// -// Clients using this transport will send requests to Sentry sequentially and -// block until a response is returned. -// -// The blocking behavior is useful in a limited set of use cases. For example, -// use it when deploying code to a Function as a Service ("Serverless") -// platform, where any work happening in a background goroutine is not -// guaranteed to execute. -// -// For most cases, prefer AsyncTransport. -type SyncTransport struct { +func validEnvelope(envelope *protocol.Envelope) bool { + if envelope == nil || envelope.Header == nil || len(envelope.Items) == 0 { + return false + } + for _, item := range envelope.Items { + if item == nil || item.Header == nil { + return false + } + } + return true +} + +// httpSender holds the delivery state shared by SyncTransport and AsyncTransport. +type httpSender struct { dsn *protocol.Dsn client *http.Client transport http.RoundTripper @@ -158,19 +162,11 @@ type SyncTransport struct { provider report.ClientReportProvider sdkInfo func() *protocol.SdkInfo - mu sync.Mutex + mu sync.RWMutex limits ratelimit.Map - - Timeout time.Duration } -func NewSyncTransport(options TransportOptions) telemetry.Transport { - dsn, err := protocol.NewDsn(options.Dsn) - if err != nil || dsn == nil { - debuglog.Printf("Transport is disabled: invalid dsn: %v\n", err) - return NewNoopTransport() - } - +func newHTTPSender(dsn *protocol.Dsn, options TransportOptions, timeout time.Duration) *httpSender { recorder := options.Recorder if recorder == nil { recorder = report.NoopRecorder() @@ -180,8 +176,7 @@ func NewSyncTransport(options TransportOptions) telemetry.Transport { provider = report.NoopProvider() } - transport := &SyncTransport{ - Timeout: defaultTimeout, + sender := &httpSender{ limits: make(ratelimit.Map), dsn: dsn, recorder: recorder, @@ -190,23 +185,55 @@ func NewSyncTransport(options TransportOptions) telemetry.Transport { } if options.HTTPTransport != nil { - transport.transport = options.HTTPTransport + sender.transport = options.HTTPTransport } else { - transport.transport = &http.Transport{ + sender.transport = &http.Transport{ Proxy: getProxyConfig(options.HTTPProxy, options.HTTPSProxy), TLSClientConfig: getTLSConfig(options), } } if options.HTTPClient != nil { - transport.client = options.HTTPClient + sender.client = options.HTTPClient } else { - transport.client = &http.Client{ - Transport: transport.transport, - Timeout: transport.Timeout, + sender.client = &http.Client{ + Transport: sender.transport, + Timeout: timeout, } } + return sender +} + +// SyncTransport is a blocking implementation of Transport. +// +// Clients using this transport will send requests to Sentry sequentially and +// block until a response is returned. +// +// The blocking behavior is useful in a limited set of use cases. For example, +// use it when deploying code to a Function as a Service ("Serverless") +// platform, where any work happening in a background goroutine is not +// guaranteed to execute. +// +// For most cases, prefer AsyncTransport. +type SyncTransport struct { + *httpSender + + Timeout time.Duration +} + +func NewSyncTransport(options TransportOptions) telemetry.Transport { + dsn, err := protocol.NewDsn(options.Dsn) + if err != nil || dsn == nil { + debuglog.Printf("Transport is disabled: invalid dsn: %v\n", err) + return NewNoopTransport() + } + + transport := &SyncTransport{ + httpSender: newHTTPSender(dsn, options, defaultTimeout), + Timeout: defaultTimeout, + } + return transport } @@ -216,56 +243,21 @@ func (t *SyncTransport) SendEnvelope(envelope *protocol.Envelope) error { func (t *SyncTransport) Close() {} -func (t *SyncTransport) IsRateLimited(category ratelimit.Category) bool { - return t.disabled(category) -} - -func (t *SyncTransport) HasCapacity() bool { return true } - func (t *SyncTransport) SendEnvelopeWithContext(ctx context.Context, envelope *protocol.Envelope) error { ctx, cancel := context.WithTimeout(ctx, t.Timeout) defer cancel() - if envelope == nil || len(envelope.Items) == 0 { - return ErrEmptyEnvelope + if !validEnvelope(envelope) { + return ErrInvalidEnvelope } - category := categoryFromEnvelope(envelope) - if t.disabled(category) { - t.recorder.RecordForEnvelope(report.ReasonRateLimitBackoff, envelope) - return nil - } - // the sync transport needs to attach client reports when available - t.provider.AttachToEnvelope(envelope) - - request, err := getSentryRequestFromEnvelope(ctx, t.dsn, envelope) - if err != nil { - debuglog.Printf("There was an issue creating the request: %v", err) - t.recorder.RecordForEnvelope(report.ReasonInternalError, envelope) - return err - } - identifier := util.EnvelopeIdentifier(envelope) debuglog.Printf( "Sending %s to %s project: %s", - identifier, + util.EnvelopeIdentifier(envelope), t.dsn.GetHost(), t.dsn.GetProjectID(), ) - - result, err := util.DoSendRequest(t.client, request, identifier) - if err != nil { - debuglog.Printf("There was an issue with sending an event: %v", err) - t.recorder.RecordForEnvelope(report.ReasonNetworkError, envelope) - return err - } - if result.IsSendError() { - t.recorder.RecordForEnvelope(report.ReasonSendError, envelope) - } - - t.mu.Lock() - t.limits.Merge(result.Limits) - t.mu.Unlock() - + t.send(ctx, envelope) return nil } @@ -277,34 +269,16 @@ func (t *SyncTransport) FlushWithContext(_ context.Context) bool { return true } -func (t *SyncTransport) disabled(c ratelimit.Category) bool { - t.mu.Lock() - defer t.mu.Unlock() - disabled := t.limits.IsRateLimited(c) - if disabled { - debuglog.Printf("Too many requests for %q, backing off till: %v", c, t.limits.Deadline(c)) - } - return disabled -} - // AsyncTransport is the default, non-blocking, implementation of Transport. // // Clients using this transport will enqueue requests in a queue and return to // the caller before any network communication has happened. Requests are sent // to Sentry sequentially from a background goroutine. type AsyncTransport struct { - dsn *protocol.Dsn - client *http.Client - transport http.RoundTripper - recorder report.ClientReportRecorder - provider report.ClientReportProvider - sdkInfo func() *protocol.SdkInfo + *httpSender queue chan *protocol.Envelope - mu sync.RWMutex - limits ratelimit.Map - // ctx bounds in-flight requests and is canceled by Close. ctx context.Context cancel context.CancelFunc @@ -334,47 +308,16 @@ func NewAsyncTransport(options TransportOptions) telemetry.Transport { return NewNoopTransport() } - recorder := options.Recorder - if recorder == nil { - recorder = report.NoopRecorder() - } - provider := options.Provider - if provider == nil { - provider = report.NoopProvider() - } - transport := &AsyncTransport{ - QueueSize: defaultQueueSize, - Timeout: defaultTimeout, - done: make(chan struct{}), - limits: make(ratelimit.Map), - dsn: dsn, - recorder: recorder, - provider: provider, - sdkInfo: options.SdkInfo, + httpSender: newHTTPSender(dsn, options, defaultTimeout), + QueueSize: defaultQueueSize, + Timeout: defaultTimeout, + done: make(chan struct{}), } transport.queue = make(chan *protocol.Envelope, transport.QueueSize) transport.flushRequest = make(chan flushRequest) - if options.HTTPTransport != nil { - transport.transport = options.HTTPTransport - } else { - transport.transport = &http.Transport{ - Proxy: getProxyConfig(options.HTTPProxy, options.HTTPSProxy), - TLSClientConfig: getTLSConfig(options), - } - } - - if options.HTTPClient != nil { - transport.client = options.HTTPClient - } else { - transport.client = &http.Client{ - Transport: transport.transport, - Timeout: transport.Timeout, - } - } - transport.start() return transport } @@ -393,19 +336,6 @@ func (t *AsyncTransport) start() { }) } -// HasCapacity reports whether the async transport queue appears to have space -// for at least one more envelope. This is a best-effort, non-blocking check. -func (t *AsyncTransport) HasCapacity() bool { - t.mu.RLock() - defer t.mu.RUnlock() - select { - case <-t.done: - return false - default: - } - return len(t.queue) < cap(t.queue) -} - func (t *AsyncTransport) SendEnvelope(envelope *protocol.Envelope) error { t.closeMu.RLock() defer t.closeMu.RUnlock() @@ -416,8 +346,8 @@ func (t *AsyncTransport) SendEnvelope(envelope *protocol.Envelope) error { default: } - if envelope == nil || len(envelope.Items) == 0 { - return ErrEmptyEnvelope + if !validEnvelope(envelope) { + return ErrInvalidEnvelope } category := categoryFromEnvelope(envelope) @@ -440,7 +370,6 @@ func (t *AsyncTransport) SendEnvelope(envelope *protocol.Envelope) error { ) return nil default: - t.recorder.RecordForEnvelope(report.ReasonQueueOverflow, envelope) return ErrTransportQueueFull } } @@ -489,10 +418,6 @@ func (t *AsyncTransport) Close() { }) } -func (t *AsyncTransport) IsRateLimited(category ratelimit.Category) bool { - return t.isRateLimited(category) -} - func (t *AsyncTransport) resolveSdkInfo() *protocol.SdkInfo { if t.sdkInfo == nil { return &protocol.SdkInfo{} @@ -580,48 +505,52 @@ func (t *AsyncTransport) drainQueue(ctx context.Context) { } func (t *AsyncTransport) sendEnvelopeHTTP(envelope *protocol.Envelope) bool { //nolint: unparam + ctx, cancel := context.WithTimeout(t.ctx, t.Timeout) + defer cancel() + return t.send(ctx, envelope) +} + +// send records every loss of an accepted envelope. +func (s *httpSender) send(ctx context.Context, envelope *protocol.Envelope) bool { category := categoryFromEnvelope(envelope) - if t.isRateLimited(category) { - t.recorder.RecordForEnvelope(report.ReasonRateLimitBackoff, envelope) + if s.isRateLimited(category) { + s.recorder.RecordForEnvelope(report.ReasonRateLimitBackoff, envelope) return false } // attach to envelope after rate-limit check - t.provider.AttachToEnvelope(envelope) - - ctx, cancel := context.WithTimeout(t.ctx, t.Timeout) - defer cancel() + s.provider.AttachToEnvelope(envelope) - request, err := getSentryRequestFromEnvelope(ctx, t.dsn, envelope) + request, err := getSentryRequestFromEnvelope(ctx, s.dsn, envelope) if err != nil { debuglog.Printf("Failed to create request from envelope: %v", err) - t.recorder.RecordForEnvelope(report.ReasonInternalError, envelope) + s.recorder.RecordForEnvelope(report.ReasonInternalError, envelope) return false } identifier := util.EnvelopeIdentifier(envelope) - result, err := util.DoSendRequest(t.client, request, identifier) + result, err := util.DoSendRequest(s.client, request, identifier) if err != nil { debuglog.Printf("HTTP request failed: %v", err) - t.recorder.RecordForEnvelope(report.ReasonNetworkError, envelope) + s.recorder.RecordForEnvelope(report.ReasonNetworkError, envelope) return false } if result.IsSendError() { - t.recorder.RecordForEnvelope(report.ReasonSendError, envelope) + s.recorder.RecordForEnvelope(report.ReasonSendError, envelope) } - t.mu.Lock() - t.limits.Merge(result.Limits) - t.mu.Unlock() + s.mu.Lock() + s.limits.Merge(result.Limits) + s.mu.Unlock() return result.Success } -func (t *AsyncTransport) isRateLimited(category ratelimit.Category) bool { - t.mu.RLock() - defer t.mu.RUnlock() - limited := t.limits.IsRateLimited(category) +func (s *httpSender) isRateLimited(category ratelimit.Category) bool { + s.mu.RLock() + defer s.mu.RUnlock() + limited := s.limits.IsRateLimited(category) if limited { - debuglog.Printf("Rate limited for category %q until %v", category, t.limits.Deadline(category)) + debuglog.Printf("Rate limited for category %q until %v", category, s.limits.Deadline(category)) } return limited } @@ -640,10 +569,6 @@ func (t *NoopTransport) SendEnvelope(_ *protocol.Envelope) error { return nil } -func (t *NoopTransport) IsRateLimited(_ ratelimit.Category) bool { - return false -} - func (t *NoopTransport) Flush(_ time.Duration) bool { return true } @@ -655,5 +580,3 @@ func (t *NoopTransport) FlushWithContext(_ context.Context) bool { func (t *NoopTransport) Close() { // Nothing to close } - -func (t *NoopTransport) HasCapacity() bool { return true } diff --git a/transport_legacy.go b/transport_legacy.go index eb1edce11..2197edb18 100644 --- a/transport_legacy.go +++ b/transport_legacy.go @@ -5,6 +5,7 @@ import ( "context" "crypto/tls" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -925,6 +926,13 @@ func (a *internalAsyncTransportAdapter) SendEvent(event *Event) { if err := a.transport.SendEnvelope(envelope); err != nil { debuglog.Printf("Error sending envelope: %v", err) + if a.recorder != nil { + reason := report.ReasonSendError + if errors.Is(err, ErrTransportQueueFull) { + reason = report.ReasonQueueOverflow + } + a.recorder.RecordForEnvelope(reason, envelope) + } } } diff --git a/transport_test.go b/transport_test.go index 22515fa49..260c8bdc8 100644 --- a/transport_test.go +++ b/transport_test.go @@ -5,6 +5,7 @@ import ( "crypto/x509" "errors" "fmt" + "io" "net" "net/http" "net/http/httptest" @@ -19,9 +20,20 @@ import ( "github.com/getsentry/sentry-go/internal/ratelimit" "github.com/getsentry/sentry-go/internal/testutils" "github.com/getsentry/sentry-go/protocol" + "github.com/getsentry/sentry-go/report" "go.uber.org/goleak" ) +func takeOutcomes(recorder *report.Aggregator) map[report.OutcomeKey]int64 { + outcomes := make(map[report.OutcomeKey]int64) + if clientReport := recorder.TakeReport(); clientReport != nil { + for _, event := range clientReport.DiscardedEvents { + outcomes[report.OutcomeKey{Reason: event.Reason, Category: event.Category}] += event.Quantity + } + } + return outcomes +} + func testEnvelope(itemType protocol.EnvelopeItemType) *protocol.Envelope { return &protocol.Envelope{ Header: &protocol.EnvelopeHeader{ @@ -74,13 +86,15 @@ func TestAsyncTransport_SendEnvelope(t *testing.T) { t.Run("success", func(t *testing.T) { tests := []struct { name string - itemType protocol.EnvelopeItemType + envelope *protocol.Envelope + wantErr error }{ - {"event", protocol.EnvelopeItemTypeEvent}, - {"transaction", protocol.EnvelopeItemTypeTransaction}, - {"check-in", protocol.EnvelopeItemTypeCheckIn}, - {"log", protocol.EnvelopeItemTypeLog}, - {"attachment", protocol.EnvelopeItemTypeAttachment}, + {"event", testEnvelope(protocol.EnvelopeItemTypeEvent), nil}, + {"transaction", testEnvelope(protocol.EnvelopeItemTypeTransaction), nil}, + {"check-in", testEnvelope(protocol.EnvelopeItemTypeCheckIn), nil}, + {"log", testEnvelope(protocol.EnvelopeItemTypeLog), nil}, + {"attachment", testEnvelope(protocol.EnvelopeItemTypeAttachment), nil}, + {"nil header", &protocol.Envelope{Items: testEnvelope(protocol.EnvelopeItemTypeEvent).Items}, ErrInvalidEnvelope}, } var count int64 @@ -100,8 +114,8 @@ func TestAsyncTransport_SendEnvelope(t *testing.T) { defer transport.Close() for _, tt := range tests { - if err := transport.SendEnvelope(testEnvelope(tt.itemType)); err != nil { - t.Errorf("send %s failed: %v", tt.name, err) + if err := transport.SendEnvelope(tt.envelope); !errors.Is(err, tt.wantErr) { + t.Errorf("send %s returned %v, want %v", tt.name, err, tt.wantErr) } } @@ -109,7 +123,7 @@ func TestAsyncTransport_SendEnvelope(t *testing.T) { t.Fatal("Flush timed out") } - expectedCount := int64(len(tests)) + expectedCount := int64(len(tests) - 1) // one invalid envelope if sent := atomic.LoadInt64(&count); sent != expectedCount { t.Errorf("expected %d sent, got %d", expectedCount, sent) } @@ -172,13 +186,13 @@ func TestAsyncTransport_SendEnvelope(t *testing.T) { t.Fatal("Flush timed out") } - if !transport.IsRateLimited(ratelimit.CategoryError) { + if !transport.isRateLimited(ratelimit.CategoryError) { t.Error("error category should be rate limited") } - if !transport.IsRateLimited(ratelimit.CategoryTransaction) { + if !transport.isRateLimited(ratelimit.CategoryTransaction) { t.Error("transaction category should be rate limited") } - if transport.IsRateLimited(ratelimit.CategoryMonitor) { + if transport.isRateLimited(ratelimit.CategoryMonitor) { t.Error("monitor category should not be rate limited") } @@ -204,14 +218,18 @@ func TestAsyncTransport_SendEnvelope(t *testing.T) { defer server.Close() dsn, _ := protocol.NewDsn("http://key@" + server.URL[7:] + "/123") + recorder := report.NewAggregator() transport := &AsyncTransport{ + httpSender: &httpSender{ + limits: make(ratelimit.Map), + dsn: dsn, + transport: &http.Transport{}, + client: &http.Client{Timeout: defaultTimeout}, + recorder: recorder, + }, QueueSize: 2, Timeout: defaultTimeout, done: make(chan struct{}), - limits: make(ratelimit.Map), - dsn: dsn, - transport: &http.Transport{}, - client: &http.Client{Timeout: defaultTimeout}, } // manually set the queue size to simulate overflow transport.queue = make(chan *protocol.Envelope, transport.QueueSize) @@ -238,6 +256,9 @@ func TestAsyncTransport_SendEnvelope(t *testing.T) { if !errors.Is(err, ErrTransportQueueFull) { t.Errorf("expected ErrTransportQueueFull, got %v", err) } + if outcomes := takeOutcomes(recorder); len(outcomes) != 0 { + t.Errorf("rejected envelope should not be recorded by the transport, got %v", outcomes) + } }) t.Run("FlushMultipleTimes", func(t *testing.T) { @@ -440,13 +461,19 @@ func TestSyncTransport_SendEnvelope(t *testing.T) { t.Run("success", func(t *testing.T) { tests := []struct { name string - itemType protocol.EnvelopeItemType + envelope *protocol.Envelope + wantErr error }{ - {"event", protocol.EnvelopeItemTypeEvent}, - {"transaction", protocol.EnvelopeItemTypeTransaction}, - {"check-in", protocol.EnvelopeItemTypeCheckIn}, - {"log", protocol.EnvelopeItemTypeLog}, - {"attachment", protocol.EnvelopeItemTypeAttachment}, + {"event", testEnvelope(protocol.EnvelopeItemTypeEvent), nil}, + {"transaction", testEnvelope(protocol.EnvelopeItemTypeTransaction), nil}, + {"check-in", testEnvelope(protocol.EnvelopeItemTypeCheckIn), nil}, + {"log", testEnvelope(protocol.EnvelopeItemTypeLog), nil}, + {"attachment", testEnvelope(protocol.EnvelopeItemTypeAttachment), nil}, + {"nil envelope", nil, ErrInvalidEnvelope}, + {"nil header", &protocol.Envelope{Items: testEnvelope(protocol.EnvelopeItemTypeEvent).Items}, ErrInvalidEnvelope}, + {"no items", &protocol.Envelope{Header: &protocol.EnvelopeHeader{}}, ErrInvalidEnvelope}, + {"nil item", &protocol.Envelope{Header: &protocol.EnvelopeHeader{}, Items: []*protocol.EnvelopeItem{nil}}, ErrInvalidEnvelope}, + {"nil item header", &protocol.Envelope{Header: &protocol.EnvelopeHeader{}, Items: []*protocol.EnvelopeItem{{}}}, ErrInvalidEnvelope}, } server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { @@ -460,55 +487,90 @@ func TestSyncTransport_SendEnvelope(t *testing.T) { defer transport.Close() for _, tt := range tests { - if err := transport.SendEnvelope(testEnvelope(tt.itemType)); err != nil { - t.Errorf("send %s failed: %v", tt.name, err) + if err := transport.SendEnvelope(tt.envelope); !errors.Is(err, tt.wantErr) { + t.Errorf("send %s returned %v, want %v", tt.name, err, tt.wantErr) } } }) t.Run("rate limited", func(t *testing.T) { + var requests atomic.Int32 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - w.Header().Add("X-Sentry-Rate-Limits", "60:error,60:transaction") + requests.Add(1) + w.Header().Add("X-Sentry-Rate-Limits", "60:error,60:transaction,60:trace_metric") w.WriteHeader(http.StatusTooManyRequests) })) defer server.Close() + recorder := report.NewAggregator() transport := NewSyncTransport(TransportOptions{ - Dsn: "http://key@" + server.URL[7:] + "/123", + Dsn: "http://key@" + server.URL[7:] + "/123", + Recorder: recorder, }) _ = transport.SendEnvelope(testEnvelope(protocol.EnvelopeItemTypeEvent)) + _ = takeOutcomes(recorder) - if !transport.IsRateLimited(ratelimit.CategoryError) { - t.Error("error category should be rate limited") - } - if !transport.IsRateLimited(ratelimit.CategoryTransaction) { - t.Error("transaction category should be rate limited") + pending, _ := (&report.ClientReport{DiscardedEvents: []report.DiscardedEvent{ + {Reason: report.ReasonBeforeSend, Category: ratelimit.CategoryError, Quantity: 2}, + }}).ToEnvelopeItem() + backoff := func(category ratelimit.Category) report.OutcomeKey { + return report.OutcomeKey{Reason: report.ReasonRateLimitBackoff, Category: category} } - if transport.IsRateLimited(ratelimit.CategoryMonitor) { - t.Error("monitor category should not be rate limited") + tests := []struct { + name string + envelope *protocol.Envelope + want map[report.OutcomeKey]int64 // nil when the envelope is sent + }{ + {"event", testEnvelope(protocol.EnvelopeItemTypeEvent), map[report.OutcomeKey]int64{backoff(ratelimit.CategoryError): 1}}, + {"transaction", testEnvelope(protocol.EnvelopeItemTypeTransaction), map[report.OutcomeKey]int64{backoff(ratelimit.CategoryTransaction): 1}}, + {"trace metric", protocol.NewEnvelope(&protocol.EnvelopeHeader{}, protocol.NewTraceMetricItem(3, []byte(`{"items":[]}`))), + map[report.OutcomeKey]int64{backoff(ratelimit.CategoryTraceMetric): 3}}, + {"client report before event", protocol.NewEnvelope(&protocol.EnvelopeHeader{}, pending, testEnvelope(protocol.EnvelopeItemTypeEvent).Items[0]), + map[report.OutcomeKey]int64{backoff(ratelimit.CategoryError): 1, {Reason: report.ReasonBeforeSend, Category: ratelimit.CategoryError}: 2}}, + {"check-in", testEnvelope(protocol.EnvelopeItemTypeCheckIn), nil}, } - - err := transport.SendEnvelope(testEnvelope(protocol.EnvelopeItemTypeEvent)) - if err != nil { - t.Errorf("rate limited envelope should return nil, got %v", err) + for _, tt := range tests { + sent := requests.Load() + if err := transport.SendEnvelope(tt.envelope); err != nil { + t.Errorf("%s: rate limited envelope should return nil, got %v", tt.name, err) + } + if gotSent := requests.Load() > sent; gotSent != (tt.want == nil) { + t.Errorf("%s: sent = %v, want %v", tt.name, gotSent, tt.want == nil) + } + if outcomes := takeOutcomes(recorder); tt.want != nil && fmt.Sprint(outcomes) != fmt.Sprint(tt.want) { + t.Errorf("%s: got outcomes %v, want %v", tt.name, outcomes, tt.want) + } } }) - t.Run("server error", func(t *testing.T) { - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - w.WriteHeader(http.StatusInternalServerError) - _, _ = w.Write([]byte("internal error")) - })) - defer server.Close() - - transport := NewSyncTransport(TransportOptions{ - Dsn: "http://key@" + server.URL[7:] + "/123", - }) + t.Run("delivery errors", func(t *testing.T) { + tests := []struct { + name string + roundTrip roundTripperFunc + reason report.DiscardReason + }{ + {"server error", func(*http.Request) (*http.Response, error) { + return &http.Response{StatusCode: http.StatusInternalServerError, Body: io.NopCloser(strings.NewReader("internal error"))}, nil + }, report.ReasonSendError}, + {"network error", func(*http.Request) (*http.Response, error) { + return nil, errors.New("connection refused") + }, report.ReasonNetworkError}, + } + for _, tt := range tests { + recorder := report.NewAggregator() + transport := NewSyncTransport(TransportOptions{ + Dsn: "https://key@sentry.io/123", Recorder: recorder, HTTPTransport: tt.roundTrip, + }) - err := transport.SendEnvelope(testEnvelope(protocol.EnvelopeItemTypeEvent)) - if err != nil { - t.Errorf("server error should not return error, got %v", err) + // The transport records losses of accepted envelopes itself. + if err := transport.SendEnvelope(testEnvelope(protocol.EnvelopeItemTypeEvent)); err != nil { + t.Errorf("%s: should not return error, got %v", tt.name, err) + } + want := map[report.OutcomeKey]int64{{Reason: tt.reason, Category: ratelimit.CategoryError}: 1} + if outcomes := takeOutcomes(recorder); fmt.Sprint(outcomes) != fmt.Sprint(want) { + t.Errorf("%s: got outcomes %v, want %v", tt.name, outcomes, want) + } } }) }