Merge pull request #11 from Netflix-Skunkworks/spawn-blocking-parse-m…

…etrics use std mutex for query analysis cache, use spawn_blocking for parsin…
Netflix-Skunkworks · May 31, 2024 · bf74725 · bf74725
2 parents f5a9fbb + b61bfbc
commit bf74725
Show file tree

Hide file tree

Showing 2 changed files with 50 additions and 15 deletions.
diff --git a/apollo-router/src/query_planner/caching_query_planner.rs b/apollo-router/src/query_planner/caching_query_planner.rs
@@ -210,7 +210,11 @@ where
         } in all_cache_keys
         {
             let context = Context::new();
-            let doc = match query_analysis.parse_document(&query, operation.as_deref()) {
+
+            let doc = match query_analysis
+                .parse_document(&query, operation.as_deref())
+                .await
+            {
                 Ok(doc) => doc,
                 Err(_) => continue,
             };
@@ -241,7 +245,10 @@ where
 
             let entry = self.cache.get(&caching_key).await;
             if entry.is_first() {
-                let doc = match query_analysis.parse_document(&query, operation.as_deref()) {
+                let doc = match query_analysis
+                    .parse_document(&query, operation.as_deref())
+                    .await
+                {
                     Ok(doc) => doc,
                     Err(error) => {
                         let e = Arc::new(QueryPlannerError::SpecError(error));

diff --git a/apollo-router/src/services/layers/query_analysis.rs b/apollo-router/src/services/layers/query_analysis.rs
@@ -10,8 +10,9 @@ use apollo_compiler::ExecutableDocument;
 use http::StatusCode;
 use lru::LruCache;
 use router_bridge::planner::UsageReporting;
+use std::sync::Mutex;
 use time::Instant;
-use tokio::sync::Mutex;
+use tracing::Instrument;
 
 use crate::context::OPERATION_KIND;
 use crate::context::OPERATION_NAME;
@@ -64,17 +65,37 @@ impl QueryAnalysisLayer {
         }
     }
 
-    pub(crate) fn parse_document(
+    pub(crate) async fn parse_document(
         &self,
         query: &str,
         operation_name: Option<&str>,
     ) -> Result<ParsedDocument, SpecError> {
-        Query::parse_document(
-            query,
-            operation_name,
-            self.schema.api_schema(),
-            &self.configuration,
-        )
+        let query = query.to_string();
+        let operation_name = operation_name.map(|o| o.to_string());
+        let schema = self.schema.clone();
+        let conf = self.configuration.clone();
+
+        tokio::task::spawn_blocking(move || {
+            let now = Instant::now();
+
+            let parsed = Query::parse_document(
+                &query,
+                operation_name.as_deref(),
+                schema.as_ref(),
+                conf.as_ref(),
+            );
+
+            let processing_seconds = now.elapsed().as_seconds_f64();
+
+            tracing::info!(
+                histogram.apollo.router.parse_validate_query = processing_seconds,
+                "operationName" = operation_name,
+            );
+
+            parsed
+        })
+        .await
+        .expect("parse_document task panicked")
     }
 
     pub(crate) async fn supergraph_request(
@@ -113,10 +134,11 @@ impl QueryAnalysisLayer {
             .query
             .clone()
             .expect("query presence was already checked");
+
         let entry = self
             .cache
             .lock()
-            .await
+            .unwrap()
             .get(&QueryAnalysisKey {
                 query: query.clone(),
                 operation_name: op_name.clone(),
@@ -126,9 +148,15 @@ impl QueryAnalysisLayer {
         let res = match entry {
             None => {
                 let span = tracing::info_span!("parse_query", "otel.kind" = "INTERNAL");
-                match span.in_scope(|| self.parse_document(&query, op_name.as_deref())) {
+
+                let result = self
+                    .parse_document(&query, op_name.as_deref())
+                    .instrument(span)
+                    .await;
+
+                match result {
                     Err(errors) => {
-                        (*self.cache.lock().await).put(
+                        (*self.cache.lock().unwrap()).put(
                             QueryAnalysisKey {
                                 query,
                                 operation_name: op_name,
@@ -176,7 +204,7 @@ impl QueryAnalysisLayer {
                             .insert(OPERATION_KIND, operation_kind.unwrap_or_default())
                             .expect("cannot insert operation kind in the context; this is a bug");
 
-                        (*self.cache.lock().await).put(
+                        (*self.cache.lock().unwrap()).put(
                             QueryAnalysisKey {
                                 query,
                                 operation_name: op_name,
@@ -192,7 +220,7 @@ impl QueryAnalysisLayer {
         };
 
         let processing_seconds = now.elapsed().as_seconds_f64();
-        tracing::info!(histogram.apollo_router_parse_validate_time = processing_seconds,);
+        tracing::info!(histogram.apollo.router.get_parsed_document = processing_seconds,);
 
         match res {
             Ok((context, doc)) => {