Skip to content
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

only send one report for a response with deferred responses #1576

Merged
merged 5 commits into from
Aug 23, 2022
Merged

only send one report for a response with deferred responses #1576

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
34ede0d
only send one report for a response with deferred responses
Aug 23, 2022
bb69135
changelog
Aug 23, 2022
6b0e9ce
Update apollo-router/src/plugins/telemetry/mod.rs
Aug 23, 2022
6845823
do not send a final deferred response on non deferred queries
Aug 23, 2022
4c9201d
Merge branch 'main' into geal/defer-studio
o0Ignition0o Aug 23, 2022
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
8 changes: 8 additions & 0 deletions NEXT_CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,5 +35,13 @@ By [@SimonSapin](https://github.com/SimonSapin) in https://github.com/apollograp

## 🚀 Features
## 🐛 Fixes

### Only send one report for a response with deferred responses ([PR #1576](https://github.com/apollographql/router/issues/1576))

The router was sending one report per response (even deferred ones), while Studio was expecting one report for the entire
response. The router now sends one report, that measures the latency of the entire operation.

By [@Geal](https://github.com/Geal) in https://github.com/apollographql/router/pull/1576

## 🛠 Maintenance
## 📚 Documentation
6 changes: 5 additions & 1 deletion apollo-router-scaffold/templates/plugin/src/plugins/{{snake_name}}.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,8 +174,12 @@ mod tests {

assert!(first_response.data.is_some());

println!("first response: {:?}", first_response);
let next = streamed_response.next_response().await;
println!("next response: {:?}", next);

// You could keep calling .next_response() until it yields None if you're expexting more parts.
assert!(streamed_response.next_response().await.is_none());
assert!(next.is_none());
Ok(())
}
}
Expand Down
2 changes: 1 addition & 1 deletion apollo-router/src/axum_http_server_factory.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -544,7 +544,7 @@ where
serde_json::to_vec(&res).unwrap().into(),
)))
.chain(once(ready(Bytes::from_static(b"\r\n"))))
}).chain(once(ready(Bytes::from_static(b"--graphql--\r\ncontent-type: application/json\r\n\r\n{\"hasNext\":false}"))))
})
.map(Ok::<_, BoxError>);

(parts, StreamBody::new(body)).into_response()
Expand Down
12 changes: 8 additions & 4 deletions apollo-router/src/plugins/telemetry/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -225,21 +225,25 @@ impl Plugin for Telemetry {
Err(e)
}
Ok(router_response) => {
let is_not_success =
let mut has_errors =
!router_response.response.status().is_success();
Ok(router_response.map(move |response_stream| {
let sender = sender.clone();
let ctx = ctx.clone();

response_stream
.map(move |response| {
let response_has_errors = !response.errors.is_empty();
if !response.errors.is_empty() {
has_errors = true;
}

if !matches!(sender, Sender::Noop) {
if !response.has_next.unwrap_or(false)
&& !matches!(sender, Sender::Noop)
{
Self::update_apollo_metrics(
&ctx,
sender.clone(),
is_not_success || response_has_errors,
has_errors,
o0Ignition0o marked this conversation as resolved.
Show resolved Hide resolved
start.elapsed(),
);
}
Expand Down
138 changes: 72 additions & 66 deletions apollo-router/src/services/supergraph_service.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,7 +152,7 @@ where
})
}
QueryPlannerContent::Plan { query, plan } => {
let is_deferred = plan.root.contains_defer();
let can_be_deferred = plan.root.contains_defer();

if let Some(err) = query.validate_variables(body, &schema).err() {
let mut res = SupergraphResponse::new_from_graphql_response(err, context);
Expand All @@ -172,75 +172,81 @@ where
.await?;

let (parts, response_stream) = http::Response::from(response).into_parts();

let stream = response_stream
.map(move |mut response: Response| {
tracing::debug_span!("format_response").in_scope(|| {
query.format_response(
&mut response,
operation_name.as_deref(),
variables.clone(),
schema.api_schema(),
)
});

match (response.path.as_ref(), response.data.as_ref()) {
(None, _) | (_, None) => {
if can_be_deferred {
response.has_next = Some(true);
}

response
}
// if the deferred response specified a path, we must extract the
//values matched by that path and create a separate response for
//each of them.
// While { "data": { "a": { "b": 1 } } } and { "data": { "b": 1 }, "path: ["a"] }
// would merge in the same ways, some clients will generate code
// that checks the specific type of the deferred response at that
// path, instead of starting from the root object, so to support
// this, we extract the value at that path.
// In particular, that means that a deferred fragment in an object
// under an array would generate one response par array element
(Some(response_path), Some(response_data)) => {
let mut sub_responses = Vec::new();
response_data.select_values_and_paths(
response_path,
|path, value| {
sub_responses
.push((path.clone(), value.clone()));
},
);

Response::builder()
.has_next(true)
.incremental(
sub_responses
.into_iter()
.map(move |(path, data)| {
IncrementalResponse::builder()
.and_label(
response.label.clone(),
)
.data(data)
.path(path)
.errors(response.errors.clone())
.extensions(
response.extensions.clone(),
)
.build()
})
.collect(),
)
.build()
}
}
});

Ok(SupergraphResponse {
context,
response: http::Response::from_parts(
parts,
response_stream
.map(move |mut response: Response| {
tracing::debug_span!("format_response").in_scope(|| {
query.format_response(
&mut response,
operation_name.as_deref(),
variables.clone(),
schema.api_schema(),
)
});

match (response.path.as_ref(), response.data.as_ref()) {
(None, _) | (_, None) => {
if is_deferred {
response.has_next = Some(true);
}

response
}
// if the deferred response specified a path, we must extract the
//values matched by that path and create a separate response for
//each of them.
// While { "data": { "a": { "b": 1 } } } and { "data": { "b": 1 }, "path: ["a"] }
// would merge in the same ways, some clients will generate code
// that checks the specific type of the deferred response at that
// path, instead of starting from the root object, so to support
// this, we extract the value at that path.
// In particular, that means that a deferred fragment in an object
// under an array would generate one response par array element
(Some(response_path), Some(response_data)) => {
let mut sub_responses = Vec::new();
response_data.select_values_and_paths(
response_path,
|path, value| {
sub_responses
.push((path.clone(), value.clone()));
},
);

Response::builder()
.has_next(true)
.incremental(
sub_responses
.into_iter()
.map(move |(path, data)| {
IncrementalResponse::builder()
.and_label(
response.label.clone(),
)
.data(data)
.path(path)
.errors(response.errors.clone())
.extensions(
response.extensions.clone(),
)
.build()
})
.collect(),
)
.build()
}
}
})
.in_current_span()
.boxed(),
if can_be_deferred {
stream.chain(once(ready(Response::builder().has_next(false).build()))).left_stream()
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice!

} else {
stream.right_stream()
}.in_current_span()
.boxed(),
)
.into(),
})
Expand Down