relay_server/

service.rs

use std::convert::Infallible;
use std::sync::Arc;
use std::time::Duration;

use crate::metrics::MetricOutcomes;
use crate::services::autoscaling::{AutoscalingMetricService, AutoscalingMetrics};
use crate::services::buffer::{
    ObservableEnvelopeBuffer, PartitionedEnvelopeBuffer, ProjectKeyPair,
};
use crate::services::cogs::{CogsService, CogsServiceRecorder};
use crate::services::config_reload::ConfigReloadService;
use crate::services::global_config::{
    GlobalConfigHandle, GlobalConfigManager, GlobalConfigService,
};
use crate::services::health_check::{HealthCheck, HealthCheckService};
use crate::services::metrics::RouterService;
#[cfg(feature = "processing")]
use crate::services::objectstore::{self, Objectstore};
use crate::services::outcome::{
    ClientReportOutcomeProducerService, NullOutcomeProducerService, OutcomeProducerService,
    TrackOutcome,
};
use crate::services::processor::{self, EnvelopeProcessor};
use crate::services::projects::cache::{ProjectCacheHandle, ProjectCacheService};
use crate::services::projects::source::ProjectSource;
use crate::services::proxy_processor::{ProxyAddrs, ProxyProcessorService};
use crate::services::relays::{RelayCache, RelayCacheService};
use crate::services::stats::RelayStats;
#[cfg(feature = "processing")]
use crate::services::store::{self, create_kafka_client};
use crate::services::upload::{self, Upload};
use crate::services::upstream::{UpstreamRelay, UpstreamRelayService};
use crate::utils::{MemoryChecker, MemoryStat, ThreadKind};
#[cfg(feature = "processing")]
use anyhow::Context;
use anyhow::Result;
use axum::extract::FromRequestParts;
use axum::http::request::Parts;
use relay_cogs::Cogs;
use relay_config::{Config, ConfigSnapshot, EmitOutcomes, RelayMode};
#[cfg(feature = "processing")]
use relay_config::{RedisConfigRef, RedisConfigsRef};
#[cfg(feature = "processing")]
use relay_redis::AsyncRedisClient;
#[cfg(feature = "processing")]
use relay_redis::redis::Script;
#[cfg(feature = "processing")]
use relay_redis::{RedisClients, RedisError, RedisScripts};
use relay_system::{Addr, AssertEmpty, Service, ServiceSpawn, ServiceSpawnExt as _, channel};
use relay_threading::ServicePool;

/// Indicates the type of failure of the server.
#[derive(Debug, thiserror::Error)]
pub enum ServiceError {
    /// GeoIp construction failed.
    #[error("could not load the Geoip Db")]
    GeoIp,

    /// Initializing the Kafka producer failed.
    #[cfg(feature = "processing")]
    #[error("could not initialize kafka producer: {0}")]
    Kafka(String),

    /// Initializing the Redis client failed.
    #[cfg(feature = "processing")]
    #[error("could not initialize redis client during startup")]
    Redis,
}

#[derive(Clone, Debug)]
pub struct Registry {
    pub cogs: Cogs,
    pub health_check: Addr<HealthCheck>,
    pub outcome_aggregator: Addr<TrackOutcome>,
    pub processor: Addr<EnvelopeProcessor>,
    pub relay_cache: Addr<RelayCache>,
    pub global_config: Addr<GlobalConfigManager>,
    pub upstream_relay: Addr<UpstreamRelay>,
    pub envelope_buffer: Arc<PartitionedEnvelopeBuffer>,
    pub project_cache_handle: ProjectCacheHandle,
    pub autoscaling: Option<Addr<AutoscalingMetrics>>,
    #[cfg(feature = "processing")]
    pub objectstore: Option<Addr<Objectstore>>,
    pub upload: Addr<Upload>,
    pub global_config_handle: GlobalConfigHandle,
}

