diff --git a/Cargo.lock b/Cargo.lock index 40ddf5ea619..f9361218716 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8118,6 +8118,7 @@ dependencies = [ "futures", "headers", "http", + "http-body", "http-body-util", "humantime", "hyper", diff --git a/Cargo.toml b/Cargo.toml index 842332340d8..b4b3f6286d3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -223,6 +223,7 @@ hex = "0.4.3" home = "0.5" hostname = "^0.3" http = "1.0" +http-body = "1.0" http-body-util= "0.1.3" humantime = "2.3" hyper = "1.0" diff --git a/crates/client-api/Cargo.toml b/crates/client-api/Cargo.toml index afbb5a2a340..3cd1f3fb466 100644 --- a/crates/client-api/Cargo.toml +++ b/crates/client-api/Cargo.toml @@ -16,6 +16,7 @@ spacetimedb-paths.workspace = true spacetimedb-schema.workspace = true base64.workspace = true +http-body.workspace = true http-body-util.workspace = true tokio = { version = "1.2", features = ["full"] } lazy_static = "1.4.0" diff --git a/crates/client-api/src/routes/mod.rs b/crates/client-api/src/routes/mod.rs index 08e1e73cb77..3a1382e0a40 100644 --- a/crates/client-api/src/routes/mod.rs +++ b/crates/client-api/src/routes/mod.rs @@ -1,4 +1,14 @@ +use std::pin::Pin; +use std::task::{Context, Poll}; + +use ::prometheus::IntCounter; +use axum::body::{Body, Bytes, HttpBody}; +use axum::extract::{MatchedPath, Request}; +use axum::middleware::Next; +use axum::response::Response; use http::header; +use http_body::Frame; +use spacetimedb::worker_metrics::WORKER_METRICS; use tower_http::cors; use crate::{Authorization, ControlStateDelegate, NodeDelegate}; @@ -18,6 +28,88 @@ use self::{database::DatabaseRoutes, identity::IdentityRoutes}; /// establish a connection to SpacetimeDB. This API call doesn't actually do anything. pub async fn ping(_auth: crate::auth::SpacetimeAuthHeader) {} +/// A body wrapper that counts bytes as they are actually transferred. +struct CountingBody { + inner: Body, + counter: IntCounter, +} + +impl HttpBody for CountingBody { + type Data = Bytes; + type Error = axum::Error; + + fn poll_frame(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll, Self::Error>>> { + let poll = Pin::new(&mut self.inner).poll_frame(cx); + if let Poll::Ready(Some(Ok(frame))) = &poll + && let Some(data) = frame.data_ref() + { + self.counter.inc_by(data.len() as u64); + } + poll + } + + fn size_hint(&self) -> http_body::SizeHint { + self.inner.size_hint() + } + + fn is_end_stream(&self) -> bool { + self.inner.is_end_stream() + } +} + +impl CountingBody { + fn wrap(counter: IntCounter) -> impl FnOnce(Body) -> Body { + move |inner| Body::new(CountingBody { inner, counter }) + } +} + +/// Returns the method as a static label value, +/// bucketing non-standard methods to keep label cardinality bounded. +fn method_label(method: &http::Method) -> &'static str { + match method.as_str() { + "GET" => "GET", + "POST" => "POST", + "PUT" => "PUT", + "DELETE" => "DELETE", + "HEAD" => "HEAD", + "OPTIONS" => "OPTIONS", + "PATCH" => "PATCH", + "CONNECT" => "CONNECT", + "TRACE" => "TRACE", + _ => "OTHER", + } +} + +/// Records request count, latency and body sizes per HTTP route. +async fn http_metrics_middleware(req: Request, next: Next) -> Response { + let Some(route) = req.extensions().get::().cloned() else { + return next.run(req).await; + }; + + let method = method_label(req.method()); + + let request_body_bytes = WORKER_METRICS.http_request_body_bytes.with_label_values(route.as_str()); + let response_body_bytes = WORKER_METRICS + .http_response_body_bytes + .with_label_values(route.as_str()); + + let req = req.map(CountingBody::wrap(request_body_bytes)); + + let start = std::time::Instant::now(); + let res = next.run(req).await; + + WORKER_METRICS + .http_requests + .with_label_values(route.as_str(), method, res.status().as_str()) + .inc(); + WORKER_METRICS + .http_request_duration + .with_label_values(route.as_str()) + .observe(start.elapsed().as_secs_f64()); + + res.map(CountingBody::wrap(response_body_bytes)) +} + #[allow(clippy::let_and_return)] pub fn router( ctx: &S, @@ -44,6 +136,11 @@ where .allow_origin(cors::Any); axum::Router::new() - .nest("/v1", router.layer(cors)) + .nest( + "/v1", + router + .layer(axum::middleware::from_fn(http_metrics_middleware)) + .layer(cors), + ) .nest("/internal", internal::router()) } diff --git a/crates/core/src/sql/execute.rs b/crates/core/src/sql/execute.rs index 7fe2fe229f8..ee24ce0024d 100644 --- a/crates/core/src/sql/execute.rs +++ b/crates/core/src/sql/execute.rs @@ -23,6 +23,7 @@ use spacetimedb_lib::metrics::ExecutionMetrics; use spacetimedb_lib::Timestamp; use spacetimedb_lib::{AlgebraicType, ProductType, ProductValue}; use spacetimedb_query::{compile_sql_stmt, execute_dml_stmt, execute_select_stmt}; +use spacetimedb_sats::bsatn; use spacetimedb_sats::raw_identifier::RawIdentifier; use tokio::sync::oneshot; @@ -129,6 +130,9 @@ fn run_inner( Ok(plan) })?; + // Charge egress, using BSATN size for parity with WebSocket queries. + metrics.bytes_sent_to_clients += rows.iter().map(|row| bsatn::to_len(row).unwrap_or(0)).sum::(); + // Update transaction metrics tx.metrics.merge(metrics); @@ -1686,4 +1690,39 @@ pub(crate) mod tests { Ok(()) } + + // Selects should charge egress for the rows they return + #[test] + fn test_select_charges_egress() -> ResultTest<()> { + let db = TestDB::durable()?; + + let table_id = db.create_table_for_test("T", &[("a", AlgebraicType::U8)], &[])?; + with_auto_commit(&db, |tx| -> Result<_, DBError> { + for i in 0..4u8 { + insert(&db, tx, table_id, &product!(i))?; + } + Ok(()) + })?; + + let rt = db.runtime().expect("runtime should be there"); + + let server = Identity::from_claims("issuer", "server"); + let auth = AuthCtx::new(server, server); + + let result = rt.block_on(run( + db.clone(), + "SELECT * FROM T".to_string(), + auth, + None, + None, + &mut vec![], + ))?; + + assert_eq!(result.rows.len(), 4); + let expected: usize = result.rows.iter().map(|row| bsatn::to_len(row).unwrap()).sum(); + assert!(expected > 0); + assert_eq!(result.metrics.bytes_sent_to_clients, expected); + + Ok(()) + } } diff --git a/crates/core/src/worker_metrics/mod.rs b/crates/core/src/worker_metrics/mod.rs index e99bd196b63..8c265665999 100644 --- a/crates/core/src/worker_metrics/mod.rs +++ b/crates/core/src/worker_metrics/mod.rs @@ -51,6 +51,26 @@ metrics_group!( #[labels(database_identity: Identity, protocol: str)] pub websocket_request_msg_size: HistogramVec, + #[name = spacetime_http_requests_total] + #[help = "The cumulative number of HTTP requests, by matched route, method and response status"] + #[labels(route: str, method: str, status: str)] + pub http_requests: IntCounterVec, + + #[name = spacetime_http_request_duration_sec] + #[help = "The time (in seconds) spent handling an HTTP request, by matched route"] + #[labels(route: str)] + pub http_request_duration: HistogramVec, + + #[name = spacetime_http_request_body_bytes_total] + #[help = "The cumulative number of HTTP request body bytes received, by matched route"] + #[labels(route: str)] + pub http_request_body_bytes: IntCounterVec, + + #[name = spacetime_http_response_body_bytes_total] + #[help = "The cumulative number of HTTP response body bytes sent, by matched route"] + #[labels(route: str)] + pub http_response_body_bytes: IntCounterVec, + #[name = jemalloc_active_bytes] #[help = "Number of bytes in jemallocs heap"] #[labels(node_id: str)]