feat(relay): Allow configuring separate http server for internal routes (#5352)

Author: Dav1dde

CHANGELOG.md

@@ -5,6 +5,7 @@
 **Features**:
 
 - Only apply non-destructive PII rules to log bodies by default. ([#5272](https://github.com/getsentry/relay/pull/5272))
+- Allow configuring separate http server for internal routes. ([#5352](https://github.com/getsentry/relay/pull/5352))
 - Infer the client ip when set to `{{auto}}` for EAP items. ([#5304](https://github.com/getsentry/relay/pull/5304))
 - Increase the default size limit for attachments to 200MiB. ([#5310](https://github.com/getsentry/relay/pull/5310))
 - Add `sentry.origin` attribute to OTLP spans. ([#5294](https://github.com/getsentry/relay/pull/5294))

relay-config/src/config.rs

@@ -487,6 +487,24 @@ pub struct Relay {
     pub host: IpAddr,
     /// The port to bind for the unencrypted relay HTTP server.
     pub port: u16,
+    /// The host the relay should bind to (network interface) for internally exposed APIs, like
+    /// health checks.
+    ///
+    /// If not configured, internal routes are exposed on the main HTTP server.
+    ///
+    /// Note: configuring the internal http server on an address which overlaps with the main
+    /// server (e.g. main on `0.0.0.0:3000` and internal on `127.0.0.1:3000`) is a misconfiguration
+    /// resulting in approximately half of the requests sent to `127.0.0.1:3000` to fail, as the handling
+    /// http server is chosen by the operating system 'at random'.
+    ///
+    /// As a best practice you should always choose different ports to avoid this issue.
+    ///
+    /// Defaults to [`Self::host`].
+    pub internal_host: Option<IpAddr>,
+    /// The port to bind for internally exposed APIs.
+    ///
+    /// Defaults to [`Self::port`].
+    pub internal_port: Option<u16>,
     /// Optional port to bind for the encrypted relay HTTPS server.
     #[serde(skip_serializing)]
     pub tls_port: Option<u16>,
@@ -512,6 +530,8 @@ impl Default for Relay {
             upstream: "https://sentry.io/".parse().unwrap(),
             host: default_host(),
             port: 3000,
+            internal_host: None,
+            internal_port: None,
             tls_port: None,
             tls_identity_path: None,
             tls_identity_password: None,
@@ -1948,6 +1968,25 @@ impl Config {
         (self.values.relay.host, self.values.relay.port).into()
     }
 
+    /// Returns the listen address for internal APIs.
+    ///
+    /// Internal APIs are APIs which do not need to be publicly exposed,
+    /// like health checks.
+    ///
+    /// Returns `None` when there is no explicit address configured for internal APIs,
+    /// and they should instead be exposed on the main [`Self::listen_addr`].
+    pub fn listen_addr_internal(&self) -> Option<SocketAddr> {
+        match (
+            self.values.relay.internal_host,
+            self.values.relay.internal_port,
+        ) {
+            (Some(host), None) => Some((host, self.values.relay.port).into()),
+            (None, Some(port)) => Some((self.values.relay.host, port).into()),
+            (Some(host), Some(port)) => Some((host, port).into()),
+            (None, None) => None,
+        }
+    }
+
     /// Returns the TLS listen address.
     pub fn tls_listen_addr(&self) -> Option<SocketAddr> {
         if self.values.relay.tls_identity_path.is_some() {

relay-server/src/endpoints/mod.rs

@@ -34,15 +34,35 @@ use crate::service::ServiceState;
 /// Size limit for internal batch endpoints.
 const BATCH_JSON_BODY_LIMIT: usize = 50_000_000; // 50 MB
 
+/// All of Relay's routes.
+///
+/// This includes [`public_routes`] as well as [`internal_routes`].
+pub fn all_routes(config: &Config) -> Router<ServiceState> {
+    public_routes_raw(config).merge(internal_routes(config))
+}
+
+/// Relay's internal routes.
+///
+/// Routes which do not need to be exposed.
 #[rustfmt::skip]
-pub fn routes(config: &Config) -> Router<ServiceState>{
-    // Relay-internal routes pointing to /api/relay/
-    let internal_routes = Router::new()
+pub fn internal_routes(_: &Config) -> Router<ServiceState>{
+    Router::new()
         .route("/api/relay/healthcheck/{kind}/", get(health_check::handle))
         .route("/api/relay/autoscaling/", get(autoscaling::handle))
         // Fallback route, but with a name, and just on `/api/relay/*`.
-        .route("/api/relay/{*not_found}", any(statics::not_found));
+        .route("/api/relay/{*not_found}", any(statics::not_found))
+}
 
+/// Relay's public routes.
+///
+/// Routes which are public API and must be exposed.
+pub fn public_routes(config: &Config) -> Router<ServiceState> {
+    // Exclude internal routes, they must be configured separately.
+    public_routes_raw(config).route("/api/relay/{*not_found}", any(statics::not_found))
+}
+
+#[rustfmt::skip]
+fn public_routes_raw(config: &Config) -> Router<ServiceState> {
     // Sentry Web API routes pointing to /api/0/relays/
     let web_routes = Router::new()
         .route("/api/0/relays/projectconfigs/", post(project_configs::handle))
@@ -93,7 +113,7 @@ pub fn routes(config: &Config) -> Router<ServiceState>{
     // NOTE: If you add a new (non-experimental) route here, please also list it in
     // https://github.com/getsentry/sentry-docs/blob/master/docs/product/relay/operating-guidelines.mdx
 
-    Router::new().merge(internal_routes)
+    Router::new()
         .merge(web_routes)
         .merge(batch_routes)
         .merge(store_routes)

relay-server/src/services/server/mod.rs

@@ -58,7 +58,7 @@ pub enum ServerError {
 type App = NormalizePath<axum::Router>;
 
 /// Build the axum application with all routes and middleware.
-fn make_app(service: ServiceState) -> App {
+fn make_app(service: ServiceState, f: impl FnOnce(&Config) -> axum::Router<ServiceState>) -> App {
     // Build the router middleware into a single service which runs _after_ routing. Service
     // builder order defines layers added first will be called first. This means:
     //  - Requests go from top to bottom
@@ -85,25 +85,14 @@ fn make_app(service: ServiceState) -> App {
                 .compress_when(SizeAbove::new(COMPRESSION_MIN_SIZE).and(DefaultPredicate::new())),
         );
 
-    let router = crate::endpoints::routes(service.config())
-        .layer(middleware)
-        .with_state(service);
+    let router = f(service.config()).layer(middleware).with_state(service);
 
     // Add middlewares that need to run _before_ routing, which need to wrap the router. This are
     // especially middlewares that modify the request path for the router:
     NormalizePath::new(router)
 }
 
-fn listen(config: &Config) -> Result<TcpListener, ServerError> {
-    // Inform the user about a removed feature.
-    if config.tls_listen_addr().is_some()
-        || config.tls_identity_password().is_some()
-        || config.tls_identity_path().is_some()
-    {
-        return Err(ServerError::TlsNotSupported);
-    }
-
-    let addr = config.listen_addr();
+fn listen(addr: SocketAddr, config: &Config) -> Result<TcpListener, ServerError> {
     let socket = match addr {
         SocketAddr::V4(_) => TcpSocket::new_v4(),
         SocketAddr::V6(_) => TcpSocket::new_v6(),
@@ -115,7 +104,7 @@ fn listen(config: &Config) -> Result<TcpListener, ServerError> {
     Ok(socket.listen(config.tcp_listen_backlog())?.into_std()?)
 }
 
-async fn serve(listener: TcpListener, app: App, config: Arc<Config>) {
+async fn serve(listener: TcpListener, app: App, config: &Config) -> std::io::Result<()> {
     let handle = Handle::new();
 
     let acceptor = self::acceptor::RelayAcceptor::new()
@@ -162,10 +151,7 @@ async fn serve(listener: TcpListener, app: App, config: Arc<Config>) {
         }
     });
 
-    server
-        .serve(service)
-        .await
-        .expect("failed to start axum server");
+    server.serve(service).await
 }
 
 /// HTTP server service.
@@ -176,16 +162,30 @@ pub struct HttpServer {
     config: Arc<Config>,
     service: ServiceState,
     listener: TcpListener,
+    internal_listener: Option<TcpListener>,
 }
 
 impl HttpServer {
     pub fn new(config: Arc<Config>, service: ServiceState) -> Result<Self, ServerError> {
-        let listener = listen(&config)?;
+        // Inform the user about a removed feature.
+        if config.tls_listen_addr().is_some()
+            || config.tls_identity_password().is_some()
+            || config.tls_identity_path().is_some()
+        {
+            return Err(ServerError::TlsNotSupported);
+        }
+
+        let listener = listen(config.listen_addr(), &config)?;
+        let internal_listener = match config.listen_addr_internal() {
+            Some(addr) => Some(listen(addr, &config)?),
+            None => None,
+        };
 
         Ok(Self {
             config,
             service,
             listener,
+            internal_listener,
         })
     }
 }
@@ -198,14 +198,32 @@ impl Service for HttpServer {
             config,
             service,
             listener,
+            internal_listener,
         } = self;
 
+        let listen_addr = config.listen_addr();
+
         relay_log::info!("spawning http server");
-        relay_log::info!("  listening on http://{}/", config.listen_addr());
+        relay_log::info!("  listening on http://{listen_addr}/");
+        if let Some(internal_addr) = config.listen_addr_internal() {
+            relay_log::info!("  listening on http://{internal_addr}/ [internal]");
+        }
         relay_statsd::metric!(counter(RelayCounters::ServerStarting) += 1);
 
-        let app = make_app(service);
-        serve(listener, app, config).await;
+        if let Some(internal_listener) = internal_listener {
+            let public = make_app(service.clone(), crate::endpoints::public_routes);
+            let internal = make_app(service, crate::endpoints::internal_routes);
+
+            tokio::try_join!(
+                serve(listener, public, &config),
+                serve(internal_listener, internal, &config),
+            )
+            .map(drop)
+        } else {
+            let app = make_app(service, crate::endpoints::all_routes);
+            serve(listener, app, &config).await
+        }
+        .expect("axum listener to not fail")
     }
 }
 
@@ -214,10 +232,13 @@ async fn emit_active_connections_metric(interval: Option<Duration>, handle: Hand
         return;
     };
 
+    let addr = handle.listening().await.map(|addr| addr.to_string());
+
     loop {
         ticker.tick().await;
         relay_statsd::metric!(
-            gauge(RelayGauges::ServerActiveConnections) = handle.connection_count() as u64
+            gauge(RelayGauges::ServerActiveConnections) = handle.connection_count() as u64,
+            addr = addr.as_deref().unwrap_or("unknown"),
         );
     }
 }

relay/src/healthcheck.rs

@@ -18,6 +18,7 @@ pub fn healthcheck(config: &Config, matches: &ArgMatches) -> Result<()> {
     let addr = matches
         .get_one::<SocketAddr>("addr")
         .copied()
+        .or_else(|| config.listen_addr_internal())
         .unwrap_or(config.listen_addr());
 
     let client = Client::builder()

tests/integration/fixtures/__init__.py

@@ -13,8 +13,15 @@ class SentryLike:
 
     default_dsn_public_key = "31a5a894b4524f74a9a8d0e27e21ba91"
 
-    def __init__(self, server_address, upstream=None, public_key=None):
+    def __init__(
+        self,
+        server_address,
+        upstream=None,
+        public_key=None,
+        internal_server_address=None,
+    ):
         self.server_address = server_address
+        self.internal_server_address = internal_server_address or server_address
         self.upstream = upstream
         self.public_key = public_key
 
@@ -79,6 +86,10 @@ def get_dsn(self, project_id, index=0):
     def url(self):
         return "http://{}:{}".format(*self.server_address)
 
+    @property
+    def internal_url(self):
+        return "http://{}:{}".format(*self.internal_server_address)
+
     def get_auth_header(self, project_id, dsn_key_idx=0, dsn_key=None):
         if dsn_key is None:
             dsn_key = self.get_dsn_public_key(project_id, dsn_key_idx)
@@ -88,11 +99,11 @@ def get_auth_header(self, project_id, dsn_key_idx=0, dsn_key=None):
             "sentry_key={}".format(dsn_key)
         )
 
-    def _wait(self, path):
+    def _wait(self, path, *, is_internal=False):
         backoff = 0.1
         while True:
             try:
-                self.get(path).raise_for_status()
+                self.get(path, is_internal=is_internal).raise_for_status()
                 break
             except Exception:
                 time.sleep(backoff)
@@ -104,7 +115,7 @@ def wait_relay_health_check(self):
         if self._health_check_passed:
             return
 
-        self._wait("/api/relay/healthcheck/ready/")
+        self._wait("/api/relay/healthcheck/ready/", is_internal=True)
         self._health_check_passed = True
 
     def __repr__(self):
@@ -551,13 +562,14 @@ def send_check_in(self, project_id, check_in):
         envelope.add_item(Item(payload=PayloadRef(json=check_in), type="check_in"))
         self.send_envelope(project_id, envelope)
 
-    def request(self, method, path, timeout=None, **kwargs):
+    def request(self, method, path, timeout=None, is_internal=False, **kwargs):
         assert path.startswith("/")
 
         if timeout is None:
             timeout = 10
 
-        return self.session.request(method, self.url + path, timeout=timeout, **kwargs)
+        url = self.url if not is_internal else self.internal_url
+        return self.session.request(method, url + path, timeout=timeout, **kwargs)
 
     def post(self, path, **kwargs):
         return self.request("post", path, **kwargs)

tests/integration/fixtures/relay.py

@@ -33,7 +33,12 @@ def __init__(
         options,
         version,
     ):
-        super().__init__(server_address, upstream, public_key)
+        super().__init__(
+            server_address,
+            upstream,
+            public_key,
+            internal_server_address=get_internal_address(options, server_address),
+        )
 
         self.process = process
         self.relay_id = relay_id
@@ -242,3 +247,10 @@ def latest_relay_version(get_relay_binary):
     ).strip()
     _the_word_relay, version = version_str.split(" ", 1)
     return version
+
+
+def get_internal_address(options, server_address):
+    relay = (options or {}).get("relay", {})
+    host = relay.get("internal_host")
+    port = relay.get("internal_port")
+    return (host or server_address[0], port or server_address[1])