/// Constructs a Tokio [`relay_system::Runtime`] configured for running [services](relay_system::Service).
pub fn create_runtime(name: &'static str, threads: usize) -> relay_system::Runtime {
    relay_system::Runtime::builder(name)
        .worker_threads(threads)
        // Relay uses `spawn_blocking` only for Redis connections within the project
        // cache, those should never exceed 100 concurrent connections
        // (limited by connection pool).
        //
        // Relay also does not use other blocking operations from Tokio which require
        // this pool, no usage of `tokio::fs` and `tokio::io::{Stdin, Stdout, Stderr}`.
        //
        // We limit the maximum amount of threads here, we've seen that Tokio
        // expands this pool very very aggressively and basically never shrinks it
        // which leads to a massive resource waste.
        .max_blocking_threads(150)
        // We also lower down the default (10s) keep alive timeout for blocking
        // threads to encourage the runtime to not keep too many idle blocking threads
        // around.
        .thread_keep_alive(Duration::from_secs(1))
        .build()
}

fn create_processor_pool(config: &ConfigSnapshot) -> Result<ServicePool> {
    // Adjust thread count for small cpu counts to not have too many idle cores
    // and distribute workload better.
    let thread_count = match config.cpu_concurrency() {
        conc @ 0..=2 => conc.max(1),
        conc @ 3..=4 => conc - 1,
        conc => conc - 2,
    };
    relay_log::info!("starting {thread_count} envelope processing workers");

    let pool = crate::utils::ThreadPoolBuilder::new("processor", tokio::runtime::Handle::current())
        .num_threads(thread_count)
        .max_concurrency(config.pool_concurrency())
        .thread_kind(ThreadKind::Worker)
        .build()?;

    Ok(pool)
}

#[cfg(feature = "processing")]
fn create_store_pool(config: &ConfigSnapshot) -> Result<ServicePool> {
    // Spawn a store worker for every 12 threads in the processor pool.
    // This ratio was found empirically and may need adjustments in the future.
    //
    // Ideally in the future the store will be single threaded again, after we move
    // all the heavy processing (de- and re-serialization) into the processor.
    let thread_count = config.cpu_concurrency().div_ceil(8);
    relay_log::info!("starting {thread_count} store workers");

    let pool = crate::utils::ThreadPoolBuilder::new("store", tokio::runtime::Handle::current())
        .num_threads(thread_count)
        .max_concurrency(config.pool_concurrency())
        .build()?;

    Ok(pool)
}

#[derive(Debug)]
struct StateInner {
    config: Arc<Config>,
    memory_checker: MemoryChecker,
    registry: Registry,
    _daemons: Vec<Addr<()>>, // services that receive no messages but need to stay alive.
    _asserter: AssertEmpty<TrackOutcome>,
}

/// Server state.
#[derive(Clone, Debug)]
pub struct ServiceState {
    inner: Arc<StateInner>,
}

