Supergraph coprocessor implementation (#3647)

Co-authored-by: o0Ignition0o <[email protected]>

.changesets/feat_geal_supergraph_coprocessor2.md

@@ -0,0 +1,33 @@
+### Supergraph coprocessor implementation ([PR #3647](https://github.com/apollographql/router/pull/3647))
+
+Coprocessors now support supergraph service interception. 
+
+On the request side, the coprocessor payload can contain:
+- method
+- headers
+- body
+- context
+- sdl
+
+On the response side, the payload can contain:
+- status_code
+- headers
+- body
+- context
+- sdl
+
+The supergraph request body contains:
+* query
+* operation name
+* variables
+* extensions
+
+The supergraph response body contains:
+* label
+* data
+* errors
+* extensions
+
+When using `@defer` or subscriptions a supergraph response may contain multiple GraphQL responses, and the coprocessor will be called for each.
+
+By [@Geal](https://github.com/Geal) in https://github.com/apollographql/router/pull/3647

CHANGELOG.md

@@ -364,7 +364,7 @@ By [@garypen](https://github.com/garypen) in https://github.com/apollographql/ro
 
 ### Spelling of `content_negociation` corrected to `content_negotiation` ([Issue #3204](https://github.com/apollographql/router/issues/3204))
 
-We had a bit of a French twist on one of our internal module names.  We won't promise it won't happen again, but  `content_negociation` is spelled as `content_negotiation` now. 😄
+We had a bit of a French twist on one of our internal module names.  We won't promise it won't happen again, but  `content_negociation` is spelled as `content_negotiation` now. 😄  
 
 Thank you for this contribution!
 
@@ -1720,7 +1720,7 @@ By [@Geal](https://github.com/Geal) in https://github.com/apollographql/router/p
 
 ### Small gzip'd responses no longer cause a panic
 
-A regression introduced in v1.17.0 — again related to compression — has been resolved.  This occurred when small responses used invalid buffer management, causing a panic.
+A regression introduced in v1.17.0 — again related to compression — has been resolved.  This occurred when small responses used invalid buffer management, causing a panic.
 
 By [@dbanty](https://github.com/dbanty) in https://github.com/apollographql/router/pull/3047
 
@@ -2254,7 +2254,7 @@ By [@BrynCooke](https://github.com/BrynCooke) in https://github.com/apollographq
 
 ### Distributed caching: Don't send Redis' `CLIENT SETNAME` ([PR #2825](https://github.com/apollographql/router/pull/2825))
 
-We won't send [the `CLIENT SETNAME` command](https://redis.io/commands/client-setname/) to connected Redis servers.  This resolves an incompatibility with some Redis-compatible servers since not all "Redis-compatible" offerings (like Google Memorystore) actually support _every_ Redis command.  We weren't actually necessitating this feature, it was just a feature that could be enabled optionally on our Redis client.  No Router functionality is impacted.
+We won't send [the `CLIENT SETNAME` command](https://redis.io/commands/client-setname/) to connected Redis servers.  This resolves an incompatibility with some Redis-compatible servers since not all "Redis-compatible" offerings (like Google Memorystore) actually support _every_ Redis command.  We weren't actually necessitating this feature, it was just a feature that could be enabled optionally on our Redis client.  No Router functionality is impacted.
 
 By [@Geal](https://github.com/Geal) in https://github.com/apollographql/router/pull/2825
 
@@ -3424,7 +3424,7 @@ By [@Geal](https://github.com/geal) in https://github.com/apollographql/router/p
 
 ### Optimize header propagation plugin's regular expression matching ([PR #2392](https://github.com/apollographql/router/pull/2392))
 
-We've changed the header propagation plugins' behavior to reduce the chance of memory allocations occurring when applying regex-based header propagation rules.
+We've changed the header propagation plugins' behavior to reduce the chance of memory allocations occurring when applying regex-based header propagation rules.
 
 By [@o0Ignition0o](https://github.com/o0Ignition0o) in https://github.com/apollographql/router/pull/2392
 
@@ -3474,7 +3474,7 @@ By [@Geal](https://github.com/geal) in https://github.com/apollographql/router/p
 
 Configuration changes will be [automatically migrated on load](https://www.apollographql.com/docs/router/configuration/overview#upgrading-your-router-configuration). However, you should update your source configuration files as these will become breaking changes in a future major release.
 
-### Defer support graduates from preview ([Issue #2368](https://github.com/apollographql/router/issues/2368))
+### Defer support graduates from preview ([Issue #2368](https://github.com/apollographql/router/issues/2368))
 
 We're pleased to announce that [`@defer` support](https://www.apollographql.com/docs/router/executing-operations/defer-support/) has been promoted to general availability in accordance with our [product launch stages](https://www.apollographql.com/docs/resources/product-launch-stages/).
 

apollo-router/src/configuration/snapshots/apollo_router__configuration__metrics__test__metrics@coprocessor.router.yaml.snap

@@ -8,6 +8,6 @@ value.apollo.router.config.coprocessor:
     opt__router__response__: true
     opt__subgraph__request__: true
     opt__subgraph__response__: true
-    opt__supergraph__request__: false
-    opt__supergraph__response__: false
+    opt__supergraph__request__: true
+    opt__supergraph__response__: true
 

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

@@ -779,6 +779,106 @@ expression: "&schema"
           },
           "additionalProperties": false
         },
+        "supergraph": {
+          "description": "The supergraph stage request/response configuration",
+          "default": {
+            "request": {
+              "headers": false,
+              "context": false,
+              "body": false,
+              "sdl": 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,
+                "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"
+                },
+                "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/configuration/testdata/metrics/coprocessor.router.yaml

@@ -15,6 +15,19 @@ coprocessor:
       body: true
       headers: true
       status_code: true
+  supergraph:
+    request:
+      headers: true
+      body: true
+      context: true
+      method: true
+      sdl: true
+    response:
+      sdl: true
+      context: true
+      body: true
+      headers: true
+      status_code: true
   subgraph:
     all:
       request:

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

@@ -36,6 +36,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;
@@ -45,6 +46,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));
 
@@ -87,6 +93,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)
     }
@@ -150,6 +163,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(),
@@ -240,6 +265,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,
@@ -677,7 +705,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"
@@ -696,7 +724,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());
@@ -858,7 +886,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
@@ -867,7 +894,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());
@@ -998,7 +1025,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
@@ -1009,7 +1035,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);
@@ -1099,7 +1125,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();