diff --git a/crates/test/src/lib.rs b/crates/test/src/lib.rs index 12f40099ce..1d8ec2c952 100644 --- a/crates/test/src/lib.rs +++ b/crates/test/src/lib.rs @@ -9,6 +9,13 @@ use tokio::sync::mpsc; #[cfg(target_os = "linux")] pub mod xdp_util; +/// Upper bound for waiting on a packet that is expected to arrive. +/// +/// Generous on purpose: shared CI runners have been observed stalling every +/// test process on the machine for ~500ms while sandboxes spin up, and this is +/// only a guard against hanging forever, not a latency assertion. +pub const RECV_TIMEOUT: u64 = 10_000; + pub const MAX_WAIT: std::time::Duration = std::time::Duration::from_secs(5); #[inline] diff --git a/crates/test/tests/mesh.rs b/crates/test/tests/mesh.rs index 8ce537543a..a146054ec4 100644 --- a/crates/test/tests/mesh.rs +++ b/crates/test/tests/mesh.rs @@ -118,7 +118,11 @@ trace_test!(relay_routing, { assert_eq!( "hello", - sandbox.timeout(1000, server_rx.recv()).await.0.unwrap() + sandbox + .timeout(RECV_TIMEOUT, server_rx.recv()) + .await + .0 + .unwrap() ); tracing::info!(%token, "received packet"); @@ -217,7 +221,7 @@ trace_test!(datacenter_discovery, { loop { let rt = sandbox - .timeout(10000, proxy_delta_rx.recv()) + .timeout(RECV_TIMEOUT, proxy_delta_rx.recv()) .await .0 .unwrap(); @@ -233,7 +237,7 @@ trace_test!(datacenter_discovery, { } loop { let rt = sandbox - .timeout(10000, proxy_delta_rx.recv()) + .timeout(RECV_TIMEOUT, proxy_delta_rx.recv()) .await .0 .unwrap(); @@ -335,7 +339,7 @@ trace_test!(filter_update, { let mut updates = 0x0; while (updates & 0x11) != 0x11 { let rt = sandbox - .timeout(10000, proxy_delta_rx.recv()) + .timeout(RECV_TIMEOUT, proxy_delta_rx.recv()) .await .0 .unwrap(); @@ -356,7 +360,11 @@ trace_test!(filter_update, { tracing::info!(len = token.len(), "received packet"); assert_eq!( "hello", - sandbox.timeout(10000, server_rx.recv()).await.0.unwrap() + sandbox + .timeout(RECV_TIMEOUT, server_rx.recv()) + .await + .0 + .unwrap() ); tracing::info!(len = token.len(), "sending bad packet"); diff --git a/crates/test/tests/proxy.rs b/crates/test/tests/proxy.rs index 23c1cf5f87..bda5040778 100644 --- a/crates/test/tests/proxy.rs +++ b/crates/test/tests/proxy.rs @@ -27,14 +27,14 @@ trace_test!(server, { client.send_to(msg.as_bytes(), addr).await.unwrap(); assert_eq!( msg, - sb.timeout(100, server1_rx.recv()) + sb.timeout(RECV_TIMEOUT, server1_rx.recv()) .await .0 .expect("should get a packet") ); assert_eq!( msg, - sb.timeout(100, server2_rx.recv()) + sb.timeout(RECV_TIMEOUT, server2_rx.recv()) .await .0 .expect("should get a packet") @@ -56,7 +56,10 @@ trace_test!(client, { let msg = "hello"; tracing::debug!(%local_addr, "sending packet"); client.send_to(msg.as_bytes(), local_addr).await.unwrap(); - assert_eq!(msg, sb.timeout(100, dest_rx.recv()).await.0.unwrap(),); + assert_eq!( + msg, + sb.timeout(RECV_TIMEOUT, dest_rx.recv()).await.0.unwrap(), + ); }); trace_test!(with_filter, { @@ -81,7 +84,7 @@ trace_test!(with_filter, { client.send_to(msg.as_bytes(), local_addr).await.unwrap(); // search for the filter strings. - let result = sb.timeout(1000, rx.recv()).await.0.unwrap(); + let result = sb.timeout(RECV_TIMEOUT, rx.recv()).await.0.unwrap(); assert!(result.starts_with(&format!("{msg}:odr:[::1]:"))); }); @@ -130,6 +133,7 @@ trace_test!(uring_receiver, { sessions: quilkin::net::sessions::SessionPool::new( vec![pending_sends.0.clone()], config.dyn_cfg.cached_filter_chain().unwrap(), + config.dyn_cfg.clusters().cloned(), usize::MAX, backend, 4, @@ -149,7 +153,10 @@ trace_test!(uring_receiver, { let msg = "hello-downstream"; tracing::debug!("sending packet"); socket.send_to(msg.as_bytes(), addr).await.unwrap(); - assert_eq!(msg, sb.timeout(200, packet_rx.recv()).await.0.unwrap()); + assert_eq!( + msg, + sb.timeout(RECV_TIMEOUT, packet_rx.recv()).await.0.unwrap() + ); }); trace_test!( @@ -196,6 +203,7 @@ trace_test!( let sessions = net::SessionPool::new( pending_sends.iter().map(|ps| ps.0.clone()).collect(), config.dyn_cfg.cached_filter_chain().unwrap(), + config.dyn_cfg.clusters().cloned(), usize::MAX, backend, 64, @@ -224,7 +232,7 @@ trace_test!( for _ in 0..WORKER_COUNT { assert_eq!( msg, - sb.timeout(20, packet_rx.recv()) + sb.timeout(RECV_TIMEOUT, packet_rx.recv()) .await .0 .expect("should receive a packet") diff --git a/crates/test/tests/uring.rs b/crates/test/tests/uring.rs index db20f26af6..bf1223d32d 100644 --- a/crates/test/tests/uring.rs +++ b/crates/test/tests/uring.rs @@ -34,21 +34,21 @@ trace_test!(fan_out, { client.send_to(msg.as_bytes(), addr).await.unwrap(); assert_eq!( msg, - sb.timeout(100, server1_rx.recv()) + sb.timeout(RECV_TIMEOUT, server1_rx.recv()) .await .0 .expect("should get a packet") ); assert_eq!( msg, - sb.timeout(100, server2_rx.recv()) + sb.timeout(RECV_TIMEOUT, server2_rx.recv()) .await .0 .expect("should get a packet") ); assert_eq!( msg, - sb.timeout(100, server3_rx.recv()) + sb.timeout(RECV_TIMEOUT, server3_rx.recv()) .await .0 .expect("should get a packet") @@ -83,7 +83,7 @@ trace_test!(refreshes_recv_ring, { client.send_to(&i.to_ne_bytes(), addr).await.unwrap(); let (len, _addr) = sb - .timeout(100, client.recv_from(&mut buf)) + .timeout(RECV_TIMEOUT, client.recv_from(&mut buf)) .await .0 .expect("should have received packet"); @@ -98,7 +98,7 @@ trace_test!(refreshes_recv_ring, { .expect("failed to send debug request"); let mut buf = [0u8; 8]; let (len, _addr) = sb - .timeout(100, client.recv_from(&mut buf)) + .timeout(RECV_TIMEOUT, client.recv_from(&mut buf)) .await .0 .expect("should have debug response packet"); @@ -164,6 +164,7 @@ trace_test!(requeues_recv, { sessions: quilkin::net::sessions::SessionPool::new( vec![pending_sends.0.clone()], config.dyn_cfg.cached_filter_chain().unwrap(), + config.dyn_cfg.clusters().cloned(), usize::MAX, backend, 4, diff --git a/crates/xds/src/locality.rs b/crates/xds/src/locality.rs index 80687b7fc9..e670af1177 100644 --- a/crates/xds/src/locality.rs +++ b/crates/xds/src/locality.rs @@ -75,6 +75,15 @@ impl Locality { self.buffer.as_str().to_owned() } + /// The full colon separated locality, borrowed. + /// + /// Unlike [`Self::colon_separated_string`] this doesn't allocate, which + /// matters where it's used as a metric label on the packet path. + #[inline] + pub fn as_str(&self) -> &str { + self.buffer.as_str() + } + #[inline] pub fn region(&self) -> &str { &self.buffer[..self.region] diff --git a/docs/src/deployment/metrics.md b/docs/src/deployment/metrics.md index e1d6e6ba42..58b34c634e 100644 --- a/docs/src/deployment/metrics.md +++ b/docs/src/deployment/metrics.md @@ -5,8 +5,9 @@ The following are metrics that Quilkin provides while in Proxy Mode. # ASN Maxmind Information If Quilkin is provided a a MaxmindDB GeoIP database, Quilkin will log the -following information in the `maxmind information` log, as well as populate -the following fields in any metrics with matching labels. +following information in the `maxmind information` log. Only `country_code`, on +`quilkin_session_active`, and `asn`, on the connection quality metrics below, are +exported as labels; the rest are too high in cardinality to be. | Field | Description | |-----------------|-----------------------------------------------| @@ -31,12 +32,24 @@ The proxy exposes the following general metrics: * `read`: when the proxy receives data from a downstream connection on the listening port. * `write`: when the proxy sends data to a downstream connection via the listening port. -* `quilkin_packets_dropped_total{reason, asn, ip_prefix}` (Counter) - - The total number of packets (not associated with any session) that were dropped by proxy. - Not that packets reflected by this metric were dropped at an earlier stage before they were associated with any session. For session based metrics, see the list of [session metrics][session-metrics] instead. - * `reason = NoConfiguredEndpoints` - * `NoConfiguredEndpoints`: No upstream endpoints were available to send the packet to. This can occur e.g if the endpoints cluster was scaled down to zero and the proxy is configured via a control plane. +* `quilkin_packets_dropped_total{event, reason, filter, destination}` (Counter) + + The total number of packets that were dropped by the proxy. + * The `reason` label is a closed set, so a breakdown built on it survives a + filter being renamed or an `errno` producing different text: + * `no_endpoint_match`: no upstream endpoint was available, or none matched the packet's routing token. + * `filter_drop`: a filter chose to drop the packet, ie the chain worked as configured. + * `filter_error`: a filter failed to process the packet. + * `socket_error`: the socket refused the packet, or the packet couldn't be built for it. + * `queue_full`: a send or receive queue was full. + * `invalid_packet`: the packet couldn't be parsed as a datagram Quilkin handles. The specific parse failure is logged rather than labelled. + * `session_limit`: the session limit was reached, so no session could be established. + * `internal`: Quilkin lost track of state it needed to forward the packet. + * The `filter` label is the filter responsible, and is empty when the drop + wasn't a filter's decision. + * The `destination` label is described under + [`quilkin_bytes_total`](#general-metrics), and is empty for packets dropped + before they were routed. * `quilkin_cluster_active` @@ -47,27 +60,46 @@ The proxy exposes the following general metrics: The number of currently active upstream endpoints. Note that this tracks the number of endpoints that the proxy knows of rather than those that it is connected to (see [Session Metrics][session-metrics] instead for those) -* `quilkin_bytes_total{event, asn, ip_prefix}` +* `quilkin_bytes_total{event, destination}` The total number of bytes sent or received * The `event` label is either: * `read`: when the proxy receives data from a downstream connection on the listening port. * `write`: when the proxy sends data to a downstream connection via the listening port. + * The `destination` label is the cluster the packet was routed to, so traffic + can be attributed to a gameserver fleet rather than only counted in aggregate. + It carries the same values as `quilkin_active_endpoints`, so the two can be + joined. + + The cluster comes from the routing decision itself, not from looking the + address up afterwards, so it is the cluster the packet actually went through + even when the same endpoint is configured in more than one. It is empty when + the destination didn't come from a cluster with a locality, which includes + destinations a filter decoded from the packet rather than selecting from the + cluster map. Note that `event=write` traffic carries the cluster of the + gameserver that sent it, not of the client it is going to. -* `quilkin_packets_total{event, asn, ip_prefix}` +* `quilkin_packets_total{event, destination}` The total number of packets sent or recieved. - * The `event` label is either: - * `read`: when the proxy receives data from a downstream connection on the listening port. - * `write`: when the proxy sends data to a downstream connection via the listening port. + * The labels are the same as [`quilkin_bytes_total`](#general-metrics). + +* `quilkin_packet_jitter{event}` -* `quilkin_packet_jitter{event, asn, ip_prefix}` + The time between packets arriving at an I/O loop (in nanoseconds). This covers + every session the loop serves, so it is a whole-proxy figure; for the + distribution across players use `quilkin_session_jitter_seconds`. - The time between receiving new packets (in nanoseconds). + The series stops being exported when no packet arrived during an aggregation + interval, rather than continuing to publish the last value the proxy saw. -* `quilkin_errors_total{event, asn, ip_prefix}` +* `quilkin_errors_total{event, reason}` The total number of errors encountered while reading a packet from the upstream endpoint. + * The `reason` label is a closed set, replacing the previous free-text `display` + label. For I/O errors it names the `errno`, eg `invalid_input`, + `no_buffer_space` or `message_too_long`, rather than interpolating the + platform's error string. * `quilkin_game_traffic_tasks` @@ -97,6 +129,88 @@ The proxy exposes the following metrics around sessions: The total number of sessions that have been created. +* `quilkin_sessions_closed_total{reason}` (Counter) + + The total number of sessions that have ended, by why they ended. A fall in + session count otherwise looks the same whether players left or their endpoints + vanished. + * `idle_timeout`: no traffic within the session TTL. UDP has no close, so this + is what a player leaving normally looks like. + * `endpoint_gone`: the endpoint the session was routed to is no longer in the + cluster map. Only reported for destinations that were in the cluster map when + the session was created, so an endpoint configured by name is never + misattributed here. + * `shutdown`: the proxy is shutting down. + +## Connection Quality Metrics + +A player's jitter and their ISP are per-player facts, so neither can be a metric +label: concurrent sessions and the thousands of ASNs seen in a day of traffic both +blow up cardinality. Instead the proxy tracks quality per session internally and +periodically exports a projection whose series count is bounded by configuration +rather than by traffic. See the `--service.udp.metrics.*` options for the +tunables. + +Note the division of labour. Whether an individual *session* is having a bad time +is a judgement the proxy makes, because it is the only thing holding that +session's packet timing. Whether an *ISP* is having a bad time is left to the +consumer of these metrics: a proxy carries on the order of a hundred concurrent +sessions spread over thousands of ASNs, so no single one sees enough of any ASN to +threshold on. The proxy exports the numerator and the denominator per ASN and +expects them to be summed across the fleet before any conclusion is drawn. + +Interarrival jitter is measured against a monotonic clock, and gaps longer than a +second are treated as the stream restarting rather than as jitter, so a player +pausing doesn't register as a player with a bad connection. + +Sessions the proxy saw no traffic for during an interval are counted, but +contribute no quality judgement. + +* `quilkin_session_jitter_seconds` (Histogram) + + The distribution of per-session interarrival jitter of downstream packets, as + the RFC 3550 estimator. A histogram rather than a mean, because a mean of 0.5 ms + is compatible with a few percent of players at 80 ms, and those are the players + worth knowing about — `histogram_quantile` over this answers "what does the 99th + percentile player at this proxy see". + + Only jitter is exported per session. Loss and round-trip time on the client's + leg need either packet sequence numbers or a client-side timestamp, and the + proxy has neither. + +* `quilkin_sessions_active_by_asn{asn}` (Gauge) + + Active sessions by the client's ASN, for the largest + `--service.udp.metrics.top-asns` ASNs at this proxy, and the denominator for the + metric below. + + Sessions belonging to any other ASN are counted under `asn="other"`, so the + breakdown always sums to the session total. Sessions whose client IP resolved to + no ASN are counted under `asn="unknown"`. Setting `--top-asns` to 0 disables + per-ASN reporting entirely. + +* `quilkin_client_sessions_degraded{asn, reason}` (Gauge) + + The number of sessions breaching a quality threshold, by the client's ASN. Only + ASNs currently breaching are exported, so a healthy proxy publishes nothing + here and the series count follows the number of ISPs actually in trouble. + * `reason = jitter`: sessions at or above `--service.udp.metrics.jitter-threshold-ms`. + + This is a count, not a verdict. For the affected share of an ISP, divide by + `quilkin_sessions_active_by_asn` — summing both across proxies first, since one + proxy's view of a single ASN is too small a sample to threshold on: + + ```promql + sum by (asn) (quilkin_client_sessions_degraded{reason="jitter"}) + / sum by (asn) (quilkin_sessions_active_by_asn) + ``` + +* `quilkin_client_sessions_degraded_total{reason}` (Counter) + + The total number of times a session was observed breaching a threshold, for + alerting on a rate without depending on a share the proxy would have to pick a + cut-off for. + ## Filter Metrics Quilkin's filters use a set of generic metric keys, to make it easier to build visualisations that can account for a dynamic set of filters that can be added, removed, or updated at runtime with different configurations. All of diff --git a/src/filters/chain.rs b/src/filters/chain.rs index 8e686a1c05..5521b917da 100644 --- a/src/filters/chain.rs +++ b/src/filters/chain.rs @@ -296,8 +296,7 @@ impl Filter for FilterChain { // has rejected, and the destinations is empty, we passthrough to all. // Which mimics the old behaviour while avoid clones in most cases. if ctx.destinations.is_empty() { - ctx.destinations - .extend(ctx.endpoints.endpoints().into_iter().map(|ep| ep.address)); + ctx.endpoints.destinations(ctx.destinations); } Ok(()) @@ -388,7 +387,18 @@ mod tests { config.filters.read(&mut context).unwrap(); let expected = endpoints_fixture.clone(); - assert_eq!(&*expected.endpoints(), &*context.destinations); + assert_eq!( + expected + .endpoints() + .into_iter() + .map(|ep| ep.address) + .collect::>(), + context + .destinations + .iter() + .map(|d| d.address.clone()) + .collect::>() + ); assert_eq!( "hello:odr:127.0.0.1:70", std::str::from_utf8(&context.contents).unwrap() @@ -445,7 +455,14 @@ mod tests { (context.contents, context.metadata) }; let expected = endpoints_fixture.clone(); - assert_eq!(expected.endpoints(), dest); + assert_eq!( + expected + .endpoints() + .into_iter() + .map(|ep| ep.address) + .collect::>(), + dest.iter().map(|d| d.address.clone()).collect::>() + ); assert_eq!(b"hello:odr:127.0.0.1:70:odr:127.0.0.1:70", &*contents); assert_eq!( "receive:receive", diff --git a/src/filters/decryptor.rs b/src/filters/decryptor.rs index bc0defdabb..7bc33bd774 100644 --- a/src/filters/decryptor.rs +++ b/src/filters/decryptor.rs @@ -109,8 +109,12 @@ impl Filter for Decryptor { }; self.decode_chacha20(nonce, edata); - ctx.destinations - .push(Self::decode_destination(edata).into()); + // The address came out of the packet, not out of the + // cluster map, so there is no cluster it was routed from + ctx.destinations.push(crate::net::Destination::new( + Self::decode_destination(edata).into(), + None, + )); Ok(()) } } @@ -286,7 +290,12 @@ mod tests { filter.read(&mut ctx).unwrap(); assert_eq!( std::net::SocketAddr::from(expected), - ctx.destinations.pop().unwrap().to_socket_addr().unwrap() + ctx.destinations + .pop() + .unwrap() + .address + .to_socket_addr() + .unwrap() ); } @@ -312,7 +321,12 @@ mod tests { filter.read(&mut ctx).unwrap(); assert_eq!( std::net::SocketAddr::from(expected), - ctx.destinations.pop().unwrap().to_socket_addr().unwrap() + ctx.destinations + .pop() + .unwrap() + .address + .to_socket_addr() + .unwrap() ); } } diff --git a/src/filters/error.rs b/src/filters/error.rs index 1ac867f31c..a7a19f334c 100644 --- a/src/filters/error.rs +++ b/src/filters/error.rs @@ -48,6 +48,44 @@ impl FilterError { Self::Custom(custom) => custom, } } + + /// The bounded drop reason this error corresponds to. + /// + /// A filter dropping a packet by design is separated from a filter failing: + /// the first is the chain working as configured, the second is a fault. + #[inline] + pub fn drop_reason(&self) -> crate::metrics::DropReason { + use crate::metrics::DropReason; + + match self { + Self::FirewallDenied | Self::Dropped | Self::RateLimitExceeded => { + DropReason::FilterDrop + } + Self::TokenRouter(tr) => tr.drop_reason(), + Self::NoValueCaptured | Self::MatchNoMetadata | Self::Custom(_) => { + DropReason::FilterError + } + Self::Io(..) => DropReason::SocketError, + } + } + + /// The filter that produced this error, empty when it didn't come from a + /// specific filter. + /// + /// Kept apart from [`Self::drop_reason`] so renaming or reconfiguring a + /// filter doesn't change the reason vocabulary. + #[inline] + pub fn filter_name(&self) -> &'static str { + match self { + Self::NoValueCaptured => "capture", + Self::TokenRouter(..) => "token_router", + Self::FirewallDenied => "firewall", + Self::MatchNoMetadata => "match", + Self::Dropped => "drop", + Self::RateLimitExceeded => "rate_limit", + Self::Io(..) | Self::Custom(..) => "", + } + } } impl std::error::Error for FilterError {} diff --git a/src/filters/load_balancer.rs b/src/filters/load_balancer.rs index 59ffc979a5..2507112ccd 100644 --- a/src/filters/load_balancer.rs +++ b/src/filters/load_balancer.rs @@ -86,7 +86,7 @@ mod tests { filter.read(&mut context).unwrap(); } - dest + dest.into_iter().map(|d| d.address).collect() } #[tokio::test] diff --git a/src/filters/load_balancer/endpoint_chooser.rs b/src/filters/load_balancer/endpoint_chooser.rs index 2476dafe24..4fa3c436d3 100644 --- a/src/filters/load_balancer/endpoint_chooser.rs +++ b/src/filters/load_balancer/endpoint_chooser.rs @@ -23,14 +23,14 @@ use std::{ hash::{Hash, Hasher}, }; -use crate::net::{ClusterMap, EndpointAddress}; +use crate::net::{ClusterMap, Destination, EndpointAddress}; /// Chooses from a set of endpoints that a proxy is connected to. pub trait EndpointChooser: Send + Sync { /// Asks for the next endpoint(s) to use. fn choose_endpoints( &self, - destinations: &mut Vec, + destinations: &mut Vec, endpoints: &ClusterMap, src: &EndpointAddress, ); @@ -52,7 +52,7 @@ impl RoundRobinEndpointChooser { impl EndpointChooser for RoundRobinEndpointChooser { fn choose_endpoints( &self, - destinations: &mut Vec, + destinations: &mut Vec, endpoints: &ClusterMap, _src: &EndpointAddress, ) { @@ -60,10 +60,8 @@ impl EndpointChooser for RoundRobinEndpointChooser { // Note: The index is guaranteed to be in range. destinations.push( endpoints - .nth_endpoint(count % endpoints.num_of_endpoints()) - .unwrap() - .address - .clone(), + .nth_destination(count % endpoints.num_of_endpoints()) + .unwrap(), ); } } @@ -74,13 +72,13 @@ pub struct RandomEndpointChooser; impl EndpointChooser for RandomEndpointChooser { fn choose_endpoints( &self, - destinations: &mut Vec, + destinations: &mut Vec, endpoints: &ClusterMap, _src: &EndpointAddress, ) { // The index is guaranteed to be in range. let index = rand::rng().random_range(0..endpoints.num_of_endpoints()); - destinations.push(endpoints.nth_endpoint(index).unwrap().address.clone()); + destinations.push(endpoints.nth_destination(index).unwrap()); } } @@ -90,7 +88,7 @@ pub struct HashEndpointChooser; impl EndpointChooser for HashEndpointChooser { fn choose_endpoints( &self, - destinations: &mut Vec, + destinations: &mut Vec, endpoints: &ClusterMap, src: &EndpointAddress, ) { @@ -98,10 +96,8 @@ impl EndpointChooser for HashEndpointChooser { src.hash(&mut hasher); destinations.push( endpoints - .nth_endpoint(hasher.finish() as usize % endpoints.num_of_endpoints()) - .unwrap() - .address - .clone(), + .nth_destination(hasher.finish() as usize % endpoints.num_of_endpoints()) + .unwrap(), ); } } diff --git a/src/filters/read.rs b/src/filters/read.rs index b17ed27bab..6a41d38922 100644 --- a/src/filters/read.rs +++ b/src/filters/read.rs @@ -17,7 +17,7 @@ #[cfg(doc)] use crate::filters::Filter; use crate::net::{ - ClusterMap, + ClusterMap, Destination, endpoint::{EndpointAddress, metadata::DynamicMetadata}, }; @@ -25,8 +25,9 @@ use crate::net::{ pub struct ReadContext<'ctx, P> { /// The upstream endpoints that the packet will be forwarded to. pub endpoints: &'ctx ClusterMap, - /// The upstream endpoints that the packet will be forwarded to. - pub destinations: &'ctx mut Vec, + /// The endpoints the packet will be forwarded to, each paired with the + /// cluster it was routed from. + pub destinations: &'ctx mut Vec, /// The source of the received packet. pub source: EndpointAddress, /// Contents of the received packet. @@ -42,7 +43,7 @@ impl<'ctx, P: super::PacketMut> ReadContext<'ctx, P> { endpoints: &'ctx ClusterMap, source: EndpointAddress, contents: P, - destinations: &'ctx mut Vec, + destinations: &'ctx mut Vec, ) -> Self { Self { endpoints, diff --git a/src/filters/token_router.rs b/src/filters/token_router.rs index 5b2735244b..095b6094cf 100644 --- a/src/filters/token_router.rs +++ b/src/filters/token_router.rs @@ -56,7 +56,7 @@ impl Filter for TokenRouter { Some(metadata::Value::Bytes(token)) => { let tok = crate::net::cluster::Token::new(token); - ctx.endpoints.addresses_for_token(tok, ctx.destinations); + ctx.endpoints.destinations_for_token(tok, ctx.destinations); if ctx.destinations.is_empty() { Err(FilterError::TokenRouter(RouterError::NoEndpointMatch { @@ -100,6 +100,18 @@ pub enum RouterError { } impl RouterError { + #[inline] + pub fn drop_reason(&self) -> crate::metrics::DropReason { + use crate::metrics::DropReason; + + match self { + // The packet carried no token to route on, so the chain is + // misconfigured or the client is wrong, not the routing table + Self::NoTokenFound => DropReason::FilterError, + Self::NoEndpointMatch { .. } => DropReason::NoEndpointMatch, + } + } + #[inline] pub fn discriminant(&self) -> &'static str { match self { @@ -315,7 +327,7 @@ mod tests { } fn with_ctx( - dest: &mut Vec, + dest: &mut Vec, test: impl FnOnce(ReadContext<'_, bytes::BytesMut>), ) { let endpoint1 = Endpoint::with_metadata( diff --git a/src/metrics.rs b/src/metrics.rs index 24c04c2f4a..bf8a6ff799 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -31,6 +31,16 @@ pub(crate) const READ: Direction = Direction::Read; pub(crate) const WRITE: Direction = Direction::Write; #[allow(dead_code)] pub(crate) const ASN_LABEL: &str = "asn"; +pub(crate) const REASON_LABEL: &str = "reason"; +/// The filter responsible for a drop, empty when the drop wasn't a filter's +/// decision. Kept separate from [`REASON_LABEL`] so renaming a filter doesn't +/// change the reason vocabulary a breakdown is built on. +pub(crate) const FILTER_LABEL: &str = "filter"; +/// The cluster a packet was destined for, ie the locality of the endpoint it was +/// routed to. Empty when the packet was dropped before it was routed to one. +/// +/// Carries the same values as `quilkin_active_endpoints`, so the two join. +pub(crate) const DESTINATION_LABEL: &str = "destination"; /// Label value for [`DIRECTION_LABEL`] for `read` events pub const READ_DIRECTION_LABEL: &str = "read"; @@ -348,6 +358,101 @@ impl Direction { Self::Write => WRITE_DIRECTION_LABEL, } } + + #[inline] + const fn index(self) -> usize { + match self { + Self::Read => 0, + Self::Write => 1, + } + } +} + +/// Why a packet was dropped. +/// +/// Deliberately a closed set: drop breakdowns are built on these values, so they +/// must survive a filter being renamed and an `errno` producing different text on +/// a different kernel. Anything variable belongs in a log, or in +/// [`FILTER_LABEL`]. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum DropReason { + /// No endpoint was available, or none matched the packet's routing token. + NoEndpointMatch, + /// A filter chose to drop the packet, ie the chain worked as configured. + FilterDrop, + /// A filter failed to process the packet. + FilterError, + /// The socket refused the packet, or the packet couldn't be built for it. + SocketError, + /// A send or receive queue was full. + QueueFull, + /// The packet couldn't be parsed as a datagram we handle. + InvalidPacket, + /// The session limit was reached, so no session could be established. + SessionLimit, + /// Quilkin lost track of state it needs to forward the packet. + Internal, +} + +impl DropReason { + #[inline] + pub fn label(self) -> &'static str { + match self { + Self::NoEndpointMatch => "no_endpoint_match", + Self::FilterDrop => "filter_drop", + Self::FilterError => "filter_error", + Self::SocketError => "socket_error", + Self::QueueFull => "queue_full", + Self::InvalidPacket => "invalid_packet", + Self::SessionLimit => "session_limit", + Self::Internal => "internal", + } + } +} + +/// The [`std::io::ErrorKind`] of `error` as a bounded label value. +/// +/// `Display` for an I/O error interpolates the raw OS string, which varies by +/// platform and libc and so can't be a label. +#[inline] +pub fn io_error_kind(error: &std::io::Error) -> &'static str { + use std::io::ErrorKind; + + // `ErrorKind::Uncategorized` can't be named, and the errnos a UDP send + // actually fails with under load land in it, so they're matched on the raw + // value before falling back + #[cfg(target_os = "linux")] + if let Some(errno) = error.raw_os_error() { + let named = match errno { + libc::ENOBUFS => Some("no_buffer_space"), + libc::EMSGSIZE => Some("message_too_long"), + libc::ENETDOWN => Some("network_down"), + libc::ENETRESET => Some("network_reset"), + libc::ENOENT => Some("not_found"), + _ => None, + }; + + if let Some(named) = named { + return named; + } + } + + match error.kind() { + ErrorKind::AddrInUse => "addr_in_use", + ErrorKind::AddrNotAvailable => "addr_not_available", + ErrorKind::BrokenPipe => "broken_pipe", + ErrorKind::ConnectionRefused => "connection_refused", + ErrorKind::ConnectionReset => "connection_reset", + ErrorKind::Interrupted => "interrupted", + // EINVAL, which a send to an address the socket can't reach produces + ErrorKind::InvalidInput => "invalid_input", + ErrorKind::NotConnected => "not_connected", + ErrorKind::OutOfMemory => "out_of_memory", + ErrorKind::PermissionDenied => "permission_denied", + ErrorKind::TimedOut => "timed_out", + ErrorKind::WouldBlock => "would_block", + _ => "other", + } } pub struct AsnInfo<'a> { @@ -427,12 +532,23 @@ pub(crate) fn phoenix_measurement_seconds( icao: crate::config::IcaoCode, direction: &str, ) -> Histogram { + /// ~2x spacing across the range one-way inter-datacenter latency occupies, + /// 0.5 ms to 0.5 s. + /// + /// The Prometheus default buckets are for HTTP durations: they put + /// everything below 5 ms in one bucket, which is where same-datacenter + /// measurements all land, and spend five buckets above 500 ms, which no + /// network path reaches. + const BUCKETS: &[f64] = &[ + 0.0005, 0.001, 0.002, 0.004, 0.008, 0.015, 0.03, 0.05, 0.08, 0.12, 0.2, 0.3, 0.5, + ]; + static PHOENIX_MEASUREMENT: Lazy = Lazy::new(|| { prometheus::register_histogram_vec_with_registry! { prometheus::histogram_opts! { "quilkin_phoenix_measurement_seconds", "Histogram of phoenix measurements for a given node", - prometheus::DEFAULT_BUCKETS.to_vec() + BUCKETS.to_vec() }, &["icao", "direction"], registry(), @@ -563,75 +679,203 @@ pub(crate) fn processing_time(direction: Direction) -> Histogram { PROCESSING_TIME.with_label_values(&[direction.label()]) } -pub(crate) fn bytes_total(direction: Direction, _asn: &AsnInfo<'_>) -> IntCounter { +pub(crate) fn bytes_total( + direction: Direction, + _asn: &AsnInfo<'_>, + destination_locality: &str, +) -> IntCounter { static BYTES_TOTAL: Lazy = Lazy::new(|| { prometheus::register_int_counter_vec_with_registry! { prometheus::opts! { "quilkin_bytes_total", "total number of bytes", }, - &[Direction::LABEL], + &[Direction::LABEL, DESTINATION_LABEL], registry(), } .unwrap() }); - BYTES_TOTAL.with_label_values(&[direction.label()]) + BYTES_TOTAL.with_label_values(&[direction.label(), destination_locality]) } #[must_use] -pub(crate) fn errors_total(direction: Direction, display: &str, _asn: &AsnInfo<'_>) -> IntCounter { +pub(crate) fn errors_total(direction: Direction, reason: &str, _asn: &AsnInfo<'_>) -> IntCounter { static ERRORS_TOTAL: Lazy = Lazy::new(|| { prometheus::register_int_counter_vec_with_registry! { prometheus::opts! { "quilkin_errors_total", "total number of errors sending packets", }, - &[Direction::LABEL, "display"], + &[Direction::LABEL, REASON_LABEL], registry(), } .unwrap() }); - ERRORS_TOTAL.with_label_values(&[direction.label(), display]) + ERRORS_TOTAL.with_label_values(&[direction.label(), reason]) } -pub(crate) fn packet_jitter(direction: Direction, _asn: &AsnInfo<'_>) -> IntGauge { - static PACKET_JITTER: Lazy = Lazy::new(|| { - prometheus::register_int_gauge_vec_with_registry! { +static PACKET_JITTER: Lazy = Lazy::new(|| { + prometheus::register_int_gauge_vec_with_registry! { + prometheus::opts! { + "quilkin_packet_jitter", + "The time between new packets", + }, + &[Direction::LABEL], + registry(), + } + .unwrap() +}); + +/// Counts observations per direction so [`remove_packet_jitter`] can tell a +/// current value from one a since-idle proxy is still publishing. +static PACKET_JITTER_OBSERVATIONS: [std::sync::atomic::AtomicU64; 2] = + [const { std::sync::atomic::AtomicU64::new(0) }; 2]; + +/// Sets `quilkin_packet_jitter` to the interarrival time of the packet just +/// processed, in nanoseconds. +/// +/// This is the interarrival time seen by an I/O loop, which covers every session +/// it serves; for the per-session distribution see +/// `quilkin_session_jitter_seconds`. +#[inline] +pub(crate) fn set_packet_jitter(direction: Direction, _asn: &AsnInfo<'_>, nanos: i64) { + PACKET_JITTER_OBSERVATIONS[direction.index()] + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + PACKET_JITTER + .with_label_values(&[direction.label()]) + .set(nanos); +} + +/// Number of times `quilkin_packet_jitter` has been set for `direction`. +#[inline] +pub(crate) fn packet_jitter_observations(direction: Direction) -> u64 { + PACKET_JITTER_OBSERVATIONS[direction.index()].load(std::sync::atomic::Ordering::Relaxed) +} + +/// Stops exporting `quilkin_packet_jitter` for `direction`. +/// +/// The metric is a gauge set per packet, so a proxy that stops receiving would +/// otherwise keep publishing its last value indefinitely. +pub(crate) fn remove_packet_jitter(direction: Direction) { + drop(PACKET_JITTER.remove_label_values(&[direction.label()])); +} + +/// Per-session interarrival jitter, in seconds. +/// +/// A histogram rather than a mean, because a cluster mean of 0.5 ms is +/// compatible with a few percent of players at 80 ms, and those are the players +/// worth knowing about. +pub(crate) fn session_jitter_seconds() -> &'static Histogram { + /// ~2x spacing across the range player connections actually occupy, 0.1 ms + /// to 0.5 s. + const BUCKETS: &[f64] = &[ + 0.0001, 0.00025, 0.0005, 0.001, 0.002, 0.004, 0.008, 0.015, 0.03, 0.06, 0.12, 0.25, 0.5, + ]; + + static SESSION_JITTER: Lazy = Lazy::new(|| { + prometheus::register_histogram_with_registry! { + prometheus::histogram_opts! { + "quilkin_session_jitter_seconds", + "Distribution of per-session interarrival jitter of downstream packets", + BUCKETS.to_vec() + }, + registry(), + } + .unwrap() + }); + + &SESSION_JITTER +} + +static SESSIONS_ACTIVE_BY_ASN: Lazy = Lazy::new(|| { + prometheus::register_int_gauge_vec_with_registry! { + prometheus::opts! { + "quilkin_sessions_active_by_asn", + "Active sessions by client ASN, for the largest ASNs at this proxy. Sessions belonging to any other ASN are counted under `asn=\"other\"`, so the breakdown sums to the session total.", + }, + &[ASN_LABEL], + registry(), + } + .unwrap() +}); + +pub(crate) fn sessions_active_by_asn(asn: &str) -> IntGauge { + SESSIONS_ACTIVE_BY_ASN.with_label_values(&[asn]) +} + +/// Stops exporting `quilkin_sessions_active_by_asn` for `asn`, used when it +/// falls out of the exported set. +pub(crate) fn remove_sessions_active_by_asn(asn: &str) { + drop(SESSIONS_ACTIVE_BY_ASN.remove_label_values(&[asn])); +} + +static CLIENT_SESSIONS_DEGRADED: Lazy = Lazy::new(|| { + prometheus::register_int_gauge_vec_with_registry! { + prometheus::opts! { + "quilkin_client_sessions_degraded", + "Sessions breaching a connection quality threshold, by client ASN. Only ASNs currently breaching are exported, so a healthy proxy publishes nothing here. Divide by `quilkin_sessions_active_by_asn` for the affected share; do it across the fleet, since one proxy sees too few sessions of any one ASN to judge it.", + }, + &[ASN_LABEL, REASON_LABEL], + registry(), + } + .unwrap() +}); + +pub(crate) fn client_sessions_degraded(asn: &str, reason: &str) -> IntGauge { + CLIENT_SESSIONS_DEGRADED.with_label_values(&[asn, reason]) +} + +/// Stops exporting `quilkin_client_sessions_degraded` for `asn`, used when it +/// recovers. +pub(crate) fn remove_client_sessions_degraded(asn: &str, reason: &str) { + drop(CLIENT_SESSIONS_DEGRADED.remove_label_values(&[asn, reason])); +} + +/// Counts degraded session observations, so a rate can be alerted on without +/// depending on a threshold the proxy would have to pick. +pub(crate) fn client_sessions_degraded_total(reason: &str) -> IntCounter { + static CLIENT_SESSIONS_DEGRADED: Lazy = Lazy::new(|| { + prometheus::register_int_counter_vec_with_registry! { prometheus::opts! { - "quilkin_packet_jitter", - "The time between new packets", + "quilkin_client_sessions_degraded_total", + "Total number of times a session was observed breaching a connection quality threshold", }, - &[Direction::LABEL], + &[REASON_LABEL], registry(), } .unwrap() }); - PACKET_JITTER.with_label_values(&[direction.label()]) + CLIENT_SESSIONS_DEGRADED.with_label_values(&[reason]) } -pub(crate) fn packets_total(direction: Direction, _asn: &AsnInfo<'_>) -> IntCounter { +pub(crate) fn packets_total( + direction: Direction, + _asn: &AsnInfo<'_>, + destination_locality: &str, +) -> IntCounter { static PACKETS_TOTAL: Lazy = Lazy::new(|| { prometheus::register_int_counter_vec_with_registry! { prometheus::opts! { "quilkin_packets_total", "Total number of packets", }, - &[Direction::LABEL], + &[Direction::LABEL, DESTINATION_LABEL], registry(), } .unwrap() }); - PACKETS_TOTAL.with_label_values(&[direction.label()]) + PACKETS_TOTAL.with_label_values(&[direction.label(), destination_locality]) } pub(crate) fn packets_dropped_total( direction: Direction, - source: &str, - _asn: &AsnInfo<'_>, + reason: DropReason, + filter: &str, + destination_locality: &str, ) -> IntCounter { static PACKETS_DROPPED: Lazy = Lazy::new(|| { prometheus::register_int_counter_vec_with_registry! { @@ -639,13 +883,30 @@ pub(crate) fn packets_dropped_total( "quilkin_packets_dropped_total", "Total number of dropped packets", }, - &[Direction::LABEL, "source"], + &[ + Direction::LABEL, + REASON_LABEL, + FILTER_LABEL, + DESTINATION_LABEL, + ], registry(), } .unwrap() }); - PACKETS_DROPPED.with_label_values(&[direction.label(), source]) + PACKETS_DROPPED.with_label_values(&[ + direction.label(), + reason.label(), + filter, + destination_locality, + ]) +} + +/// [`packets_dropped_total`] for drops with no filter or destination to +/// attribute, ie packets dropped before they were routed. +#[inline] +pub(crate) fn packets_dropped(direction: Direction, reason: DropReason) -> IntCounter { + packets_dropped_total(direction, reason, "", "") } pub(crate) fn provider_task_failures_total(provider_task: &str) -> IntCounter { @@ -848,6 +1109,91 @@ mod tests { .any(|m| m.get_label().iter().any(|l| l.value() == label)) } + #[test] + fn drops_are_labelled_with_a_bounded_reason() { + use crate::filters::FilterError; + use crate::net::PipelineError; + + // The vocabulary a drop breakdown is built on, which must not shift when + // a filter is renamed or an errno differs + assert_eq!( + PipelineError::NoUpstreamEndpoints.drop_reason(), + DropReason::NoEndpointMatch + ); + assert_eq!( + PipelineError::Filter(FilterError::Dropped).drop_reason(), + DropReason::FilterDrop + ); + assert_eq!( + PipelineError::Filter(FilterError::NoValueCaptured).drop_reason(), + DropReason::FilterError + ); + assert_eq!( + PipelineError::Filter(FilterError::Custom("anything at all")).drop_reason(), + DropReason::FilterError + ); + assert_eq!( + PipelineError::Io(std::io::Error::from_raw_os_error(22)).drop_reason(), + DropReason::SocketError + ); + + // A filter's identity is a separate label, so it never widens the reasons + assert_eq!( + PipelineError::Filter(FilterError::FirewallDenied).filter_name(), + "firewall" + ); + assert_eq!( + PipelineError::Io(std::io::Error::from_raw_os_error(22)).filter_name(), + "" + ); + + // EINVAL, which used to reach the label as "Invalid argument (os error 22)" + assert_eq!( + io_error_kind(&std::io::Error::from_raw_os_error(22)), + "invalid_input" + ); + + // The errnos a UDP send fails with under load. `ErrorKind` puts both in + // its uncategorised bucket, so without naming them the most common real + // send failures would be indistinguishable. + #[cfg(target_os = "linux")] + { + assert_eq!( + io_error_kind(&std::io::Error::from_raw_os_error(libc::ENOBUFS)), + "no_buffer_space" + ); + assert_eq!( + io_error_kind(&std::io::Error::from_raw_os_error(libc::EMSGSIZE)), + "message_too_long" + ); + } + } + + #[test] + fn dropped_packets_carry_the_full_label_set() { + packets_dropped_total(READ, DropReason::FilterDrop, "firewall", "eu-north1").inc(); + + let rendered = registry() + .gather() + .iter() + .filter(|mf| mf.name() == "quilkin_packets_dropped_total") + .flat_map(|mf| mf.get_metric()) + .any(|m| { + let labels: std::collections::HashMap<_, _> = m + .get_label() + .iter() + .map(|l| (l.name(), l.value())) + .collect(); + + labels.get(DIRECTION_LABEL) == Some(&"read") + && labels.get(REASON_LABEL) == Some(&"filter_drop") + && labels.get(FILTER_LABEL) == Some(&"firewall") + && labels.get(DESTINATION_LABEL) == Some(&"eu-north1") + }); + + assert!(rendered); + } + #[test] fn apply_clusters_prunes_removed_localities() { let clusters = crate::config::Watch::new(crate::net::ClusterMap::default()); diff --git a/src/net.rs b/src/net.rs index 37caae4f48..cd6fb83396 100644 --- a/src/net.rs +++ b/src/net.rs @@ -42,7 +42,7 @@ cfg_select! { pub use { self::{ - cluster::ClusterMap, + cluster::{ClusterMap, Destination}, endpoint::{Endpoint, EndpointAddress}, error::PipelineError, packet::{Packet, PacketMut, PacketQueue, PacketQueueReceiver, PacketQueueSender, queue}, diff --git a/src/net/cluster.rs b/src/net/cluster.rs index 5d4dd38799..6993643c38 100644 --- a/src/net/cluster.rs +++ b/src/net/cluster.rs @@ -306,6 +306,16 @@ impl EndpointSet { self.endpoints.contains_key(&ep.address) } + #[inline] + pub fn contains_address(&self, address: &EndpointAddress) -> bool { + self.endpoints.contains_key(address) + } + + #[inline] + pub fn addresses(&self) -> impl Iterator { + self.endpoints.keys() + } + /// Unique version for this endpoint set #[inline] pub fn version(&self) -> EndpointSetVersion { @@ -518,7 +528,8 @@ impl EndpointSet { pub fn corrosion_apply( &mut self, ss: corrosion::pubsub::SubscriptionStream, - token_map: &DashMap>, + locality: &Option, + token_map: &DashMap>, subm: &mut corrosion::persistent::SubMetrics, ) -> (ChangeId, isize) { use corrosion::{ @@ -572,7 +583,7 @@ impl EndpointSet { { let mut tm = token_map.entry(tok.0).or_default(); - tm.insert(addr.clone()); + tm.insert(Destination::new(addr.clone(), locality.clone())); } this.entry(tok.0).or_default().insert(addr.clone()); @@ -582,7 +593,7 @@ impl EndpointSet { let tok = Token::new(tok); let remove = if let Some(mut tm) = token_map.get_mut(&tok.0) { - tm.remove(addr); + tm.remove(&Destination::new(addr.clone(), locality.clone())); tm.is_empty() } else { false @@ -689,13 +700,48 @@ impl EndpointSet { pub struct ClusterMap { map: DashMap, EndpointSet, S>, localities: DashMap, Option>, - token_map: DashMap>, + token_map: DashMap>, num_endpoints: AtomicUsize, version: AtomicU64, } type DashMapRef<'inner> = dashmap::mapref::one::Ref<'inner, Option, EndpointSet>; +/// An endpoint that routing selected, paired with the cluster it was selected +/// from. +/// +/// The cluster travels with the address because it is only knowable at the point +/// of the routing decision: the same address may be configured in more than one +/// locality, so recovering it afterwards from the address alone is guesswork. +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub struct Destination { + pub address: EndpointAddress, + /// Locality of the cluster the endpoint was routed from, `None` for a + /// cluster configured without one. + pub cluster: Option, +} + +impl Destination { + #[inline] + pub fn new(address: EndpointAddress, cluster: Option) -> Self { + Self { address, cluster } + } + + /// The cluster as a metric label value, empty when there isn't one. + #[inline] + pub fn cluster_label(&self) -> &str { + self.cluster + .as_ref() + .map_or("", |locality| locality.as_str()) + } +} + +impl From for Destination { + fn from(address: EndpointAddress) -> Self { + Self::new(address, None) + } +} + impl ClusterMap { pub fn new() -> Self { Self::default() @@ -748,18 +794,15 @@ where self.version.fetch_add(1, Relaxed); - for (token_hash, addrs) in token_map_diff { - if let Some(addrs) = addrs { - self.token_map.insert(token_hash, addrs); - } else { - self.token_map.remove(&token_hash); - } - } + self.apply_token_diff(&locality, token_map_diff); } else { - for (token_hash, addrs) in &endpoint_set.token_map { - self.token_map - .insert(*token_hash, addrs.iter().cloned().collect()); - } + self.apply_token_diff( + &locality, + endpoint_set + .token_map + .iter() + .map(|(token, addrs)| (*token, Some(addrs.iter().cloned().collect()))), + ); let new_len = endpoint_set.len(); self.map.insert(locality, endpoint_set); @@ -819,18 +862,15 @@ where self.version.fetch_add(1, Relaxed); - for (token_hash, addrs) in token_map_diff { - if let Some(addrs) = addrs { - self.token_map.insert(token_hash, addrs); - } else { - self.token_map.remove(&token_hash); - } - } + self.apply_token_diff(&locality, token_map_diff); } else { - for (token_hash, addrs) in &cluster.token_map { - self.token_map - .insert(*token_hash, addrs.iter().cloned().collect()); - } + self.apply_token_diff( + &locality, + cluster + .token_map + .iter() + .map(|(token, addrs)| (*token, Some(addrs.iter().cloned().collect()))), + ); upserted.extend(cluster.to_map()); @@ -874,6 +914,70 @@ where }) } + /// Applies `locality`'s token map changes to the aggregated one, tagging each + /// entry with the cluster that contributed it so routing can report it. + /// + /// Routing tokens are globally unique, so a locality's entry for a token is + /// the entire entry and replacing it wholesale is correct. + #[inline] + fn apply_token_diff( + &self, + locality: &Option, + diff: impl IntoIterator>)>, + ) { + for (token_hash, addresses) in diff { + match addresses { + Some(addresses) => { + self.token_map.insert( + token_hash, + addresses + .into_iter() + .map(|address| Destination::new(address, locality.clone())) + .collect(), + ); + } + None => { + self.token_map.remove(&token_hash); + } + } + } + } + + /// Whether any cluster currently contains an endpoint at `address`. + /// + /// Used to tell an endpoint disappearing from underneath a session apart from + /// a player going quiet, so it runs once per session rather than per packet. + /// Which cluster a packet was routed to is carried by [`Destination`], since + /// only the routing decision knows that. + /// + /// Note that an endpoint configured by name is keyed by that name, so it + /// isn't found by the address the name resolves to. + #[inline] + pub fn contains_endpoint(&self, address: &std::net::SocketAddr) -> bool { + let address = EndpointAddress::from(*address); + let mut localities = self + .map + .iter() + .filter(|entry| entry.value().contains_address(&address)); + + let Some(first) = localities.next() else { + return false; + }; + + // A misconfiguration rather than something to resolve: an endpoint in two + // clusters makes traffic attribution and capacity accounting ambiguous + if let Some(second) = localities.next() { + tracing::warn!( + %address, + first = ?first.key(), + second = ?second.key(), + "endpoint is present in more than one cluster" + ); + } + + true + } + /// Applies a batch of updates to the `ClusterMap` /// /// BEWARE: This method does not keep the global `token_map` up to date, as this is part of the @@ -1024,11 +1128,42 @@ where self.do_remove_locality(locality) } - pub fn addresses_for_token(&self, token: Token, addrs: &mut Vec) { + pub fn destinations_for_token(&self, token: Token, destinations: &mut Vec) { if let Some(ma) = self.token_map.get(&token.0) { - addrs.extend(ma.value().iter().cloned()); + destinations.extend(ma.value().iter().cloned()); } } + + /// Every endpoint, paired with the cluster it belongs to. + #[inline] + pub fn destinations(&self, destinations: &mut Vec) { + for entry in self.map.iter() { + destinations.extend( + entry + .value() + .addresses() + .map(|address| Destination::new(address.clone(), entry.key().clone())), + ); + } + } + + /// The `index`th endpoint, paired with the cluster it belongs to. + #[inline] + pub fn nth_destination(&self, mut index: usize) -> Option { + for entry in self.map.iter() { + let set = entry.value(); + if index < set.len() { + return set + .addresses() + .nth(index) + .map(|address| Destination::new(address.clone(), entry.key().clone())); + } + + index -= set.len(); + } + + None + } } impl ClusterMap @@ -1051,7 +1186,7 @@ where .map .entry(Some((*CORRO).clone())) .or_insert_with(|| EndpointSet::new(BTreeSet::default())) - .corrosion_apply(ss, &self.token_map, subm); + .corrosion_apply(ss, &Some((*CORRO).clone()), &self.token_map, subm); // If we don't update the num_endpoints and it's 0, no filter will run! if diff >= 0 { @@ -1237,11 +1372,17 @@ where fn from(map: DashMap, EndpointSet, S>) -> Self { let num_endpoints = AtomicUsize::new(map.iter().map(|kv| kv.value().len()).sum()); - let token_map = DashMap::>::default(); + let token_map = DashMap::>::default(); let localities = DashMap::default(); for es in &map { for (token_hash, addrs) in &es.value().token_map { - token_map.insert(*token_hash, addrs.iter().cloned().collect()); + token_map.insert( + *token_hash, + addrs + .iter() + .map(|address| Destination::new(address.clone(), es.key().clone())) + .collect(), + ); } localities.insert(es.key().clone(), None); @@ -1474,6 +1615,64 @@ mod tests { use super::*; + #[test] + fn contains_endpoint() { + let se1 = Locality::with_region("se-1"); + let localised: std::net::SocketAddr = (Ipv4Addr::LOCALHOST, 7777).into(); + let unlocalised: std::net::SocketAddr = (Ipv4Addr::LOCALHOST, 7778).into(); + + let cluster = ClusterMap::new(); + cluster.insert(None, Some(se1), [Endpoint::new(localised.into())].into()); + cluster.insert(None, None, [Endpoint::new(unlocalised.into())].into()); + + assert!(cluster.contains_endpoint(&localised)); + // Present, in a cluster configured without a locality + assert!(cluster.contains_endpoint(&unlocalised)); + assert!(!cluster.contains_endpoint(&(Ipv4Addr::LOCALHOST, 7779).into())); + } + + #[test] + fn destinations_carry_the_cluster_they_were_routed_from() { + let se1 = Locality::with_region("se-1"); + let de1 = Locality::with_region("de-1"); + + let endpoint = |port: u16, token: &[u8]| { + Endpoint::with_metadata( + (Ipv4Addr::LOCALHOST, port).into(), + crate::net::endpoint::Metadata { + tokens: [token.to_vec()].into(), + }, + ) + }; + + let cluster = ClusterMap::new(); + cluster.insert(None, Some(se1.clone()), [endpoint(7777, b"abc")].into()); + cluster.insert(None, Some(de1.clone()), [endpoint(7778, b"def")].into()); + + let resolve = |token: &[u8]| { + let mut destinations = Vec::new(); + cluster.destinations_for_token(Token::new(token), &mut destinations); + destinations + }; + + // Tokens are globally unique, so each resolves within exactly one cluster, + // and the cluster travels with the address the router picked + assert_eq!( + resolve(b"abc"), + [Destination::new( + (Ipv4Addr::LOCALHOST, 7777).into(), + Some(se1) + )] + ); + assert_eq!( + resolve(b"def"), + [Destination::new( + (Ipv4Addr::LOCALHOST, 7778).into(), + Some(de1) + )] + ); + } + #[test] fn merge() { let nl1 = Locality::with_region("nl-1"); diff --git a/src/net/error.rs b/src/net/error.rs index 830fe2b342..09116cfa17 100644 --- a/src/net/error.rs +++ b/src/net/error.rs @@ -35,12 +35,14 @@ impl PipelineError { direction: crate::metrics::Direction, asn_info: &crate::metrics::AsnInfo<'_>, ) { - if matches!( - self, - PipelineError::Io(_) | PipelineError::Filter(crate::filters::FilterError::Io(_)) - ) { - crate::metrics::errors_total(direction, &self.to_string(), asn_info).inc(); - } + let io = match self { + PipelineError::Io(io) | PipelineError::Filter(crate::filters::FilterError::Io(io)) => { + io + } + _ => return, + }; + + crate::metrics::errors_total(direction, crate::metrics::io_error_kind(io), asn_info).inc(); } pub fn discriminant(&self) -> &'static str { @@ -53,6 +55,32 @@ impl PipelineError { Self::SessionLimit => "session limit", } } + + /// The bounded drop reason this error corresponds to. + #[inline] + pub fn drop_reason(&self) -> crate::metrics::DropReason { + use crate::metrics::DropReason; + + match self { + Self::NoUpstreamEndpoints => DropReason::NoEndpointMatch, + Self::Filter(fe) => fe.drop_reason(), + Self::Session(_) => DropReason::Internal, + Self::Io(_) => DropReason::SocketError, + // Spoofed or misdirected source, ie not a packet we can parse as + // belonging to a client + Self::DisallowedSourceIP(_) => DropReason::InvalidPacket, + Self::SessionLimit => DropReason::SessionLimit, + } + } + + /// The filter that produced this error, empty when it didn't come from one. + #[inline] + pub fn filter_name(&self) -> &'static str { + match self { + Self::Filter(fe) => fe.filter_name(), + _ => "", + } + } } impl std::error::Error for PipelineError {} diff --git a/src/net/io/completion/io_uring.rs b/src/net/io/completion/io_uring.rs index b43e32d2ab..bb7bafcb06 100644 --- a/src/net/io/completion/io_uring.rs +++ b/src/net/io/completion/io_uring.rs @@ -216,7 +216,7 @@ pub enum PacketProcessorCtx { config: Arc, sessions: Arc, worker_id: usize, - destinations: Vec, + destinations: Vec, }, SessionPool { pool: Arc, @@ -239,8 +239,11 @@ fn process_packet( } => { let received_at = UtcTimestamp::now(); if let Some(last_received_at) = last_received_at { - metrics::packet_jitter(metrics::READ, &metrics::EMPTY) - .set((received_at - *last_received_at).nanos()); + metrics::set_packet_jitter( + metrics::READ, + &metrics::EMPTY, + (received_at - *last_received_at).nanos(), + ); } *last_received_at = Some(received_at); @@ -630,7 +633,7 @@ impl IoUringLoop { data.put_u16_ne(rb.count); data.put_u16_ne(rb.len(id)); data.put_u32_ne(alloced); - loop_ctx.enqueue_send(SendPacket { destination: packet.source, data: data.freeze(), asn_info: None }); + loop_ctx.enqueue_send(SendPacket { destination: packet.source, data: data.freeze(), asn_info: None, cluster: std::sync::Arc::from("") }); continue; } } @@ -666,15 +669,20 @@ impl IoUringLoop { }; let asn_info = zs.asn_info.as_ref().into(); + let locality = &*zs.cluster; if ret < 0 { - let source = - std::io::Error::from_raw_os_error(-ret).to_string(); - metrics::errors_total(send_dir, &source, &asn_info).inc(); - metrics::packets_dropped_total(send_dir, &source, &asn_info) - .inc(); + let error = std::io::Error::from_raw_os_error(-ret); + metrics::errors_total(send_dir, metrics::io_error_kind(&error), &asn_info).inc(); + metrics::packets_dropped_total( + send_dir, + metrics::DropReason::SocketError, + "", + locality, + ) + .inc(); } else if ret as usize != zs.data.len() { - metrics::packets_total(send_dir, &asn_info).inc(); + metrics::packets_total(send_dir, &asn_info, locality).inc(); metrics::errors_total( send_dir, "sent bytes != packet length", @@ -682,8 +690,9 @@ impl IoUringLoop { ) .inc(); } else { - metrics::packets_total(send_dir, &asn_info).inc(); - metrics::bytes_total(send_dir, &asn_info).inc_by(ret as u64); + metrics::packets_total(send_dir, &asn_info, locality).inc(); + metrics::bytes_total(send_dir, &asn_info, locality) + .inc_by(ret as u64); } } diff --git a/src/net/io/nic/xdp/process.rs b/src/net/io/nic/xdp/process.rs index 6a83165cd3..7eff8c7110 100644 --- a/src/net/io/nic/xdp/process.rs +++ b/src/net/io/nic/xdp/process.rs @@ -2,10 +2,9 @@ use crate::{ filters::{self, Filter as _}, metrics::{self, AsnInfo}, net::{ - EndpointAddress, error::PipelineError, maxmind_db::{self, IpNetEntry}, - sessions::inner_metrics as session_metrics, + sessions::{inner_metrics as session_metrics, quality as session_quality}, }, time::UtcTimestamp, }; @@ -201,7 +200,7 @@ pub struct State { /// or servers (upstream) pub external_port: NetworkU16, pub qcmp_port: NetworkU16, - pub destinations: Vec, + pub destinations: Vec, pub addr_to_asn: AsnCache, pub sessions: Arc, pub local_ipv4: std::net::Ipv4Addr, @@ -217,8 +216,8 @@ impl State { &self, server_addr: SocketAddr, port: NetworkU16, - ) -> Option<(SocketAddr, AsnInfo<'_>)> { - let addr = self.sessions.lookup_client(server_addr, port)?; + ) -> Option<(SocketAddr, AsnInfo<'_>, Arc)> { + let (addr, cluster) = self.sessions.lookup_client(server_addr, port)?; let entry = self .addr_to_asn .get(&addr.ip(), self.last_receive.unix_nanos()) @@ -227,7 +226,7 @@ impl State { asn: asn.as_str(), }); - Some((addr, entry)) + Some((addr, entry, cluster)) } /// Retrieves or creates a session, ie a mapping of a server endpoint + port @@ -237,6 +236,7 @@ impl State { &mut self, client_addr: SocketAddr, server_addr: SocketAddr, + cluster: &str, ) -> (NetworkU16, AsnInfo<'_>, IpAddresses) { let ips = self.ips(server_addr.ip()); let asn = self.addr_to_asn.get_or_insert_with( @@ -251,9 +251,9 @@ impl State { }, ); - let port = self - .sessions - .get_or_create(client_addr, server_addr, asn.map(|(ipe, _)| ipe)); + let port = + self.sessions + .get_or_create(client_addr, server_addr, asn.map(|(ipe, _)| ipe), cluster); ( port, @@ -391,6 +391,9 @@ struct ClientInfo { created_at: Instant, /// The port used to identify this unique session to the IP owning this map port: NetworkU16, + /// Interarrival jitter of this session, folded into aggregate metrics by + /// [`crate::net::sessions::quality::spawn_aggregator`] + quality: session_quality::SessionQualityHandle, } struct PortMapper { @@ -398,15 +401,20 @@ struct PortMapper { /// to the server endpoint `Self` is associated with client_to_port: Arc>>, port_to_client: Arc>, + /// The cluster the server endpoint this maps to belongs to, as routing + /// reported it when the first session to that endpoint was created. Held here + /// rather than per client, since it's a property of the server. + cluster: Arc, port: AtomicU16, } impl PortMapper { #[inline] - fn new() -> Self { + fn new(cluster: &str) -> Self { Self { client_to_port: Arc::new(Default::default()), port_to_client: Arc::new(parking_lot::RwLock::new(PortMap::new())), + cluster: Arc::from(cluster), port: AtomicU16::new(EPHEMERAL_RANGE_END), } } @@ -418,7 +426,11 @@ impl PortMapper { asn: Option<&IpNetEntry>, ) -> Option { match self.client_to_port.lock().entry(client_addr) { - Entry::Occupied(entry) => Some(entry.get().port), + Entry::Occupied(entry) => { + let client = entry.get(); + client.quality.record_arrival(); + Some(client.port) + } Entry::Vacant(entry) => { let port = self.port.fetch_add(1, Ordering::Relaxed); @@ -434,6 +446,7 @@ impl PortMapper { let port = port.into(); entry.insert(ClientInfo { + quality: session_quality::SessionQualityHandle::register(asn), asn_info: asn.cloned(), created_at: Instant::now(), port, @@ -457,6 +470,7 @@ impl Drop for PortMapper { for client_info in lock.values() { session_metrics::active_sessions(client_info.asn_info.as_ref()).dec(); + session_metrics::sessions_closed_total(session_metrics::CloseReason::IdleTimeout).inc(); session_metrics::duration_secs() .observe(now.duration_since(client_info.created_at).as_secs_f64()); } @@ -480,10 +494,17 @@ impl SessionState { /// Attempts to lookup a client endpoint based on the server endpoint that sent /// the packet to the specified port #[inline] - fn lookup_client(&self, server_addr: SocketAddr, port: NetworkU16) -> Option { - self.sessions - .get(&server_addr) - .and_then(|pm| pm.get_client(port)) + fn lookup_client( + &self, + server_addr: SocketAddr, + port: NetworkU16, + ) -> Option<(SocketAddr, Arc)> { + // The cluster lives on the port mapper, which is keyed by the server + // address, so it comes back from the lookup already being done here + let pm = self.sessions.get(&server_addr)?; + let client = pm.get_client(port)?; + + Some((client, pm.cluster.clone())) } /// Retrieves the port used to forward packets from the specified client @@ -495,13 +516,14 @@ impl SessionState { client_addr: SocketAddr, server_addr: SocketAddr, asn: Option<&IpNetEntry>, + cluster: &str, ) -> NetworkU16 { let port = match self.sessions.entry(server_addr) { crate::collections::ttl::Entry::Occupied(entry) => { entry.get().get_or_alloc(client_addr, asn) } crate::collections::ttl::Entry::Vacant(entry) => { - let pm = PortMapper::new(); + let pm = PortMapper::new(cluster); let port = pm.get_or_alloc(client_addr, asn); entry.insert(pm); port @@ -519,7 +541,7 @@ impl SessionState { // the client endpoint is any longer, or, slightly worse, a packet gets // redirected to a different client. self.sessions.remove(server_addr); - self.get_or_create(client_addr, server_addr, asn) + self.get_or_create(client_addr, server_addr, asn, cluster) } } @@ -621,7 +643,7 @@ pub fn process_packets( // This indicates a packet that is split, which we don't handle _at all_ // right now, and only the first buffer has headers, so check before parsing if buffer.is_continued() { - metrics::packets_dropped_total(metrics::READ, "split packet", &metrics::EMPTY).inc(); + metrics::packets_dropped(metrics::READ, metrics::DropReason::InvalidPacket).inc(); umem.free_packet(buffer); continue; } @@ -630,7 +652,7 @@ pub fn process_packets( Ok(headers) => headers, Err(reason) => { tracing::debug!(reason, length = buffer.len(), "dropped unparsable packet"); - metrics::packets_dropped_total(metrics::READ, reason, &metrics::EMPTY).inc(); + metrics::packets_dropped(metrics::READ, metrics::DropReason::InvalidPacket).inc(); umem.free_packet(buffer); continue; } @@ -667,9 +689,14 @@ pub fn process_packets( umem.free_packet(packet); } Err((error, packet)) => { - let discriminant = error.discriminant(); error.inc_system_errors_total(direction, &metrics::EMPTY); - metrics::packets_dropped_total(direction, discriminant, &metrics::EMPTY).inc(); + metrics::packets_dropped_total( + direction, + error.drop_reason(), + error.filter_name(), + "", + ) + .inc(); umem.free_packet(packet); } @@ -677,15 +704,17 @@ pub fn process_packets( } if had_read { - metrics::packet_jitter(metrics::READ, &metrics::EMPTY).set(jitter); + metrics::set_packet_jitter(metrics::READ, &metrics::EMPTY, jitter); } } #[inline] +#[allow(clippy::too_many_arguments)] fn push_packet( direction: metrics::Direction, packet: Packet, asn: AsnInfo<'_>, + cluster: &str, data_length: usize, res: Result<(), PacketError>, tx_slab: &mut StackSlab, @@ -694,17 +723,16 @@ fn push_packet( match res { Ok(()) => { if let Some(packet) = tx_slab.push_front(packet) { - metrics::packets_dropped_total(direction, "tx slab full", &metrics::EMPTY).inc(); + metrics::packets_dropped(direction, metrics::DropReason::QueueFull).inc(); umem.free_packet(packet); } else { - metrics::packets_total(direction, &asn).inc(); - metrics::bytes_total(direction, &asn).inc_by(data_length as u64); + metrics::packets_total(direction, &asn, cluster).inc(); + metrics::bytes_total(direction, &asn, cluster).inc_by(data_length as u64); } } Err(err) => { - let discriminant = err.discriminant(); - metrics::errors_total(direction, discriminant, &metrics::EMPTY).inc(); - metrics::packets_dropped_total(direction, discriminant, &metrics::EMPTY).inc(); + metrics::errors_total(direction, err.discriminant(), &metrics::EMPTY).inc(); + metrics::packets_dropped(direction, metrics::DropReason::SocketError).inc(); umem.free_packet(packet); } } @@ -746,10 +774,11 @@ fn process_client_packet( // a new packet for each destination, only modifying the headers if !state.destinations.is_empty() { while let Some(daddr) = state.destinations.pop() { - let Ok(dest_addr) = daddr.to_socket_addr() else { + let Ok(dest_addr) = daddr.address.to_socket_addr() else { continue; }; - let (source, asn, ips) = state.session(source_addr, dest_addr); + let cluster = daddr.cluster_label(); + let (source, asn, ips) = state.session(source_addr, dest_addr, cluster); let mut headers = UdpHeaders { eth, @@ -776,6 +805,7 @@ fn process_client_packet( metrics::Direction::Read, new_packet, asn, + cluster, data_length, res, tx_slab, @@ -784,10 +814,11 @@ fn process_client_packet( } } - let Ok(dest_addr) = dest_addr.to_socket_addr() else { + let cluster = dest_addr.cluster_label(); + let Ok(dest_addr) = dest_addr.address.to_socket_addr() else { return Ok(Some(packet.buffer)); }; - let (source, asn, ips) = state.session(source_addr, dest_addr); + let (source, asn, ips) = state.session(source_addr, dest_addr, cluster); let mut headers = UdpHeaders { eth, @@ -808,6 +839,7 @@ fn process_client_packet( metrics::Direction::Read, packet.buffer, asn, + cluster, data_length, res, tx_slab, @@ -829,13 +861,14 @@ fn process_server_packet( let mut server_addr = packet.headers.source_address(); server_addr.set_ip(server_addr.ip().to_canonical()); - let Some((client_addr, asn)) = state.lookup_client(server_addr, packet.headers.udp.destination) + let Some((client_addr, asn, cluster)) = + state.lookup_client(server_addr, packet.headers.udp.destination) else { tracing::debug!(address = %server_addr, "received traffic from a server that has no downstream"); return Ok(Some(packet.buffer)); }; - metrics::packet_jitter(metrics::Direction::Write, &asn).set(jitter); + metrics::set_packet_jitter(metrics::Direction::Write, &asn, jitter); let mut ctx = filters::WriteContext::new(server_addr.into(), client_addr.into(), packet); @@ -863,6 +896,7 @@ fn process_server_packet( metrics::Direction::Write, packet.buffer, asn, + &cluster, packet.headers.data_length(), res, tx_slab, diff --git a/src/net/io/poll/tokio.rs b/src/net/io/poll/tokio.rs index 5370614d72..b164f95d71 100644 --- a/src/net/io/poll/tokio.rs +++ b/src/net/io/poll/tokio.rs @@ -172,20 +172,27 @@ fn spawn_poll_listener_impl( let destination = packet.destination; let (result, _) = ps_send_to(&send_socket, packet.data, destination).await; let asn_info = packet.asn_info.as_ref().into(); + let locality = &*packet.cluster; match result { Ok(size) => { - crate::metrics::packets_total(crate::metrics::WRITE, &asn_info).inc(); - crate::metrics::bytes_total(crate::metrics::WRITE, &asn_info) + crate::metrics::packets_total( + crate::metrics::WRITE, + &asn_info, + locality, + ) + .inc(); + crate::metrics::bytes_total(crate::metrics::WRITE, &asn_info, locality) .inc_by(size as u64); } Err(error) => { - let source = error.to_string(); - crate::metrics::errors_total(crate::metrics::WRITE, &source, &asn_info) + let kind = crate::metrics::io_error_kind(&error); + crate::metrics::errors_total(crate::metrics::WRITE, kind, &asn_info) .inc(); crate::metrics::packets_dropped_total( crate::metrics::WRITE, - &source, - &asn_info, + crate::metrics::DropReason::SocketError, + "", + locality, ) .inc(); } @@ -223,11 +230,11 @@ fn spawn_poll_listener_impl( let packet = crate::net::packet::DownstreamPacket { contents: buffer, source, filters }; if let Some(last_received_at) = last_received_at { - crate::metrics::packet_jitter( + crate::metrics::set_packet_jitter( crate::metrics::READ, &crate::metrics::EMPTY, - ) - .set((received_at - last_received_at).nanos()); + (received_at - last_received_at).nanos(), + ); } last_received_at = Some(received_at); @@ -314,26 +321,32 @@ pub fn spawn_session( ); let (result, _) = ps_send_to(&socket2, packet.data, destination).await; let asn_info = packet.asn_info.as_ref().into(); + let locality = &*packet.cluster; match result { Ok(size) => { - crate::metrics::packets_total(crate::metrics::READ, &asn_info) - .inc(); - crate::metrics::bytes_total(crate::metrics::READ, &asn_info) - .inc_by(size as u64); - } - Err(error) => { - tracing::trace!(%error, "sending packet upstream failed"); - let source = error.to_string(); - crate::metrics::errors_total( + crate::metrics::packets_total( crate::metrics::READ, - &source, &asn_info, + locality, ) .inc(); - crate::metrics::packets_dropped_total( + crate::metrics::bytes_total( crate::metrics::READ, - &source, &asn_info, + locality, + ) + .inc_by(size as u64); + } + Err(error) => { + tracing::trace!(%error, "sending packet upstream failed"); + let kind = crate::metrics::io_error_kind(&error); + crate::metrics::errors_total(crate::metrics::READ, kind, &asn_info) + .inc(); + crate::metrics::packets_dropped_total( + crate::metrics::READ, + crate::metrics::DropReason::SocketError, + "", + locality, ) .inc(); } diff --git a/src/net/packet.rs b/src/net/packet.rs index 64c14a20f2..5a4dc9c810 100644 --- a/src/net/packet.rs +++ b/src/net/packet.rs @@ -114,7 +114,7 @@ impl DownstreamPacket<'_, P> { worker_id: usize, config: &Arc, sessions: &S, - destinations: &mut Vec, + destinations: &mut Vec, ) { tracing::trace!( id = worker_id, @@ -125,10 +125,14 @@ impl DownstreamPacket<'_, P> { let timer = metrics::processing_time(metrics::READ).start_timer(); if let Err(error) = self.process_inner(config, sessions, destinations) { - let discriminant = error.discriminant(); - error.inc_system_errors_total(metrics::READ, &metrics::EMPTY); - metrics::packets_dropped_total(metrics::READ, discriminant, &metrics::EMPTY).inc(); + metrics::packets_dropped_total( + metrics::READ, + error.drop_reason(), + error.filter_name(), + "", + ) + .inc(); } timer.stop_and_record(); @@ -140,7 +144,7 @@ impl DownstreamPacket<'_, P> { self, config: &Arc, sessions: &S, - destinations: &mut Vec, + destinations: &mut Vec, ) -> Result<(), PipelineError> { let Some(clusters) = config .dyn_cfg @@ -191,18 +195,18 @@ impl DownstreamPacket<'_, P> { { let session_key = SessionKey { source: self.source, - dest: dest.to_socket_addr()?, + dest: dest.address.to_socket_addr()?, }; - sessions.send(session_key, contents)?; + sessions.send(session_key, contents, dest.cluster)?; } else { - for epa in destinations.drain(0..) { + for dest in destinations.drain(0..) { let session_key = SessionKey { source: self.source, - dest: epa.to_socket_addr()?, + dest: dest.address.to_socket_addr()?, }; - sessions.send(session_key, contents.clone())?; + sessions.send(session_key, contents.clone(), dest.cluster)?; } } @@ -217,7 +221,7 @@ pub fn bench_process_packet( source: std::net::SocketAddr, config: &Arc, filter_chain: &crate::filters::FilterChain, - destinations: &mut Vec, + destinations: &mut Vec, ) { struct Noop; impl crate::net::sessions::SessionManager for Noop { @@ -225,6 +229,7 @@ pub fn bench_process_packet( &self, _key: crate::net::sessions::SessionKey, _contents: bytes::Bytes, + _cluster: Option, ) -> Result<(), crate::net::PipelineError> { Ok(()) } @@ -305,6 +310,7 @@ mod tests { let session_manager = SessionPool::new( vec![], cached_filter_chain, + None, usize::MAX, crate::net::io::UdpBackend::default(), ); diff --git a/src/net/packet/queue.rs b/src/net/packet/queue.rs index 057b064f92..433a3dd49d 100644 --- a/src/net/packet/queue.rs +++ b/src/net/packet/queue.rs @@ -127,4 +127,8 @@ pub struct SendPacket { pub data: bytes::Bytes, /// The asn info for the sender, used for metrics pub asn_info: Option, + /// The cluster this packet belongs to, as a label value, so traffic can be + /// attributed to a gameserver fleet. Empty when the destination didn't come + /// from a cluster with a locality. + pub cluster: std::sync::Arc, } diff --git a/src/net/phoenix.rs b/src/net/phoenix.rs index 89dfd9a297..cdf4cfc359 100644 --- a/src/net/phoenix.rs +++ b/src/net/phoenix.rs @@ -247,7 +247,24 @@ impl From<(i64, i64)> for DistanceMeasure { } } +/// Upper bound on a plausible one-way latency between two datacenters. +/// +/// A peer that replies with a zero or badly skewed timestamp yields a leg on the +/// order of the unix epoch in nanos. Feeding that to a cumulative histogram +/// corrupts its `_sum` permanently and wrecks the coordinate solve, so +/// measurements outside the range are discarded rather than clamped. +const MAX_PLAUSIBLE_LATENCY: Duration = Duration::from_secs(10); + impl DistanceMeasure { + /// Whether both legs fall in the range a network path can produce. + #[inline] + pub fn is_plausible(self) -> bool { + let max = MAX_PLAUSIBLE_LATENCY.as_nanos() as i64; + let leg = |nanos: i64| (0..=max).contains(&nanos); + + leg(self.incoming.nanos()) && leg(self.outgoing.nanos()) + } + #[inline] pub fn total_nanos(self) -> i64 { self.incoming.nanos() + self.outgoing.nanos() @@ -476,6 +493,15 @@ impl Phoenix { }; match result { + Ok(distance) if !distance.is_plausible() => { + tracing::warn!( + %address, + incoming_nanos = distance.incoming.nanos(), + outgoing_nanos = distance.outgoing.nanos(), + "discarding implausible measurement, the peer's timestamps are wrong" + ); + node.increase_error_estimate(); + } Ok(distance) => { crate::metrics::phoenix_measurement_seconds(node.icao_code, "incoming") .observe(distance.incoming.duration().as_secs_f64()); @@ -991,6 +1017,37 @@ mod tests { ); } + #[test] + fn implausible_measurements_are_rejected() { + assert!(DistanceMeasure::from((500_000, 1_500_000)).is_plausible()); + assert!(DistanceMeasure::default().is_plausible()); + + // A peer replying with a zero timestamp makes a leg the size of the unix + // epoch in nanos + let epoch_nanos = 1_730_000_000_000_000_000; + assert!(!DistanceMeasure::from((500_000, epoch_nanos)).is_plausible()); + assert!(!DistanceMeasure::from((epoch_nanos, 500_000)).is_plausible()); + // Clock skew the other way + assert!(!DistanceMeasure::from((-1, 500_000)).is_plausible()); + } + + #[tokio::test] + async fn implausible_measurements_are_not_recorded() { + // A peer replying with a zero timestamp makes a leg the size of the unix + // epoch in nanos + let epoch_nanos = 1_730_000_000_000_000_000; + let address: SocketAddr = "127.0.0.1:8080".parse().unwrap(); + let phoenix = Phoenix::new(MockMeasurement { + latencies: HashMap::from([(address, DistanceMeasure::from((500_000, epoch_nanos)))]), + }); + phoenix.add_node(address, abcd()); + + phoenix.measure_all_nodes().await; + + // Discarded rather than clamped, so it can't skew the coordinate solve + assert!(phoenix.ordered_nodes_by_latency().is_empty()); + } + #[tokio::test] async fn successful_measurements() { let latencies = HashMap::from([ diff --git a/src/net/sessions.rs b/src/net/sessions.rs index 269380da94..ab3bbfe11e 100644 --- a/src/net/sessions.rs +++ b/src/net/sessions.rs @@ -40,13 +40,33 @@ use crate::{ use parking_lot::RwLock; pub(crate) mod inner_metrics; +pub mod quality; pub type SessionMap = crate::collections::ttl::TtlMap; +/// What the send path needs from a session, all of it resolved when the session +/// was created. +pub(crate) struct SessionRoute { + /// `GeoIP` information for the client, for metrics. + pub asn_info: Option, + /// The cluster routing selected the destination from, as a label value. + /// Empty for a cluster configured without a locality, or for a destination + /// that didn't come from the cluster map at all. + pub destination: Arc, + pub pending_sends: PacketQueueSender, +} + /// Responsible for managing sending processed traffic to its destination and /// tracking metrics and other information about the session. pub trait SessionManager { - fn send(&self, key: SessionKey, contents: bytes::Bytes) -> Result<(), super::PipelineError>; + /// Sends `contents` upstream, `cluster` being the locality routing selected + /// the destination from. + fn send( + &self, + key: SessionKey, + contents: bytes::Bytes, + cluster: Option, + ) -> Result<(), super::PipelineError>; } #[derive(PartialEq, Eq, Hash)] @@ -94,6 +114,10 @@ pub struct SessionPool { downstream_sends: Vec, downstream_index: atomic::AtomicUsize, cached_filter_chain: CachedFilterChain, + /// Used to tell an endpoint disappearing from underneath a session apart from + /// a player going quiet. Sessions are never reported as `endpoint_gone` when + /// unset. + clusters: Option>, max_sessions: usize, backend: crate::net::io::UdpBackend, pub ring_buffer_len: u16, @@ -106,6 +130,10 @@ struct SocketStorage { destination_to_sources: HashMap<(SocketAddr, u16), SocketAddr>, sources_to_asn_info: HashMap, sockets_to_destination: HashMap>, + /// The cluster each destination belongs to, recorded when a session to it is + /// created, since a packet arriving from upstream carries no routing decision + /// to read it from. + destination_to_cluster: HashMap>, } impl SessionPool { @@ -115,6 +143,7 @@ impl SessionPool { pub fn new( downstream_sends: Vec, cached_filter_chain: CachedFilterChain, + clusters: Option>, max_sessions: usize, backend: crate::net::io::UdpBackend, ring_buffer_len: u16, @@ -129,6 +158,7 @@ impl SessionPool { downstream_sends, downstream_index: atomic::AtomicUsize::new(0), cached_filter_chain, + clusters, max_sessions, backend, ring_buffer_len, @@ -139,7 +169,8 @@ impl SessionPool { fn create_new_session_from_new_socket( self: &Arc, key: SessionKey, - ) -> Result<(Option, PacketQueueSender), super::PipelineError> { + cluster: Option, + ) -> Result { tracing::trace!(source=%key.source, dest=%key.dest, "creating new socket for session"); let raw_socket = crate::net::raw_socket_with_reuse(0)?; let port = raw_socket @@ -159,7 +190,7 @@ impl SessionPool { self.ports_to_sockets .write() .insert(port, pending_sends.clone()); - self.create_session_from_existing_socket(key, pending_sends, port) + self.create_session_from_existing_socket(key, pending_sends, port, cluster) } pub(crate) fn process_received_upstream_packet( @@ -172,7 +203,7 @@ impl SessionPool { ) { let received_at = UtcTimestamp::now(); recv_addr.set_ip(recv_addr.ip().to_canonical()); - let (downstream_addr, asn_info): (SocketAddr, Option) = { + let (downstream_addr, asn_info, cluster) = { let storage = self.storage.read(); let Some(downstream_addr) = storage.destination_to_sources.get(&(recv_addr, port)) else { @@ -181,20 +212,38 @@ impl SessionPool { }; let asn_info = storage.sources_to_asn_info.get(downstream_addr); - (*downstream_addr, asn_info.map(MetricsIpNetEntry::from)) + ( + *downstream_addr, + asn_info.map(MetricsIpNetEntry::from), + storage + .destination_to_cluster + .get(&recv_addr) + .cloned() + .unwrap_or_else(|| Arc::from("")), + ) }; let asn_metric_info = asn_info.as_ref().into(); if let Some(last_received_at) = last_received_at { - metrics::packet_jitter(metrics::WRITE, &asn_metric_info) - .set((received_at - *last_received_at).nanos()); + metrics::set_packet_jitter( + metrics::WRITE, + &asn_metric_info, + (received_at - *last_received_at).nanos(), + ); } *last_received_at = Some(received_at); let result = { let _timer = metrics::processing_time(metrics::WRITE).start_timer(); - Self::process_recv_packet(recv_addr, downstream_addr, asn_info, packet, filters) + Self::process_recv_packet( + recv_addr, + downstream_addr, + asn_info, + cluster, + packet, + filters, + ) }; match result { @@ -208,14 +257,18 @@ impl SessionPool { self.downstream_sends.get_unchecked(index).push(packet); } } - Err((asn_info, error)) => { + Err((asn_info, cluster, error)) => { error.log(); - let discriminant = error.discriminant(); let asn_metric_info = asn_info.as_ref().into(); - metrics::packets_dropped_total(metrics::WRITE, discriminant, &asn_metric_info) - .inc(); - metrics::errors_total(metrics::WRITE, discriminant, &asn_metric_info).inc(); + metrics::packets_dropped_total( + metrics::WRITE, + error.drop_reason(), + error.filter_name(), + &cluster, + ) + .inc(); + metrics::errors_total(metrics::WRITE, error.discriminant(), &asn_metric_info).inc(); } } } @@ -227,14 +280,20 @@ impl SessionPool { pub(crate) fn get( self: &Arc, key @ SessionKey { dest, .. }: SessionKey, - ) -> Result<(Option, PacketQueueSender), super::PipelineError> { + cluster: Option, + ) -> Result { tracing::trace!(source=%key.source, dest=%key.dest, "SessionPool::get"); // If we already have a session for the key pairing, return that session. if let Some(entry) = self.session_map.get(&key) { - return Ok(( - entry.asn_info.as_ref().map(MetricsIpNetEntry::from), - entry.pending_sends.clone(), - )); + // The only point on the downstream path holding the session, so also + // where its jitter estimate is updated + entry.quality.record_arrival(); + + return Ok(SessionRoute { + asn_info: entry.asn_info.as_ref().map(MetricsIpNetEntry::from), + destination: entry.destination.clone(), + pending_sends: entry.pending_sends.clone(), + }); } if self.session_map.len() >= self.max_sessions { @@ -256,7 +315,7 @@ impl SessionPool { let no_sockets = self.ports_to_sockets.read().is_empty(); return if no_sockets { // Initial case where we have no allocated or reserved sockets. - self.create_new_session_from_new_socket(key) + self.create_new_session_from_new_socket(key, cluster) } else { // Where we have no allocated sockets for a destination, assign // the first available one. @@ -268,7 +327,7 @@ impl SessionPool { .map(|(port, socket)| (*port, socket.clone())) .ok_or(SessionError::MissingAllocatedSocket)?; - self.create_session_from_existing_socket(key, sender, port) + self.create_session_from_existing_socket(key, sender, port, cluster) }; }; @@ -287,10 +346,10 @@ impl SessionPool { .get_mut(&dest) .ok_or(SessionError::MissingDestinationSocket)? .insert(port); - self.create_session_from_existing_socket(key, socket, port) + self.create_session_from_existing_socket(key, socket, port, cluster) } else { drop(storage); - self.create_new_session_from_new_socket(key) + self.create_new_session_from_new_socket(key, cluster) } } @@ -300,8 +359,23 @@ impl SessionPool { key: SessionKey, pending_sends: PacketQueueSender, socket_port: u16, - ) -> Result<(Option, PacketQueueSender), super::PipelineError> { + cluster: Option, + ) -> Result { tracing::trace!(source=%key.source, dest=%key.dest, "reusing socket for session"); + // Interned once per session rather than per packet: the label is the + // cluster routing chose, and a session's destination doesn't change + let destination: Arc = cluster + .as_ref() + .map_or("", |locality| locality.as_str()) + .into(); + + // Resolved once here rather than at close, so a destination that was never + // in the cluster map can't later look like one that vanished from it + let tracked = self + .clusters + .as_ref() + .is_some_and(|clusters| clusters.read().contains_endpoint(&key.dest)); + let asn_info = { let mut storage = self.storage.write(); storage @@ -318,6 +392,14 @@ impl SessionPool { .destination_to_sources .insert((key.dest, socket_port), key.source); + // The upstream receive path has no routing decision to read the + // cluster from, so it reads it back from here + drop( + storage + .destination_to_cluster + .insert(key.dest, destination.clone()), + ); + let asn_info = crate::net::maxmind_db::MaxmindDb::lookup(key.source.ip()); if let Some(asn_info) = &asn_info { @@ -337,33 +419,44 @@ impl SessionPool { socket_port, self.clone(), asn_info, + destination.clone(), + tracked, ); tracing::trace!("inserting session into map"); self.session_map.insert(key, session); tracing::trace!("session inserted"); - Ok((asn_metrics_info, pending_sends)) + Ok(SessionRoute { + asn_info: asn_metrics_info, + destination, + pending_sends, + }) } /// Processes a packet that is received by this session. + #[allow(clippy::type_complexity)] fn process_recv_packet( source: SocketAddr, dest: SocketAddr, asn_info: Option, + cluster: Arc, packet: P, filters: &crate::filters::FilterChain, - ) -> Result, Error)> { + ) -> Result, Arc, Error)> { tracing::trace!(%source, %dest, length = packet.len(), "received packet from upstream"); let mut context = crate::filters::WriteContext::new(source.into(), dest.into(), packet); if let Err(err) = filters.write(&mut context) { - return Err((asn_info, err.into())); + return Err((asn_info, cluster, err.into())); } Ok(SendPacket { data: context.contents.freeze(), destination: dest, asn_info, + // The traffic is from the gameserver this session is routed to, so it + // belongs to the same cluster as the downstream direction + cluster, }) } @@ -378,8 +471,9 @@ impl SessionPool { self: &Arc, key: SessionKey, packet: bytes::Bytes, + cluster: Option, ) -> Result<(), super::PipelineError> { - self.send_inner(key, packet)?; + self.send_inner(key, packet, cluster)?; Ok(()) } @@ -389,15 +483,21 @@ impl SessionPool { self: &Arc, key: SessionKey, packet: bytes::Bytes, + cluster: Option, ) -> Result { - let (asn_info, sender) = self.get(key)?; + let SessionRoute { + asn_info, + destination, + pending_sends, + } = self.get(key, cluster)?; - sender.push(SendPacket { + pending_sends.push(SendPacket { destination: key.dest, data: packet, asn_info, + cluster: destination, }); - Ok(sender) + Ok(pending_sends) } /// Spawns a session I/O loop for the given socket, dispatching to the @@ -482,13 +582,24 @@ impl SessionPool { // Not asserted because the source might not have GeoIP info. storage.sources_to_asn_info.remove(source); storage.destination_to_sources.remove(&(*dest, port)); + + // Only once no session is left using the destination, since the locality + // is shared by all of them + if !storage.destination_to_sockets.contains_key(dest) { + storage.destination_to_cluster.remove(dest); + } tracing::trace!("socket released"); } } impl SessionManager for Arc { - fn send(&self, key: SessionKey, contents: bytes::Bytes) -> Result<(), super::PipelineError> { - SessionPool::send(self, key, contents) + fn send( + &self, + key: SessionKey, + contents: bytes::Bytes, + cluster: Option, + ) -> Result<(), super::PipelineError> { + SessionPool::send(self, key, contents, cluster) } } @@ -513,6 +624,14 @@ pub struct Session { pending_sends: PacketQueueSender, /// The `GeoIP` information of the source. asn_info: Option, + /// The cluster routing selected the destination from, as a label value. + destination: Arc, + /// Whether the destination was in the cluster map when the session was + /// created. When it wasn't, its absence later says nothing. + tracked: bool, + /// Interarrival jitter of this session, folded into aggregate metrics by + /// [`quality::spawn_aggregator`]. + quality: quality::SessionQualityHandle, /// The socket pool of the session. pool: Arc, } @@ -524,13 +643,18 @@ impl Session { socket_port: u16, pool: Arc, asn_info: Option, + destination: Arc, + tracked: bool, ) -> Self { let s = Self { key, pending_sends, pool, socket_port, + quality: quality::SessionQualityHandle::register(asn_info.as_ref()), asn_info, + destination, + tracked, created_at: Instant::now(), }; @@ -556,10 +680,35 @@ impl Session { inner_metrics::active_sessions(self.asn_info.as_ref()) } + /// Why this session is ending. + /// + /// UDP has no close, so a player leaving is indistinguishable from one going + /// quiet and is reported as an idle timeout. What is worth separating out is + /// the endpoint having gone away underneath the session, and the proxy itself + /// going away. + fn close_reason(&self) -> inner_metrics::CloseReason { + if crate::metrics::shutdown_initiated().get() != 0 { + return inner_metrics::CloseReason::Shutdown; + } + + // Only meaningful for a destination that was in the cluster map when the + // session was created + if self.tracked + && let Some(clusters) = &self.pool.clusters + && !clusters.read().contains_endpoint(&self.key.dest) + { + return inner_metrics::CloseReason::EndpointGone; + } + + inner_metrics::CloseReason::IdleTimeout + } + fn release(&mut self) { + let reason = self.close_reason(); self.active_session_metric().dec(); + inner_metrics::sessions_closed_total(reason).inc(); inner_metrics::duration_secs().observe(self.created_at.elapsed().as_secs() as f64); - tracing::debug!(source = %self.key.source, dest_address = %self.key.dest, "Session closed"); + tracing::debug!(source = %self.key.source, dest_address = %self.key.dest, ?reason, "Session closed"); SessionPool::release_socket(self.pool.clone(), self.key, self.socket_port); } } @@ -597,6 +746,20 @@ impl Error { Self::Filter(fe) => fe.discriminant(), } } + + #[inline] + pub fn drop_reason(&self) -> crate::metrics::DropReason { + match self { + Self::Filter(fe) => fe.drop_reason(), + } + } + + #[inline] + pub fn filter_name(&self) -> &'static str { + match self { + Self::Filter(fe) => fe.filter_name(), + } + } } impl Loggable for Error { @@ -620,6 +783,7 @@ mod tests { SessionPool::new( vec![pending_sends.clone()], fake.cached(), + None, usize::MAX, backend, 64, @@ -628,6 +792,123 @@ mod tests { ) } + /// A pool with a cluster map holding `dest`, so destination attribution and + /// endpoint-gone detection are both live. + async fn new_pool_with_cluster( + dest: SocketAddr, + locality: Option, + ) -> ( + Arc, + crate::config::Watch, + ) { + let backend = crate::net::io::UdpBackend::default(); + let (pending_sends, _srecv) = crate::net::queue(1, backend).unwrap(); + let fake = crate::config::filter::FilterChainConfig::default(); + + let clusters = crate::config::Watch::new(crate::net::ClusterMap::default()); + clusters.read().insert( + None, + locality, + [crate::net::endpoint::Endpoint::new(dest.into())].into(), + ); + + let pool = SessionPool::new( + vec![pending_sends], + fake.cached(), + Some(clusters.clone()), + usize::MAX, + backend, + 64, + ); + + (pool, clusters) + } + + #[tokio::test] + async fn sessions_carry_the_cluster_routing_chose() { + let dest: SocketAddr = (std::net::Ipv4Addr::LOCALHOST, 8090u16).into(); + let locality = crate::net::endpoint::Locality::with_region("session-locality-test"); + let (pool, _clusters) = new_pool_with_cluster(dest, Some(locality.clone())).await; + + let key: SessionKey = ((std::net::Ipv4Addr::LOCALHOST, 8091u16).into(), dest).into(); + let route = pool.get(key, Some(locality.clone())).unwrap(); + + assert_eq!(&*route.destination, locality.to_string().as_str()); + } + + #[tokio::test] + async fn a_destination_outside_the_cluster_map_is_not_tracked() { + let known: SocketAddr = (std::net::Ipv4Addr::LOCALHOST, 8092u16).into(); + let (pool, _clusters) = new_pool_with_cluster(known, None).await; + + let unknown: SocketAddr = (std::net::Ipv4Addr::LOCALHOST, 8093u16).into(); + let key: SessionKey = ((std::net::Ipv4Addr::LOCALHOST, 8094u16).into(), unknown).into(); + drop(pool.get(key, None).unwrap()); + + // Never in the cluster map, so its absence at close says nothing + assert!(!pool.session_map.get(&key).unwrap().tracked); + } + + #[tokio::test] + async fn a_destination_routed_without_a_cluster_still_detects_it_vanishing() { + let dest: SocketAddr = (std::net::Ipv4Addr::LOCALHOST, 8099u16).into(); + let locality = crate::net::endpoint::Locality::with_region("no-cluster-gone-test"); + let (pool, clusters) = new_pool_with_cluster(dest, Some(locality.clone())).await; + + // Routed without a cluster, as the decryptor filter does when it decodes a + // destination out of the packet itself + let key: SessionKey = ((std::net::Ipv4Addr::LOCALHOST, 8100u16).into(), dest).into(); + let route = pool.get(key, None).unwrap(); + + assert!(route.destination.is_empty()); + + clusters.read().remove_locality(None, &Some(locality)); + assert_eq!( + pool.session_map.get(&key).unwrap().close_reason(), + inner_metrics::CloseReason::EndpointGone + ); + } + + #[tokio::test] + async fn a_session_whose_endpoint_vanished_closes_as_endpoint_gone() { + let dest: SocketAddr = (std::net::Ipv4Addr::LOCALHOST, 8095u16).into(); + let locality = crate::net::endpoint::Locality::with_region("endpoint-gone-test"); + let (pool, clusters) = new_pool_with_cluster(dest, Some(locality.clone())).await; + + let key: SessionKey = ((std::net::Ipv4Addr::LOCALHOST, 8096u16).into(), dest).into(); + drop(pool.get(key, None).unwrap()); + + let session = pool.session_map.get(&key).unwrap(); + assert_eq!( + session.close_reason(), + inner_metrics::CloseReason::IdleTimeout + ); + + clusters.read().remove_locality(None, &Some(locality)); + assert_eq!( + session.close_reason(), + inner_metrics::CloseReason::EndpointGone + ); + } + + #[tokio::test] + async fn a_session_records_arrivals_for_its_jitter_estimate() { + let (pool, _receiver) = new_pool().await; + let key: SessionKey = ( + (std::net::Ipv4Addr::LOCALHOST, 8097u16).into(), + (std::net::Ipv4Addr::UNSPECIFIED, 8098u16).into(), + ) + .into(); + + // The first `get` creates the session, subsequent ones are packets + drop(pool.get(key, None).unwrap()); + drop(pool.get(key, None).unwrap()); + drop(pool.get(key, None).unwrap()); + + let session = pool.session_map.get(&key).unwrap(); + assert_eq!(session.quality.packets_since_last_sample(), 2); + } + #[tokio::test] async fn insert_and_release_single_socket() { let (pool, _receiver) = new_pool().await; @@ -637,7 +918,7 @@ mod tests { ) .into(); - let _session = pool.get(key).unwrap(); + let _session = pool.get(key, None).unwrap(); assert!(pool.drop_session(key).await); @@ -658,8 +939,8 @@ mod tests { ) .into(); - let _session1 = pool.get(key1).unwrap(); - let _session2 = pool.get(key2).unwrap(); + let _session1 = pool.get(key1, None).unwrap(); + let _session2 = pool.get(key2, None).unwrap(); assert!(pool.drop_session(key1).await); assert!(!pool.has_no_allocated_sockets()); @@ -683,8 +964,8 @@ mod tests { ) .into(); - let _socket1 = pool.get(key1).unwrap(); - let _socket2 = pool.get(key2).unwrap(); + let _socket1 = pool.get(key1, None).unwrap(); + let _socket2 = pool.get(key2, None).unwrap(); assert_ne!( pool.session_map.get(&key1).unwrap().socket_port, pool.session_map.get(&key2).unwrap().socket_port @@ -708,8 +989,8 @@ mod tests { ) .into(); - let _socket1 = pool.get(key1).unwrap(); - let _socket2 = pool.get(key2).unwrap(); + let _socket1 = pool.get(key1, None).unwrap(); + let _socket2 = pool.get(key2, None).unwrap(); assert_eq!( pool.session_map.get(&key1).unwrap().socket_port, @@ -731,13 +1012,13 @@ mod tests { ) .into(); - let socket1 = pool.get(key1).unwrap(); + let socket1 = pool.get(key1, None).unwrap(); let task = tokio::spawn(async move { drop(socket1); }); - let _socket2 = pool.get(key2).unwrap(); + let _socket2 = pool.get(key2, None).unwrap(); task.await.unwrap(); } @@ -756,13 +1037,13 @@ mod tests { ) .into(); - let socket1 = pool.get(key1).unwrap(); + let socket1 = pool.get(key1, None).unwrap(); let task = tokio::spawn(async move { drop(socket1); }); - let _socket2 = pool.get(key2).unwrap(); + let _socket2 = pool.get(key2, None).unwrap(); task.await.unwrap(); } @@ -775,6 +1056,7 @@ mod tests { SessionPool::new( vec![pending_sends.clone()], fake.cached(), + None, limit, backend, 64, @@ -795,11 +1077,11 @@ mod tests { async fn session_limit_rejects_new_sessions_at_capacity() { let (pool, _receiver) = pool_with_limit(2); - assert!(pool.get(key(8080, 9000)).is_ok()); - assert!(pool.get(key(8081, 9000)).is_ok()); + assert!(pool.get(key(8080, 9000), None).is_ok()); + assert!(pool.get(key(8081, 9000), None).is_ok()); assert!(matches!( - pool.get(key(8082, 9000)), + pool.get(key(8082, 9000), None), Err(super::super::PipelineError::SessionLimit) )); } @@ -808,32 +1090,32 @@ mod tests { async fn session_limit_allows_existing_session_at_capacity() { let (pool, _receiver) = pool_with_limit(1); - assert!(pool.get(key(8080, 9000)).is_ok()); + assert!(pool.get(key(8080, 9000), None).is_ok()); // Limit hit — new session rejected. assert!(matches!( - pool.get(key(8081, 9000)), + pool.get(key(8081, 9000), None), Err(super::super::PipelineError::SessionLimit) )); // But re-fetching the existing session must still succeed. - assert!(pool.get(key(8080, 9000)).is_ok()); + assert!(pool.get(key(8080, 9000), None).is_ok()); } #[tokio::test] async fn session_limit_recovers_after_session_removed() { let (pool, _receiver) = pool_with_limit(1); - assert!(pool.get(key(8080, 9000)).is_ok()); + assert!(pool.get(key(8080, 9000), None).is_ok()); assert!(matches!( - pool.get(key(8081, 9000)), + pool.get(key(8081, 9000), None), Err(super::super::PipelineError::SessionLimit) )); pool.drop_session(key(8080, 9000)).await; // Slot freed — a new session should be accepted. - assert!(pool.get(key(8081, 9000)).is_ok()); + assert!(pool.get(key(8081, 9000), None).is_ok()); } #[tokio::test] @@ -853,7 +1135,7 @@ mod tests { let msg = b"helloworld"; let pending = pool - .send_inner(key, bytes::Bytes::from_static(msg)) + .send_inner(key, bytes::Bytes::from_static(msg), None) .unwrap(); let pending = pending.swap(Vec::new()); diff --git a/src/net/sessions/inner_metrics.rs b/src/net/sessions/inner_metrics.rs index b4c84f09a7..7a1c1783e5 100644 --- a/src/net/sessions/inner_metrics.rs +++ b/src/net/sessions/inner_metrics.rs @@ -78,6 +78,47 @@ pub(crate) fn sessions_rejected_total() -> &'static IntCounter { &SESSIONS_REJECTED_TOTAL } +/// Why a session ended. +/// +/// A fall in session count looks the same whether players left or their +/// endpoints vanished, and those want different responses. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum CloseReason { + /// No traffic within the session TTL. UDP has no close, so this is how a + /// player leaving normally looks. + IdleTimeout, + /// The endpoint the session was routed to is no longer in the cluster map. + EndpointGone, + /// The proxy is shutting down. + Shutdown, +} + +impl CloseReason { + fn label(self) -> &'static str { + match self { + Self::IdleTimeout => "idle_timeout", + Self::EndpointGone => "endpoint_gone", + Self::Shutdown => "shutdown", + } + } +} + +pub(crate) fn sessions_closed_total(reason: CloseReason) -> prometheus::IntCounter { + static SESSIONS_CLOSED: Lazy = Lazy::new(|| { + prometheus::register_int_counter_vec_with_registry! { + Opts::new( + "quilkin_sessions_closed_total", + "total number of sessions closed, by the reason they ended", + ), + &[crate::metrics::REASON_LABEL], + crate::metrics::registry(), + } + .unwrap() + }); + + SESSIONS_CLOSED.with_label_values(&[reason.label()]) +} + pub(crate) fn duration_secs() -> &'static Histogram { static DURATION_SECS: Lazy = Lazy::new(|| { register( diff --git a/src/net/sessions/quality.rs b/src/net/sessions/quality.rs new file mode 100644 index 0000000000..82762f099f --- /dev/null +++ b/src/net/sessions/quality.rs @@ -0,0 +1,775 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +//! Per-session connection quality, and the aggregation that turns it into +//! bounded metrics. +//! +//! A session's jitter and its client's ASN are per-player facts, so neither can +//! be a metric label: concurrent sessions and the ~9700 ASNs seen in a day of +//! traffic both blow up cardinality. Instead every session registers a +//! [`SessionQuality`] here, the I/O paths stamp packet arrivals onto it, and +//! [`spawn_aggregator`] periodically folds the whole registry into a jitter +//! histogram plus per-ASN aggregates whose series count is bounded by +//! configuration rather than by traffic. + +use std::{ + collections::{HashMap, HashSet}, + sync::{ + Arc, + atomic::{AtomicI64, AtomicU64, Ordering::Relaxed}, + }, + time::Duration, +}; + +use once_cell::sync::Lazy; + +use crate::{metrics, net::maxmind_db::IpNetEntry}; + +/// Divisor of the RFC 3550 interarrival jitter estimator, which weights the +/// estimate towards recent packet pairs. +const JITTER_GAIN: i64 = 16; + +/// Interarrival gap beyond which a packet pair is treated as the stream +/// restarting rather than as jitter. +/// +/// RFC 3550 assumes a continuous media stream. Game traffic pauses — loading +/// screens, alt-tabs, backgrounded mobile clients — and feeding a multi-second +/// gap to the estimator spikes it by gap/16, which against a 30 ms threshold +/// reads as a badly degraded player rather than one who stopped sending. +const MAX_INTERARRIVAL_NANOS: i64 = 1_000_000_000; + +/// Monotonic nanoseconds since the first call. +/// +/// Interarrival must not be measured against the wall clock: an NTP step +/// backwards yields a negative delta, and the estimator turns that into a jitter +/// spike the size of the step. +#[inline] +fn monotonic_nanos() -> i64 { + static START: Lazy = Lazy::new(std::time::Instant::now); + + // i64 nanos covers 292 years of uptime + START.elapsed().as_nanos() as i64 +} + +/// Packets a session needs within an aggregation interval for its jitter +/// estimate to be worth recording. Two packets give one interarrival delta and +/// no variation to compare it against. +const MIN_PACKETS_FOR_JITTER: u64 = 3; + +/// Label value used for sessions whose client IP resolved to no ASN, either +/// because no maxmind database is loaded or because the address isn't in it. +const UNKNOWN_ASN: &str = "unknown"; + +/// Label value carrying the sessions of every ASN outside the exported top N, +/// so per-ASN shares still sum to the session total. +const REMAINDER_ASN: &str = "other"; + +/// Interarrival jitter of one session, and the ASN of the client it belongs to. +/// +/// Updated from the I/O paths on every downstream packet and read by the +/// aggregator, so all access is via relaxed atomics. Packets for one session +/// normally arrive on a single worker, but nothing guarantees it; a concurrent +/// update perturbs the estimate for one packet pair and is not worth locking +/// the hot path to prevent. +pub struct SessionQuality { + /// ASN of the client, `None` when it couldn't be resolved. + asn: Option, + /// Arrival of the most recent downstream packet in unix nanos, 0 before the + /// first one. + last_arrival: AtomicI64, + /// Interarrival delta of the previous packet pair, in nanos. + last_delta: AtomicI64, + /// RFC 3550 interarrival jitter estimate, in nanos. + jitter: AtomicI64, + /// Downstream packets seen since the last aggregation. + packets: AtomicU64, +} + +impl SessionQuality { + fn new(asn: Option) -> Self { + Self { + asn, + last_arrival: AtomicI64::new(0), + last_delta: AtomicI64::new(0), + jitter: AtomicI64::new(0), + packets: AtomicU64::new(0), + } + } + + /// Records the arrival of a downstream packet, updating the session's jitter + /// estimate. + #[inline] + pub fn record_arrival(&self) { + self.record_arrival_at(monotonic_nanos()); + } + + /// [`Self::record_arrival`] against a caller supplied monotonic reading, so + /// the estimator can be exercised without the clock. + #[inline] + fn record_arrival_at(&self, now: i64) { + self.packets.fetch_add(1, Relaxed); + + let previous_arrival = self.last_arrival.swap(now, Relaxed); + if previous_arrival == 0 { + return; + } + + // A non-positive delta means two threads interleaved on this session, and + // an oversized one means the stream paused. Neither is jitter, and both + // would inflate the estimate, so the pair is dropped and the next one + // measures from here. + let delta = now - previous_arrival; + if delta <= 0 || delta > MAX_INTERARRIVAL_NANOS { + self.last_delta.store(0, Relaxed); + return; + } + + let previous_delta = self.last_delta.swap(delta, Relaxed); + if previous_delta == 0 { + return; + } + + // J += (|D(i-1, i)| - J) / 16, per RFC 3550 A.8. Both operands are + // non-negative and bounded by MAX_INTERARRIVAL_NANOS, so the estimate + // cannot go negative and overflow the unsigned conversion at sampling. + let deviation = (delta - previous_delta).abs(); + let jitter = self.jitter.load(Relaxed); + self.jitter + .store(jitter + (deviation - jitter) / JITTER_GAIN, Relaxed); + } + + #[cfg(test)] + pub(crate) fn packets_since_last_sample(&self) -> u64 { + self.packets.load(Relaxed) + } + + /// Jitter estimate in nanos, and the packets seen since the previous call, + /// which this resets. + fn take_sample(&self) -> (i64, u64) { + (self.jitter.load(Relaxed), self.packets.swap(0, Relaxed)) + } +} + +/// Every live session's quality state, keyed by a registration id. +/// +/// Sessions insert and remove themselves once each, so the cost lands on +/// session churn rather than on the packet path. +static REGISTRY: Lazy>> = Lazy::new(<_>::default); + +static NEXT_ID: AtomicU64 = AtomicU64::new(0); + +/// Keeps a session's [`SessionQuality`] in the registry for as long as the +/// session lives. +pub struct SessionQualityHandle { + id: u64, + quality: Arc, +} + +impl SessionQualityHandle { + /// Registers quality tracking for a session whose client resolved to `asn`. + pub fn register(asn: Option<&IpNetEntry>) -> Self { + // ASNs are 32-bit, the wider maxmind field is truncated rather than + // dropped so a malformed database can't lose the whole session + Self::register_for_asn(asn.map(|entry| entry.id as u32)) + } + + fn register_for_asn(asn: Option) -> Self { + let id = NEXT_ID.fetch_add(1, Relaxed); + let quality = Arc::new(SessionQuality::new(asn)); + REGISTRY.insert(id, quality.clone()); + + Self { id, quality } + } +} + +impl std::ops::Deref for SessionQualityHandle { + type Target = SessionQuality; + + #[inline] + fn deref(&self) -> &Self::Target { + &self.quality + } +} + +impl Drop for SessionQualityHandle { + fn drop(&mut self) { + drop(REGISTRY.remove(&self.id)); + } +} + +/// Tunables for [`spawn_aggregator`]. +#[derive(Clone, Copy, Debug)] +pub struct AggregationConfig { + /// How often the registry is folded into metrics. + pub interval: Duration, + /// Fraction of sessions whose jitter is recorded into the histogram each + /// interval, in `0.0..=1.0`. Lowering it trades resolution for time spent in + /// the aggregation. + pub sample_fraction: f64, + /// Number of client ASNs to report, largest first, with the rest folded into + /// a remainder bucket. 0 disables per-ASN reporting. + pub top_asns: usize, + /// Jitter at or above which a session counts as degraded. + pub jitter_threshold: Duration, +} + +impl Default for AggregationConfig { + fn default() -> Self { + Self { + interval: Duration::from_secs(15), + sample_fraction: 1.0, + top_asns: 32, + jitter_threshold: Duration::from_millis(30), + } + } +} + +impl AggregationConfig { + /// Rejects values that would silently produce meaningless metrics. + pub fn validate(&self) -> eyre::Result<()> { + if self.interval.is_zero() { + eyre::bail!("session metrics interval must be at least a second"); + } + + if !(0.0..=1.0).contains(&self.sample_fraction) { + eyre::bail!( + "session metrics sample fraction must be between 0 and 1, got {}", + self.sample_fraction + ); + } + + if self.jitter_threshold.is_zero() { + eyre::bail!("session metrics jitter threshold must be non-zero"); + } + + Ok(()) + } +} + +/// The `reason` label value for the only quality signal the proxy can derive. +/// +/// Loss and RTT on the client's leg need either packet sequence numbers or a +/// client-side timestamp, and the proxy has neither. +const REASON_JITTER: &str = "jitter"; + +/// What one interval measured for a single ASN. +#[derive(Default)] +struct AsnSample { + /// Every session of this ASN, whether or not it carried traffic. + sessions: usize, + /// Sessions that carried enough packets to judge. A session gone quiet says + /// nothing about its client's connection. + judged: usize, + /// Judged sessions at or above the jitter threshold. + degraded: usize, +} + +impl AsnSample { + fn merge(&mut self, other: &Self) { + self.sessions += other.sessions; + self.judged += other.judged; + self.degraded += other.degraded; + } +} + +/// One session's contribution, copied out of the registry before any metric is +/// touched. +struct Sample { + asn: Option, + jitter_nanos: i64, + packets: u64, +} + +/// Folds the session registry into metrics, holding the state that spans +/// intervals. +/// +/// Whether an individual session is having a bad time is a judgement the proxy +/// can make, since it holds that session's packet timing. Whether an *ISP* is +/// having a bad time is not: a pod carries around a hundred concurrent sessions +/// spread over thousands of ASNs, so no single proxy sees enough of any one ASN +/// to threshold on. This exports the numerator and denominator per ASN and leaves +/// that decision to whatever sums them across the fleet. +struct Aggregator { + config: AggregationConfig, + /// ASN label values currently exported, so an ASN dropping out of the + /// reported set has its series removed rather than left at a stale value. + exported_asns: HashSet, + /// ASN label values currently exported as having degraded sessions. + exported_degraded: HashSet, + /// `quilkin_packet_jitter` observation counts per direction as of the + /// previous interval, used to spot a gauge that has gone stale. + last_jitter_observations: [u64; 2], + /// Reused across intervals so folding the registry doesn't allocate + /// proportionally to the session count every time. + samples: Vec, +} + +impl Aggregator { + fn new(config: AggregationConfig) -> Self { + Self { + config, + exported_asns: HashSet::new(), + exported_degraded: HashSet::new(), + last_jitter_observations: [0, 0], + samples: Vec::new(), + } + } + + fn tick(&mut self) { + // Copied out first: iterating the registry holds a shard lock, and session + // registration needs that same lock on the packet path, so no metric work + // happens while it's held + self.samples.clear(); + self.samples.extend(REGISTRY.iter().map(|entry| { + let (jitter_nanos, packets) = entry.value().take_sample(); + Sample { + asn: entry.value().asn, + jitter_nanos, + packets, + } + })); + + let jitter_threshold = self.config.jitter_threshold.as_nanos() as i64; + let mut per_asn: HashMap, AsnSample> = HashMap::new(); + + for sample in &self.samples { + let entry = per_asn.entry(sample.asn).or_default(); + entry.sessions += 1; + + if sample.packets < MIN_PACKETS_FOR_JITTER { + continue; + } + + entry.judged += 1; + + if sample.jitter_nanos >= jitter_threshold { + entry.degraded += 1; + } + + if sampled(self.config.sample_fraction) { + // Non-negative and bounded by construction, see `record_arrival` + metrics::session_jitter_seconds() + .observe(Duration::from_nanos(sample.jitter_nanos as u64).as_secs_f64()); + } + } + + self.update_asn_metrics(&per_asn); + self.prune_stale_jitter_gauges(); + } + + /// Exports session and degraded-session counts for the largest `top_asns` + /// ASNs, with everything else folded into a remainder bucket so both sum to + /// the proxy's totals. + fn update_asn_metrics(&mut self, per_asn: &HashMap, AsnSample>) { + let mut buckets: Vec<(String, AsnSample)> = Vec::new(); + + if self.config.top_asns > 0 { + let mut ranked: Vec<(String, &AsnSample)> = per_asn + .iter() + .map(|(asn, sample)| (asn_label(*asn), sample)) + .collect(); + // Ties broken by label so a pod's reported set doesn't churn between + // equally sized ASNs from one interval to the next + ranked.sort_unstable_by(|(a_label, a), (b_label, b)| { + b.sessions + .cmp(&a.sessions) + .then_with(|| a_label.cmp(b_label)) + }); + + let remainder = ranked.split_off(self.config.top_asns.min(ranked.len())); + buckets.extend( + ranked + .into_iter() + .map(|(label, sample)| (label, AsnSample { ..*sample })), + ); + + if !remainder.is_empty() { + let mut folded = AsnSample::default(); + for (_, sample) in remainder { + folded.merge(sample); + } + buckets.push((REMAINDER_ASN.to_owned(), folded)); + } + } + + let mut active = HashSet::with_capacity(buckets.len()); + let mut degraded = HashSet::new(); + let mut degraded_observations = 0; + + for (label, sample) in buckets { + metrics::sessions_active_by_asn(&label).set(sample.sessions as i64); + + // Only exported while non-zero, so a healthy fleet publishes nothing + // here and the series count follows the number of ISPs in trouble + if sample.degraded > 0 { + metrics::client_sessions_degraded(&label, REASON_JITTER) + .set(sample.degraded as i64); + degraded_observations += sample.degraded; + degraded.insert(label.clone()); + } + + active.insert(label); + } + + for stale in self.exported_asns.difference(&active) { + metrics::remove_sessions_active_by_asn(stale); + } + for stale in self.exported_degraded.difference(°raded) { + metrics::remove_client_sessions_degraded(stale, REASON_JITTER); + } + + self.exported_asns = active; + self.exported_degraded = degraded; + + if degraded_observations > 0 { + metrics::client_sessions_degraded_total(REASON_JITTER) + .inc_by(degraded_observations as u64); + } + } + + /// `quilkin_packet_jitter` is a gauge set per packet, so a proxy that stops + /// receiving keeps publishing the last value it saw indefinitely. Drop the + /// series when nothing observed it in the interval instead. + fn prune_stale_jitter_gauges(&mut self) { + for (index, direction) in [metrics::READ, metrics::WRITE].into_iter().enumerate() { + let observations = metrics::packet_jitter_observations(direction); + if observations == self.last_jitter_observations[index] { + metrics::remove_packet_jitter(direction); + } + self.last_jitter_observations[index] = observations; + } + } +} + +#[inline] +fn sampled(fraction: f64) -> bool { + fraction >= 1.0 || rand::random::() < fraction +} + +fn asn_label(asn: Option) -> String { + match asn { + Some(asn) => asn.to_string(), + None => UNKNOWN_ASN.to_owned(), + } +} + +/// Spawns the task that periodically folds session quality into metrics. +/// +/// The registry and the metrics it writes are process-wide, so a second +/// aggregator would fight the first over the same series. Repeat calls are +/// refused rather than silently adopting the first caller's configuration. +pub fn spawn_aggregator( + config: AggregationConfig, + shutdown: &mut crate::signal::ShutdownHandler, +) -> eyre::Result<()> { + static SPAWNED: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false); + + config.validate()?; + + if SPAWNED.swap(true, Relaxed) { + tracing::warn!("session metrics aggregation is already running, ignoring configuration"); + return Ok(()); + } + + // Registered up front so the series exists for any proxy serving UDP, rather + // than appearing only once a session has enough packets to sample + let _ = metrics::session_jitter_seconds(); + + let mut aggregator = Aggregator::new(config); + let finished = shutdown.push("session_metrics"); + let mut srx = shutdown.shutdown_rx(); + + tokio::spawn(async move { + let mut interval = tokio::time::interval(config.interval); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + + loop { + tokio::select! { + _ = interval.tick() => aggregator.tick(), + _ = srx.changed() => break, + } + } + + // So a scrape during drain doesn't see a value from before shutdown + aggregator.tick(); + drop(finished.send(Ok(()))); + }); + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn quality(asn: Option) -> SessionQuality { + SessionQuality { + asn, + last_arrival: AtomicI64::new(0), + last_delta: AtomicI64::new(0), + jitter: AtomicI64::new(0), + packets: AtomicU64::new(0), + } + } + + /// Feeds a sequence of interarrival gaps in milliseconds. + fn arrivals(q: &SessionQuality, gaps_ms: &[i64]) { + let mut now = 1; + for gap in gaps_ms { + now += gap * 1_000_000; + q.record_arrival_at(now); + } + } + + #[test] + fn steady_arrivals_have_no_jitter() { + let q = quality(None); + arrivals(&q, &[20; 10]); + + let (jitter, packets) = q.take_sample(); + assert_eq!(jitter, 0); + assert_eq!(packets, 10); + // The counter is a per-interval window + assert_eq!(q.take_sample().1, 0); + } + + #[test] + fn varying_arrivals_accumulate_jitter() { + let q = quality(Some(1)); + arrivals(&q, &[20, 60, 20, 70, 15, 80, 20, 65, 25, 75]); + + assert!(q.take_sample().0 > 0); + } + + #[test] + fn a_pause_in_the_stream_is_not_jitter() { + let q = quality(None); + // Steady, then the player alt-tabs for 30s, then steady again + arrivals(&q, &[20; 20]); + arrivals(&q, &[30_000]); + arrivals(&q, &[20; 20]); + + // Fed to the estimator the gap would read as ~1.9s of jitter, which + // against a 30ms threshold is a badly degraded player rather than one who + // stopped sending + assert_eq!(q.take_sample().0, 0); + } + + #[test] + fn a_clock_going_backwards_is_not_jitter() { + let q = quality(None); + arrivals(&q, &[20; 10]); + + // Two workers interleaving on one session, or a wall clock being stepped + q.record_arrival_at(1); + arrivals(&q, &[20; 10]); + + assert_eq!(q.take_sample().0, 0); + } + + #[test] + fn jitter_never_goes_negative() { + let q = quality(None); + // Alternating extremes, then settling: the estimate must stay in range for + // the unsigned conversion at sampling to be sound + arrivals(&q, &[1, 999, 1, 999, 1, 999]); + arrivals(&q, &[20; 50]); + + let (jitter, _) = q.take_sample(); + assert!((0..=MAX_INTERARRIVAL_NANOS).contains(&jitter), "{jitter}"); + } + + #[test] + fn handle_registration_is_scoped_to_the_session() { + let before = REGISTRY.len(); + let handle = SessionQualityHandle::register(None); + assert_eq!(REGISTRY.len(), before + 1); + + handle.record_arrival(); + drop(handle); + assert_eq!(REGISTRY.len(), before); + } + + /// `(asn, sessions, judged, degraded)` + fn samples(entries: &[(Option, usize, usize, usize)]) -> HashMap, AsnSample> { + entries + .iter() + .map(|(asn, sessions, judged, degraded)| { + ( + *asn, + AsnSample { + sessions: *sessions, + judged: *judged, + degraded: *degraded, + }, + ) + }) + .collect() + } + + #[test] + fn top_asns_are_capped_and_the_rest_land_in_the_remainder() { + let mut aggregator = Aggregator::new(AggregationConfig { + top_asns: 2, + ..<_>::default() + }); + + aggregator.update_asn_metrics(&samples(&[ + (Some(1), 10, 10, 0), + (Some(2), 5, 5, 0), + (Some(3), 3, 3, 0), + (Some(4), 1, 1, 0), + ])); + + assert_eq!( + aggregator.exported_asns, + ["1", "2", REMAINDER_ASN] + .into_iter() + .map(String::from) + .collect() + ); + assert_eq!(metrics::sessions_active_by_asn("1").get(), 10); + // Everything below the top 2 reconciles against the session total + assert_eq!(metrics::sessions_active_by_asn(REMAINDER_ASN).get(), 4); + } + + #[test] + fn degraded_sessions_are_reported_against_a_denominator() { + let mut aggregator = Aggregator::new(AggregationConfig { + top_asns: 4, + ..<_>::default() + }); + + // Six of this ASN's twenty judged sessions are having a bad time. Whether + // 30% is bad enough to act on is not a decision one pod can make, so both + // numbers are exported and neither is thresholded here. + aggregator.update_asn_metrics(&samples(&[(Some(5), 25, 20, 6)])); + + assert_eq!(metrics::sessions_active_by_asn("5").get(), 25); + assert_eq!( + metrics::client_sessions_degraded("5", REASON_JITTER).get(), + 6 + ); + assert!(aggregator.exported_degraded.contains("5")); + } + + #[test] + fn a_healthy_asn_exports_no_degraded_series() { + let mut aggregator = Aggregator::new(AggregationConfig { + top_asns: 4, + ..<_>::default() + }); + + aggregator.update_asn_metrics(&samples(&[(Some(6), 30, 30, 2)])); + assert!(aggregator.exported_degraded.contains("6")); + + // Recovered, so the series goes away rather than sitting at a stale count + aggregator.update_asn_metrics(&samples(&[(Some(6), 30, 30, 0)])); + assert!(aggregator.exported_degraded.is_empty()); + assert!(aggregator.exported_asns.contains("6")); + } + + #[test] + fn quiet_sessions_are_counted_but_not_judged() { + let mut aggregator = Aggregator::new(AggregationConfig { + top_asns: 1, + ..<_>::default() + }); + + // Sixty sessions of which twenty carried traffic: the session gauge counts + // them all, so it reconciles with `quilkin_session_active` + aggregator.update_asn_metrics(&samples(&[(Some(7), 60, 20, 15)])); + + assert_eq!(metrics::sessions_active_by_asn("7").get(), 60); + assert_eq!( + metrics::client_sessions_degraded("7", REASON_JITTER).get(), + 15 + ); + } + + #[test] + fn per_asn_reporting_can_be_turned_off() { + let mut aggregator = Aggregator::new(AggregationConfig { + top_asns: 0, + ..<_>::default() + }); + + aggregator.update_asn_metrics(&samples(&[(Some(8), 10, 10, 4)])); + + assert!(aggregator.exported_asns.is_empty()); + assert!(aggregator.exported_degraded.is_empty()); + } + + #[test] + fn a_tick_folds_registered_sessions_into_the_asn_gauges() { + let mut aggregator = Aggregator::new(AggregationConfig { + top_asns: 4, + jitter_threshold: Duration::from_millis(30), + ..<_>::default() + }); + + let steady = SessionQualityHandle::register_for_asn(Some(424242)); + let jittery = SessionQualityHandle::register_for_asn(Some(424242)); + let quiet = SessionQualityHandle::register_for_asn(Some(424242)); + + arrivals(&steady, &[20; 30]); + arrivals(&jittery, &[5, 90, 5, 90, 5, 90, 5, 90, 5, 90]); + // Below MIN_PACKETS_FOR_JITTER, so counted but not judged + jittery.record_arrival(); + quiet.record_arrival(); + + aggregator.tick(); + + assert_eq!(metrics::sessions_active_by_asn("424242").get(), 3); + assert_eq!( + metrics::client_sessions_degraded("424242", REASON_JITTER).get(), + 1 + ); + + drop((steady, jittery, quiet)); + aggregator.tick(); + + // Series withdrawn once the sessions are gone + assert!(!aggregator.exported_asns.contains("424242")); + assert!(!aggregator.exported_degraded.contains("424242")); + } + + #[test] + fn configuration_is_rejected_rather_than_clamped() { + assert!(AggregationConfig::default().validate().is_ok()); + + for invalid in [ + AggregationConfig { + sample_fraction: 5.0, + ..<_>::default() + }, + AggregationConfig { + sample_fraction: -1.0, + ..<_>::default() + }, + AggregationConfig { + interval: Duration::ZERO, + ..<_>::default() + }, + AggregationConfig { + jitter_threshold: Duration::ZERO, + ..<_>::default() + }, + ] { + assert!(invalid.validate().is_err(), "{invalid:?}"); + } + } +} diff --git a/src/service.rs b/src/service.rs index 0c0779e94e..28f5ea4dc8 100644 --- a/src/service.rs +++ b/src/service.rs @@ -100,6 +100,8 @@ pub struct Service { default_value_t = 256 )] pub session_pool_ring_buffer: u16, + #[clap(flatten)] + pub session_metrics: SessionMetricsCli, /// The UDP I/O backend to use. /// auto selects the best available: kernel (XDP) -> completion (io-uring) -> poll (epoll). #[clap( @@ -249,6 +251,72 @@ pub struct Service { pub type Finalizer = Box; +/// Options for the aggregation that turns per-session connection quality into +/// bounded metrics. +/// +/// Per-player and per-ISP breakdowns can't be labels, so the proxy aggregates +/// them internally and exports a projection whose series count these options +/// bound. +#[derive(Debug, Clone, clap::Parser)] +#[command(next_help_heading = "Session Metrics Options")] +pub struct SessionMetricsCli { + /// How often, in seconds, per-session connection quality is folded into + /// metrics. + #[clap( + long = "service.udp.metrics.interval", + env = "QUILKIN_SERVICE_UDP_METRICS_INTERVAL", + default_value_t = 15 + )] + pub interval_secs: u64, + /// Fraction of sessions whose jitter is recorded into + /// `quilkin_session_jitter_seconds` each interval. + #[clap( + long = "service.udp.metrics.sample-fraction", + env = "QUILKIN_SERVICE_UDP_METRICS_SAMPLE_FRACTION", + default_value_t = 1.0 + )] + pub sample_fraction: f64, + /// Number of client ASNs to report, largest first, with the rest counted + /// under `asn="other"`. 0 disables per-ASN reporting. + #[clap( + long = "service.udp.metrics.top-asns", + env = "QUILKIN_SERVICE_UDP_METRICS_TOP_ASNS", + default_value_t = 32 + )] + pub top_asns: usize, + /// Jitter, in milliseconds, at or above which a session counts as degraded. + #[clap( + long = "service.udp.metrics.jitter-threshold-ms", + env = "QUILKIN_SERVICE_UDP_METRICS_JITTER_THRESHOLD_MS", + default_value_t = 30 + )] + pub jitter_threshold_ms: u64, +} + +impl Default for SessionMetricsCli { + fn default() -> Self { + Self { + interval_secs: 15, + sample_fraction: 1.0, + top_asns: 32, + jitter_threshold_ms: 30, + } + } +} + +impl From<&SessionMetricsCli> for crate::net::sessions::quality::AggregationConfig { + fn from(cli: &SessionMetricsCli) -> Self { + // Deliberately not clamped: `AggregationConfig::validate` rejects values + // that would silently produce meaningless metrics + Self { + interval: std::time::Duration::from_secs(cli.interval_secs), + sample_fraction: cli.sample_fraction, + top_asns: cli.top_asns, + jitter_threshold: std::time::Duration::from_millis(cli.jitter_threshold_ms), + } + } +} + pub struct ServicePorts { pub mds: Option, pub phoenix: Option, @@ -271,6 +339,7 @@ impl Default for Service { udp_enabled: <_>::default(), udp_port: 7777, udp_workers: std::num::NonZeroUsize::new(num_cpus::get()).unwrap(), + session_metrics: <_>::default(), udp_session_limit: 10_000, udp_ring_buffer: 2048, session_pool_ring_buffer: 256, @@ -309,6 +378,12 @@ impl Service { self } + /// Sets how often per-session connection quality is folded into metrics. + pub fn session_metrics_interval(mut self, interval_secs: u64) -> Self { + self.session_metrics.interval_secs = interval_secs; + self + } + /// Sets the UDP service port. pub fn udp_port(mut self, port: u16) -> Self { self.udp_port = port; @@ -728,6 +803,15 @@ impl Service { return Ok(()); } + // Only a proxy has sessions to aggregate; a QCMP-only instance would + // otherwise run the task over an empty registry + if self.udp_enabled { + crate::net::sessions::quality::spawn_aggregator( + (&self.session_metrics).into(), + shutdown, + )?; + } + let resolved_backend = self.udp_backend.resolve(); tracing::info!( port=%self.udp_port, @@ -876,6 +960,10 @@ impl Service { let sessions = SessionPool::new( session_sends, cached_filters, + // Used to tell an endpoint disappearing from underneath a session + // apart from a player going quiet. The `destination` label doesn't + // need it: the cluster arrives with the routing decision. + config.dyn_cfg.clusters().cloned(), self.udp_session_limit, backend, self.session_pool_ring_buffer, diff --git a/src/test.rs b/src/test.rs index 08e0ba3869..4eaf9bfd7c 100644 --- a/src/test.rs +++ b/src/test.rs @@ -325,6 +325,8 @@ impl TestHelper { .qcmp_port(0) .phoenix() .phoenix_port(0) + // Fast enough for a test to observe an aggregation without waiting + .session_metrics_interval(1) .spawn_services(&config, shutdown) .await .expect("failed to spawn services"); diff --git a/tests/metrics.rs b/tests/metrics.rs index 38767c7a6b..dfd42955b9 100644 --- a/tests/metrics.rs +++ b/tests/metrics.rs @@ -97,4 +97,16 @@ async fn metrics_server() { let write_regex = regex::Regex::new(r#"quilkin_packets_total\{.*event="write".*\} 2"#).unwrap(); assert!(read_regex.is_match(&response)); assert!(write_regex.is_match(&response)); + + // The proxy in front of the echo server routes to a cluster with no locality, + // so the label is present and empty rather than absent + assert!( + regex::Regex::new(r#"quilkin_packets_total\{.*destination=""#) + .unwrap() + .is_match(&response) + ); + + // Registered by the aggregation, which spawns with the UDP service, so a + // proxy exports the distribution whether or not it currently has players + assert!(response.contains("quilkin_session_jitter_seconds_bucket")); }