Supergraph coprocessor implementation

.changesets/feat_geal_supergraph_coprocessor.md

@@ -0,0 +1,5 @@
+### Supergraph coprocessor implementation ([PR #3408](https://github.com/apollographql/router/pull/3408))
+
+This adds support for coprocessors at the supergraph service level. Supergraph plugins work on the request side with a parsed GraphQL request object, so the query and operation name, variables and extensions are directly accessible. On the response side, they handle GraphQL response objects, with label, data, path, errors, extensions. The supergraph response contains a stream of GraphQL responses, which can contain multiple elements if the query uses `@defer` or subscriptions. When configured to observe the responses, the coprocessor will be called for each of the deferred responses.
+
+By [@Geal](https://github.com/Geal) in https://github.com/apollographql/router/pull/3408

apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__schema_generation.snap

@@ -770,6 +770,113 @@ expression: "&schema"
           },
           "additionalProperties": false
         },
+        "supergraph": {
+          "description": "The supergraph stage request/response configuration",
+          "default": {
+            "request": {
+              "headers": false,
+              "context": false,
+              "body": false,
+              "sdl": false,
+              "path": false,
+              "method": false
+            },
+            "response": {
+              "headers": false,
+              "context": false,
+              "body": false,
+              "sdl": false,
+              "status_code": false
+            }
+          },
+          "type": "object",
+          "properties": {
+            "request": {
+              "description": "The request configuration",
+              "default": {
+                "headers": false,
+                "context": false,
+                "body": false,
+                "sdl": false,
+                "path": false,
+                "method": false
+              },
+              "type": "object",
+              "properties": {
+                "body": {
+                  "description": "Send the body",
+                  "default": false,
+                  "type": "boolean"
+                },
+                "context": {
+                  "description": "Send the context",
+                  "default": false,
+                  "type": "boolean"
+                },
+                "headers": {
+                  "description": "Send the headers",
+                  "default": false,
+                  "type": "boolean"
+                },
+                "method": {
+                  "description": "Send the method",
+                  "default": false,
+                  "type": "boolean"
+                },
+                "path": {
+                  "description": "Send the path",
+                  "default": false,
+                  "type": "boolean"
+                },
+                "sdl": {
+                  "description": "Send the SDL",
+                  "default": false,
+                  "type": "boolean"
+                }
+              },
+              "additionalProperties": false
+            },
+            "response": {
+              "description": "What information is passed to a router request/response stage",
+              "default": {
+                "headers": false,
+                "context": false,
+                "body": false,
+                "sdl": false,
+                "status_code": false
+              },
+              "type": "object",
+              "properties": {
+                "body": {
+                  "description": "Send the body",
+                  "default": false,
+                  "type": "boolean"
+                },
+                "context": {
+                  "description": "Send the context",
+                  "default": false,
+                  "type": "boolean"
+                },
+                "headers": {
+                  "description": "Send the headers",
+                  "default": false,
+                  "type": "boolean"
+                },
+                "sdl": {
+                  "description": "Send the SDL",
+                  "default": false,
+                  "type": "boolean"
+                },
+                "status_code": {
+                  "description": "Send the HTTP status",
+                  "default": false,
+                  "type": "boolean"
+                }
+              },
+              "additionalProperties": false
+            }
+          }
+        },
         "timeout": {
           "description": "The timeout for external requests",
           "default": {

apollo-router/src/plugins/coprocessor.rs → apollo-router/src/plugins/coprocessor/mod.rs

@@ -35,6 +35,7 @@ use crate::layers::ServiceBuilderExt;
 use crate::plugin::Plugin;
 use crate::plugin::PluginInit;
 use crate::register_plugin;
+use crate::services;
 use crate::services::external::Control;
 use crate::services::external::Externalizable;
 use crate::services::external::PipelineStep;
@@ -44,6 +45,11 @@ use crate::services::router;
 use crate::services::subgraph;
 use crate::tracer::TraceId;
 
+#[cfg(test)]
+mod test;
+
+mod supergraph;
+
 pub(crate) const EXTERNAL_SPAN_NAME: &str = "external_plugin";
 const POOL_IDLE_TIMEOUT_DURATION: Option<Duration> = Some(Duration::from_secs(5));
 
@@ -86,6 +92,13 @@ impl Plugin for CoprocessorPlugin<HTTPClientService> {
         self.router_service(service)
     }
 
+    fn supergraph_service(
+        &self,
+        service: services::supergraph::BoxService,
+    ) -> services::supergraph::BoxService {
+        self.supergraph_service(service)
+    }
+
     fn subgraph_service(&self, name: &str, service: subgraph::BoxService) -> subgraph::BoxService {
         self.subgraph_service(name, service)
     }
@@ -149,6 +162,18 @@ where
         )
     }
 
