add support of metrics (#738)

Signed-off-by: Benjamin Coenen <5719034+bnjjj@users.noreply.github.com>

CHANGELOG.md

@@ -52,6 +52,43 @@ We reworked the way query plans are generated before being cached, which lead to
 
  -->
 
+ # [v0.1.0-preview.3] (unreleased) - 2022-mm-dd
+## ❗ BREAKING ❗
+## 🚀 Features
+- **Add support of metrics in `apollo.telemetry` plugin** ([#738](https://github.com/apollographql/router/pull/738))
+
+  The Router will now compute different metrics you can expose via Prometheus or OTLP exporter.
+
+  Example of configuration to export an endpoint (configured with the path `/plugins/apollo.telemetry/metrics`) with metrics in `Prometheus` format:
+
+  ```yaml
+  telemetry:
+    metrics:
+      exporter:
+        prometheus:
+          # By setting this endpoint you enable the prometheus exporter
+          # All our endpoints exposed by plugins are namespaced by the name of the plugin
+          # Then to access to this prometheus endpoint, the full url path will be `/plugins/apollo.telemetry/metrics`
+          endpoint: "/metrics"
+    ```
+
+- **Add experimental support of `custom_endpoint` method in `Plugin` trait** ([#738](https://github.com/apollographql/router/pull/738))
+
+  The `custom_endpoint` method lets you declare a new endpoint exposed for your plugin. For now it's only accessible for official `apollo.` plugins and for `experimental.`. The return type of this method is a Tower [`Service`]().
+
+## 🐛 Fixes
+- **Trim the query to better detect an empty query** ([PR #738](https://github.com/apollographql/router/pull/738))
+
+  Before this fix, if you wrote a query with only whitespaces inside, it wasn't detected as an empty query.
+
+- **Keep the original context in `RouterResponse` when returning an error** ([PR #738](https://github.com/apollographql/router/pull/738))
+
+  This fix keeps the original http request in `RouterResponse` when there is an error.
+
+
+## 🛠 Maintenance
+## 📚 Documentation
+
 # [v0.1.0-preview.2] - 2022-04-01
 ## ❗ BREAKING ❗
 

apollo-router-core/src/layers/ensure_query_presence.rs

@@ -20,7 +20,7 @@ where
             |req: RouterRequest| {
                 // A query must be available at this point
                 let query = req.context.request.body().query.as_ref();
-                if query.is_none() || query.unwrap().is_empty() {
+                if query.is_none() || query.unwrap().trim().is_empty() {
                     let res = plugin_utils::RouterResponse::builder()
                         .errors(vec![crate::Error {
                             message: "Must provide query string.".to_string(),

apollo-router-core/src/plugin.rs

@@ -1,16 +1,22 @@
+use crate::services::ServiceBuilderExt;
 use crate::{
-    ExecutionRequest, ExecutionResponse, QueryPlannerRequest, QueryPlannerResponse, RouterRequest,
-    RouterResponse, SubgraphRequest, SubgraphResponse,
+    http_compat, ExecutionRequest, ExecutionResponse, QueryPlannerRequest, QueryPlannerResponse,
+    ResponseBody, RouterRequest, RouterResponse, SubgraphRequest, SubgraphResponse,
 };
 use async_trait::async_trait;
+use bytes::Bytes;
+use futures::future::BoxFuture;
 use once_cell::sync::Lazy;
 use schemars::gen::SchemaGenerator;
 use schemars::JsonSchema;
 use serde::{de::DeserializeOwned, Deserialize};
 use std::collections::HashMap;
 use std::sync::Mutex;
+use std::task::{Context, Poll};
+use tower::buffer::future::ResponseFuture;
+use tower::buffer::Buffer;
 use tower::util::BoxService;
-use tower::BoxError;
+use tower::{BoxError, Service, ServiceBuilder};
 
 type InstanceFactory = fn(&serde_json::Value) -> Result<Box<dyn DynPlugin>, BoxError>;
 
@@ -101,6 +107,10 @@ pub trait Plugin: Send + Sync + 'static + Sized {
         service
     }
 
+    fn custom_endpoint(&self) -> Option<Handler> {
+        None
+    }
+
     fn name(&self) -> &'static str {
         get_type_of(self)
     }
@@ -137,7 +147,7 @@ pub trait DynPlugin: Send + Sync + 'static {
         _name: &str,
         service: BoxService<SubgraphRequest, SubgraphResponse, BoxError>,
     ) -> BoxService<SubgraphRequest, SubgraphResponse, BoxError>;
-
+    fn custom_endpoint(&self) -> Option<Handler>;
     fn name(&self) -> &'static str;
 }
 
@@ -184,6 +194,9 @@ where
     ) -> BoxService<SubgraphRequest, SubgraphResponse, BoxError> {
         self.subgraph_service(name, service)
     }
+    fn custom_endpoint(&self) -> Option<Handler> {
+        self.custom_endpoint()
+    }
 
     fn name(&self) -> &'static str {
         self.name()
@@ -221,3 +234,53 @@ macro_rules! register_plugin {
         }
     };
 }
+
+#[derive(Clone)]
+pub struct Handler {
+    service: Buffer<
+        BoxService<http_compat::Request<Bytes>, http_compat::Response<ResponseBody>, BoxError>,
+        http_compat::Request<Bytes>,
+    >,
+}
+
+impl Handler {
+    pub fn new(
+        service: BoxService<
+            http_compat::Request<Bytes>,
+            http_compat::Response<ResponseBody>,
+            BoxError,
+        >,
+    ) -> Self {
+        Self {
+            service: ServiceBuilder::new().buffered().service(service),
+        }
+    }
+}
+
+impl Service<http_compat::Request<Bytes>> for Handler {
+    type Response = http_compat::Response<ResponseBody>;
+    type Error = BoxError;
+    type Future = ResponseFuture<BoxFuture<'static, Result<Self::Response, Self::Error>>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.service.poll_ready(cx)
+    }
+
+    fn call(&mut self, req: http_compat::Request<Bytes>) -> Self::Future {
+        self.service.call(req)
+    }
+}
+
+impl From<BoxService<http_compat::Request<Bytes>, http_compat::Response<ResponseBody>, BoxError>>
+    for Handler
+{
+    fn from(
+        original: BoxService<
+            http_compat::Request<Bytes>,
+            http_compat::Response<ResponseBody>,
+            BoxError,
+        >,
+    ) -> Self {
+        Self::new(original)
+    }
+}

apollo-router-core/src/plugins/traffic_shaping.rs

@@ -7,7 +7,7 @@ use tower::{BoxError, ServiceBuilder, ServiceExt};
 
 use crate::deduplication::QueryDeduplicationLayer;
 use crate::plugin::Plugin;
-use crate::{register_plugin, SubgraphRequest, SubgraphResponse};
+use crate::{register_plugin, ServiceBuilderExt, SubgraphRequest, SubgraphResponse};
 
 #[derive(PartialEq, Debug, Clone, Deserialize, JsonSchema)]
 struct Shaping {
@@ -61,7 +61,7 @@ impl Plugin for TrafficShaping {
                     //Buffer is required because dedup layer requires a clone service.
                     ServiceBuilder::new()
                         .layer(QueryDeduplicationLayer::default())
-                        .buffer(20_000)
+                        .buffered()
                 }))
                 .service(service)
                 .boxed()

apollo-router-core/src/services/mod.rs

@@ -13,6 +13,7 @@ use std::ops::ControlFlow;
 use std::str::FromStr;
 use std::sync::Arc;
 use tokio::sync::RwLock;
+use tower::buffer::BufferLayer;
 use tower::layer::util::Stack;
 use tower::{BoxError, ServiceBuilder};
 use tower_service::Service;
@@ -24,6 +25,8 @@ mod router_service;
 mod tower_subgraph_service;
 pub use tower_subgraph_service::TowerSubgraphService;
 
+pub(crate) const DEFAULT_BUFFER_SIZE: usize = 20_000;
+
 impl From<http_compat::Request<Request>> for RouterRequest {
     fn from(http_request: http_compat::Request<Request>) -> Self {
         Self {
@@ -38,6 +41,7 @@ pub enum ResponseBody {
     GraphQL(Response),
     RawJSON(serde_json::Value),
     RawString(String),
+    Text(String),
 }
 
 impl TryFrom<ResponseBody> for Response {
@@ -52,6 +56,7 @@ impl TryFrom<ResponseBody> for Response {
             ResponseBody::RawString(_) => {
                 Err("wrong ResponseBody kind: expected Response, found RawString")
             }
+            ResponseBody::Text(_) => Err("wrong ResponseBody kind: expected Response, found Text"),
         }
     }
 }
@@ -68,6 +73,7 @@ impl TryFrom<ResponseBody> for String {
                 Err("wrong ResponseBody kind: expected RawString, found GraphQL")
             }
             ResponseBody::RawString(res) => Ok(res),
+            ResponseBody::Text(_) => Err("wrong ResponseBody kind: expected RawString, found Text"),
         }
     }
 }
@@ -84,6 +90,7 @@ impl TryFrom<ResponseBody> for serde_json::Value {
             ResponseBody::RawString(_) => {
                 Err("wrong ResponseBody kind: expected RawJSON, found RawString")
             }
+            ResponseBody::Text(_) => Err("wrong ResponseBody kind: expected RawJSON, found Text"),
         }
     }
 }
@@ -236,6 +243,7 @@ pub trait ServiceBuilderExt<L>: Sized {
     {
         self.layer(AsyncCheckpointLayer::new(async_checkpoint_fn))
     }
+    fn buffered<Request>(self) -> ServiceBuilder<Stack<BufferLayer<Request>, L>>;
     fn layer<T>(self, layer: T) -> ServiceBuilder<Stack<T, L>>;
 }
 
@@ -244,4 +252,8 @@ impl<L> ServiceBuilderExt<L> for ServiceBuilder<L> {
     fn layer<T>(self, layer: T) -> ServiceBuilder<Stack<T, L>> {
         ServiceBuilder::layer(self, layer)
     }
+
+    fn buffered<Request>(self) -> ServiceBuilder<Stack<BufferLayer<Request>, L>> {
+        self.buffer(DEFAULT_BUFFER_SIZE)
+    }
 }