Skip to content

Navigation Menu

Sign in
Sign up

feat(core): Make transport channel capacity configurable #1311

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
szokeasaurusrex wants to merge 1 commit into master
base: master
Choose a base branch
Loading
from szokeasaurusrex/transport-channel-capacity
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
View file Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

### Features

- Added a `transport_channel_capacity` option to [`ClientOptions`](https://docs.rs/sentry-core/0.49.3/sentry_core/struct.ClientOptions.html), making the transport's channel capacity configurable. When unset, each transport uses its own default (currently `30`), which is subject to change ([#1311](https://github.com/getsentry/sentry-rust/issues/1311)).
- The Tower integration's [`SentryHttpLayer`](https://docs.rs/sentry-tower/0.49.3/sentry_tower/struct.SentryHttpLayer.html) now records the [`http.response.status_code`](https://getsentry.github.io/sentry-conventions/attributes/http/) attribute on transactions ([#1253](https://github.com/getsentry/sentry-rust/pull/1253)).

### Deprecations
Expand Down
2 changes: 2 additions & 0 deletions sentry-core/src/client/mod.rs
View file Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -631,6 +631,7 @@ fn build_envelope_sender(client_options: &ClientOptions) -> EnvelopeSender {
http_proxy,
https_proxy,
accept_invalid_certs,
transport_channel_capacity,
..
} = client_options;

Expand All @@ -643,6 +644,7 @@ fn build_envelope_sender(client_options: &ClientOptions) -> EnvelopeSender {
https_proxy: https_proxy.clone(),
accept_invalid_certs: *accept_invalid_certs,
client_report_recorder,
transport_channel_capacity: *transport_channel_capacity,
};

transport_factory.create_transport_with_options(options)
Expand Down
60 changes: 60 additions & 0 deletions sentry-core/src/clientoptions.rs
View file Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
use std::borrow::Cow;
use std::fmt;
use std::num::NonZeroUsize;
use std::sync::Arc;
use std::time::Duration;

Expand Down Expand Up @@ -245,6 +246,19 @@ pub struct ClientOptions {
///
/// See [`max_request_body_size`](method@ClientOptions::max_request_body_size) for details.
pub max_request_body_size: MaxRequestBodySize,
/// The maximum number of commands the transport channel can queue.
///
/// The channel primarily carries envelopes, which are sent to Sentry on a background thread.
/// If the channel is full — for example, in high-throughput scenarios — new envelopes are
/// dropped and recorded as queue-overflow client reports, so increasing this value trades
/// memory usage for reliability. Control commands, such as flushing and shutdown, also count
/// against this capacity.
///
/// If left unset, each transport uses its own default. The current default is `30` for all
/// built-in transports, but this is subject to change.
///
/// See [`transport_channel_capacity`](method@ClientOptions::transport_channel_capacity).
pub transport_channel_capacity: Option<NonZeroUsize>,
/// Deprecated. Setting this to `false` only disables automatic log capture by the
/// log-capturing integrations (`log` and `tracing` with the `logs` feature); it does not
/// disable logs captured manually via [`Hub::capture_log`](crate::Hub::capture_log) and the
Expand Down Expand Up @@ -680,6 +694,27 @@ impl ClientOptions {
}
}

/// Sets the
/// [transport channel capacity](field@ClientOptions::transport_channel_capacity).
///
/// The smallest usable channel capacity is `1`. If `0` is passed, the capacity is clamped
/// to `1` and a debug message is emitted.
#[inline]
pub fn transport_channel_capacity(self, transport_channel_capacity: usize) -> Self {
#[cfg_attr(not(feature = "client"), expect(clippy::unnecessary_lazy_evaluations))]
let transport_channel_capacity = NonZeroUsize::new(transport_channel_capacity)
.unwrap_or_else(|| {
#[cfg(feature = "client")]
sentry_debug!("cannot set transport channel capacity to 0; clamping to 1");
NonZeroUsize::MIN
})
.into();
Self {
transport_channel_capacity,
..self
}
}

/// Deprecated. Setting [`enable_logs`](field@ClientOptions::enable_logs) to `false` only
/// disables automatic log capture by the log-capturing integrations (`log` and `tracing`
/// with the `logs` feature); it does not disable logs captured manually via
Expand Down Expand Up @@ -825,6 +860,10 @@ impl fmt::Debug for ClientOptions {
.field("http_proxy", &self.http_proxy)
.field("https_proxy", &self.https_proxy)
.field("shutdown_timeout", &self.shutdown_timeout)
.field(
"transport_channel_capacity",
&self.transport_channel_capacity,
)
.field("accept_invalid_certs", &self.accept_invalid_certs)
.field("auto_session_tracking", &self.auto_session_tracking)
.field("session_mode", &self.session_mode)
Expand Down Expand Up @@ -877,6 +916,7 @@ impl Default for ClientOptions {
session_mode: SessionMode::Application,
user_agent: Cow::Borrowed(USER_AGENT),
max_request_body_size: MaxRequestBodySize::Medium,
transport_channel_capacity: None,
#[expect(deprecated, reason = "still need to set deprecated fields")]
enable_logs: true,
before_send_log: None,
Expand All @@ -902,3 +942,23 @@ impl<T: IntoDsn> From<T> for ClientOptions {
}
}
}

#[cfg(test)]
mod tests {
use super::*;

#[test]
fn transport_channel_capacity_stores_value() {
let options = ClientOptions::new().transport_channel_capacity(42);
assert_eq!(
options.transport_channel_capacity,
Some(NonZeroUsize::new(42).unwrap())
);
}

#[test]
fn transport_channel_capacity_clamps_zero() {
let options = ClientOptions::new().transport_channel_capacity(0);
assert_eq!(options.transport_channel_capacity, Some(NonZeroUsize::MIN));
}
}
13 changes: 13 additions & 0 deletions sentry-core/src/transport/options.rs
View file Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
//! Includes the [`TransportOptions`] struct.

use std::borrow::Cow;
use std::num::NonZeroUsize;

use sentry_types::Dsn;

Expand All @@ -26,6 +27,14 @@ pub struct TransportOptions {
/// A handle for recording lost Sentry data.
#[cfg(feature = "client")]
pub client_report_recorder: ClientReportRecorder,
/// The maximum number of commands the transport channel can queue.
///
/// The channel primarily carries envelopes; control commands, such as flushing and shutdown,
/// also count against this capacity.
///
/// If `None`, the transport uses its own default. The current default is `30` for all
/// built-in transports, but this is subject to change.
pub transport_channel_capacity: Option<NonZeroUsize>,
}

impl TransportOptions {
Expand All @@ -43,6 +52,7 @@ impl TransportOptions {
https_proxy,
accept_invalid_certs,
user_agent,
transport_channel_capacity,
..
} = options;

Expand All @@ -54,6 +64,7 @@ impl TransportOptions {
accept_invalid_certs: *accept_invalid_certs,
#[cfg(feature = "client")]
client_report_recorder: ClientReportRecorder::new_no_op(),
transport_channel_capacity: *transport_channel_capacity,
})
}

