relay_config/
config.rs

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