relay_config/
config.rs

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