Skip to main content

relay_config/
config.rs

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