Skip to main content

relay_config/
config.rs

1use std::collections::{BTreeMap, HashMap};
2use std::error::Error;
3use std::io::Write;
4use std::net::{IpAddr, SocketAddr};
5use std::num::{NonZeroU8, NonZeroU16};
6use std::path::{Path, PathBuf};
7use std::str::FromStr;
8use std::sync::{Arc, Mutex, PoisonError};
9use std::time::Duration;
10use std::{env, fmt, fs, io};
11
12use anyhow::Context;
13use arc_swap::ArcSwap;
14use relay_auth::{PublicKey, RelayId, SecretKey, generate_key_pair, generate_relay_id};
15use relay_common::Dsn;
16use relay_kafka::{
17    ConfigError as KafkaConfigError, KafkaConfigParam, KafkaTopic, KafkaTopicConfig,
18    TopicAssignments,
19};
20use relay_metrics::MetricNamespace;
21use serde::de::{DeserializeOwned, Unexpected, Visitor};
22use serde::{Deserialize, Deserializer, Serialize, Serializer};
23use uuid::Uuid;
24
25use crate::aggregator::{AggregatorServiceConfig, ScopedAggregatorConfig};
26use crate::byte_size::ByteSize;
27use crate::upstream::UpstreamDescriptor;
28use crate::{RedisConfig, RedisConfigs, RedisConfigsRef, build_redis_configs};
29
30const DEFAULT_NETWORK_OUTAGE_GRACE_PERIOD: u64 = 10;
31
32static CONFIG_YAML_HEADER: &str = r###"# Please see the relevant documentation.
33# Performance tuning: https://docs.sentry.io/product/relay/operating-guidelines/
34# All config options: https://docs.sentry.io/product/relay/options/
35"###;
36
37/// Indicates config related errors.
38#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
39#[non_exhaustive]
40pub enum ConfigErrorKind {
41    /// Failed to open the file.
42    CouldNotOpenFile,
43    /// Failed to save a file.
44    CouldNotWriteFile,
45    /// Parsing YAML failed.
46    BadYaml,
47    /// Parsing JSON failed.
48    BadJson,
49    /// Invalid config value
50    InvalidValue,
51    /// The user attempted to run Relay with processing enabled, but uses a binary that was
52    /// compiled without the processing feature.
53    ProcessingNotAvailable,
54}
55
56impl fmt::Display for ConfigErrorKind {
57    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
58        match self {
59            Self::CouldNotOpenFile => write!(f, "could not open config file"),
60            Self::CouldNotWriteFile => write!(f, "could not write config file"),
61            Self::BadYaml => write!(f, "could not parse yaml config file"),
62            Self::BadJson => write!(f, "could not parse json config file"),
63            Self::InvalidValue => write!(f, "invalid config value"),
64            Self::ProcessingNotAvailable => write!(
65                f,
66                "was not compiled with processing, cannot enable processing"
67            ),
68        }
69    }
70}
71
72/// Defines the source of a config error
73#[derive(Debug, Default)]
74enum ConfigErrorSource {
75    /// An error occurring independently.
76    #[default]
77    None,
78    /// An error originating from a configuration file.
79    File(PathBuf),
80    /// An error originating in a field override (an env var, or a CLI parameter).
81    FieldOverride(String),
82}
83
84impl fmt::Display for ConfigErrorSource {
85    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
86        match self {
87            ConfigErrorSource::None => Ok(()),
88            ConfigErrorSource::File(file_name) => {
89                write!(f, " (file {})", file_name.display())
90            }
91            ConfigErrorSource::FieldOverride(name) => write!(f, " (field {name})"),
92        }
93    }
94}
95
96/// Indicates config related errors.
97#[derive(Debug)]
98pub struct ConfigError {
99    source: ConfigErrorSource,
100    kind: ConfigErrorKind,
101}
102
103impl ConfigError {
104    #[inline]
105    fn new(kind: ConfigErrorKind) -> Self {
106        Self {
107            source: ConfigErrorSource::None,
108            kind,
109        }
110    }
111
112    #[inline]
113    fn field(field: &'static str) -> Self {
114        Self {
115            source: ConfigErrorSource::FieldOverride(field.to_owned()),
116            kind: ConfigErrorKind::InvalidValue,
117        }
118    }
119
120    #[inline]
121    fn file(kind: ConfigErrorKind, p: impl AsRef<Path>) -> Self {
122        Self {
123            source: ConfigErrorSource::File(p.as_ref().to_path_buf()),
124            kind,
125        }
126    }
127
128    /// Returns the error kind of the error.
129    pub fn kind(&self) -> ConfigErrorKind {
130        self.kind
131    }
132}
133
134impl fmt::Display for ConfigError {
135    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
136        write!(f, "{}{}", self.kind(), self.source)
137    }
138}
139
140impl Error for ConfigError {}
141
142enum ConfigFormat {
143    Yaml,
144    Json,
145}
146
147impl ConfigFormat {
148    pub fn extension(&self) -> &'static str {
149        match self {
150            ConfigFormat::Yaml => "yml",
151            ConfigFormat::Json => "json",
152        }
153    }
154}
155
156trait ConfigObject: DeserializeOwned + Serialize {
157    /// The format in which to serialize this configuration.
158    fn format() -> ConfigFormat;
159
160    /// The basename of the config file.
161    fn name() -> &'static str;
162
163    /// The full filename of the config file, including the file extension.
164    fn path(base: &Path) -> PathBuf {
165        base.join(format!("{}.{}", Self::name(), Self::format().extension()))
166    }
167
168    /// Loads the config file from a file within the given directory location.
169    fn load(base: &Path) -> anyhow::Result<Self> {
170        let path = Self::path(base);
171
172        let f = fs::File::open(&path)
173            .with_context(|| ConfigError::file(ConfigErrorKind::CouldNotOpenFile, &path))?;
174        let f = io::BufReader::new(f);
175
176        let mut source = {
177            let file = serde_vars::FileSource::default()
178                .with_variable_prefix("${file:")
179                .with_variable_suffix("}")
180                .with_base_path(base);
181            let env = serde_vars::EnvSource::default()
182                .with_variable_prefix("${")
183                .with_variable_suffix("}");
184            (file, env)
185        };
186        match Self::format() {
187            ConfigFormat::Yaml => {
188                serde_vars::deserialize(serde_yaml::Deserializer::from_reader(f), &mut source)
189                    .with_context(|| ConfigError::file(ConfigErrorKind::BadYaml, &path))
190            }
191            ConfigFormat::Json => {
192                serde_vars::deserialize(&mut serde_json::Deserializer::from_reader(f), &mut source)
193                    .with_context(|| ConfigError::file(ConfigErrorKind::BadJson, &path))
194            }
195        }
196    }
197
198    /// Writes the configuration to a file within the given directory location.
199    fn save(&self, base: &Path) -> anyhow::Result<()> {
200        let path = Self::path(base);
201        let mut options = fs::OpenOptions::new();
202        options.write(true).truncate(true).create(true);
203
204        // Remove all non-user permissions for the newly created file
205        #[cfg(unix)]
206        {
207            use std::os::unix::fs::OpenOptionsExt;
208            options.mode(0o600);
209        }
210
211        let mut f = options
212            .open(&path)
213            .with_context(|| ConfigError::file(ConfigErrorKind::CouldNotWriteFile, &path))?;
214
215        match Self::format() {
216            ConfigFormat::Yaml => {
217                f.write_all(CONFIG_YAML_HEADER.as_bytes())?;
218                serde_yaml::to_writer(&mut f, self)
219                    .with_context(|| ConfigError::file(ConfigErrorKind::CouldNotWriteFile, &path))?
220            }
221            ConfigFormat::Json => serde_json::to_writer_pretty(&mut f, self)
222                .with_context(|| ConfigError::file(ConfigErrorKind::CouldNotWriteFile, &path))?,
223        }
224
225        f.write_all(b"\n").ok();
226
227        Ok(())
228    }
229}
230
231/// Structure used to hold information about configuration overrides via
232/// CLI parameters or environment variables
233#[derive(Debug, Default, Clone)]
234pub struct OverridableConfig {
235    /// The operation mode of this relay.
236    pub mode: Option<String>,
237    /// The instance type of this relay.
238    pub instance: Option<String>,
239    /// The log level of this relay.
240    pub log_level: Option<String>,
241    /// The log format of this relay.
242    pub log_format: Option<String>,
243    /// The upstream relay or sentry instance.
244    pub upstream: Option<String>,
245    /// Alternate upstream provided through a Sentry DSN. Key and project will be ignored.
246    pub upstream_dsn: Option<String>,
247    /// The host the relay should bind to (network interface).
248    pub host: Option<String>,
249    /// The port to bind for the unencrypted relay HTTP server.
250    pub port: Option<String>,
251    /// "true" if processing is enabled "false" otherwise
252    pub processing: Option<String>,
253    /// the kafka bootstrap.servers configuration string
254    pub kafka_url: Option<String>,
255    /// the redis server url
256    pub redis_url: Option<String>,
257    /// The globally unique ID of the relay.
258    pub id: Option<String>,
259    /// The secret key of the relay
260    pub secret_key: Option<String>,
261    /// The public key of the relay
262    pub public_key: Option<String>,
263    /// Outcome source
264    pub outcome_source: Option<String>,
265    /// shutdown timeout
266    pub shutdown_timeout: Option<String>,
267    /// Server name reported in the Sentry SDK.
268    pub server_name: Option<String>,
269}
270
271/// The relay credentials
272#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
273pub struct Credentials {
274    /// The secret key of the relay
275    pub secret_key: SecretKey,
276    /// The public key of the relay
277    pub public_key: PublicKey,
278    /// The globally unique ID of the relay.
279    pub id: RelayId,
280}
281
282impl Credentials {
283    /// Generates new random credentials.
284    pub fn generate() -> Self {
285        relay_log::info!("generating new relay credentials");
286        let (secret_key, public_key) = generate_key_pair();
287        Self {
288            secret_key,
289            public_key,
290            id: generate_relay_id(),
291        }
292    }
293
294    /// Serializes this configuration to JSON.
295    pub fn to_json_string(&self) -> anyhow::Result<String> {
296        serde_json::to_string(self)
297            .with_context(|| ConfigError::new(ConfigErrorKind::CouldNotWriteFile))
298    }
299}
300
301impl ConfigObject for Credentials {
302    fn format() -> ConfigFormat {
303        ConfigFormat::Json
304    }
305    fn name() -> &'static str {
306        "credentials"
307    }
308}
309
310/// Information on a downstream Relay.
311#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
312#[serde(rename_all = "camelCase")]
313pub struct RelayInfo {
314    /// The public key that this Relay uses to authenticate and sign requests.
315    pub public_key: PublicKey,
316
317    /// Marks an internal relay that has privileged access to more project configuration.
318    #[serde(default)]
319    pub internal: bool,
320}
321
322impl RelayInfo {
323    /// Creates a new RelayInfo
324    pub fn new(public_key: PublicKey) -> Self {
325        Self {
326            public_key,
327            internal: false,
328        }
329    }
330}
331
332/// The operation mode of a relay.
333#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
334#[serde(rename_all = "camelCase")]
335pub enum RelayMode {
336    /// This relay acts as a proxy for all requests and events.
337    ///
338    /// Events are normalized and rate limits from the upstream are enforced, but the relay will not
339    /// fetch project configurations from the upstream or perform PII stripping. All events are
340    /// accepted unless overridden on the file system.
341    Proxy,
342
343    /// Project configurations are managed by the upstream.
344    ///
345    /// Project configurations are always fetched from the upstream, unless they are statically
346    /// overridden in the file system. This relay must be allowed in the upstream Sentry. This is
347    /// only possible, if the upstream is Sentry directly, or another managed Relay.
348    Managed,
349}
350
351impl<'de> Deserialize<'de> for RelayMode {
352    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
353    where
354        D: Deserializer<'de>,
355    {
356        let s = String::deserialize(deserializer)?;
357        match s.as_str() {
358            "proxy" => Ok(RelayMode::Proxy),
359            "managed" => Ok(RelayMode::Managed),
360            "static" => Err(serde::de::Error::custom(
361                "Relay mode 'static' has been removed. Please use 'managed' or 'proxy' instead.",
362            )),
363            other => Err(serde::de::Error::unknown_variant(
364                other,
365                &["proxy", "managed"],
366            )),
367        }
368    }
369}
370
371impl fmt::Display for RelayMode {
372    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
373        match self {
374            RelayMode::Proxy => write!(f, "proxy"),
375            RelayMode::Managed => write!(f, "managed"),
376        }
377    }
378}
379
380/// The instance type of Relay.
381#[derive(Clone, Copy, Debug, Eq, PartialEq, Deserialize, Serialize)]
382#[serde(rename_all = "camelCase")]
383pub enum RelayInstance {
384    /// This Relay is run as a default instance.
385    Default,
386
387    /// This Relay is run as a canary instance where experiments can be run.
388    Canary,
389}
390
391impl RelayInstance {
392    /// Returns `true` if the [`RelayInstance`] is of type [`RelayInstance::Canary`].
393    pub fn is_canary(&self) -> bool {
394        matches!(self, RelayInstance::Canary)
395    }
396}
397
398impl fmt::Display for RelayInstance {
399    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
400        match self {
401            RelayInstance::Default => write!(f, "default"),
402            RelayInstance::Canary => write!(f, "canary"),
403        }
404    }
405}
406
407impl FromStr for RelayInstance {
408    type Err = fmt::Error;
409
410    fn from_str(s: &str) -> Result<Self, Self::Err> {
411        match s {
412            "canary" => Ok(RelayInstance::Canary),
413            _ => Ok(RelayInstance::Default),
414        }
415    }
416}
417
418/// Error returned when parsing an invalid [`RelayMode`].
419#[derive(Clone, Copy, Debug, Eq, PartialEq)]
420pub struct ParseRelayModeError;
421
422impl fmt::Display for ParseRelayModeError {
423    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
424        write!(f, "Relay mode must be one of: managed or proxy")
425    }
426}
427
428impl Error for ParseRelayModeError {}
429
430impl FromStr for RelayMode {
431    type Err = ParseRelayModeError;
432
433    fn from_str(s: &str) -> Result<Self, Self::Err> {
434        match s {
435            "proxy" => Ok(RelayMode::Proxy),
436            "managed" => Ok(RelayMode::Managed),
437            _ => Err(ParseRelayModeError),
438        }
439    }
440}
441
442/// Returns `true` if this value is equal to `Default::default()`.
443fn is_default<T: Default + PartialEq>(t: &T) -> bool {
444    *t == T::default()
445}
446
447/// Checks if we are running in docker.
448fn is_docker() -> bool {
449    if fs::metadata("/.dockerenv").is_ok() {
450        return true;
451    }
452
453    fs::read_to_string("/proc/self/cgroup").is_ok_and(|s| s.contains("/docker"))
454}
455
456/// Default value for the "bind" configuration.
457fn default_host() -> IpAddr {
458    if is_docker() {
459        // Docker images rely on this service being exposed
460        "0.0.0.0".parse().unwrap()
461    } else {
462        "127.0.0.1".parse().unwrap()
463    }
464}
465
466/// Controls responses from the readiness health check endpoint based on authentication.
467///
468/// Independent of the the readiness condition, shutdown always switches Relay into unready state.
469#[derive(Clone, Copy, Debug, Eq, PartialEq, Deserialize, Serialize)]
470#[serde(rename_all = "lowercase")]
471#[derive(Default)]
472pub enum ReadinessCondition {
473    /// (default) Relay is ready when authenticated and connected to the upstream.
474    ///
475    /// Before authentication has succeeded and during network outages, Relay responds as not ready.
476    /// Relay reauthenticates based on the `http.auth_interval` parameter. During reauthentication,
477    /// Relay remains ready until authentication fails.
478    ///
479    /// Authentication is only required for Relays in managed mode. Other Relays will only check for
480    /// network outages.
481    #[default]
482    Authenticated,
483    /// Relay reports readiness regardless of the authentication and networking state.
484    Always,
485}
486
487/// Relay specific configuration values.
488#[derive(Serialize, Deserialize, Debug, Clone)]
489#[serde(default)]
490pub struct Relay {
491    /// The operation mode of this Relay.
492    pub mode: RelayMode,
493    /// The instance type of this Relay.
494    pub instance: RelayInstance,
495    /// The upstream Relay or Sentry instance.
496    pub upstream: UpstreamDescriptor,
497    /// The upstream advertised to downstream Relay instances.
498    ///
499    /// This value will be advertised to downstream Relays as the upstream to use when forwarding
500    /// data. It can be used for traffic routing and balancing, it must not redirect to a different
501    /// Sentry instance.
502    ///
503    /// Downstream Relays will treat the advertised upstream as the same logical component as this instance
504    /// and re-use already established authentication keys.
505    pub advertised_upstream: Option<UpstreamDescriptor>,
506    /// The host the relay should bind to (network interface).
507    pub host: IpAddr,
508    /// The port to bind for the unencrypted relay HTTP server.
509    pub port: u16,
510    /// The host the relay should bind to (network interface) for internally exposed APIs, like
511    /// health checks.
512    ///
513    /// If not configured, internal routes are exposed on the main HTTP server.
514    ///
515    /// Note: configuring the internal http server on an address which overlaps with the main
516    /// server (e.g. main on `0.0.0.0:3000` and internal on `127.0.0.1:3000`) is a misconfiguration
517    /// resulting in approximately half of the requests sent to `127.0.0.1:3000` to fail, as the handling
518    /// http server is chosen by the operating system 'at random'.
519    ///
520    /// As a best practice you should always choose different ports to avoid this issue.
521    ///
522    /// Defaults to [`Self::host`].
523    pub internal_host: Option<IpAddr>,
524    /// The port to bind for internally exposed APIs.
525    ///
526    /// Defaults to [`Self::port`].
527    pub internal_port: Option<u16>,
528    /// Optional port to bind for the encrypted relay HTTPS server.
529    #[serde(skip_serializing)]
530    pub tls_port: Option<u16>,
531    /// The path to the identity (DER-encoded PKCS12) to use for TLS.
532    #[serde(skip_serializing)]
533    pub tls_identity_path: Option<PathBuf>,
534    /// Password for the PKCS12 archive.
535    #[serde(skip_serializing)]
536    pub tls_identity_password: Option<String>,
537    /// Always override project IDs from the URL and DSN with the identifier used at the upstream.
538    ///
539    /// Enable this setting for Relays used to redirect traffic to a migrated Sentry instance.
540    /// Validation of project identifiers can be safely skipped in these cases.
541    #[serde(skip_serializing_if = "is_default")]
542    pub override_project_ids: bool,
543    /// Interval in seconds for Relay to check if its configuration changed.
544    ///
545    /// If configured Relay will periodically check its configuration for changes
546    /// and hot reload it.
547    ///
548    /// Hot reloading is only supported for a limited set of values.
549    ///
550    /// Defaults to `None` / off.
551    pub config_reload_interval: Option<u64>,
552}
553
554impl Default for Relay {
555    fn default() -> Self {
556        Relay {
557            mode: RelayMode::Managed,
558            instance: RelayInstance::Default,
559            upstream: "https://sentry.io/".parse().unwrap(),
560            advertised_upstream: None,
561            host: default_host(),
562            port: 3000,
563            internal_host: None,
564            internal_port: None,
565            tls_port: None,
566            tls_identity_path: None,
567            tls_identity_password: None,
568            override_project_ids: false,
569            config_reload_interval: None,
570        }
571    }
572}
573
574/// Control the metrics.
575#[derive(Serialize, Deserialize, Debug, Clone)]
576#[serde(default)]
577pub struct Metrics {
578    /// Hostname and port of the statsd server.
579    ///
580    /// Defaults to `None`.
581    pub statsd: Option<String>,
582    /// Buffer size used for metrics sent to the statsd socket.
583    ///
584    /// Defaults to `None`.
585    pub statsd_buffer_size: Option<usize>,
586    /// Common prefix that should be added to all metrics.
587    ///
588    /// Defaults to `"sentry.relay"`.
589    pub prefix: String,
590    /// Default tags to apply to all metrics.
591    pub default_tags: BTreeMap<String, String>,
592    /// Tag name to report the hostname to for each metric. Defaults to not sending such a tag.
593    pub hostname_tag: Option<String>,
594    /// Interval for periodic metrics emitted from Relay.
595    ///
596    /// Setting it to `0` seconds disables the periodic metrics.
597    /// Defaults to 5 seconds.
598    pub periodic_secs: u64,
599}
600
601impl Default for Metrics {
602    fn default() -> Self {
603        Metrics {
604            statsd: None,
605            statsd_buffer_size: None,
606            prefix: "sentry.relay".into(),
607            default_tags: BTreeMap::new(),
608            hostname_tag: None,
609            periodic_secs: 5,
610        }
611    }
612}
613
614/// Controls various limits
615#[derive(Serialize, Deserialize, Debug, Clone)]
616#[serde(default)]
617pub struct Limits {
618    /// How many requests can be sent concurrently from Relay to the upstream before Relay starts
619    /// buffering.
620    pub max_concurrent_requests: usize,
621    /// How many queries can be sent concurrently from Relay to the upstream before Relay starts
622    /// buffering.
623    ///
624    /// The concurrency of queries is additionally constrained by `max_concurrent_requests`.
625    pub max_concurrent_queries: usize,
626    /// The maximum payload size for events.
627    pub max_event_size: ByteSize,
628    /// The maximum size for each attachment.
629    pub max_attachment_size: ByteSize,
630    /// The maximum amount of attachments in a single envelope.
631    pub max_attachment_count: usize,
632    /// The maximum combined size for all attachments in an envelope or request.
633    pub max_attachments_size: ByteSize,
634    /// The maximum size for a TUS upload request body.
635    pub max_upload_size: ByteSize,
636    /// The maximum combined size for all client reports in an envelope or request.
637    pub max_client_reports_size: ByteSize,
638    /// The maximum number of client report items per envelope.
639    pub max_client_reports_count: usize,
640    /// The maximum payload size for a monitor check-in.
641    pub max_check_in_size: ByteSize,
642    /// The maximum payload size for an entire envelopes. Individual limits still apply.
643    pub max_envelope_size: ByteSize,
644    /// The maximum combined size for all sessions in an envelope in bytes.
645    pub max_sessions_size: ByteSize,
646    /// The maximum number of session items per envelope.
647    pub max_session_count: usize,
648    /// The maximum payload size for general API requests.
649    pub max_api_payload_size: ByteSize,
650    /// The maximum payload size for file uploads and chunks.
651    pub max_api_file_upload_size: ByteSize,
652    /// The maximum payload size for chunks
653    pub max_api_chunk_upload_size: ByteSize,
654    /// The maximum payload size for a profile
655    pub max_profile_size: ByteSize,
656    /// The maximum payload size for a trace metric.
657    pub max_trace_metric_size: ByteSize,
658    /// The maximum payload size for a log.
659    pub max_log_size: ByteSize,
660    /// The maximum payload size for a span.
661    pub max_span_size: ByteSize,
662    /// The maximum amount of standalone transaction spans per envelope.
663    pub max_standalone_span_count: usize,
664    /// The maximum payload size for an item container.
665    pub max_container_size: ByteSize,
666    /// The maximum payload size for a statsd metric.
667    pub max_statsd_size: ByteSize,
668    /// The maximum payload size for metric buckets.
669    pub max_metric_buckets_size: ByteSize,
670    /// The maximum payload size for a compressed replay.
671    pub max_replay_compressed_size: ByteSize,
672    /// The maximum payload size for an uncompressed replay.
673    #[serde(alias = "max_replay_size")]
674    max_replay_uncompressed_size: ByteSize,
675    /// The maximum size for a replay recording Kafka message.
676    pub max_replay_message_size: ByteSize,
677    /// The byte size limit up to which Relay will retain
678    /// keys of invalid/removed attributes.
679    ///
680    /// This is only relevant for EAP items (spans, logs, …).
681    /// In principle, we want to record all deletions of attributes,
682    /// but we have to institute some limit to protect our infrastructure
683    /// against excessive metadata sizes.
684    ///
685    /// Defaults to 10KiB.
686    pub max_removed_attribute_key_size: ByteSize,
687    /// The maximum number of threads to spawn for CPU and web work, each.
688    ///
689    /// The total number of threads spawned will roughly be `2 * max_thread_count`. Defaults to
690    /// the number of logical CPU cores on the host.
691    pub max_thread_count: usize,
692    /// Controls the maximum concurrency of each worker thread.
693    ///
694    /// Increasing the concurrency, can lead to a better utilization of worker threads by
695    /// increasing the amount of I/O done concurrently.
696    //
697    /// Currently has no effect on defaults to `1`.
698    pub max_pool_concurrency: usize,
699    /// The maximum number of seconds a query is allowed to take across retries. Individual requests
700    /// have lower timeouts. Defaults to 30 seconds.
701    pub query_timeout: u64,
702    /// The maximum number of seconds to wait for pending envelopes after receiving a shutdown
703    /// signal.
704    pub shutdown_timeout: u64,
705    /// Server keep-alive timeout in seconds.
706    ///
707    /// By default, keep-alive is set to 5 seconds.
708    pub keepalive_timeout: u64,
709    /// Server idle timeout in seconds.
710    ///
711    /// The idle timeout limits the amount of time a connection is kept open without activity.
712    /// Setting this too short may abort connections before Relay is able to send a response.
713    ///
714    /// By default there is no idle timeout.
715    pub idle_timeout: Option<u64>,
716    /// Sets the maximum number of concurrent connections.
717    ///
718    /// Upon reaching the limit, the server will stop accepting connections.
719    ///
720    /// By default there is no limit.
721    pub max_connections: Option<usize>,
722    /// The TCP listen backlog.
723    ///
724    /// Configures the TCP listen backlog for the listening socket of Relay.
725    /// See [`man listen(2)`](https://man7.org/linux/man-pages/man2/listen.2.html)
726    /// for a more detailed description of the listen backlog.
727    ///
728    /// Defaults to `1024`, a value [google has been using for a long time](https://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git/commit/?id=19f92a030ca6d772ab44b22ee6a01378a8cb32d4).
729    pub tcp_listen_backlog: u32,
730}
731
732impl Default for Limits {
733    fn default() -> Self {
734        Limits {
735            max_concurrent_requests: 100,
736            max_concurrent_queries: 5,
737            max_event_size: ByteSize::mebibytes(1),
738            max_attachment_size: ByteSize::mebibytes(200),
739            max_attachment_count: 30,
740            max_attachments_size: ByteSize::mebibytes(200),
741            max_upload_size: ByteSize::mebibytes(1024),
742            max_client_reports_size: ByteSize::kibibytes(100),
743            max_client_reports_count: 100,
744            max_check_in_size: ByteSize::kibibytes(100),
745            max_envelope_size: ByteSize::mebibytes(200),
746            max_sessions_size: ByteSize::mebibytes(10),
747            max_session_count: 100,
748            max_api_payload_size: ByteSize::mebibytes(20),
749            max_api_file_upload_size: ByteSize::mebibytes(40),
750            max_api_chunk_upload_size: ByteSize::mebibytes(100),
751            max_profile_size: ByteSize::mebibytes(50),
752            max_trace_metric_size: ByteSize::mebibytes(1),
753            max_log_size: ByteSize::mebibytes(2),
754            max_span_size: ByteSize::mebibytes(10),
755            max_standalone_span_count: 25,
756            max_container_size: ByteSize::mebibytes(12),
757            max_statsd_size: ByteSize::mebibytes(1),
758            max_metric_buckets_size: ByteSize::mebibytes(1),
759            max_replay_compressed_size: ByteSize::mebibytes(10),
760            max_replay_uncompressed_size: ByteSize::mebibytes(100),
761            max_replay_message_size: ByteSize::mebibytes(15),
762            max_thread_count: num_cpus::get(),
763            max_pool_concurrency: 1,
764            query_timeout: 30,
765            shutdown_timeout: 10,
766            keepalive_timeout: 5,
767            idle_timeout: None,
768            max_connections: None,
769            tcp_listen_backlog: 1024,
770            max_removed_attribute_key_size: ByteSize::kibibytes(10),
771        }
772    }
773}
774
775/// Controls traffic steering.
776#[derive(Debug, Default, Deserialize, Serialize, Clone)]
777#[serde(default)]
778pub struct Routing {
779    /// Accept and forward unknown Envelope items to the upstream.
780    ///
781    /// Forwarding unknown items should be enabled in most cases to allow proxying traffic for newer
782    /// SDK versions. The upstream in Sentry makes the final decision on which items are valid. If
783    /// this is disabled, just the unknown items are removed from Envelopes, and the rest is
784    /// processed as usual.
785    ///
786    /// Defaults to `true` for all Relay modes other than processing mode. In processing mode, this
787    /// is disabled by default since the item cannot be handled.
788    pub accept_unknown_items: Option<bool>,
789}
790
791/// Http content encoding for both incoming and outgoing web requests.
792#[derive(Clone, Copy, Debug, Default, Deserialize, Serialize)]
793#[serde(rename_all = "lowercase")]
794pub enum HttpEncoding {
795    /// Identity function without no compression.
796    ///
797    /// This is the default encoding and does not require the presence of the `content-encoding`
798    /// HTTP header.
799    #[default]
800    Identity,
801    /// Compression using a [zlib](https://en.wikipedia.org/wiki/Zlib) structure with
802    /// [deflate](https://en.wikipedia.org/wiki/DEFLATE) encoding.
803    ///
804    /// These structures are defined in [RFC 1950](https://datatracker.ietf.org/doc/html/rfc1950)
805    /// and [RFC 1951](https://datatracker.ietf.org/doc/html/rfc1951).
806    Deflate,
807    /// A format using the [Lempel-Ziv coding](https://en.wikipedia.org/wiki/LZ77_and_LZ78#LZ77)
808    /// (LZ77), with a 32-bit CRC.
809    ///
810    /// This is the original format of the UNIX gzip program. The HTTP/1.1 standard also recommends
811    /// that the servers supporting this content-encoding should recognize `x-gzip` as an alias, for
812    /// compatibility purposes.
813    Gzip,
814    /// A format using the [Brotli](https://en.wikipedia.org/wiki/Brotli) algorithm.
815    Br,
816    /// A format using the [Zstd](https://en.wikipedia.org/wiki/Zstd) compression algorithm.
817    Zstd,
818}
819
820impl HttpEncoding {
821    /// Parses a [`HttpEncoding`] from its `content-encoding` header value.
822    pub fn parse(str: &str) -> Self {
823        let str = str.trim();
824        if str.eq_ignore_ascii_case("zstd") {
825            Self::Zstd
826        } else if str.eq_ignore_ascii_case("br") {
827            Self::Br
828        } else if str.eq_ignore_ascii_case("gzip") || str.eq_ignore_ascii_case("x-gzip") {
829            Self::Gzip
830        } else if str.eq_ignore_ascii_case("deflate") {
831            Self::Deflate
832        } else {
833            Self::Identity
834        }
835    }
836
837    /// Returns the value for the `content-encoding` HTTP header.
838    ///
839    /// Returns `None` for [`Identity`](Self::Identity), and `Some` for other encodings.
840    pub fn name(&self) -> Option<&'static str> {
841        match self {
842            Self::Identity => None,
843            Self::Deflate => Some("deflate"),
844            Self::Gzip => Some("gzip"),
845            Self::Br => Some("br"),
846            Self::Zstd => Some("zstd"),
847        }
848    }
849}
850
851/// Controls authentication with upstream.
852#[derive(Serialize, Deserialize, Debug, Clone)]
853#[serde(default)]
854pub struct Http {
855    /// Timeout for upstream requests in seconds.
856    ///
857    /// This timeout covers the time from sending the request until receiving response headers.
858    /// Neither the connection process and handshakes, nor reading the response body is covered in
859    /// this timeout.
860    pub timeout: u32,
861    /// Timeout for establishing connections with the upstream in seconds.
862    ///
863    /// This includes SSL handshakes. Relay reuses connections when the upstream supports connection
864    /// keep-alive. Connections are retained for a maximum 75 seconds, or 15 seconds of inactivity.
865    pub connection_timeout: u32,
866    /// Maximum interval between failed request retries in seconds.
867    pub max_retry_interval: u32,
868    /// The custom HTTP Host header to send to the upstream.
869    pub host_header: Option<String>,
870    /// The interval in seconds at which Relay attempts to reauthenticate with the upstream server.
871    ///
872    /// Re-authentication happens even when Relay is idle. If authentication fails, Relay reverts
873    /// back into startup mode and tries to establish a connection. During this time, incoming
874    /// envelopes will be buffered.
875    ///
876    /// Defaults to `600` (10 minutes).
877    pub auth_interval: Option<u64>,
878    /// The maximum time of experiencing uninterrupted network failures until Relay considers that
879    /// it has encountered a network outage in seconds.
880    ///
881    /// During a network outage relay will try to reconnect and will buffer all upstream messages
882    /// until it manages to reconnect.
883    pub outage_grace_period: u64,
884    /// The time Relay waits before retrying an upstream request, in seconds.
885    ///
886    /// This time is only used before going into a network outage mode.
887    pub retry_delay: u64,
888    /// The interval in seconds for continued failed project fetches at which Relay will error.
889    ///
890    /// A successful fetch resets this interval. Relay does nothing during long
891    /// times without emitting requests.
892    pub project_failure_interval: u64,
893    /// Content encoding to apply to upstream store requests.
894    ///
895    /// By default, Relay applies `zstd` content encoding to compress upstream requests. Compression
896    /// can be disabled to reduce CPU consumption, but at the expense of increased network traffic.
897    ///
898    /// This setting applies to all store requests of SDK data, including events, transactions,
899    /// envelopes and sessions. At the moment, this does not apply to Relay's internal queries.
900    ///
901    /// Available options are:
902    ///
903    ///  - `identity`: Disables compression.
904    ///  - `deflate`: Compression using a zlib header with deflate encoding.
905    ///  - `gzip` (default): Compression using gzip.
906    ///  - `br`: Compression using the brotli algorithm.
907    ///  - `zstd`: Compression using the zstd algorithm.
908    pub encoding: HttpEncoding,
909    /// Submit metrics globally through a shared endpoint.
910    ///
911    /// As opposed to regular envelopes which are sent to an endpoint inferred from the project's
912    /// DSN, this submits metrics to the global endpoint with Relay authentication.
913    ///
914    /// This option does not have any effect on processing mode.
915    pub global_metrics: bool,
916    /// Controls whether the forward endpoint is enabled.
917    ///
918    /// The forward endpoint forwards unknown API requests to the upstream.
919    ///
920    /// Relay instances with processing enabled are expected to support the latest API and do never
921    /// support forwarding requests to Sentry.
922    pub forward: bool,
923    /// Enables an async DNS resolver through the `hickory-dns` crate, which uses an LRU cache for
924    /// the resolved entries. This helps to limit the amount of requests made to the upstream DNS
925    /// server (important for K8s infrastructure).
926    pub dns_cache: bool,
927}
928
929impl Default for Http {
930    fn default() -> Self {
931        Http {
932            timeout: 5,
933            connection_timeout: 3,
934            max_retry_interval: 60, // 1 minute
935            host_header: None,
936            auth_interval: Some(600), // 10 minutes
937            outage_grace_period: DEFAULT_NETWORK_OUTAGE_GRACE_PERIOD,
938            retry_delay: 1,
939            project_failure_interval: 90,
940            encoding: HttpEncoding::Zstd,
941            global_metrics: false,
942            forward: true,
943            dns_cache: true,
944        }
945    }
946}
947
948/// Strategy used to assign envelopes to buffer partitions.
949#[derive(Clone, Copy, Debug, Eq, PartialEq, Default, Deserialize, Serialize)]
950#[serde(rename_all = "snake_case")]
951pub enum EnvelopeSpoolPartitioning {
952    /// Envelopes with the same project key pair land on the same partition.
953    ///
954    /// Keeps per-project state, disk files, and event ordering co-located on one partition.
955    ProjectKeyPair,
956    /// Envelopes are distributed across partitions in a round-robin fashion (default).
957    ///
958    /// This prevents "hot" partitions when a single project pair dominates traffic, but has
959    /// trade-offs:
960    /// - Per-project LIFO ordering is no longer preserved across partitions.
961    /// - Per-partition memory footprint grows since every partition sees every project.
962    #[default]
963    RoundRobin,
964}
965
966/// Persistent buffering configuration for incoming envelopes.
967#[derive(Debug, Serialize, Deserialize, Clone)]
968#[serde(default)]
969pub struct EnvelopeSpool {
970    /// The path of the SQLite database file(s) which persist the data.
971    ///
972    /// Based on the number of partitions, more database files will be created within the same path.
973    ///
974    /// If not set, the envelopes will be buffered in memory.
975    pub path: Option<PathBuf>,
976    /// The maximum size of the buffer to keep, in bytes.
977    ///
978    /// When the on-disk buffer reaches this size, new envelopes will be dropped.
979    ///
980    /// Defaults to 500MB.
981    pub max_disk_size: ByteSize,
982    /// Size of the batch of compressed envelopes that are spooled to disk at once.
983    ///
984    /// Note that this is the size after which spooling will be triggered but it does not guarantee
985    /// that exactly this size will be spooled, it can be greater or equal.
986    ///
987    /// Defaults to 10 KiB.
988    pub batch_size_bytes: ByteSize,
989    /// Maximum time between receiving the envelope and processing it.
990    ///
991    /// When envelopes spend too much time in the buffer (e.g. because their project cannot be loaded),
992    /// they are dropped.
993    ///
994    /// Defaults to 24h.
995    pub max_envelope_delay_secs: u64,
996    /// The refresh frequency in ms of how frequently disk usage is updated by querying SQLite
997    /// internal page stats.
998    ///
999    /// Defaults to 100ms.
1000    pub disk_usage_refresh_frequency_ms: u64,
1001    /// The relative memory usage above which the buffer service will stop dequeueing envelopes.
1002    ///
1003    /// Only applies when [`Self::path`] is set.
1004    ///
1005    /// This value should be lower than [`Health::max_memory_percent`] to prevent flip-flopping.
1006    ///
1007    /// Warning: This threshold can cause the buffer service to deadlock when the buffer consumes
1008    /// excessive memory (as influenced by [`Self::batch_size_bytes`]).
1009    ///
1010    /// This scenario arises when the buffer stops spooling due to reaching the
1011    /// [`Self::max_backpressure_memory_percent`] limit, but the batch threshold for spooling
1012    /// ([`Self::batch_size_bytes`]) is never reached. As a result, no data is spooled, memory usage
1013    /// continues to grow, and the system becomes deadlocked.
1014    ///
1015    /// ### Example
1016    /// Suppose the system has 1GB of available memory and is configured to spool only after
1017    /// accumulating 10GB worth of envelopes. If Relay consumes 900MB of memory, it will stop
1018    /// unspooling due to reaching the [`Self::max_backpressure_memory_percent`] threshold.
1019    ///
1020    /// However, because the buffer hasn't accumulated the 10GB needed to trigger spooling,
1021    /// no data will be offloaded. Memory usage keeps increasing until it hits the
1022    /// [`Health::max_memory_percent`] threshold, e.g., at 950MB. At this point:
1023    ///
1024    /// - No more envelopes are accepted.
1025    /// - The buffer remains stuck, as unspooling won’t resume until memory drops below 900MB which
1026    ///   will not happen.
1027    /// - A deadlock occurs, with the system unable to recover without manual intervention.
1028    ///
1029    /// Defaults to 90% (5% less than max memory).
1030    pub max_backpressure_memory_percent: f32,
1031    /// Number of partitions of the buffer.
1032    ///
1033    /// A partition is a separate instance of the buffer which has its own isolated queue, stacks
1034    /// and other resources.
1035    ///
1036    /// Defaults to 1.
1037    pub partitions: NonZeroU8,
1038    /// Strategy used to assign envelopes to buffer partitions.
1039    ///
1040    /// Defaults to partitioning by `ProjectKeyPair`, which keeps all envelopes of a given project
1041    /// pair on the same partition. See [`EnvelopeSpoolPartitioning`] for alternatives and
1042    /// trade-offs.
1043    pub partitioning: EnvelopeSpoolPartitioning,
1044    /// Whether the database defined in `path` is on an ephemeral storage disk.
1045    ///
1046    /// With `ephemeral: true`, Relay does not spool in-flight data to disk
1047    /// during graceful shutdown. Instead, it attempts to process all data before it terminates.
1048    ///
1049    /// Defaults to `false`.
1050    pub ephemeral: bool,
1051}
1052
1053impl Default for EnvelopeSpool {
1054    fn default() -> Self {
1055        Self {
1056            path: None,
1057            max_disk_size: ByteSize::mebibytes(500),
1058            batch_size_bytes: ByteSize::kibibytes(10),
1059            max_envelope_delay_secs: 24 * 60 * 60,
1060            disk_usage_refresh_frequency_ms: 100,
1061            max_backpressure_memory_percent: 0.8,
1062            partitions: NonZeroU8::new(1).unwrap(),
1063            partitioning: EnvelopeSpoolPartitioning::default(),
1064            ephemeral: false,
1065        }
1066    }
1067}
1068
1069/// Persistent buffering configuration.
1070#[derive(Debug, Serialize, Deserialize, Default, Clone)]
1071#[serde(default)]
1072pub struct Spool {
1073    /// Configuration for envelope spooling.
1074    pub envelopes: EnvelopeSpool,
1075}
1076
1077/// Controls internal caching behavior.
1078#[derive(Serialize, Deserialize, Debug, Clone)]
1079#[serde(default)]
1080pub struct Cache {
1081    /// The full project state will be requested by this Relay if set to `true`.
1082    pub project_request_full_config: bool,
1083    /// The cache timeout for project configurations in seconds.
1084    pub project_expiry: u32,
1085    /// Continue using project state this many seconds after cache expiry while a new state is
1086    /// being fetched. This is added on top of `project_expiry`.
1087    ///
1088    /// Default is 2 minutes.
1089    pub project_grace_period: u32,
1090    /// Refresh a project after the specified seconds.
1091    ///
1092    /// The time must be between expiry time and the grace period.
1093    ///
1094    /// By default there are no refreshes enabled.
1095    pub project_refresh_interval: Option<u32>,
1096    /// The cache timeout for downstream relay info (public keys) in seconds.
1097    pub relay_expiry: u32,
1098    /// Unused cache timeout for envelopes.
1099    ///
1100    /// The envelope buffer is instead controlled by `envelope_buffer_size`, which controls the
1101    /// maximum number of envelopes in the buffer. A time based configuration may be re-introduced
1102    /// at a later point.
1103    #[serde(alias = "event_expiry")]
1104    envelope_expiry: u32,
1105    /// The maximum amount of envelopes to queue before dropping them.
1106    #[serde(alias = "event_buffer_size")]
1107    envelope_buffer_size: u32,
1108    /// The cache timeout for non-existing entries.
1109    pub miss_expiry: u32,
1110    /// The buffer timeout for batched project config queries before sending them upstream in ms.
1111    pub batch_interval: u32,
1112    /// The buffer timeout for batched queries of downstream relays in ms. Defaults to 100ms.
1113    pub downstream_relays_batch_interval: u32,
1114    /// The maximum number of project configs to fetch from Sentry at once. Defaults to 500.
1115    ///
1116    /// `cache.batch_interval` controls how quickly batches are sent, this controls the batch size.
1117    pub batch_size: usize,
1118    /// Interval for watching local cache override files in seconds.
1119    pub file_interval: u32,
1120    /// Interval for fetching new global configs from the upstream, in seconds.
1121    pub global_config_fetch_interval: u32,
1122}
1123
1124impl Default for Cache {
1125    fn default() -> Self {
1126        Cache {
1127            project_request_full_config: false,
1128            project_expiry: 300,       // 5 minutes
1129            project_grace_period: 120, // 2 minutes
1130            project_refresh_interval: None,
1131            relay_expiry: 3600,   // 1 hour
1132            envelope_expiry: 600, // 10 minutes
1133            envelope_buffer_size: 1000,
1134            miss_expiry: 60,                       // 1 minute
1135            batch_interval: 100,                   // 100ms
1136            downstream_relays_batch_interval: 100, // 100ms
1137            batch_size: 500,
1138            file_interval: 10,                // 10 seconds
1139            global_config_fetch_interval: 10, // 10 seconds
1140        }
1141    }
1142}
1143
1144/// Controls Sentry-internal event processing.
1145#[derive(Serialize, Deserialize, Debug, Clone)]
1146#[serde(default)]
1147pub struct Processing {
1148    /// True if the Relay should do processing. Defaults to `false`.
1149    pub enabled: bool,
1150    /// GeoIp DB file source.
1151    pub geoip_path: Option<PathBuf>,
1152    /// Maximum future timestamp of ingested events.
1153    pub max_secs_in_future: u32,
1154    /// Maximum age of ingested sessions. Older sessions will be dropped.
1155    pub max_session_secs_in_past: u32,
1156    /// Kafka producer configurations.
1157    pub kafka_config: Vec<KafkaConfigParam>,
1158    /// Additional kafka producer configurations.
1159    ///
1160    /// The `kafka_config` is the default producer configuration used for all topics. A secondary
1161    /// kafka config can be referenced in `topics:` like this:
1162    ///
1163    /// ```yaml
1164    /// secondary_kafka_configs:
1165    ///   mycustomcluster:
1166    ///     - name: 'bootstrap.servers'
1167    ///       value: 'sentry_kafka_metrics:9093'
1168    ///
1169    /// topics:
1170    ///   transactions: ingest-transactions
1171    ///   metrics:
1172    ///     name: ingest-metrics
1173    ///     config: mycustomcluster
1174    /// ```
1175    ///
1176    /// Then metrics will be produced to an entirely different Kafka cluster.
1177    pub secondary_kafka_configs: BTreeMap<String, Vec<KafkaConfigParam>>,
1178    /// Kafka topic names.
1179    pub topics: TopicAssignments,
1180    /// Whether to validate the supplied topics by calling Kafka's metadata endpoints.
1181    pub kafka_validate_topics: bool,
1182    /// Redis hosts to connect to for storing state for rate limits.
1183    pub redis: Option<RedisConfigs>,
1184    /// Maximum chunk size of attachments for Kafka.
1185    pub attachment_chunk_size: ByteSize,
1186    /// Prefix to use when looking up project configs in Redis. Defaults to "relayconfig".
1187    pub projectconfig_cache_prefix: String,
1188    /// Maximum rate limit to report to clients.
1189    pub max_rate_limit: Option<u32>,
1190    /// Configures the quota cache ratio between `0.0` and `1.0`.
1191    ///
1192    /// The quota cache, caches the specified ratio of remaining quota in memory to reduce the
1193    /// amount of synchronizations required with Redis.
1194    ///
1195    /// The ratio is applied to the (per second) rate of the quota, not the total limit.
1196    /// For example a quota with limit 100 with a 10 second window is treated equally to a quota of
1197    /// 10 with a 1 second window.
1198    ///
1199    /// By default quota caching is disabled.
1200    pub quota_cache_ratio: Option<f32>,
1201    /// Relative amount of the total quota limit to which quota caching is applied.
1202    ///
1203    /// If exceeded, the rate limiter will no longer cache the quota and sync with Redis on every call instead.
1204    /// Lowering this value reduces the probability of incorrectly over-accepting.
1205    ///
1206    /// Must be between `0.0` and `1.0`, by default there is no limit configured.
1207    pub quota_cache_max: Option<f32>,
1208    /// Configuration for the objectstore service.
1209    #[serde(alias = "upload")]
1210    pub objectstore: ObjectstoreServiceConfig,
1211}
1212
1213impl Default for Processing {
1214    /// Constructs a disabled processing configuration.
1215    fn default() -> Self {
1216        Self {
1217            enabled: false,
1218            geoip_path: None,
1219            max_secs_in_future: 60,                  // 1 minute
1220            max_session_secs_in_past: 5 * 24 * 3600, // 5 days
1221            kafka_config: Vec::new(),
1222            secondary_kafka_configs: BTreeMap::new(),
1223            topics: TopicAssignments::default(),
1224            kafka_validate_topics: false,
1225            redis: None,
1226            attachment_chunk_size: ByteSize::mebibytes(1),
1227            projectconfig_cache_prefix: "relayconfig".to_owned(),
1228            max_rate_limit: Some(300), // 5 minutes
1229            quota_cache_ratio: None,
1230            quota_cache_max: None,
1231            objectstore: ObjectstoreServiceConfig::default(),
1232        }
1233    }
1234}
1235
1236/// Configuration for normalization in this Relay.
1237#[derive(Debug, Default, Serialize, Deserialize, Clone)]
1238#[serde(default)]
1239pub struct Normalization {
1240    /// Level of normalization for Relay to apply to incoming data.
1241    pub level: NormalizationLevel,
1242}
1243
1244/// Configuration for the level of normalization this Relay should do.
1245#[derive(Copy, Clone, Debug, Default, Serialize, Deserialize, Eq, PartialEq)]
1246#[serde(rename_all = "lowercase")]
1247pub enum NormalizationLevel {
1248    /// Runs normalization, excluding steps that break future compatibility.
1249    ///
1250    /// Processing Relays run [`NormalizationLevel::Full`] if this option is set.
1251    #[default]
1252    Default,
1253    /// Run full normalization.
1254    ///
1255    /// It includes steps that break future compatibility and should only run in
1256    /// the last layer of relays.
1257    Full,
1258}
1259
1260/// Configuration options for objectstore's auth scheme.
1261#[derive(Serialize, Deserialize, Clone)]
1262pub struct ObjectstoreAuthConfig {
1263    /// Identifier for the private key used to sign objectstore's tokens. Must correspond to a
1264    /// public key configured in objectstore.
1265    pub key_id: String,
1266
1267    /// EdDSA private key used to sign Objectstore's tokens, in PEM format.
1268    pub signing_key: String,
1269}
1270
1271impl fmt::Debug for ObjectstoreAuthConfig {
1272    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1273        f.debug_struct("ObjectstoreAuthConfig")
1274            .field("key_id", &self.key_id)
1275            .field("signing_key", &"[redacted]")
1276            .finish()
1277    }
1278}
1279
1280/// Configuration values for the objectstore service.
1281#[derive(Serialize, Deserialize, Debug, Clone)]
1282#[serde(default)]
1283pub struct ObjectstoreServiceConfig {
1284    /// The base URL for the objectstore service.
1285    ///
1286    /// This defaults to [`None`], which means that the service will be disabled,
1287    /// unless a proper configuration is provided.
1288    pub objectstore_url: Option<String>,
1289
1290    /// Maximum concurrency of uploads.
1291    pub max_concurrent_requests: usize,
1292
1293    /// Maximum size of the service input queue when `max_concurrent_requests` is saturated.
1294    ///
1295    /// The service will loadshed if this threshold is reached.
1296    pub max_backlog: usize,
1297
1298    /// Maximum duration of an attachment upload in seconds. Uploads that take longer are discarded.
1299    ///
1300    /// NOTE: This timeout applies to attachments that are already in-memory. Streaming uploads
1301    /// might take longer and are restricted independently by [`Self::stream_timeout`].
1302    pub timeout: u64,
1303
1304    /// Maximum duration of an upload stream.
1305    ///
1306    /// Streams get a larger default timeout because their duration depends on the client
1307    /// as well as the server.
1308    pub stream_timeout: u64,
1309
1310    /// Time between upload attempts.
1311    pub retry_delay: f64,
1312
1313    /// Maximum number of attempts made to upload.
1314    pub max_attempts: NonZeroU16,
1315
1316    /// Whether event attachment payloads may be sent through Kafka if objectstore upload fails.
1317    ///
1318    /// When disabled, failed event attachments are dropped with an `upload_failed`
1319    /// outcome instead of falling back to Store's Kafka attachment path.
1320    pub fallback_to_kafka: bool,
1321
1322    /// Configuration values for objectstore's auth scheme.
1323    pub auth: Option<ObjectstoreAuthConfig>,
1324}
1325
1326impl Default for ObjectstoreServiceConfig {
1327    fn default() -> Self {
1328        Self {
1329            objectstore_url: None,
1330            max_concurrent_requests: 10,
1331            max_backlog: 20,
1332            timeout: 60,
1333            stream_timeout: 5 * 60, // synced with `Upload::timeout`
1334            retry_delay: 1.0,
1335            max_attempts: NonZeroU16::new(5).unwrap(),
1336            fallback_to_kafka: true,
1337            auth: None,
1338        }
1339    }
1340}
1341
1342/// Determines how to emit outcomes.
1343/// For compatibility reasons, this can either be true, false or AsClientReports
1344#[derive(Copy, Clone, Debug, PartialEq, Eq)]
1345
1346pub enum EmitOutcomes {
1347    /// Do not emit any outcomes.
1348    None,
1349    /// Emit outcomes as client reports.
1350    AsClientReports,
1351    /// Emit outcomes as outcomes.
1352    AsOutcomes,
1353}
1354
1355impl EmitOutcomes {
1356    /// Returns true of outcomes are emitted via http, kafka, or client reports.
1357    pub fn any(&self) -> bool {
1358        !matches!(self, EmitOutcomes::None)
1359    }
1360}
1361
1362impl Serialize for EmitOutcomes {
1363    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
1364    where
1365        S: Serializer,
1366    {
1367        // For compatibility, serialize None and AsOutcomes as booleans.
1368        match self {
1369            Self::None => serializer.serialize_bool(false),
1370            Self::AsClientReports => serializer.serialize_str("as_client_reports"),
1371            Self::AsOutcomes => serializer.serialize_bool(true),
1372        }
1373    }
1374}
1375
1376struct EmitOutcomesVisitor;
1377
1378impl Visitor<'_> for EmitOutcomesVisitor {
1379    type Value = EmitOutcomes;
1380
1381    fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
1382        formatter.write_str("true, false, 'as_client_reports'")
1383    }
1384
1385    fn visit_bool<E>(self, v: bool) -> Result<Self::Value, E>
1386    where
1387        E: serde::de::Error,
1388    {
1389        Ok(if v {
1390            EmitOutcomes::AsOutcomes
1391        } else {
1392            EmitOutcomes::None
1393        })
1394    }
1395
1396    fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
1397    where
1398        E: serde::de::Error,
1399    {
1400        match v {
1401            "as_client_reports" => Ok(EmitOutcomes::AsClientReports),
1402            _ => Err(E::invalid_value(Unexpected::Str(v), &self)),
1403        }
1404    }
1405}
1406
1407impl<'de> Deserialize<'de> for EmitOutcomes {
1408    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
1409    where
1410        D: Deserializer<'de>,
1411    {
1412        deserializer.deserialize_any(EmitOutcomesVisitor)
1413    }
1414}
1415
1416/// Outcome generation specific configuration values.
1417#[derive(Serialize, Deserialize, Debug, Clone)]
1418#[serde(default)]
1419pub struct Outcomes {
1420    /// Controls whether outcomes will be emitted when processing is disabled.
1421    /// Processing relays always emit outcomes (for backwards compatibility).
1422    /// Can take the following values: false, "as_client_reports", true
1423    pub emit_outcomes: EmitOutcomes,
1424    /// The maximum number of outcomes that are batched before being sent
1425    /// via http to the upstream (only applies to non processing relays).
1426    pub batch_size: usize,
1427    /// The maximum time interval (in milliseconds) that an outcome may be batched
1428    /// via http to the upstream (only applies to non processing relays).
1429    pub batch_interval: u64,
1430    /// Defines the source string registered in the outcomes originating from
1431    /// this Relay (typically something like the region or the layer).
1432    pub source: Option<String>,
1433}
1434
1435impl Default for Outcomes {
1436    fn default() -> Self {
1437        Outcomes {
1438            emit_outcomes: EmitOutcomes::AsClientReports,
1439            batch_size: 1000,
1440            batch_interval: 500,
1441            source: None,
1442        }
1443    }
1444}
1445
1446/// Minimal version of a config for dumping out.
1447#[derive(Serialize, Deserialize, Debug, Default)]
1448pub struct MinimalConfig {
1449    /// The relay part of the config.
1450    pub relay: Relay,
1451}
1452
1453impl MinimalConfig {
1454    /// Saves the config in the given config folder as config.yml
1455    pub fn save_in_folder<P: AsRef<Path>>(&self, p: P) -> anyhow::Result<()> {
1456        let path = p.as_ref();
1457        if fs::metadata(path).is_err() {
1458            fs::create_dir_all(path)
1459                .with_context(|| ConfigError::file(ConfigErrorKind::CouldNotOpenFile, path))?;
1460        }
1461        self.save(path)
1462    }
1463}
1464
1465impl ConfigObject for MinimalConfig {
1466    fn format() -> ConfigFormat {
1467        ConfigFormat::Yaml
1468    }
1469
1470    fn name() -> &'static str {
1471        "config"
1472    }
1473}
1474
1475/// Alternative serialization of RelayInfo for config file using snake case.
1476mod config_relay_info {
1477    use serde::ser::SerializeMap;
1478
1479    use super::*;
1480
1481    // Uses snake_case as opposed to camelCase.
1482    #[derive(Debug, Serialize, Deserialize, Clone)]
1483    struct RelayInfoConfig {
1484        public_key: PublicKey,
1485        #[serde(default)]
1486        internal: bool,
1487    }
1488
1489    impl From<RelayInfoConfig> for RelayInfo {
1490        fn from(v: RelayInfoConfig) -> Self {
1491            RelayInfo {
1492                public_key: v.public_key,
1493                internal: v.internal,
1494            }
1495        }
1496    }
1497
1498    impl From<RelayInfo> for RelayInfoConfig {
1499        fn from(v: RelayInfo) -> Self {
1500            RelayInfoConfig {
1501                public_key: v.public_key,
1502                internal: v.internal,
1503            }
1504        }
1505    }
1506
1507    pub(super) fn deserialize<'de, D>(des: D) -> Result<HashMap<RelayId, RelayInfo>, D::Error>
1508    where
1509        D: Deserializer<'de>,
1510    {
1511        let map = HashMap::<RelayId, RelayInfoConfig>::deserialize(des)?;
1512        Ok(map.into_iter().map(|(k, v)| (k, v.into())).collect())
1513    }
1514
1515    pub(super) fn serialize<S>(elm: &HashMap<RelayId, RelayInfo>, ser: S) -> Result<S::Ok, S::Error>
1516    where
1517        S: Serializer,
1518    {
1519        let mut map = ser.serialize_map(Some(elm.len()))?;
1520
1521        for (k, v) in elm {
1522            map.serialize_entry(k, &RelayInfoConfig::from(v.clone()))?;
1523        }
1524
1525        map.end()
1526    }
1527}
1528
1529/// Authentication options.
1530#[derive(Serialize, Deserialize, Debug, Clone)]
1531#[serde(default)]
1532pub struct AuthConfig {
1533    /// Controls responses from the readiness health check endpoint based on authentication.
1534    #[serde(skip_serializing_if = "is_default")]
1535    pub ready: ReadinessCondition,
1536
1537    /// Statically authenticated downstream relays.
1538    #[serde(with = "config_relay_info")]
1539    pub static_relays: HashMap<RelayId, RelayInfo>,
1540
1541    /// How old a signature can be before it is considered invalid, in seconds.
1542    ///
1543    /// Defaults to 5 minutes.
1544    pub signature_max_age: u64,
1545}
1546
1547impl Default for AuthConfig {
1548    fn default() -> Self {
1549        Self {
1550            ready: ReadinessCondition::default(),
1551            static_relays: HashMap::new(),
1552            signature_max_age: 300, // 5 minutes
1553        }
1554    }
1555}
1556
1557/// GeoIp database configuration options.
1558#[derive(Serialize, Deserialize, Debug, Default, Clone)]
1559pub struct GeoIpConfig {
1560    /// The path to GeoIP database.
1561    pub path: Option<PathBuf>,
1562}
1563
1564/// Settings to control Relay's health checks.
1565///
1566/// After breaching one of the configured thresholds, Relay will
1567/// return an `unhealthy` status from its health endpoint.
1568#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
1569#[serde(default)]
1570pub struct Health {
1571    /// Interval to refresh internal health checks.
1572    ///
1573    /// Shorter intervals will decrease the time it takes the health check endpoint to report
1574    /// issues, but can also increase sporadic unhealthy responses.
1575    ///
1576    /// Defaults to `3000`` (3 seconds).
1577    pub refresh_interval_ms: u64,
1578    /// Maximum memory watermark in bytes.
1579    ///
1580    /// By default, there is no absolute limit set and the watermark
1581    /// is only controlled by setting [`Self::max_memory_percent`].
1582    pub max_memory_bytes: Option<ByteSize>,
1583    /// Maximum memory watermark as a percentage of maximum system memory.
1584    ///
1585    /// Defaults to `0.95` (95%).
1586    pub max_memory_percent: f32,
1587    /// Health check probe timeout in milliseconds.
1588    ///
1589    /// Any probe exceeding the timeout will be considered failed.
1590    /// This limits the max execution time of Relay health checks.
1591    ///
1592    /// Defaults to 900 milliseconds.
1593    pub probe_timeout_ms: u64,
1594    /// The refresh frequency of memory stats which are used to poll memory
1595    /// usage of Relay.
1596    ///
1597    /// The implementation of memory stats guarantees that the refresh will happen at
1598    /// least every `x` ms since memory readings are lazy and are updated only if needed.
1599    pub memory_stat_refresh_frequency_ms: u64,
1600}
1601
1602impl Default for Health {
1603    fn default() -> Self {
1604        Self {
1605            refresh_interval_ms: 3000,
1606            max_memory_bytes: None,
1607            max_memory_percent: 0.95,
1608            probe_timeout_ms: 900,
1609            memory_stat_refresh_frequency_ms: 100,
1610        }
1611    }
1612}
1613
1614/// COGS configuration.
1615#[derive(Serialize, Deserialize, Debug, Clone)]
1616#[serde(default)]
1617pub struct Cogs {
1618    /// Maximium amount of COGS measurements allowed to backlog.
1619    ///
1620    /// Any additional COGS measurements recorded will be dropped.
1621    ///
1622    /// Defaults to `10_000`.
1623    pub max_queue_size: u64,
1624    /// Relay COGS resource id.
1625    ///
1626    /// All Relay related COGS measurements are emitted with this resource id.
1627    ///
1628    /// Defaults to `relay_service`.
1629    pub relay_resource_id: String,
1630}
1631
1632impl Default for Cogs {
1633    fn default() -> Self {
1634        Self {
1635            max_queue_size: 10_000,
1636            relay_resource_id: "relay_service".to_owned(),
1637        }
1638    }
1639}
1640
1641/// Configuration for the upload service.
1642#[derive(Debug, Clone, Serialize, Deserialize)]
1643#[serde(default)]
1644pub struct Upload {
1645    /// Maximum number of uploads that the service accepts.
1646    ///
1647    /// Additional uploads will be rejected.
1648    pub max_concurrent_requests: usize,
1649    /// Maximum time spent trying to upload, in seconds.
1650    pub timeout: u64,
1651    /// The maximum time between creating the upload and uploading the data / the attachment placeholder.
1652    ///
1653    /// In seconds.
1654    pub max_age: i64,
1655
1656    /// Credentials used for signing & verifying upload locations.
1657    ///
1658    /// If omitted, relay's default [`Credentials`] are used.
1659    pub credentials: Option<UploadCredentials>,
1660}
1661
1662impl Default for Upload {
1663    fn default() -> Self {
1664        Self {
1665            max_concurrent_requests: 100,
1666            timeout: 5 * 60,  // five minutes
1667            max_age: 60 * 60, // 1h
1668            credentials: None,
1669        }
1670    }
1671}
1672
1673/// Credentials used for signing & verifying upload locations.
1674#[derive(Clone, Serialize, Deserialize)]
1675pub struct UploadCredentials {
1676    /// Key used to sign upload locations.
1677    #[cfg(feature = "processing")]
1678    pub signing_key: SecretKey,
1679
1680    /// Key used to verify upload locations.
1681    pub verification_key: PublicKey,
1682}
1683
1684impl fmt::Debug for UploadCredentials {
1685    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1686        let Self {
1687            #[cfg(feature = "processing")]
1688                signing_key: _,
1689            verification_key,
1690        } = self;
1691        let mut b = f.debug_struct("UploadCredentials");
1692        #[cfg(feature = "processing")]
1693        b.field("signing_key", &"[redacted]");
1694        b.field("verification_key", verification_key).finish()
1695    }
1696}
1697
1698/// All configuration values that can be deserialized from `config.yml`.
1699#[derive(Serialize, Deserialize, Debug, Default, Clone)]
1700#[serde(default)]
1701#[allow(missing_docs)]
1702pub struct ConfigValues {
1703    pub relay: Relay,
1704    pub http: Http,
1705    pub cache: Cache,
1706    pub spool: Spool,
1707    pub limits: Limits,
1708    pub logging: relay_log::LogConfig,
1709    pub routing: Routing,
1710    pub metrics: Metrics,
1711    pub sentry: relay_log::SentryConfig,
1712    pub processing: Processing,
1713    pub outcomes: Outcomes,
1714    pub aggregator: AggregatorServiceConfig,
1715    pub secondary_aggregators: Vec<ScopedAggregatorConfig>,
1716    pub auth: AuthConfig,
1717    pub geoip: GeoIpConfig,
1718    pub normalization: Normalization,
1719    pub health: Health,
1720    pub cogs: Cogs,
1721    pub upload: Upload,
1722}
1723
1724impl ConfigObject for ConfigValues {
1725    fn format() -> ConfigFormat {
1726        ConfigFormat::Yaml
1727    }
1728
1729    fn name() -> &'static str {
1730        "config"
1731    }
1732}
1733
1734#[derive(Default, Clone)]
1735struct ConfigInner {
1736    /// Relay's config values.
1737    values: ConfigValues,
1738    /// Configured Relay credentials.
1739    ///
1740    /// Credentials may be missing for proxy mode.
1741    credentials: Option<Credentials>,
1742}
1743
1744impl ConfigInner {
1745    fn apply_overrides(&mut self, overrides: &OverridableConfig) -> anyhow::Result<()> {
1746        if let Some(log_level) = &overrides.log_level {
1747            self.values.logging.level = log_level.parse()?;
1748        }
1749
1750        if let Some(log_format) = &overrides.log_format {
1751            self.values.logging.format = log_format.parse()?;
1752        }
1753
1754        let relay = &mut self.values.relay;
1755        if let Some(mode) = &overrides.mode {
1756            relay.mode = mode
1757                .parse::<RelayMode>()
1758                .with_context(|| ConfigError::field("mode"))?;
1759        }
1760        if let Some(deployment) = &overrides.instance {
1761            relay.instance = deployment
1762                .parse::<RelayInstance>()
1763                .with_context(|| ConfigError::field("deployment"))?;
1764        }
1765        if let Some(upstream) = &overrides.upstream {
1766            relay.upstream = upstream
1767                .parse::<UpstreamDescriptor>()
1768                .with_context(|| ConfigError::field("upstream"))?;
1769        } else if let Some(upstream_dsn) = &overrides.upstream_dsn {
1770            relay.upstream = upstream_dsn
1771                .parse::<Dsn>()
1772                .map(|dsn| UpstreamDescriptor::from_dsn(&dsn))
1773                .with_context(|| ConfigError::field("upstream_dsn"))?;
1774        }
1775        if let Some(host) = &overrides.host {
1776            relay.host = host
1777                .parse::<IpAddr>()
1778                .with_context(|| ConfigError::field("host"))?;
1779        }
1780        if let Some(port) = &overrides.port {
1781            relay.port = port
1782                .as_str()
1783                .parse()
1784                .with_context(|| ConfigError::field("port"))?;
1785        }
1786
1787        let processing = &mut self.values.processing;
1788        if let Some(enabled) = &overrides.processing {
1789            match enabled.to_lowercase().as_str() {
1790                "true" | "1" => processing.enabled = true,
1791                "false" | "0" | "" => processing.enabled = false,
1792                _ => return Err(ConfigError::field("processing").into()),
1793            }
1794        }
1795        if let Some(redis) = overrides.redis_url.clone() {
1796            processing.redis = Some(RedisConfigs::Unified(RedisConfig::single(redis)))
1797        }
1798        if let Some(kafka_url) = overrides.kafka_url.clone() {
1799            let existing = processing
1800                .kafka_config
1801                .iter_mut()
1802                .find(|e| e.name == "bootstrap.servers");
1803
1804            if let Some(config_param) = existing {
1805                config_param.value = kafka_url;
1806            } else {
1807                self.values.processing.kafka_config.push(KafkaConfigParam {
1808                    name: "bootstrap.servers".to_owned(),
1809                    value: kafka_url,
1810                })
1811            }
1812        }
1813
1814        if overrides.outcome_source.is_some() {
1815            self.values.outcomes.source = overrides.outcome_source.clone();
1816        }
1817
1818        if let Some(shutdown_timeout) = &overrides.shutdown_timeout
1819            && let Ok(shutdown_timeout) = shutdown_timeout.parse::<u64>()
1820        {
1821            self.values.limits.shutdown_timeout = shutdown_timeout;
1822        }
1823
1824        if let Some(server_name) = overrides.server_name.clone() {
1825            self.values.sentry.server_name = Some(server_name.into());
1826        }
1827
1828        let id = if let Some(id) = &overrides.id {
1829            let id = Uuid::parse_str(id).with_context(|| ConfigError::field("id"))?;
1830            Some(id)
1831        } else {
1832            None
1833        };
1834        let public_key = if let Some(public_key) = &overrides.public_key {
1835            let public_key = public_key
1836                .parse::<PublicKey>()
1837                .with_context(|| ConfigError::field("public_key"))?;
1838            Some(public_key)
1839        } else {
1840            None
1841        };
1842
1843        let secret_key = if let Some(secret_key) = &overrides.secret_key {
1844            let secret_key = secret_key
1845                .parse::<SecretKey>()
1846                .with_context(|| ConfigError::field("secret_key"))?;
1847            Some(secret_key)
1848        } else {
1849            None
1850        };
1851
1852        if let Some(credentials) = &mut self.credentials {
1853            //we have existing credentials we may override some entries
1854            if let Some(id) = id {
1855                credentials.id = id;
1856            }
1857            if let Some(public_key) = public_key {
1858                credentials.public_key = public_key;
1859            }
1860            if let Some(secret_key) = secret_key {
1861                credentials.secret_key = secret_key
1862            }
1863        } else {
1864            //no existing credentials we may only create the full credentials
1865            match (id, public_key, secret_key) {
1866                (Some(id), Some(public_key), Some(secret_key)) => {
1867                    self.credentials = Some(Credentials {
1868                        secret_key,
1869                        public_key,
1870                        id,
1871                    })
1872                }
1873                (None, None, None) => {
1874                    // nothing provided, we'll just leave the credentials None, maybe we
1875                    // don't need them in the current command or we'll override them later
1876                }
1877                _ => {
1878                    return Err(ConfigError::field("incomplete credentials").into());
1879                }
1880            }
1881        }
1882
1883        Ok(())
1884    }
1885
1886    /// Merges reloadable parts of the config from `other` into `self`.
1887    ///
1888    /// Returns `true` if any parts of the config were updated.
1889    fn reload_with(&mut self, other: &Self) -> bool {
1890        let mut changed = false;
1891
1892        if self.values.health != other.values.health {
1893            relay_log::debug!("updating health");
1894            self.values.health = other.values.health.clone();
1895            changed = true;
1896        }
1897
1898        changed
1899    }
1900}
1901
1902/// Relay's Configuration.
1903pub struct Config {
1904    /// A mutex to serialize all write accesses to `inner`.
1905    ///
1906    /// Accessing the arc swap is done with compare and swap, but we still want serialized
1907    /// access and operations to guarantee consistency when dealing with e.g. the filesystem.
1908    ///
1909    /// The mutex must be acquired before changing `inner`. Methods taking `&mut` can omit
1910    /// this, as the `&mut` requirement already satisfies that there is no concurrent access
1911    /// possible.
1912    inner_access: Mutex<()>,
1913    /// The actual config.
1914    inner: ArcSwap<ConfigInner>,
1915    /// A list of overrides applied to the config, in order.
1916    ///
1917    /// When re-loading the configuration these overrides need to be applied again
1918    /// in the same order as they were applied originally.
1919    overrides: Vec<OverridableConfig>,
1920    /// Path from which the config is loaded.
1921    ///
1922    /// This is Relay's configuration directory.
1923    path: PathBuf,
1924}
1925
1926impl fmt::Debug for Config {
1927    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1928        let inner = self.inner.load();
1929
1930        f.debug_struct("Config")
1931            .field("path", &self.path)
1932            // Only print specific parts of `inner` to not leak the credentials.
1933            .field("values", &inner.values)
1934            .finish()
1935    }
1936}
1937
1938impl Config {
1939    /// Loads a config from a given config folder.
1940    pub fn from_path<P: AsRef<Path>>(path: P) -> anyhow::Result<Config> {
1941        let path = env::current_dir()
1942            .map(|x| x.join(path.as_ref()))
1943            .unwrap_or_else(|_| path.as_ref().to_path_buf());
1944
1945        let inner = ConfigInner {
1946            values: ConfigValues::load(&path)?,
1947            credentials: match Credentials::path(&path).exists() {
1948                true => Some(Credentials::load(&path)?),
1949                false => None,
1950            },
1951        };
1952
1953        let config = Config {
1954            inner_access: Mutex::new(()),
1955            inner: ArcSwap::from_pointee(inner),
1956            overrides: Vec::new(),
1957            path: path.clone(),
1958        };
1959
1960        if cfg!(not(feature = "processing")) && config.current().processing_enabled() {
1961            return Err(ConfigError::file(ConfigErrorKind::ProcessingNotAvailable, &path).into());
1962        }
1963
1964        Ok(config)
1965    }
1966
1967    /// Creates a config from a JSON value.
1968    ///
1969    /// This is mostly useful for tests.
1970    pub fn from_json_value(value: serde_json::Value) -> anyhow::Result<Config> {
1971        Ok(Config {
1972            inner_access: Mutex::new(()),
1973            inner: ArcSwap::from_pointee(ConfigInner {
1974                values: serde_json::from_value(value)
1975                    .with_context(|| ConfigError::new(ConfigErrorKind::BadJson))?,
1976                credentials: None,
1977            }),
1978            overrides: Vec::new(),
1979            path: PathBuf::new(),
1980        })
1981    }
1982
1983    /// Override configuration with values coming from other sources (e.g. env variables or
1984    /// command line parameters).
1985    ///
1986    /// If applying the overrides fails, the config may be left in an inconsistent state.
1987    pub fn apply_override(&mut self, overrides: OverridableConfig) -> anyhow::Result<&mut Self> {
1988        // We could introduce a config builder which operates on mutable configs, which would eliminate
1989        // the need for this `try_rcu` dance here.
1990        crate::utils::try_rcu(&self.inner, |inner| {
1991            let mut new = ConfigInner::clone(inner);
1992            new.apply_overrides(&overrides)?;
1993            Ok::<_, anyhow::Error>(Arc::new(new))
1994        })?;
1995
1996        // Overrides successfully applied.
1997        self.overrides.push(overrides);
1998
1999        Ok(self)
2000    }
2001
2002    /// Checks if the config is already initialized.
2003    pub fn config_exists<P: AsRef<Path>>(path: P) -> bool {
2004        fs::metadata(ConfigValues::path(path.as_ref())).is_ok()
2005    }
2006
2007    /// Returns the filename of the config file.
2008    pub fn path(&self) -> &Path {
2009        &self.path
2010    }
2011
2012    /// Dumps out a YAML string of the values.
2013    pub fn to_yaml_string(&self) -> anyhow::Result<String> {
2014        serde_yaml::to_string(&self.inner.load().values)
2015            .with_context(|| ConfigError::new(ConfigErrorKind::CouldNotWriteFile))
2016    }
2017
2018    /// Set new credentials.
2019    ///
2020    /// This also writes the credentials back to the file, if this config was loaded from the file-system.
2021    pub fn replace_credentials(
2022        &mut self,
2023        credentials: Option<Credentials>,
2024    ) -> anyhow::Result<bool> {
2025        if self.inner.load().credentials == credentials {
2026            return Ok(false);
2027        }
2028
2029        if !self.path.is_empty() {
2030            match &credentials {
2031                Some(creds) => {
2032                    creds.save(&self.path)?;
2033                }
2034                None => {
2035                    let path = Credentials::path(&self.path);
2036                    if fs::metadata(&path).is_ok() {
2037                        fs::remove_file(&path).with_context(|| {
2038                            ConfigError::file(ConfigErrorKind::CouldNotWriteFile, &path)
2039                        })?;
2040                    }
2041                }
2042            }
2043        }
2044
2045        // Note: there is never anyone racing on the `ArcSwap` as long as `Self` borrowed mutably.
2046        //
2047        // We can improve this if we split out mutable operations into a separate struct and only
2048        // once `frozen()` we change to an `ArcSwap` internally.
2049        self.inner.rcu(|inner| {
2050            let mut inner = ConfigInner::clone(inner);
2051            inner.credentials = credentials.clone();
2052            Arc::new(inner)
2053        });
2054
2055        Ok(true)
2056    }
2057
2058    /// Reloads the configuration from disk.
2059    ///
2060    /// Returns `true` if the configuration changed.
2061    ///
2062    /// In order for a config to be reloadable it must've been loaded from a path. The original
2063    /// config will be re-read and reloadable parts of the config will be replaced with their updated
2064    /// values.
2065    ///
2066    /// If the reload fails for any reason, the current config is untouched.
2067    pub fn reload(&self) -> anyhow::Result<bool> {
2068        if self.path.is_empty() {
2069            return Ok(false);
2070        }
2071
2072        let _access = self
2073            .inner_access
2074            .lock()
2075            .unwrap_or_else(PoisonError::into_inner);
2076
2077        let mut new_config = Self::from_path(&self.path)?;
2078        for overrides in &self.overrides {
2079            new_config.apply_override(overrides.clone())?;
2080        }
2081        let new_config = new_config.current();
2082
2083        let mut changed = false;
2084
2085        // Since we do have the `_access` lock, this will always succeed on the first try.
2086        self.inner.rcu(|inner| {
2087            let mut new_inner = ConfigInner::clone(inner);
2088            changed = new_inner.reload_with(&new_config.inner);
2089            match changed {
2090                true => Arc::new(new_inner),
2091                false => Arc::clone(inner),
2092            }
2093        });
2094
2095        Ok(changed)
2096    }
2097
2098    /// Acquires a current [`snapshot`](ConfigSnapshot) of the config.
2099    ///
2100    /// A snapshot is the way to actually consume values from the config. A snapshot should ideally be acquired
2101    /// once per unit of work.
2102    pub fn current(&self) -> ConfigSnapshot {
2103        let inner = self.inner.load();
2104        ConfigSnapshot { inner }
2105    }
2106}
2107
2108impl Default for Config {
2109    fn default() -> Self {
2110        Self {
2111            inner_access: Mutex::new(()),
2112            inner: ArcSwap::from_pointee(Default::default()),
2113            overrides: Vec::new(),
2114            path: PathBuf::new(),
2115        }
2116    }
2117}
2118
2119/// A config snapshot is a point in time snapshot of the [`Config`].
2120///
2121/// The [`Config`] may change over time, to guarantee a consistent view of the config
2122/// a snapshot must be acquired first.
2123///
2124/// The snapshot should not be stored in a long lasting datastructure. As a rule of thumb it should
2125/// only exist on the stack.
2126pub struct ConfigSnapshot {
2127    inner: arc_swap::Guard<Arc<ConfigInner>>,
2128}
2129
2130impl ConfigSnapshot {
2131    /// Returns `true` if the config is ready to use.
2132    pub fn has_credentials(&self) -> bool {
2133        self.inner.credentials.is_some()
2134    }
2135
2136    /// Return the current credentials.
2137    pub fn credentials(&self) -> Option<&Credentials> {
2138        self.inner.credentials.as_ref()
2139    }
2140
2141    /// Returns the secret key if set.
2142    pub fn secret_key(&self) -> Option<&SecretKey> {
2143        self.inner.credentials.as_ref().map(|x| &x.secret_key)
2144    }
2145
2146    /// Returns the public key if set.
2147    pub fn public_key(&self) -> Option<&PublicKey> {
2148        self.inner.credentials.as_ref().map(|x| &x.public_key)
2149    }
2150
2151    /// Returns the relay ID.
2152    pub fn relay_id(&self) -> Option<&RelayId> {
2153        self.inner.credentials.as_ref().map(|x| &x.id)
2154    }
2155
2156    /// Returns the relay mode.
2157    pub fn relay_mode(&self) -> RelayMode {
2158        self.inner.values.relay.mode
2159    }
2160
2161    /// Returns the instance type of relay.
2162    pub fn relay_instance(&self) -> RelayInstance {
2163        self.inner.values.relay.instance
2164    }
2165
2166    /// Returns the upstream target as descriptor.
2167    pub fn upstream(&self) -> &UpstreamDescriptor {
2168        &self.inner.values.relay.upstream
2169    }
2170
2171    /// Returns the advertised upstream for downstream instances as descriptor.
2172    pub fn advertised_upstream(&self) -> Option<&UpstreamDescriptor> {
2173        self.inner.values.relay.advertised_upstream.as_ref()
2174    }
2175
2176    /// Returns the custom HTTP "Host" header.
2177    pub fn http_host_header(&self) -> Option<&str> {
2178        self.inner.values.http.host_header.as_deref()
2179    }
2180
2181    /// Returns the listen address.
2182    pub fn listen_addr(&self) -> SocketAddr {
2183        (self.inner.values.relay.host, self.inner.values.relay.port).into()
2184    }
2185
2186    /// Returns the listen address for internal APIs.
2187    ///
2188    /// Internal APIs are APIs which do not need to be publicly exposed,
2189    /// like health checks.
2190    ///
2191    /// Returns `None` when there is no explicit address configured for internal APIs,
2192    /// and they should instead be exposed on the main [`Self::listen_addr`].
2193    pub fn listen_addr_internal(&self) -> Option<SocketAddr> {
2194        match (
2195            self.inner.values.relay.internal_host,
2196            self.inner.values.relay.internal_port,
2197        ) {
2198            (Some(host), None) => Some((host, self.inner.values.relay.port).into()),
2199            (None, Some(port)) => Some((self.inner.values.relay.host, port).into()),
2200            (Some(host), Some(port)) => Some((host, port).into()),
2201            (None, None) => None,
2202        }
2203    }
2204
2205    /// Returns the TLS listen address.
2206    pub fn tls_listen_addr(&self) -> Option<SocketAddr> {
2207        if self.inner.values.relay.tls_identity_path.is_some() {
2208            let port = self.inner.values.relay.tls_port.unwrap_or(3443);
2209            Some((self.inner.values.relay.host, port).into())
2210        } else {
2211            None
2212        }
2213    }
2214
2215    /// Returns the path to the identity bundle
2216    pub fn tls_identity_path(&self) -> Option<&Path> {
2217        self.inner.values.relay.tls_identity_path.as_deref()
2218    }
2219
2220    /// Returns the password for the identity bundle
2221    pub fn tls_identity_password(&self) -> Option<&str> {
2222        self.inner.values.relay.tls_identity_password.as_deref()
2223    }
2224
2225    /// Returns `true` when project IDs should be overriden rather than validated.
2226    ///
2227    /// Defaults to `false`, which requires project ID validation.
2228    pub fn override_project_ids(&self) -> bool {
2229        self.inner.values.relay.override_project_ids
2230    }
2231
2232    /// Returns the interval to check for configuration changes.
2233    ///
2234    /// `None` if the config should never be reloaded.
2235    pub fn config_reload_interval(&self) -> Option<Duration> {
2236        let interval = self.inner.values.relay.config_reload_interval?;
2237        Some(match interval {
2238            // Useful for tests to be able to configure this to a very small value,
2239            // but we also don't want it to be actually 0.
2240            0 => Duration::from_millis(50),
2241            secs => Duration::from_secs(secs),
2242        })
2243    }
2244
2245    /// Returns `true` if Relay requires authentication for readiness.
2246    ///
2247    /// See [`ReadinessCondition`] for more information.
2248    pub fn requires_auth(&self) -> bool {
2249        match self.inner.values.auth.ready {
2250            ReadinessCondition::Authenticated => self.relay_mode() == RelayMode::Managed,
2251            ReadinessCondition::Always => false,
2252        }
2253    }
2254
2255    /// Returns the interval at which Realy should try to re-authenticate with the upstream.
2256    ///
2257    /// Always disabled in processing mode.
2258    pub fn http_auth_interval(&self) -> Option<Duration> {
2259        if self.processing_enabled() {
2260            return None;
2261        }
2262
2263        match self.inner.values.http.auth_interval {
2264            None | Some(0) => None,
2265            Some(secs) => Some(Duration::from_secs(secs)),
2266        }
2267    }
2268
2269    /// The maximum time of experiencing uninterrupted network failures until Relay considers that
2270    /// it has encountered a network outage.
2271    pub fn http_outage_grace_period(&self) -> Duration {
2272        Duration::from_secs(self.inner.values.http.outage_grace_period)
2273    }
2274
2275    /// Time Relay waits before retrying an upstream request.
2276    ///
2277    /// Before going into a network outage, Relay may fail to make upstream
2278    /// requests. This is the time Relay waits before retrying the same request.
2279    pub fn http_retry_delay(&self) -> Duration {
2280        Duration::from_secs(self.inner.values.http.retry_delay)
2281    }
2282
2283    /// Time of continued project request failures before Relay emits an error.
2284    pub fn http_project_failure_interval(&self) -> Duration {
2285        Duration::from_secs(self.inner.values.http.project_failure_interval)
2286    }
2287
2288    /// Content encoding of upstream requests.
2289    pub fn http_encoding(&self) -> HttpEncoding {
2290        self.inner.values.http.encoding
2291    }
2292
2293    /// Returns whether metrics should be sent globally through a shared endpoint.
2294    pub fn http_global_metrics(&self) -> bool {
2295        self.inner.values.http.global_metrics
2296    }
2297
2298    /// Returns `true` if Relay supports forwarding unknown API requests.
2299    ///
2300    /// Relay instances with processing enabled are expected to support the latest API and do never
2301    /// support forwarding requests to Sentry.
2302    pub fn http_forward(&self) -> bool {
2303        self.inner.values.http.forward && !self.processing_enabled()
2304    }
2305
2306    /// Returns whether this Relay should emit outcomes.
2307    ///
2308    /// This is `true` either if `outcomes.emit_outcomes` is explicitly enabled, or if this Relay is
2309    /// in processing mode.
2310    pub fn emit_outcomes(&self) -> EmitOutcomes {
2311        if self.processing_enabled() {
2312            return EmitOutcomes::AsOutcomes;
2313        }
2314        self.inner.values.outcomes.emit_outcomes
2315    }
2316
2317    /// Returns the maximum number of outcomes that are batched before being sent
2318    pub fn outcome_batch_size(&self) -> usize {
2319        self.inner.values.outcomes.batch_size
2320    }
2321
2322    /// Returns the maximum interval that an outcome may be batched
2323    pub fn outcome_batch_interval(&self) -> Duration {
2324        Duration::from_millis(self.inner.values.outcomes.batch_interval)
2325    }
2326
2327    /// The originating source of the outcome
2328    pub fn outcome_source(&self) -> Option<&str> {
2329        self.inner.values.outcomes.source.as_deref()
2330    }
2331
2332    /// Returns logging configuration.
2333    pub fn logging(&self) -> &relay_log::LogConfig {
2334        &self.inner.values.logging
2335    }
2336
2337    /// Returns logging configuration.
2338    pub fn sentry(&self) -> &relay_log::SentryConfig {
2339        &self.inner.values.sentry
2340    }
2341
2342    /// Returns the addresses for statsd metrics.
2343    pub fn statsd_addr(&self) -> Option<&str> {
2344        self.inner.values.metrics.statsd.as_deref()
2345    }
2346
2347    /// Returns the addresses for statsd metrics.
2348    pub fn statsd_buffer_size(&self) -> Option<usize> {
2349        self.inner.values.metrics.statsd_buffer_size
2350    }
2351
2352    /// Return the prefix for statsd metrics.
2353    pub fn metrics_prefix(&self) -> &str {
2354        &self.inner.values.metrics.prefix
2355    }
2356
2357    /// Returns the default tags for statsd metrics.
2358    pub fn metrics_default_tags(&self) -> &BTreeMap<String, String> {
2359        &self.inner.values.metrics.default_tags
2360    }
2361
2362    /// Returns the name of the hostname tag that should be attached to each outgoing metric.
2363    pub fn metrics_hostname_tag(&self) -> Option<&str> {
2364        self.inner.values.metrics.hostname_tag.as_deref()
2365    }
2366
2367    /// Returns the interval for periodic metrics emitted from Relay.
2368    ///
2369    /// `None` if periodic metrics are disabled.
2370    pub fn metrics_periodic_interval(&self) -> Option<Duration> {
2371        match self.inner.values.metrics.periodic_secs {
2372            0 => None,
2373            secs => Some(Duration::from_secs(secs)),
2374        }
2375    }
2376
2377    /// Returns the default timeout for all upstream HTTP requests.
2378    pub fn http_timeout(&self) -> Duration {
2379        Duration::from_secs(self.inner.values.http.timeout.into())
2380    }
2381
2382    /// Returns the connection timeout for all upstream HTTP requests.
2383    pub fn http_connection_timeout(&self) -> Duration {
2384        Duration::from_secs(self.inner.values.http.connection_timeout.into())
2385    }
2386
2387    /// Returns the failed upstream request retry interval.
2388    pub fn http_max_retry_interval(&self) -> Duration {
2389        Duration::from_secs(self.inner.values.http.max_retry_interval.into())
2390    }
2391
2392    /// Returns `true` if relay should use an in-process cache for DNS lookups.
2393    pub fn http_dns_cache(&self) -> bool {
2394        self.inner.values.http.dns_cache
2395    }
2396
2397    /// Returns the expiry timeout for cached projects.
2398    pub fn project_cache_expiry(&self) -> Duration {
2399        Duration::from_secs(self.inner.values.cache.project_expiry.into())
2400    }
2401
2402    /// Returns `true` if the full project state should be requested from upstream.
2403    pub fn request_full_project_config(&self) -> bool {
2404        self.inner.values.cache.project_request_full_config
2405    }
2406
2407    /// Returns the expiry timeout for cached relay infos (public keys).
2408    pub fn relay_cache_expiry(&self) -> Duration {
2409        Duration::from_secs(self.inner.values.cache.relay_expiry.into())
2410    }
2411
2412    /// Returns the maximum number of buffered envelopes
2413    pub fn envelope_buffer_size(&self) -> usize {
2414        self.inner
2415            .values
2416            .cache
2417            .envelope_buffer_size
2418            .try_into()
2419            .unwrap_or(usize::MAX)
2420    }
2421
2422    /// Returns the expiry timeout for cached misses before trying to refetch.
2423    pub fn cache_miss_expiry(&self) -> Duration {
2424        Duration::from_secs(self.inner.values.cache.miss_expiry.into())
2425    }
2426
2427    /// Returns the grace period for project caches.
2428    pub fn project_grace_period(&self) -> Duration {
2429        Duration::from_secs(self.inner.values.cache.project_grace_period.into())
2430    }
2431
2432    /// Returns the refresh interval for a project.
2433    ///
2434    /// Validates the refresh time to be between the grace period and expiry.
2435    pub fn project_refresh_interval(&self) -> Option<Duration> {
2436        self.inner
2437            .values
2438            .cache
2439            .project_refresh_interval
2440            .map(Into::into)
2441            .map(Duration::from_secs)
2442    }
2443
2444    /// Returns the duration in which batchable project config queries are
2445    /// collected before sending them in a single request.
2446    pub fn query_batch_interval(&self) -> Duration {
2447        Duration::from_millis(self.inner.values.cache.batch_interval.into())
2448    }
2449
2450    /// Returns the duration in which downstream relays are requested from upstream.
2451    pub fn downstream_relays_batch_interval(&self) -> Duration {
2452        Duration::from_millis(
2453            self.inner
2454                .values
2455                .cache
2456                .downstream_relays_batch_interval
2457                .into(),
2458        )
2459    }
2460
2461    /// Returns the interval in seconds in which local project configurations should be reloaded.
2462    pub fn local_cache_interval(&self) -> Duration {
2463        Duration::from_secs(self.inner.values.cache.file_interval.into())
2464    }
2465
2466    /// Returns the interval in seconds in which fresh global configs should be
2467    /// fetched from  upstream.
2468    pub fn global_config_fetch_interval(&self) -> Duration {
2469        Duration::from_secs(self.inner.values.cache.global_config_fetch_interval.into())
2470    }
2471
2472    /// Returns the path of the buffer file if the `cache.persistent_envelope_buffer.path` is configured.
2473    ///
2474    /// In case a partition with id > 0 is supplied, the filename of the envelopes path will be
2475    /// suffixed with `.{partition_id}`.
2476    pub fn spool_envelopes_path(&self, partition_id: u8) -> Option<PathBuf> {
2477        let mut path = self
2478            .inner
2479            .values
2480            .spool
2481            .envelopes
2482            .path
2483            .as_ref()
2484            .map(|path| path.to_owned())?;
2485
2486        if partition_id == 0 {
2487            return Some(path);
2488        }
2489
2490        let file_name = path.file_name().and_then(|f| f.to_str())?;
2491        let new_file_name = format!("{file_name}.{partition_id}");
2492        path.set_file_name(new_file_name);
2493
2494        Some(path)
2495    }
2496
2497    /// The maximum size of the buffer, in bytes.
2498    pub fn spool_envelopes_max_disk_size(&self) -> usize {
2499        self.inner.values.spool.envelopes.max_disk_size.as_bytes()
2500    }
2501
2502    /// Number of encoded envelope bytes that need to be accumulated before
2503    /// flushing one batch to disk.
2504    pub fn spool_envelopes_batch_size_bytes(&self) -> usize {
2505        self.inner
2506            .values
2507            .spool
2508            .envelopes
2509            .batch_size_bytes
2510            .as_bytes()
2511    }
2512
2513    /// Returns the time after which we drop envelopes as a [`Duration`] object.
2514    pub fn spool_envelopes_max_age(&self) -> Duration {
2515        Duration::from_secs(self.inner.values.spool.envelopes.max_envelope_delay_secs)
2516    }
2517
2518    /// Returns the refresh frequency for disk usage monitoring as a [`Duration`] object.
2519    pub fn spool_disk_usage_refresh_frequency_ms(&self) -> Duration {
2520        Duration::from_millis(
2521            self.inner
2522                .values
2523                .spool
2524                .envelopes
2525                .disk_usage_refresh_frequency_ms,
2526        )
2527    }
2528
2529    /// Returns the relative memory usage up to which the disk buffer will unspool envelopes.
2530    pub fn spool_max_backpressure_memory_percent(&self) -> f32 {
2531        self.inner
2532            .values
2533            .spool
2534            .envelopes
2535            .max_backpressure_memory_percent
2536    }
2537
2538    /// Returns the number of partitions for the buffer.
2539    pub fn spool_partitions(&self) -> NonZeroU8 {
2540        self.inner.values.spool.envelopes.partitions
2541    }
2542
2543    /// Returns the strategy used to assign envelopes to buffer partitions.
2544    pub fn spool_partitioning(&self) -> EnvelopeSpoolPartitioning {
2545        self.inner.values.spool.envelopes.partitioning
2546    }
2547
2548    /// Returns `true` if the data is stored on ephemeral disks.
2549    pub fn spool_ephemeral(&self) -> bool {
2550        self.inner.values.spool.envelopes.ephemeral
2551    }
2552
2553    /// Returns the maximum size of an event payload in bytes.
2554    pub fn max_event_size(&self) -> usize {
2555        self.inner.values.limits.max_event_size.as_bytes()
2556    }
2557
2558    /// Returns the maximum size of each attachment.
2559    pub fn max_attachment_size(&self) -> usize {
2560        self.inner.values.limits.max_attachment_size.as_bytes()
2561    }
2562
2563    /// The maximum amount of attachments in a single envelope.
2564    pub fn max_attachment_count(&self) -> usize {
2565        self.inner.values.limits.max_attachment_count
2566    }
2567
2568    /// Returns the maximum combined size of attachments or payloads containing attachments
2569    /// (minidump, unreal, standalone attachments) in bytes.
2570    pub fn max_attachments_size(&self) -> usize {
2571        self.inner.values.limits.max_attachments_size.as_bytes()
2572    }
2573
2574    /// Returns the maximum size of a TUS upload request body.
2575    pub fn max_upload_size(&self) -> usize {
2576        self.inner.values.limits.max_upload_size.as_bytes()
2577    }
2578
2579    /// Returns the maximum number of client reports per envelope.
2580    pub fn max_client_reports_count(&self) -> usize {
2581        self.inner.values.limits.max_client_reports_count
2582    }
2583
2584    /// Returns the maximum combined size of client reports in bytes.
2585    pub fn max_client_reports_size(&self) -> usize {
2586        self.inner.values.limits.max_client_reports_size.as_bytes()
2587    }
2588
2589    /// Returns the maximum payload size of a monitor check-in in bytes.
2590    pub fn max_check_in_size(&self) -> usize {
2591        self.inner.values.limits.max_check_in_size.as_bytes()
2592    }
2593
2594    /// Returns the maximum payload size of a log in bytes.
2595    pub fn max_log_size(&self) -> usize {
2596        self.inner.values.limits.max_log_size.as_bytes()
2597    }
2598
2599    /// Returns the maximum payload size of a span in bytes.
2600    pub fn max_span_size(&self) -> usize {
2601        self.inner.values.limits.max_span_size.as_bytes()
2602    }
2603
2604    /// Returns the maximum amount of standalone transaction spans per envelope.
2605    pub fn max_standalone_span_count(&self) -> usize {
2606        self.inner.values.limits.max_standalone_span_count
2607    }
2608
2609    /// Returns the maximum payload size of an item container in bytes.
2610    pub fn max_container_size(&self) -> usize {
2611        self.inner.values.limits.max_container_size.as_bytes()
2612    }
2613
2614    /// Returns the maximum size of an envelope payload in bytes.
2615    ///
2616    /// Individual item size limits still apply.
2617    pub fn max_envelope_size(&self) -> usize {
2618        self.inner.values.limits.max_envelope_size.as_bytes()
2619    }
2620
2621    /// Returns the maximum number of sessions per envelope.
2622    pub fn max_session_count(&self) -> usize {
2623        self.inner.values.limits.max_session_count
2624    }
2625
2626    /// Returns the maximum combined size for all sessions in an envelope in bytes.
2627    pub fn max_sessions_size(&self) -> usize {
2628        self.inner.values.limits.max_sessions_size.as_bytes()
2629    }
2630
2631    /// Returns the maximum payload size of a statsd metric in bytes.
2632    pub fn max_statsd_size(&self) -> usize {
2633        self.inner.values.limits.max_statsd_size.as_bytes()
2634    }
2635
2636    /// Returns the maximum payload size of metric buckets in bytes.
2637    pub fn max_metric_buckets_size(&self) -> usize {
2638        self.inner.values.limits.max_metric_buckets_size.as_bytes()
2639    }
2640
2641    /// Returns the maximum payload size for general API requests.
2642    pub fn max_api_payload_size(&self) -> usize {
2643        self.inner.values.limits.max_api_payload_size.as_bytes()
2644    }
2645
2646    /// Returns the maximum payload size for file uploads and chunks.
2647    pub fn max_api_file_upload_size(&self) -> usize {
2648        self.inner.values.limits.max_api_file_upload_size.as_bytes()
2649    }
2650
2651    /// Returns the maximum payload size for chunks
2652    pub fn max_api_chunk_upload_size(&self) -> usize {
2653        self.inner
2654            .values
2655            .limits
2656            .max_api_chunk_upload_size
2657            .as_bytes()
2658    }
2659
2660    /// Returns the maximum payload size for a profile
2661    pub fn max_profile_size(&self) -> usize {
2662        self.inner.values.limits.max_profile_size.as_bytes()
2663    }
2664
2665    /// Returns the maximum payload size for a trace metric.
2666    pub fn max_trace_metric_size(&self) -> usize {
2667        self.inner.values.limits.max_trace_metric_size.as_bytes()
2668    }
2669
2670    /// Returns the maximum payload size for a compressed replay.
2671    pub fn max_replay_compressed_size(&self) -> usize {
2672        self.inner
2673            .values
2674            .limits
2675            .max_replay_compressed_size
2676            .as_bytes()
2677    }
2678
2679    /// Returns the maximum payload size for an uncompressed replay.
2680    pub fn max_replay_uncompressed_size(&self) -> usize {
2681        self.inner
2682            .values
2683            .limits
2684            .max_replay_uncompressed_size
2685            .as_bytes()
2686    }
2687
2688    /// Returns the maximum message size for an uncompressed replay.
2689    ///
2690    /// This is greater than max_replay_compressed_size because
2691    /// it can include additional metadata about the replay in
2692    /// addition to the recording.
2693    pub fn max_replay_message_size(&self) -> usize {
2694        self.inner.values.limits.max_replay_message_size.as_bytes()
2695    }
2696
2697    /// Returns the maximum number of active requests
2698    pub fn max_concurrent_requests(&self) -> usize {
2699        self.inner.values.limits.max_concurrent_requests
2700    }
2701
2702    /// Returns the maximum number of active queries
2703    pub fn max_concurrent_queries(&self) -> usize {
2704        self.inner.values.limits.max_concurrent_queries
2705    }
2706
2707    /// Returns the maximum combined size of keys of invalid attributes.
2708    pub fn max_removed_attribute_key_size(&self) -> usize {
2709        self.inner
2710            .values
2711            .limits
2712            .max_removed_attribute_key_size
2713            .as_bytes()
2714    }
2715
2716    /// The maximum number of seconds a query is allowed to take across retries.
2717    pub fn query_timeout(&self) -> Duration {
2718        Duration::from_secs(self.inner.values.limits.query_timeout)
2719    }
2720
2721    /// The maximum number of seconds to wait for pending envelopes after receiving a shutdown
2722    /// signal.
2723    pub fn shutdown_timeout(&self) -> Duration {
2724        Duration::from_secs(self.inner.values.limits.shutdown_timeout)
2725    }
2726
2727    /// Returns the server keep-alive timeout in seconds.
2728    ///
2729    /// By default keep alive is set to a 5 seconds.
2730    pub fn keepalive_timeout(&self) -> Duration {
2731        Duration::from_secs(self.inner.values.limits.keepalive_timeout)
2732    }
2733
2734    /// Returns the server idle timeout in seconds.
2735    pub fn idle_timeout(&self) -> Option<Duration> {
2736        self.inner
2737            .values
2738            .limits
2739            .idle_timeout
2740            .map(Duration::from_secs)
2741    }
2742
2743    /// Returns the maximum connections.
2744    pub fn max_connections(&self) -> Option<usize> {
2745        self.inner.values.limits.max_connections
2746    }
2747
2748    /// TCP listen backlog to configure on Relay's listening socket.
2749    pub fn tcp_listen_backlog(&self) -> u32 {
2750        self.inner.values.limits.tcp_listen_backlog
2751    }
2752
2753    /// Returns the number of cores to use for thread pools.
2754    pub fn cpu_concurrency(&self) -> usize {
2755        self.inner.values.limits.max_thread_count
2756    }
2757
2758    /// Returns the number of tasks that can run concurrently in the worker pool.
2759    pub fn pool_concurrency(&self) -> usize {
2760        self.inner.values.limits.max_pool_concurrency
2761    }
2762
2763    /// Returns the maximum size of a project config query.
2764    pub fn query_batch_size(&self) -> usize {
2765        self.inner.values.cache.batch_size
2766    }
2767
2768    /// True if the Relay should do processing.
2769    pub fn processing_enabled(&self) -> bool {
2770        self.inner.values.processing.enabled
2771    }
2772
2773    /// Level of normalization for Relay to apply to incoming data.
2774    pub fn normalization_level(&self) -> NormalizationLevel {
2775        self.inner.values.normalization.level
2776    }
2777
2778    /// The path to the GeoIp database required for event processing.
2779    pub fn geoip_path(&self) -> Option<&Path> {
2780        self.inner.values.geoip.path.as_deref().or(self
2781            .inner
2782            .values
2783            .processing
2784            .geoip_path
2785            .as_deref())
2786    }
2787
2788    /// Maximum future timestamp of ingested data.
2789    ///
2790    /// Events past this timestamp will be adjusted to `now()`. Sessions will be dropped.
2791    pub fn max_secs_in_future(&self) -> i64 {
2792        self.inner.values.processing.max_secs_in_future.into()
2793    }
2794
2795    /// Maximum age of ingested sessions. Older sessions will be dropped.
2796    pub fn max_session_secs_in_past(&self) -> i64 {
2797        self.inner.values.processing.max_session_secs_in_past.into()
2798    }
2799
2800    /// Configuration name and list of Kafka configuration parameters for a given topic.
2801    pub fn kafka_configs(
2802        &self,
2803        topic: KafkaTopic,
2804    ) -> Result<KafkaTopicConfig<'_>, KafkaConfigError> {
2805        self.inner
2806            .values
2807            .processing
2808            .topics
2809            .get(topic)
2810            .kafka_configs(
2811                &self.inner.values.processing.kafka_config,
2812                &self.inner.values.processing.secondary_kafka_configs,
2813            )
2814    }
2815
2816    /// Whether to validate the topics against Kafka.
2817    pub fn kafka_validate_topics(&self) -> bool {
2818        self.inner.values.processing.kafka_validate_topics
2819    }
2820
2821    /// All unused but configured topic assignments.
2822    pub fn unused_topic_assignments(&self) -> &relay_kafka::Unused {
2823        &self.inner.values.processing.topics.unused
2824    }
2825
2826    /// Configuration of the objectstore service.
2827    pub fn objectstore(&self) -> &ObjectstoreServiceConfig {
2828        &self.inner.values.processing.objectstore
2829    }
2830
2831    /// Configuration of the upload service.
2832    pub fn upload(&self) -> &Upload {
2833        &self.inner.values.upload
2834    }
2835
2836    /// Returns the key used to sign upload locations.
2837    #[cfg(feature = "processing")]
2838    pub fn upload_signing_key(&self) -> Option<&SecretKey> {
2839        self.upload()
2840            .credentials
2841            .as_ref()
2842            .map(|c| &c.signing_key)
2843            .or(self.credentials().map(|c| &c.secret_key))
2844    }
2845
2846    /// Returns the key used to verify upload locations.
2847    #[cfg(feature = "processing")]
2848    pub fn upload_verification_key(&self) -> Option<&PublicKey> {
2849        self.upload()
2850            .credentials
2851            .as_ref()
2852            .map(|c| &c.verification_key)
2853            .or(self.credentials().map(|c| &c.public_key))
2854    }
2855
2856    /// Redis servers to connect to for project configs, rate limiting, and metrics metadata.
2857    pub fn redis(&self) -> Option<RedisConfigsRef<'_>> {
2858        let redis_configs = self.inner.values.processing.redis.as_ref()?;
2859
2860        Some(build_redis_configs(
2861            redis_configs,
2862            self.cpu_concurrency() as u32,
2863            self.pool_concurrency() as u32,
2864        ))
2865    }
2866
2867    /// Chunk size of attachments in bytes.
2868    pub fn attachment_chunk_size(&self) -> usize {
2869        self.inner
2870            .values
2871            .processing
2872            .attachment_chunk_size
2873            .as_bytes()
2874    }
2875
2876    /// Maximum metrics batch size in bytes.
2877    pub fn metrics_max_batch_size_bytes(&self) -> usize {
2878        self.inner.values.aggregator.max_flush_bytes
2879    }
2880
2881    /// Default prefix to use when looking up project configs in Redis. This is only done when
2882    /// Relay is in processing mode.
2883    pub fn projectconfig_cache_prefix(&self) -> &str {
2884        &self.inner.values.processing.projectconfig_cache_prefix
2885    }
2886
2887    /// Maximum rate limit to report to clients in seconds.
2888    pub fn max_rate_limit(&self) -> Option<u64> {
2889        self.inner.values.processing.max_rate_limit.map(u32::into)
2890    }
2891
2892    /// Amount of remaining quota which is cached in memory.
2893    pub fn quota_cache_ratio(&self) -> Option<f32> {
2894        self.inner.values.processing.quota_cache_ratio
2895    }
2896
2897    /// Maximum limit (ratio) for the in memory quota cache.
2898    pub fn quota_cache_max(&self) -> Option<f32> {
2899        self.inner.values.processing.quota_cache_max
2900    }
2901
2902    /// Interval to refresh internal health checks.
2903    pub fn health_refresh_interval(&self) -> Duration {
2904        Duration::from_millis(self.inner.values.health.refresh_interval_ms)
2905    }
2906
2907    /// Maximum memory watermark in bytes.
2908    pub fn health_max_memory_watermark_bytes(&self) -> u64 {
2909        self.inner
2910            .values
2911            .health
2912            .max_memory_bytes
2913            .as_ref()
2914            .map_or(u64::MAX, |b| b.as_bytes() as u64)
2915    }
2916
2917    /// Maximum memory watermark as a percentage of maximum system memory.
2918    pub fn health_max_memory_watermark_percent(&self) -> f32 {
2919        self.inner.values.health.max_memory_percent
2920    }
2921
2922    /// Health check probe timeout.
2923    pub fn health_probe_timeout(&self) -> Duration {
2924        Duration::from_millis(self.inner.values.health.probe_timeout_ms)
2925    }
2926
2927    /// Refresh frequency for polling new memory stats.
2928    pub fn memory_stat_refresh_frequency_ms(&self) -> u64 {
2929        self.inner.values.health.memory_stat_refresh_frequency_ms
2930    }
2931
2932    /// Maximum amount of COGS measurements buffered in memory.
2933    pub fn cogs_max_queue_size(&self) -> u64 {
2934        self.inner.values.cogs.max_queue_size
2935    }
2936
2937    /// Resource ID to use for Relay COGS measurements.
2938    pub fn cogs_relay_resource_id(&self) -> &str {
2939        &self.inner.values.cogs.relay_resource_id
2940    }
2941
2942    /// Returns configuration for the default metrics aggregator.
2943    pub fn default_aggregator_config(&self) -> &AggregatorServiceConfig {
2944        &self.inner.values.aggregator
2945    }
2946
2947    /// Returns configuration for non-default metrics aggregator.
2948    pub fn secondary_aggregator_configs(&self) -> &Vec<ScopedAggregatorConfig> {
2949        &self.inner.values.secondary_aggregators
2950    }
2951
2952    /// Returns aggregator config for a given metrics namespace.
2953    pub fn aggregator_config_for(&self, namespace: MetricNamespace) -> &AggregatorServiceConfig {
2954        for entry in &self.inner.values.secondary_aggregators {
2955            if entry.condition.matches(Some(namespace)) {
2956                return &entry.config;
2957            }
2958        }
2959        &self.inner.values.aggregator
2960    }
2961
2962    /// Return the statically configured Relays.
2963    pub fn static_relays(&self) -> &HashMap<RelayId, RelayInfo> {
2964        &self.inner.values.auth.static_relays
2965    }
2966
2967    /// Returns the max age a signature is considered valid, in seconds.
2968    pub fn signature_max_age(&self) -> Duration {
2969        Duration::from_secs(self.inner.values.auth.signature_max_age)
2970    }
2971
2972    /// Returns `true` if unknown items should be accepted and forwarded.
2973    pub fn accept_unknown_items(&self) -> bool {
2974        let forward = self.inner.values.routing.accept_unknown_items;
2975        forward.unwrap_or_else(|| !self.processing_enabled())
2976    }
2977}
2978
2979impl fmt::Debug for ConfigSnapshot {
2980    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2981        f.debug_struct("ConfigSnapshot")
2982            .field("values", &self.inner.values)
2983            .finish()
2984    }
2985}
2986
2987#[cfg(test)]
2988mod tests {
2989    use super::*;
2990
2991    /// Regression test for renaming the envelope buffer flags.
2992    #[test]
2993    fn test_event_buffer_size() {
2994        let yaml = r###"
2995cache:
2996    event_buffer_size: 1000000
2997    event_expiry: 1800
2998"###;
2999
3000        let values: ConfigValues = serde_yaml::from_str(yaml).unwrap();
3001        assert_eq!(values.cache.envelope_buffer_size, 1_000_000);
3002        assert_eq!(values.cache.envelope_expiry, 1800);
3003    }
3004
3005    #[cfg(feature = "processing")]
3006    #[test]
3007    fn test_upload_secret_key_from_file() {
3008        let path = env::temp_dir().join(Uuid::new_v4().to_string());
3009        fs::create_dir(&path).unwrap();
3010        fs::write(
3011            path.join("my_secret.txt"),
3012            "U3LSQM5NorvgnoYHW_aZpc_43nuuh3lhs3zjjcBwaks",
3013        )
3014        .unwrap();
3015        fs::write(
3016            ConfigValues::path(&path),
3017            r#"
3018    upload:
3019        credentials:
3020            signing_key: ${file:my_secret.txt}
3021            verification_key: "VNS8haF0VTnuMMDR2t-f7AgnmUcXmcdzV3SVksSk34s""#,
3022        )
3023        .unwrap();
3024
3025        let config = Config::from_path(&path).unwrap().current();
3026
3027        fs::remove_dir_all(path).unwrap();
3028
3029        let signing_key = &config.upload().credentials.as_ref().unwrap().signing_key;
3030        assert_eq!(
3031            signing_key.to_string(),
3032            "U3LSQM5NorvgnoYHW_aZpc_43nuuh3lhs3zjjcBwaks"
3033        );
3034    }
3035
3036    #[test]
3037    fn test_emit_outcomes() {
3038        for (serialized, deserialized) in &[
3039            ("true", EmitOutcomes::AsOutcomes),
3040            ("false", EmitOutcomes::None),
3041            ("\"as_client_reports\"", EmitOutcomes::AsClientReports),
3042        ] {
3043            let value: EmitOutcomes = serde_json::from_str(serialized).unwrap();
3044            assert_eq!(value, *deserialized);
3045            assert_eq!(serde_json::to_string(&value).unwrap(), *serialized);
3046        }
3047    }
3048
3049    #[test]
3050    fn test_emit_outcomes_invalid() {
3051        assert!(serde_json::from_str::<EmitOutcomes>("asdf").is_err());
3052    }
3053}