Expand All @@ -73,6 +84,7 @@ impl TransportOptions {
accept_invalid_certs,
#[cfg(feature = "client")]
client_report_recorder: _,
transport_channel_capacity,
} = self;

let dsn = Some(dsn);
Expand All @@ -83,6 +95,7 @@ impl TransportOptions {
http_proxy,
https_proxy,
accept_invalid_certs,
transport_channel_capacity,
..Default::default()
}
}
Expand Down
4 changes: 3 additions & 1 deletion sentry/src/transports/curl.rs
View file Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ use sentry_core::TransportOptions;

use super::{
thread::{TransportThread, TransportThreadOptions},
RateLimiter, HTTP_PAYLOAD_TOO_LARGE, HTTP_PAYLOAD_TOO_LARGE_MESSAGE,
RateLimiter, DEFAULT_CHANNEL_CAPACITY, HTTP_PAYLOAD_TOO_LARGE, HTTP_PAYLOAD_TOO_LARGE_MESSAGE,
};

use crate::{sentry_debug, types::Scheme, ClientOptions, Envelope, Transport};
Expand Down Expand Up @@ -83,6 +83,7 @@ impl CurlHttpTransport {
https_proxy,
accept_invalid_certs,
client_report_recorder,
transport_channel_capacity,
..
},
client,
Expand Down Expand Up @@ -222,6 +223,7 @@ impl CurlHttpTransport {

let thread = TransportThreadOptions::new(send_fn)
.with_client_report_recorder(client_report_recorder)
.with_capacity(transport_channel_capacity.unwrap_or(DEFAULT_CHANNEL_CAPACITY))
.spawn_thread();
Self { thread }
}
Expand Down
6 changes: 6 additions & 0 deletions sentry/src/transports/mod.rs
View file Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@
use sentry_core::TransportOptions;

use crate::{Transport, TransportFactory};
#[cfg(all(sentry_any_http_transport, not(sentry_embedded_svc_http)))]
use std::num::NonZeroUsize;
use std::sync::Arc;

#[cfg(feature = "httpdate")]
Expand Down Expand Up @@ -54,6 +56,10 @@ pub(crate) const HTTP_PAYLOAD_TOO_LARGE: u16 = 413;
pub(crate) const HTTP_PAYLOAD_TOO_LARGE_MESSAGE: &str =
"Envelope was discarded due to size limits (HTTP 413).";

/// The default transport channel capacity.
#[cfg(all(sentry_any_http_transport, not(sentry_embedded_svc_http)))]
pub(crate) const DEFAULT_CHANNEL_CAPACITY: NonZeroUsize = NonZeroUsize::new(30).unwrap();

