add support of rate limit and timeout

NEXT_CHANGELOG.md

@@ -62,6 +62,21 @@ traffic_shaping:
 
 By [@bnjjj](https://github.com/bnjjj) in https://github.com/apollographql/router/pull/1347
 
+### Update span attributes to be compliant with the opentelemetry for GraphQL specs ([PR #1449](https://github.com/apollographql/router/pull/1449))
+
+Change attribute name `query` to `graphql.document` and `operation_name` to `graphql.operation.name` in spans.
+
+By [@bnjjj](https://github.com/bnjjj) in https://github.com/apollographql/router/pull/1449 
+
+### Configuration handling enhancements ([PR #1454](https://github.com/apollographql/router/pull/1454))
+
+Router config handling now:
+* Allows completely empty configuration without error.
+* Prevents unknown tags at the root of the configuration from being silently ignored.
+
+By [@bryncooke](https://github.com/bryncooke) in https://github.com/apollographql/router/pull/1454
+
+
 ## 🛠 Maintenance
 
 ## 📚 Documentation

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

@@ -466,7 +466,7 @@ mod test {
             test:
                 global_rate_limit:
                     capacity: 1
-                    interval: 1sec
+                    interval: 300ms
                 timeout: 500ms
         "#,
         )
@@ -491,7 +491,7 @@ mod test {
             .oneshot(SubgraphRequest::fake_builder().build())
             .await
             .unwrap();
-        tokio::time::sleep(Duration::from_millis(1000)).await;
+        tokio::time::sleep(Duration::from_millis(300)).await;
         let _response = plugin
             .subgraph_service("test", test_service.boxed())
             .oneshot(SubgraphRequest::fake_builder().build())
@@ -506,7 +506,7 @@ mod test {
         router:
             global_rate_limit:
                 capacity: 1
-                interval: 1sec
+                interval: 300ms
             timeout: 500ms
         "#,
         )
@@ -536,7 +536,7 @@ mod test {
             .oneshot(RouterRequest::fake_builder().build().unwrap())
             .await
             .is_err());
-        tokio::time::sleep(Duration::from_millis(1000)).await;
+        tokio::time::sleep(Duration::from_millis(300)).await;
         let _response = plugin
             .router_service(mock_service.clone().boxed())
             .oneshot(RouterRequest::fake_builder().build().unwrap())

apollo-router/src/plugins/traffic_shaping/rate/layer.rs

@@ -1,10 +1,11 @@
 use std::num::NonZeroU64;
+use std::sync::atomic::AtomicU64;
 use std::sync::atomic::AtomicUsize;
 use std::sync::Arc;
-use std::sync::RwLock;
 use std::time::Duration;
+use std::time::SystemTime;
+use std::time::UNIX_EPOCH;
 
-use tokio::time::Instant;
 use tower::Layer;
 
 use super::Rate;
@@ -14,7 +15,7 @@ use super::RateLimit;
 #[derive(Debug, Clone)]
 pub(crate) struct RateLimitLayer {
     rate: Rate,
-    window_start: Arc<RwLock<Instant>>,
+    window_start: Arc<AtomicU64>,
     previous_nb_requests: Arc<AtomicUsize>,
     current_nb_requests: Arc<AtomicUsize>,
 }
@@ -25,7 +26,12 @@ impl RateLimitLayer {
         let rate = Rate::new(num, per);
         RateLimitLayer {
             rate,
-            window_start: Arc::new(RwLock::new(Instant::now())),
+            window_start: Arc::new(AtomicU64::new(
+                SystemTime::now()
+                    .duration_since(UNIX_EPOCH)
+                    .expect("system time must be after EPOCH")
+                    .as_millis() as u64,
+            )),
             previous_nb_requests: Arc::default(),
             current_nb_requests: Arc::new(AtomicUsize::new(1)),
         }

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

@@ -15,7 +15,7 @@ impl Rate {
     ///
     /// This function panics if `num` or `per` is 0.
     pub(crate) fn new(num: NonZeroU64, per: Duration) -> Self {
-        assert!(per > Duration::from_millis(0));
+        assert!(per > Duration::default());
 
         Rate {
             num: num.into(),

apollo-router/src/plugins/traffic_shaping/rate/service.rs

@@ -1,12 +1,14 @@
+use std::sync::atomic::AtomicU64;
 use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
-use std::sync::RwLock;
 use std::task::Context;
 use std::task::Poll;
+use std::time::Duration;
+use std::time::SystemTime;
+use std::time::UNIX_EPOCH;
 
 use futures::ready;
-use tokio::time::Instant;
 use tower::Service;
 
 use super::future::ResponseFuture;
@@ -17,7 +19,10 @@ use crate::plugins::traffic_shaping::rate::error::RateLimited;
 pub(crate) struct RateLimit<T> {
     pub(crate) inner: T,
     pub(crate) rate: Rate,
-    pub(crate) window_start: Arc<RwLock<Instant>>,
+    /// We're using an atomic u64 because it's basically a timestamp in milliseconds for the start of the window
+    /// Instead of using an Instant which is not thread safe we're using an atomic u64
+    /// It's ok to have an u64 because we just care about milliseconds for this use case
+    pub(crate) window_start: Arc<AtomicU64>,
     pub(crate) previous_nb_requests: Arc<AtomicUsize>,
     pub(crate) current_nb_requests: Arc<AtomicUsize>,
 }
@@ -32,25 +37,43 @@ where
     type Future = ResponseFuture<S::Future>;
 
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        let mut window_start = self.window_start.read().unwrap().elapsed();
-        let time_unit = self.rate.per();
+        let time_unit = self.rate.per().as_millis() as u64;
 
-        if window_start > time_unit {
-            let new_window_start = Instant::now();
-            *self.window_start.write().unwrap() = new_window_start;
-            window_start = new_window_start.elapsed();
+        let updated =
+            self.window_start
+                .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |window_start| {
+                    let duration_start = Duration::from_millis(window_start);
+                    let duration_now = Duration::from_millis(
+                        SystemTime::now()
+                            .duration_since(UNIX_EPOCH)
+                            .expect("system time must be after EPOCH")
+                            .as_millis() as u64,
+                    );
+                    if duration_now.saturating_sub(duration_start) > self.rate.per() {
+                        Some(
+                            SystemTime::now()
+                                .duration_since(UNIX_EPOCH)
+                                .expect("system time must be after EPOCH")
+                                .as_millis() as u64,
+                        )
+                    } else {
+                        None
+                    }
+                });
+        // If it has been updated
+        if let Ok(_updated_window_start) = updated {
             self.previous_nb_requests.swap(
                 self.current_nb_requests.load(Ordering::SeqCst),
                 Ordering::SeqCst,
             );
             self.current_nb_requests.swap(1, Ordering::SeqCst);
         }
+
         let estimated_cap = (self.previous_nb_requests.load(Ordering::SeqCst)
             * (time_unit
-                .checked_sub(window_start)
+                .checked_sub(self.window_start.load(Ordering::SeqCst))
                 .unwrap_or_default()
-                .as_millis()
-                / time_unit.as_millis()) as usize)
+                / time_unit) as usize)
             + self.current_nb_requests.load(Ordering::SeqCst);
 
         if estimated_cap as u64 > self.rate.num() {