Skip to content

[Error Counting] Use Builder for all router::Response Creation #7699

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rregitsky merged 17 commits into from
Jun 17, 2025
Merged

[Error Counting] Use Builder for all router::Response Creation #7699

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
a7d1fe5
Pull response refactor files from original branch
rregitsky Jun 16, 2025
2960e89
remove error counting changes
rregitsky Jun 16, 2025
4ecc35f
remove rhai/tests.rs
rregitsky Jun 16, 2025
e0d73ce
add missed subscription file
rregitsky Jun 16, 2025
2a492ba
remove todo
rregitsky Jun 16, 2025
eca21e8
Merge remote-tracking branch 'origin/dev' into rreg/err_count_respons…
rregitsky Jun 16, 2025
88da6ff
response_cache invalidation_endpoint
rregitsky Jun 16, 2025
246dbca
lint
rregitsky Jun 16, 2025
fb439c4
Merge branch 'dev' into rreg/err_count_response_refactor_PULSR_1504
rregitsky Jun 17, 2025
6cca62c
remove todos
rregitsky Jun 17, 2025
b448efd
missed one last conversion
rregitsky Jun 17, 2025
5dd8a3d
Merge branch 'dev' into rreg/err_count_response_refactor_PULSR_1504
rregitsky Jun 17, 2025
4b998ee
comment change to retrigger tests
rregitsky Jun 17, 2025
0c3b4f0
use http_response_builder for successful cache invalidation
rregitsky Jun 17, 2025
2ca6b1f
lint
rregitsky Jun 17, 2025
18bace3
Swap another accepted response to http_request_builder
rregitsky Jun 17, 2025
8ddcf78
Merge remote-tracking branch 'origin/dev' into rreg/err_count_respons…
rregitsky Jun 17, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 18 additions & 15 deletions apollo-router/src/axum_factory/listeners.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -550,14 +550,17 @@ mod tests {
.unwrap();

let endpoint = service_fn(|req: router::Request| async move {
Ok::<_, BoxError>(router::Response {
response: http::Response::builder()
.body::<crate::services::router::Body>(body::from_bytes(
"this is a test".to_string(),
))
Ok::<_, BoxError>(
router::Response::http_response_builder()
.response(
http::Response::builder().body::<crate::services::router::Body>(
body::from_bytes("this is a test".to_string()),
)?,
)
.context(req.context)
.build()
.unwrap(),
context: req.context,
})
)
})
.boxed();

Expand Down Expand Up @@ -591,14 +594,14 @@ mod tests {
.build()
.unwrap();
let endpoint = service_fn(|req: router::Request| async move {
Ok::<_, BoxError>(router::Response {
response: http::Response::builder()
.body::<crate::services::router::Body>(body::from_bytes(
"this is a test".to_string(),
))
.unwrap(),
context: req.context,
})
router::Response::http_response_builder()
.response(
http::Response::builder().body::<crate::services::router::Body>(
body::from_bytes("this is a test".to_string()),
)?,
)
.context(req.context)
.build()
})
.boxed();

Expand Down
116 changes: 68 additions & 48 deletions apollo-router/src/plugins/cache/invalidation_endpoint.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ use super::invalidation::Invalidation;
use super::invalidation::InvalidationOrigin;
use crate::ListenAddr;
use crate::configuration::subgraph::SubgraphConfiguration;
use crate::graphql;
use crate::plugins::cache::invalidation::InvalidationRequest;
use crate::plugins::telemetry::consts::OTEL_STATUS_CODE;
use crate::plugins::telemetry::consts::OTEL_STATUS_CODE_ERROR;
Expand Down Expand Up @@ -108,13 +109,16 @@ impl Service<router::Request> for InvalidationService {
let (parts, body) = req.router_request.into_parts();
if !parts.headers.contains_key(AUTHORIZATION) {
Span::current().record(OTEL_STATUS_CODE, OTEL_STATUS_CODE_ERROR);
return Ok(router::Response {
response: http::Response::builder()
.status(StatusCode::UNAUTHORIZED)
.body(router::body::from_bytes("Missing authorization header"))
.map_err(BoxError::from)?,
context: req.context,
});
return router::Response::error_builder()
.status_code(StatusCode::UNAUTHORIZED)
.error(
graphql::Error::builder()
.message(String::from("Missing authorization header"))
.extension_code(StatusCode::UNAUTHORIZED.to_string())
.build(),
)
.context(req.context)
.build();
}
match parts.method {
Method::POST => {
Expand Down Expand Up @@ -156,66 +160,82 @@ impl Service<router::Request> for InvalidationService {
if !valid_shared_key {
Span::current()
.record(OTEL_STATUS_CODE, OTEL_STATUS_CODE_ERROR);
return Ok(router::Response {
response: http::Response::builder()
.status(StatusCode::UNAUTHORIZED)
.body(router::body::from_bytes(
"Invalid authorization header",
))
.map_err(BoxError::from)?,
context: req.context,
});
return router::Response::error_builder()
.status_code(StatusCode::UNAUTHORIZED)
.error(
graphql::Error::builder()
.message("Invalid authorization header")
.extension_code(
StatusCode::UNAUTHORIZED.to_string(),
)
.build(),
)
.context(req.context)
.build();
}
match invalidation
.invalidate(InvalidationOrigin::Endpoint, body)
.instrument(tracing::info_span!("invalidate"))
.await
{
Ok(count) => Ok(router::Response {
response: http::Response::builder()
.status(StatusCode::ACCEPTED)
.body(router::body::from_bytes(serde_json::to_string(
&json!({
"count": count
}),
)?))
.map_err(BoxError::from)?,
context: req.context,
}),
Ok(count) => router::Response::http_response_builder()
.response(
http::Response::builder()
.status(StatusCode::ACCEPTED)
.body(router::body::from_bytes(
serde_json::to_string(&json!({
"count": count
}))?,
))
.map_err(BoxError::from)?,
)
.context(req.context)
.build(),
Err(err) => {
Span::current()
.record(OTEL_STATUS_CODE, OTEL_STATUS_CODE_ERROR);
Ok(router::Response {
response: http::Response::builder()
.status(StatusCode::BAD_REQUEST)
.body(router::body::from_bytes(err.to_string()))
.map_err(BoxError::from)?,
context: req.context,
})
router::Response::error_builder()
.status_code(StatusCode::BAD_REQUEST)
.error(
graphql::Error::builder()
.message(err.to_string())
.extension_code(
StatusCode::BAD_REQUEST.to_string(),
)
.build(),
)
.context(req.context)
.build()
}
}
}
Err(err) => {
Span::current().record(OTEL_STATUS_CODE, OTEL_STATUS_CODE_ERROR);
Ok(router::Response {
response: http::Response::builder()
.status(StatusCode::BAD_REQUEST)
.body(router::body::from_bytes(err))
.map_err(BoxError::from)?,
context: req.context,
})
router::Response::error_builder()
.status_code(StatusCode::BAD_REQUEST)
.error(
graphql::Error::builder()
.message(err)
.extension_code(StatusCode::BAD_REQUEST.to_string())
.build(),
)
.context(req.context)
.build()
}
}
}
_ => {
Span::current().record(OTEL_STATUS_CODE, OTEL_STATUS_CODE_ERROR);
Ok(router::Response {
response: http::Response::builder()
.status(StatusCode::METHOD_NOT_ALLOWED)
.body(router::body::from_bytes("".to_string()))
.map_err(BoxError::from)?,
context: req.context,
})
router::Response::error_builder()
.status_code(StatusCode::METHOD_NOT_ALLOWED)
.error(
graphql::Error::builder()
.message("".to_string())
.extension_code(StatusCode::METHOD_NOT_ALLOWED.to_string())
.build(),
)
.context(req.context)
.build()
}
}
}
Expand Down
10 changes: 5 additions & 5 deletions apollo-router/src/plugins/coprocessor/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1104,11 +1104,11 @@ where
.map(|b| b.map(http_body::Frame::data).map_err(axum::Error::new)),
));

