relay_server/services/server/
mod.rs

1use std::net::{SocketAddr, TcpListener};
2use std::sync::Arc;
3use std::time::Duration;
4
5use axum::ServiceExt;
6use axum::extract::Request;
7use axum::http::{HeaderName, HeaderValue, header};
8use axum_server::Handle;
9use hyper_util::rt::TokioTimer;
10use relay_config::{Config, ConfigSnapshot};
11use relay_system::{Controller, Service, Shutdown};
12use sentry::integrations::tower::{NewSentryLayer, SentryHttpLayer};
13use tokio::net::TcpSocket;
14use tower::ServiceBuilder;
15use tower_http::compression::predicate::SizeAbove;
16use tower_http::compression::{CompressionLayer, DefaultPredicate, Predicate};
17use tower_http::set_header::SetResponseHeaderLayer;
18
19use crate::constants;
20use crate::middlewares::{self, CatchPanicLayer, NormalizePath};
21use crate::service::ServiceState;
22use crate::statsd::{RelayCounters, RelayGauges};
23
24mod acceptor;
25mod concurrency_limit;
26mod io;
27
28/// Set the number of keep-alive retransmissions to be carried out before declaring that remote end
29/// is not available.
30const KEEPALIVE_RETRIES: u32 = 5;
31
32/// Set a timeout for reading client request headers. If a client does not transmit the entire
33/// header within this time, the connection is closed.
34const CLIENT_HEADER_TIMEOUT: Duration = Duration::from_secs(5);
35
36/// Only compress responses above this configured size, in bytes.
37///
38/// Small responses don't benefit from compression,
39/// additionally the envelope endpoint which returns the event id
40/// should not be compressed due to a bug in the Unity SDK.
41const COMPRESSION_MIN_SIZE: u16 = 128;
42
43/// Indicates the type of failure of the server.
44#[allow(clippy::enum_variant_names)]
45#[derive(Debug, thiserror::Error)]
46pub enum ServerError {
47    /// Binding failed.
48    #[error("bind to interface failed")]
49    BindFailed(#[from] std::io::Error),
50}
51
52type App = NormalizePath<axum::Router>;
53
54/// Build the axum application with all routes and middleware.
55fn make_app(
56    service: ServiceState,
57    f: impl FnOnce(&ConfigSnapshot) -> axum::Router<ServiceState>,
58) -> App {
59    // Build the router middleware into a single service which runs _after_ routing. Service
60    // builder order defines layers added first will be called first. This means:
61    //  - Requests go from top to bottom
62    //  - Responses go from bottom to top
63    let middleware = ServiceBuilder::new()
64        .layer(axum::middleware::from_fn_with_state(
65            service.clone(),
66            middlewares::metrics,
67        ))
68        .layer(CatchPanicLayer::custom(middlewares::handle_panic))
69        .layer(SetResponseHeaderLayer::overriding(
70            header::SERVER,
71            HeaderValue::from_static(constants::SERVER),
72        ))
73        .layer(SetResponseHeaderLayer::overriding(
74            HeaderName::from_static("cross-origin-resource-policy"),
75            HeaderValue::from_static("cross-origin"),
76        ))
77        .layer(NewSentryLayer::new_from_top())
78        .layer(SentryHttpLayer::new().enable_transaction())
79        .layer(middlewares::trace_http_layer())
80        .map_request(middlewares::remove_empty_encoding)
81        .layer(
82            CompressionLayer::new()
83                .compress_when(SizeAbove::new(COMPRESSION_MIN_SIZE).and(DefaultPredicate::new())),
84        );
85
86    let router = f(&service.config()).layer(middleware).with_state(service);
87
88    // Add middlewares that need to run _before_ routing, which need to wrap the router. This are
89    // especially middlewares that modify the request path for the router:
90    NormalizePath::new(router)
91}
92
93fn listen(addr: SocketAddr, config: &ConfigSnapshot) -> Result<TcpListener, ServerError> {
94    let socket = match addr {
95        SocketAddr::V4(_) => TcpSocket::new_v4(),
96        SocketAddr::V6(_) => TcpSocket::new_v6(),
97    }?;
98
99    #[cfg(all(unix, not(target_os = "solaris"), not(target_os = "illumos")))]
100    socket.set_reuseport(true)?;
101    socket.bind(addr)?;
102    Ok(socket.listen(config.tcp_listen_backlog())?.into_std()?)
103}
104
105async fn serve(listener: TcpListener, app: App, config: &ConfigSnapshot) -> std::io::Result<()> {
106    let handle = Handle::new();
107
108    let acceptor = self::acceptor::RelayAcceptor::new()
109        .tcp_keepalive(config.keepalive_timeout(), KEEPALIVE_RETRIES)
110        .idle_timeout(config.idle_timeout());
111
112    let mut server = axum_server::from_tcp(listener)?
113        .acceptor(acceptor)
114        .handle(handle.clone());
115
116    server
117        .http_builder()
118        .http1()
119        .timer(TokioTimer::new())
120        .half_close(true)
121        .keep_alive(true)
122        .header_read_timeout(CLIENT_HEADER_TIMEOUT)
123        .writev(true);
124
125    server
126        .http_builder()
127        .http2()
128        .timer(TokioTimer::new())
129        .keep_alive_timeout(config.keepalive_timeout());
130
131    let service = ServiceBuilder::new()
132        .layer(concurrency_limit::ConcurrencyLimitLayer::new(
133            config.max_connections(),
134        ))
135        .service(ServiceExt::<Request>::into_make_service_with_connect_info::<SocketAddr>(app));
136
137    relay_system::spawn!(emit_active_connections_metric(
138        config.metrics_periodic_interval(),
139        handle.clone(),
140    ));
141
142    relay_system::spawn!(async move {
143        let Shutdown { timeout } = Controller::shutdown_handle().notified().await;
144        relay_log::info!("Shutting down HTTP server");
145
146        match timeout {
147            Some(timeout) => handle.graceful_shutdown(Some(timeout)),
148            None => handle.shutdown(),
149        }
150    });
151
152    server.serve(service).await
153}
154
155/// HTTP server service.
156///
157/// This is the main HTTP server of Relay which hosts all [services](ServiceState) and dispatches
158/// incoming traffic to them. The server stops when a [`Shutdown`] is triggered.
159pub struct HttpServer {
160    config: Arc<Config>,
161    service: ServiceState,
162    listener: TcpListener,
163    internal_listener: Option<TcpListener>,
164}
165
166impl HttpServer {
167    pub fn new(config: Arc<Config>, service: ServiceState) -> Result<Self, ServerError> {
168        let current_config = config.current();
169
170        let listener = listen(current_config.listen_addr(), &current_config)?;
171        let internal_listener = match current_config.listen_addr_internal() {
172            Some(addr) => Some(listen(addr, &current_config)?),
173            None => None,
174        };
175
176        Ok(Self {
177            config,
178            service,
179            listener,
180            internal_listener,
181        })
182    }
183}
184
185impl Service for HttpServer {
186    type Interface = ();
187
188    async fn run(self, _rx: relay_system::Receiver<Self::Interface>) {
189        let Self {
190            config,
191            service,
192            listener,
193            internal_listener,
194        } = self;
195        let current_config = config.current();
196
197        let listen_addr = current_config.listen_addr();
198
199        relay_log::info!("spawning http server");
200        relay_log::info!("  listening on http://{listen_addr}/");
201        if let Some(internal_addr) = current_config.listen_addr_internal() {
202            relay_log::info!("  listening on http://{internal_addr}/ [internal]");
203        }
204        relay_statsd::metric!(counter(RelayCounters::ServerStarting) += 1);
205
206        if let Some(internal_listener) = internal_listener {
207            let public = make_app(service.clone(), crate::endpoints::public_routes);
208            let internal = make_app(service, crate::endpoints::internal_routes);
209
210            tokio::try_join!(
211                serve(listener, public, &current_config),
212                serve(internal_listener, internal, &current_config),
213            )
214            .map(drop)
215        } else {
216            let app = make_app(service, crate::endpoints::all_routes);
217            serve(listener, app, &current_config).await
218        }
219        .expect("axum listener to not fail")
220    }
221}
222
223async fn emit_active_connections_metric(interval: Option<Duration>, handle: Handle<SocketAddr>) {
224    let Some(mut ticker) = interval.map(tokio::time::interval) else {
225        return;
226    };
227
228    let addr = handle.listening().await.map(|addr| addr.to_string());
229
230    loop {
231        ticker.tick().await;
232        relay_statsd::metric!(
233            gauge(RelayGauges::ServerActiveConnections) = handle.connection_count() as u64,
234            addr = addr.as_deref().unwrap_or("unknown"),
235        );
236    }
237}
relay_server/services/server/mod.rs

relay_server/services/server/
mod.rs