#[cfg(feature = "reqwest")]
type DefaultTransport = ReqwestHttpTransport;

Expand Down
4 changes: 3 additions & 1 deletion sentry/src/transports/reqwest.rs
View file Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ use sentry_core::TransportOptions;

use super::{
tokio_thread::{TransportThread, TransportThreadOptions},
RateLimiter, HTTP_PAYLOAD_TOO_LARGE, HTTP_PAYLOAD_TOO_LARGE_MESSAGE,
RateLimiter, DEFAULT_CHANNEL_CAPACITY, HTTP_PAYLOAD_TOO_LARGE, HTTP_PAYLOAD_TOO_LARGE_MESSAGE,
};

use crate::{sentry_debug, ClientOptions, Envelope, Transport};
Expand Down Expand Up @@ -84,6 +84,7 @@ impl ReqwestHttpTransport {
https_proxy,
accept_invalid_certs,
client_report_recorder,
transport_channel_capacity,
..
},
client,
Expand Down Expand Up @@ -192,6 +193,7 @@ impl ReqwestHttpTransport {

let thread = TransportThreadOptions::new(send_fn)
.with_client_report_recorder(client_report_recorder)
.with_capacity(transport_channel_capacity.unwrap_or(DEFAULT_CHANNEL_CAPACITY))
.spawn_thread();
Self { thread }
}
Expand Down
54 changes: 53 additions & 1 deletion sentry/src/transports/thread.rs
View file Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
use std::num::NonZeroUsize;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{sync_channel, SyncSender, TrySendError};
use std::sync::Arc;
Expand All @@ -7,6 +8,7 @@ use std::time::Duration;
use sentry_core::client_report::{Reason as ClientReportReason, Recorder as ClientReportRecorder};

use super::ratelimit::{RateLimiter, RateLimitingCategory};
use super::DEFAULT_CHANNEL_CAPACITY;
#[cfg(doc)]
use super::{StdTransportThread, StdTransportThreadOptions}; // so we can use pub re-exports in docs
use crate::{sentry_debug, Envelope};
Expand Down Expand Up @@ -35,6 +37,8 @@ pub struct TransportThread {
pub struct TransportThreadOptions<F> {
send_fn: F,
client_report_recorder: ClientReportRecorder,
/// The transport channel capacity. Defaults to [`DEFAULT_CHANNEL_CAPACITY`].
channel_capacity: NonZeroUsize,
}

impl<F> TransportThreadOptions<F> {
Expand All @@ -43,6 +47,7 @@ impl<F> TransportThreadOptions<F> {
Self {
send_fn,
client_report_recorder: Default::default(),
channel_capacity: DEFAULT_CHANNEL_CAPACITY,
}
}

Expand All @@ -53,6 +58,31 @@ impl<F> TransportThreadOptions<F> {
..self
}
}

/// Sets the transport channel capacity to the provided value.
pub(super) fn with_capacity(self, channel_capacity: NonZeroUsize) -> Self {
Self {
channel_capacity,
..self
}
}

/// Sets the transport channel capacity to the provided value.
///
/// The smallest usable channel capacity is `1`; if `0` is passed to this function, we clamp
/// the capacity to `1`.
pub fn with_channel_capacity(self, channel_capacity: usize) -> Self {
let channel_capacity = channel_capacity.try_into().unwrap_or_else(|_| {
sentry_debug!(
"Cannot initialize transport with channel capacity of 0, using channel capacity
of {} instead.",
NonZeroUsize::MIN
);
NonZeroUsize::MIN
});

self.with_capacity(channel_capacity)
}
}

impl<F> TransportThreadOptions<F>
Expand Down Expand Up @@ -85,8 +115,9 @@ impl TransportThread {
let TransportThreadOptions {
send_fn: mut send,
client_report_recorder,
channel_capacity,
} = options;
let (sender, receiver) = sync_channel(30);
let (sender, receiver) = sync_channel(channel_capacity.into());
let shutdown = Arc::new(AtomicBool::new(false));
let shutdown_worker = shutdown.clone();
let handle_client_report_recorder = client_report_recorder.clone();
Expand Down Expand Up @@ -182,3 +213,24 @@ impl Drop for TransportThread {
}
}
}

#[cfg(test)]
mod tests {
use super::*;

fn noop_options() -> TransportThreadOptions<impl FnMut(Envelope, &mut RateLimiter)> {
TransportThreadOptions::new(|_: Envelope, _: &mut RateLimiter| {})
}

#[test]
fn with_channel_capacity_stores_capacity() {
let options = noop_options().with_channel_capacity(42);
assert_eq!(options.channel_capacity.get(), 42);
}

#[test]
fn with_channel_capacity_clamps_zero() {
let options = noop_options().with_channel_capacity(0);
assert_eq!(options.channel_capacity.get(), 1);
}
}
Loading
Loading

AltStyle によって変換されたページ (->オリジナル) /