// Finally, return a response which has a Body that wraps our stream of response chunks.
Ok(router::Response {
context,
response: http::Response::from_parts(parts, final_stream),
})
// Finally, return a response which has a Body that wraps our stream of response chunks
router::Response::http_response_builder()
.context(context)
.response(http::Response::from_parts(parts, final_stream))
.build()
}
// -----------------------------------------------------------------------------------------------------

Expand Down
70 changes: 40 additions & 30 deletions apollo-router/src/plugins/fleet_detector.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -270,19 +270,22 @@ impl PluginPrivate for FleetDetector {
context: req.context,
})
// Count the number of response bytes from the router to clients
.map_response(move |res: router::Response| router::Response {
response: res.response.map(move |body| {
router::body::from_result_stream(body.into_data_stream().inspect(|res| {
if let Ok(bytes) = res {
u64_counter!(
"apollo.router.operations.response_size",
"Total number of response bytes to clients",
bytes.len() as u64
);
}
.map_response(move |res: router::Response| {
router::Response::http_response_builder()
.response(res.response.map(move |body| {
router::body::from_result_stream(body.into_data_stream().inspect(|res| {
if let Ok(bytes) = res {
u64_counter!(
"apollo.router.operations.response_size",
"Total number of response bytes to clients",
bytes.len() as u64
);
}
}))
}))
}),
context: res.context,
.context(res.context)
.build()
.unwrap()
})
.boxed()
}
Expand Down Expand Up @@ -532,6 +535,7 @@ mod tests {
use tower::Service as _;

use super::*;
use crate::graphql;
use crate::metrics::FutureMetricsExt as _;
use crate::metrics::collect_metrics;
use crate::metrics::test_utils::MetricType;
Expand All @@ -551,15 +555,19 @@ mod tests {
.expect_call()
.times(1)
.returning(|req: router::Request| {
Ok(router::Response {
context: req.context,
response: http::Response::builder()
.status(StatusCode::BAD_REQUEST)
.header("content-type", "application/json")
// making sure the request body is consumed
.body(req.router_request.into_body())
.unwrap(),
})
// making sure the request body is consumed
req.router_request.into_body();
router::Response::error_builder()
.context(req.context)
.status_code(StatusCode::BAD_REQUEST)
.header("content-type", "application/json")
.error(
graphql::Error::builder()
.message("bad request")
.extension_code(StatusCode::BAD_REQUEST.to_string())
.build(),
)
.build()
});
let mut bad_request_router_service =
plugin.router_service(mock_bad_request_service.boxed());
Expand Down Expand Up @@ -609,15 +617,17 @@ mod tests {
.expect_call()
.times(1)
.returning(|req: router::Request| {
Ok(router::Response {
context: req.context,
response: http::Response::builder()
.status(StatusCode::BAD_REQUEST)
.header("content-type", "application/json")
// making sure the request body is consumed
.body(req.router_request.into_body())
.unwrap(),
})
router::Response::http_response_builder()
.context(req.context)
.response(
http::Response::builder()
.status(StatusCode::BAD_REQUEST)
.header("content-type", "application/json")
// making sure the request body is consumed
.body(req.router_request.into_body())
.unwrap(),
)
.build()
});
let mut bad_request_router_service =
plugin.router_service(mock_bad_request_service.boxed());
Expand Down
10 changes: 5 additions & 5 deletions apollo-router/src/plugins/healthcheck/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -290,14 +290,14 @@ impl PluginPrivate for HealthCheck {
};
tracing::trace!(?health, request = ?req.router_request, "health check");
async move {
Ok(router::Response {
response: http::Response::builder().status(status_code).body(
router::Response::http_response_builder()
.response(http::Response::builder().status(status_code).body(
router::body::from_bytes(
serde_json::to_vec(&health).map_err(BoxError::from)?,
),
)?,
context: req.context,
})
)?)
.context(req.context)
.build()
}
})
.boxed(),
Expand Down
10 changes: 5 additions & 5 deletions apollo-router/src/plugins/record_replay/record.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,13 +127,13 @@ impl Plugin for Record {

let stream = stream.into_data_stream().chain(after_complete);

Ok(router::Response {
context: res.context,
response: http::Response::from_parts(
router::Response::http_response_builder()
.context(res.context)
.response(http::Response::from_parts(
parts,
router::body::from_result_stream(stream),
),
})
))
.build()
}
})
.service(service)
Expand Down
Loading