impl ServiceState {
    /// Starts all services and returns addresses to all of them.
    pub async fn start(
        handle: &relay_system::Handle,
        services: &dyn ServiceSpawn,
        config: Arc<Config>,
    ) -> Result<Self> {
        let upstream_relay = services.start(UpstreamRelayService::new(config.clone()));
        let current_config = config.current();

        #[cfg(feature = "processing")]
        let redis_clients = current_config
            .redis()
            .filter(|_| current_config.processing_enabled())
            .map(create_redis_clients)
            .transpose()
            .context(ServiceError::Redis)?;

        // If we have Redis configured, we want to initialize all the scripts by loading them in
        // the scripts cache if not present. Our custom ConnectionLike implementation relies on this
        // initialization to work properly since it assumes that scripts are loaded across all Redis
        // instances.
        #[cfg(feature = "processing")]
        if let Some(redis_clients) = &redis_clients {
            initialize_redis_scripts_for_client(redis_clients)
                .await
                .context(ServiceError::Redis)?;
        }

        // We create an instance of `MemoryStat` which can be supplied composed with any arbitrary
        // configuration object down the line.
        let memory_stat = MemoryStat::new(current_config.memory_stat_refresh_frequency_ms());

        // Create an address for the `EnvelopeProcessor`, which can be injected into the
        // other services.
        let (envelope_processor, envelope_processor_rx) = match current_config.relay_mode() {
            RelayMode::Proxy => channel(ProxyProcessorService::name()),
            RelayMode::Managed => channel("envelope_processor"),
        };

        let (aggregator, aggregator_rx) = channel(RouterService::name());

        // Use a separate metrics router for outcomes to prevent cycles in the service graph:
        let (outcome_metrics_router, outcome_metrics_router_rx) = channel("outcome_metrics_router");
        let (dummy_outcomes, _asserter) = Addr::never(); // outcomes never generate outcomes

        let outcome_aggregator = match current_config.emit_outcomes() {
            EmitOutcomes::None => services.start(NullOutcomeProducerService::new()),
            _ => match current_config.relay_mode() {
                RelayMode::Proxy => services.start(ClientReportOutcomeProducerService::new(
                    &current_config,
                    envelope_processor.clone(),
                )),
                RelayMode::Managed => services.start(OutcomeProducerService::new(
                    Arc::clone(&config),
                    outcome_metrics_router,
                )),
            },
        };

        let (global_config, global_config_rx) =
            GlobalConfigService::new(config.clone(), upstream_relay.clone());
        let global_config_handle = global_config.handle();
        // The global config service must start before dependant services are
        // started. Messages like subscription requests to the global config
        // service fail if the service is not running.
        let global_config = services.start(global_config);

        let project_source = ProjectSource::start_in(
            services,
            Arc::clone(&config),
            upstream_relay.clone(),
            #[cfg(feature = "processing")]
            redis_clients.clone(),
        )
        .await;
        let project_cache_handle =
            ProjectCacheService::new(Arc::clone(&config), project_source).start_in(services);

        let cogs = CogsService::new(&current_config);
        let cogs = Cogs::new(CogsServiceRecorder::new(
            &current_config,
            services.start(cogs),
        ));

        let metric_outcomes = MetricOutcomes::new(outcome_aggregator.clone());

        #[cfg(feature = "processing")]
        let store_pool = create_store_pool(&current_config)?;

        #[cfg(feature = "processing")]
        let mut maybe_kafka_client = None;

        #[cfg(feature = "processing")]
        let store = current_config
            .processing_enabled()
            .then(|| {
                let kafka_client = create_kafka_client(&current_config)?;
                maybe_kafka_client = Some(kafka_client.clone());
                let service = store::create_service(
                    store_pool.clone(),
                    kafka_client.clone(),
                    config.clone(),
                    global_config_handle.clone(),
                    metric_outcomes.clone(),
                );
                Ok::<_, anyhow::Error>(services.start(service))
            })
            .transpose()?;

        #[cfg(feature = "processing")]
        let objectstore = objectstore::create_service(current_config.objectstore(), store.clone())?
            .map(|s| services.start(s));

        let envelope_buffer = PartitionedEnvelopeBuffer::create(
            current_config.spool_partitions(),
            config.clone(),
            memory_stat.clone(),
            global_config_rx.clone(),
            project_cache_handle.clone(),
            envelope_processor.clone(),
            outcome_aggregator.clone(),
            services,
        );

        let (processor_pool, aggregator_handle, autoscaling) = match current_config.relay_mode() {
            RelayMode::Proxy => {
                services.start_with(
                    ProxyProcessorService::new(
                        config.clone(),
                        project_cache_handle.clone(),
                        ProxyAddrs {
                            outcome_aggregator: outcome_aggregator.clone(),
                            upstream_relay: upstream_relay.clone(),
                        },
                    ),
                    envelope_processor_rx,
                );
                (None, None, None)
            }
            RelayMode::Managed => {
                let processor_pool = create_processor_pool(&current_config)?;
                // Share the quota cache between envelope and bucket processing.
                #[cfg(feature = "processing")]
                let rate_limiter =
                    processor::create_rate_limiter(&current_config, redis_clients.clone());

                let bucket_processor = services.start(processor::bucket::create_service(
                    processor_pool.clone(),
                    config.clone(),
                    global_config_handle.clone(),
                    project_cache_handle.clone(),
                    cogs.clone(),
                    #[cfg(feature = "processing")]
                    rate_limiter.clone(),
                    processor::bucket::Addrs {
                        outcome_aggregator: outcome_aggregator.clone(),
                        upstream_relay: upstream_relay.clone(),
                        #[cfg(feature = "processing")]
                        store_forwarder: store.clone(),
                    },
                ));

                let router = RouterService::new(
                    handle.clone(),
                    current_config.default_aggregator_config().clone(),
                    current_config.secondary_aggregator_configs().clone(),
                    Some(bucket_processor.recipient()),
                    project_cache_handle.clone(),
                );
                let router_handle = router.handle();
                services.start_with(router, aggregator_rx);

                // Start the secondary outcome pipeline to prevent cycles in the service graph.
                // +-------------------+
                // | EnvelopeProcessor |------------------------+
                // +-------------------+                        |
                //           |                                  |
                //           v                                  v
                // +-------------------+          +---------------------------+
                // | MetricsAggregator |     +--->|      OutcomeProducer      |
                // +-------------------+     |    +---------------------------+
                //           |               |                  |
                //           v               |                  v
                // +-------------------+     |    +---------------------------+
                // |  BucketProcessor  |-----+    | outcome MetricsAggregator |
                // +-------------------+     |    +---------------------------+
                //           |               |                  |
                //           v               |                  v
                // +-------------------+     |    +---------------------------+
                // |       Store       |-----+    |  outcome BucketProcessor  |
                // +-------------------+          +---------------------------+
                //                                              |
                //                                              v
                //                                +---------------------------+
                //                                |       outcome Store       |
                //                                +---------------------------+
                {
                    #[cfg(feature = "processing")]
                    let outcome_store = {
                        if current_config.processing_enabled() {
                            let (addr, rx) = channel("outcome_store");
                            let service = store::create_service(
                                store_pool.clone(),
                                maybe_kafka_client
                                    .clone()
                                    .expect("kafka client created by store"),
                                config.clone(),
                                global_config_handle.clone(),
                                MetricOutcomes::new(dummy_outcomes.clone()),
                            );
                            services.start_with(service, rx);
                            Some(addr)
                        } else {
                            None
                        }
                    };

                    let (outcome_bucket_processor, outcome_bucket_rx) =
                        channel("outcome_bucket_processor");
                    let outcome_bucket_service = processor::bucket::create_service(
                        processor_pool.clone(),
                        config.clone(),
                        global_config_handle.clone(),
                        project_cache_handle.clone(),
                        cogs.clone(),
                        #[cfg(feature = "processing")]
                        rate_limiter.clone(),
                        processor::bucket::Addrs {
                            outcome_aggregator: dummy_outcomes,
                            upstream_relay: upstream_relay.clone(),
                            #[cfg(feature = "processing")]
                            store_forwarder: outcome_store,
                        },
                    );
                    services.start_with(outcome_bucket_service, outcome_bucket_rx);

                    // Give the outcome metrics router the full config, because the user
                    // may have configured specifics for outcome metrics.
                    let outcome_metrics_router = RouterService::prefixed(
                        handle.clone(),
                        "outcomes_",
                        current_config.default_aggregator_config().clone(),
                        current_config.secondary_aggregator_configs().clone(),
                        Some(outcome_bucket_processor.recipient()),
                        project_cache_handle.clone(),
                    );
                    services.start_with(outcome_metrics_router, outcome_metrics_router_rx);
                }

                services.start_with(
                    processor::envelope::create_service(
                        processor_pool.clone(),
                        config.clone(),
                        global_config_handle.clone(),
                        project_cache_handle.clone(),
                        cogs.clone(),
                        #[cfg(feature = "processing")]
                        rate_limiter,
                        processor::envelope::Addrs {
                            outcome_aggregator: outcome_aggregator.clone(),
                            upstream_relay: upstream_relay.clone(),
                            #[cfg(feature = "processing")]
                            objectstore: objectstore.clone(),
                            #[cfg(feature = "processing")]
                            store_forwarder: store,
                            aggregator: aggregator.clone(),
                        },
                        metric_outcomes.clone(),
                    ),
                    envelope_processor_rx,
                );

                let autoscaling = services.start(AutoscalingMetricService::new(
                    memory_stat.clone(),
                    envelope_buffer.clone(),
                    handle.clone(),
                    processor_pool.clone(),
                ));

                (Some(processor_pool), Some(router_handle), Some(autoscaling))
            }
        };

        let health_check = services.start(HealthCheckService::new(
            config.clone(),
            MemoryChecker::new(memory_stat.clone(), config.clone()),
            aggregator_handle,
            upstream_relay.clone(),
            envelope_buffer.clone(),
        ));

        let stats = services.start(RelayStats::new(
            config.clone(),
            handle.clone(),
            upstream_relay.clone(),
            #[cfg(feature = "processing")]
            redis_clients.clone(),
            processor_pool,
            #[cfg(feature = "processing")]
            store_pool,
        ));

        let relay_cache = services.start(RelayCacheService::new(
            config.clone(),
            upstream_relay.clone(),
        ));

        let upload = services.start(upload::create_service(
            &config,
            &upstream_relay,
            #[cfg(feature = "processing")]
            &objectstore,
        ));

        let config_reload = services.start(ConfigReloadService::new(
            config.clone(),
            #[cfg(feature = "processing")]
            maybe_kafka_client,
        ));

        let registry = Registry {
            cogs,
            processor: envelope_processor,
            health_check,
            outcome_aggregator,
            relay_cache,
            global_config,
            project_cache_handle,
            upstream_relay,
            envelope_buffer,
            autoscaling,
            #[cfg(feature = "processing")]
            objectstore,
            upload,
            global_config_handle,
        };

        let state = StateInner {
            config: config.clone(),
            memory_checker: MemoryChecker::new(memory_stat, config.clone()),
            registry,
            _daemons: vec![stats, config_reload],
            _asserter,
        };

        Ok(ServiceState {
            inner: Arc::new(state),
        })
    }

