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