+    fn supergraph_service(
+        &self,
+        service: services::supergraph::BoxService,
+    ) -> services::supergraph::BoxService {
+        self.configuration.supergraph.as_service(
+            self.http_client.clone(),
+            service,
+            self.configuration.url.clone(),
+            self.sdl.clone(),
+        )
+    }
+
     fn subgraph_service(&self, name: &str, service: subgraph::BoxService) -> subgraph::BoxService {
         self.configuration.subgraph.all.as_service(
             self.http_client.clone(),
@@ -239,6 +264,9 @@ struct Conf {
     /// The router stage request/response configuration
     #[serde(default)]
     router: RouterStage,
+    /// The supergraph stage request/response configuration
+    #[serde(default)]
+    supergraph: supergraph::SupergraphStage,
     /// The subgraph stage request/response configuration
     #[serde(default)]
     subgraph: SubgraphStages,
@@ -676,7 +704,7 @@ where
     // If first is None, or contains an error we return an error
     let opt_first: Option<Bytes> = first.and_then(|f| f.ok());
     let bytes = match opt_first {
-        Some(b) => b.to_vec(),
+        Some(b) => b,
         None => {
             tracing::error!(
                 "Coprocessor cannot convert body into future due to problem with first part"
@@ -695,7 +723,7 @@ where
         .transpose()?;
     let body_to_send = response_config
         .body
-        .then(|| String::from_utf8(bytes.clone()))
+        .then(|| std::str::from_utf8(&bytes).map(|s| s.to_string()))
         .transpose()?;
     let status_to_send = response_config.status_code.then(|| parts.status.as_u16());
     let context_to_send = response_config.context.then(|| response.context.clone());
@@ -857,7 +885,6 @@ where
     // First, extract the data we need from our request and prepare our
     // external call. Use our configuration to figure out which data to send.
     let (parts, body) = request.subgraph_request.into_parts();
-    let bytes = Bytes::from(serde_json::to_vec(&body)?);
 
     let headers_to_send = request_config
         .headers
@@ -866,7 +893,7 @@ where
 
     let body_to_send = request_config
         .body
-        .then(|| serde_json::from_slice::<serde_json::Value>(&bytes))
+        .then(|| serde_json::to_value(&body))
         .transpose()?;
     let context_to_send = request_config.context.then(|| request.context.clone());
     let uri = request_config.uri.then(|| parts.uri.to_string());
@@ -997,7 +1024,6 @@ where
     // external call. Use our configuration to figure out which data to send.
 
     let (parts, body) = response.response.into_parts();
-    let bytes = Bytes::from(serde_json::to_vec(&body)?);
 
     let headers_to_send = response_config
         .headers
@@ -1008,7 +1034,7 @@ where
 
     let body_to_send = response_config
         .body
-        .then(|| serde_json::from_slice::<serde_json::Value>(&bytes))
+        .then(|| serde_json::to_value(&body))
         .transpose()?;
     let context_to_send = response_config.context.then(|| response.context.clone());
     let service_name = response_config.service_name.then_some(service_name);
@@ -1098,7 +1124,7 @@ fn validate_coprocessor_output<T>(
 }
 
 /// Convert a HeaderMap into a HashMap
-pub(super) fn externalize_header_map(
+pub(crate) fn externalize_header_map(
     input: &HeaderMap<HeaderValue>,
 ) -> Result<HashMap<String, Vec<String>>, BoxError> {
     let mut output = HashMap::new();