    /// Returns a snapshot of the Relay configuration.
    pub fn config(&self) -> ConfigSnapshot {
        self.inner.config.current()
    }

    /// Returns a reference to the [`Cogs`] tracking.
    pub fn cogs(&self) -> &Cogs {
        &self.inner.registry.cogs
    }

    /// Returns a reference to the [`MemoryChecker`] which is a [`Config`] aware wrapper on the
    /// [`MemoryStat`] which gives utility methods to determine whether memory usage is above
    /// thresholds set in the [`Config`].
    pub fn memory_checker(&self) -> &MemoryChecker {
        &self.inner.memory_checker
    }

    pub fn autoscaling(&self) -> Option<&Addr<AutoscalingMetrics>> {
        self.inner.registry.autoscaling.as_ref()
    }

    /// Returns `true` if all envelope buffer partitions are empty, in memory and on disk.
    pub fn envelope_buffer_is_empty(&self) -> bool {
        self.inner.registry.envelope_buffer.is_empty()
    }

    /// Returns the V2 envelope buffer, if present.
    pub fn envelope_buffer(&self, project_key_pair: ProjectKeyPair) -> &ObservableEnvelopeBuffer {
        self.inner.registry.envelope_buffer.buffer(project_key_pair)
    }

    /// Returns a [`ProjectCacheHandle`].
    pub fn project_cache_handle(&self) -> &ProjectCacheHandle {
        &self.inner.registry.project_cache_handle
    }

    /// Returns the address of the [`RelayCache`] service.
    pub fn relay_cache(&self) -> &Addr<RelayCache> {
        &self.inner.registry.relay_cache
    }

    /// Returns the address of the [`HealthCheck`] service.
    pub fn health_check(&self) -> &Addr<HealthCheck> {
        &self.inner.registry.health_check
    }

    /// Returns the address of the [`UpstreamRelay`] service.
    pub fn upstream_relay(&self) -> &Addr<UpstreamRelay> {
        &self.inner.registry.upstream_relay
    }

    /// Returns the address of the [`EnvelopeProcessor`] service.
    pub fn processor(&self) -> &Addr<EnvelopeProcessor> {
        &self.inner.registry.processor
    }

    /// Returns the address of the [`GlobalConfigService`] service.
    pub fn global_config(&self) -> &Addr<GlobalConfigManager> {
        &self.inner.registry.global_config
    }

    /// Returns the address of the [`TrackOutcome`] service.
    pub fn outcome_aggregator(&self) -> &Addr<TrackOutcome> {
        &self.inner.registry.outcome_aggregator
    }

    #[cfg(feature = "processing")]
    /// Returns the address of the [`Objectstore`] service.
    pub fn objectstore(&self) -> Option<&Addr<Objectstore>> {
        self.inner.registry.objectstore.as_ref()
    }

    /// Returns the address of the [`Upload`] service.
    pub fn upload(&self) -> &Addr<Upload> {
        &self.inner.registry.upload
    }

    pub fn global_config_handle(&self) -> &GlobalConfigHandle {
        &self.inner.registry.global_config_handle
    }
}

/// Creates Redis clients from the given `configs`.
///
/// If `configs` is [`Unified`](RedisConfigsRef::Unified), one client is created and then cloned
/// for project configs, cardinality, and quotas, meaning that they really use the same client.
///
/// If it is [`Individual`](RedisConfigsRef::Individual), an actual separate client
/// is created for each use case.
#[cfg(feature = "processing")]
pub fn create_redis_clients(configs: RedisConfigsRef<'_>) -> Result<RedisClients, RedisError> {
    const PROJECT_CONFIG_REDIS_CLIENT: &str = "projectconfig";
    const QUOTA_REDIS_CLIENT: &str = "quotas";
    const UNIFIED_REDIS_CLIENT: &str = "unified";

    match configs {
        RedisConfigsRef::Unified(unified) => {
            let client = create_async_redis_client(UNIFIED_REDIS_CLIENT, &unified)?;

            Ok(RedisClients {
                project_configs: client.clone(),
                quotas: client,
            })
        }
        RedisConfigsRef::Individual {
            project_configs,
            quotas,
        } => {
            let project_configs =
                create_async_redis_client(PROJECT_CONFIG_REDIS_CLIENT, &project_configs)?;
            let quotas = create_async_redis_client(QUOTA_REDIS_CLIENT, &quotas)?;

            Ok(RedisClients {
                project_configs,
                quotas,
            })
        }
    }
}

#[cfg(feature = "processing")]
fn create_async_redis_client(
    name: &'static str,
    config: &RedisConfigRef<'_>,
) -> Result<AsyncRedisClient, RedisError> {
    match config {
        RedisConfigRef::Cluster {
            cluster_nodes,
            options,
        } => AsyncRedisClient::cluster(name, cluster_nodes.iter().map(|s| s.as_str()), options),
        RedisConfigRef::Single { server, options } => {
            AsyncRedisClient::single(name, server, options)
        }
    }
}

#[cfg(feature = "processing")]
async fn initialize_redis_scripts_for_client(
    redis_clients: &RedisClients,
) -> Result<(), RedisError> {
    let scripts = RedisScripts::all();

    let RedisClients {
        project_configs,
        quotas,
    } = redis_clients;

    initialize_redis_scripts(project_configs, &scripts).await?;
    initialize_redis_scripts(quotas, &scripts).await?;

    Ok(())
}

#[cfg(feature = "processing")]
async fn initialize_redis_scripts(
    client: &AsyncRedisClient,
    scripts: &[&Script],
) -> Result<(), RedisError> {
    let mut connection = client.get_connection().await?;

    for script in scripts {
        // We load on all instances without checking if the script is already in cache because of a
        // limitation in the connection implementation.
        script
            .prepare_invoke()
            .load_async(&mut connection)
            .await
            .map_err(RedisError::Redis)?;
    }

    Ok(())
}

impl FromRequestParts<Self> for ServiceState {
    type Rejection = Infallible;

    async fn from_request_parts(_: &mut Parts, state: &Self) -> Result<Self, Self::Rejection> {
        Ok(state.clone())
    }
}