all: Have GraphQlRunner manage effort tracking

Author: lutter

graph/src/components/graphql.rs

@@ -1,5 +1,6 @@
 use futures::prelude::*;
 
+use crate::data::graphql::effort::QueryEffort;
 use crate::data::query::{Query, QueryError, QueryResult};
 use crate::data::subscription::{Subscription, SubscriptionError, SubscriptionResult};
 
@@ -8,6 +9,7 @@ use failure::format_err;
 use failure::Error;
 use futures03::compat::Future01CompatExt;
 use graphql_parser::query as q;
+use std::sync::Arc;
 
 /// Future for query results.
 pub type QueryResultFuture = Box<dyn Future<Item = QueryResult, Error = QueryError> + Send>;
@@ -47,4 +49,6 @@ pub trait GraphQlRunner: Send + Sync + 'static {
                 }
             })
     }
+
+    fn effort(&self) -> Arc<QueryEffort>;
 }

graphql/src/query/mod.rs

@@ -5,6 +5,8 @@ use std::sync::{atomic::AtomicBool, Arc};
 use std::time::Instant;
 use uuid::Uuid;
 
+use graph::data::graphql::effort::QueryEffort;
+
 use crate::execution::*;
 use crate::schema::ast as sast;
 
@@ -30,6 +32,8 @@ where
 
     /// Maximum value for the `first` argument.
     pub max_first: u32,
+
+    pub effort: Arc<QueryEffort>,
 }
 
 /// Executes a query and returns a result.
@@ -74,13 +78,15 @@ where
     // Execute top-level `query { ... }` and `{ ... }` expressions.
     let start = Instant::now();
     let result = execute_root_selection_set(&ctx, selection_set, query_type, block_ptr);
+    let elapsed = start.elapsed();
+    options.effort.add(query.shape_hash, elapsed);
     if *graph::log::LOG_GQL_TIMING {
         info!(
             query_logger,
             "Query timing (GraphQL)";
             "query" => &query.query_text,
             "variables" => &query.variables_text,
-            "query_time_ms" => start.elapsed().as_millis(),
+            "query_time_ms" => elapsed.as_millis(),
             "cached" => ctx.cached.load(std::sync::atomic::Ordering::SeqCst),
             "block" => block_ptr.map(|b| b.number).unwrap_or(0),
             "complexity" => &query.complexity

graphql/src/runner.rs

@@ -11,6 +11,7 @@ use crate::prelude::{
 };
 use crate::query::execute_query;
 use crate::subscription::execute_prepared_subscription;
+use graph::data::graphql::effort::QueryEffort;
 use graph::prelude::{
     o, shape_hash, EthereumBlockPointer, GraphQlRunner as GraphQlRunnerTrait, Logger, Query,
     QueryExecutionError, QueryResult, QueryResultFuture, Store, StoreError, SubgraphDeploymentId,
@@ -24,6 +25,7 @@ pub struct GraphQlRunner<S> {
     logger: Logger,
     store: Arc<S>,
     expensive: HashMap<u64, Arc<q::Document>>,
+    effort: Arc<QueryEffort>,
 }
 
 lazy_static! {
@@ -47,6 +49,8 @@ lazy_static! {
         .map(|s| u32::from_str(&s)
             .unwrap_or_else(|_| panic!("failed to parse env var GRAPH_GRAPHQL_MAX_FIRST")))
         .unwrap_or(1000);
+    static ref WINDOW_SIZE: Duration = Duration::from_secs(300);
+    static ref BIN_SIZE: Duration = Duration::from_secs(1);
 }
 
 impl<S> GraphQlRunner<S>
@@ -63,6 +67,7 @@ where
             logger: logger.new(o!("component" => "GraphQlRunner")),
             store,
             expensive,
+            effort: Arc::new(QueryEffort::new(*WINDOW_SIZE, *BIN_SIZE)),
         }
     }
 
@@ -132,6 +137,7 @@ where
                     resolver,
                     deadline: GRAPHQL_QUERY_TIMEOUT.map(|t| Instant::now() + t),
                     max_first: max_first.unwrap_or(*GRAPHQL_MAX_FIRST),
+                    effort: self.effort.clone(),
                 },
             ) {
                 Err(errs) => errors.extend(errs),
@@ -210,4 +216,8 @@ where
 
         Box::new(future::result(result))
     }
+
+    fn effort(&self) -> Arc<QueryEffort> {
+        self.effort.clone()
+    }
 }

graphql/tests/introspection.rs

@@ -4,7 +4,9 @@ extern crate pretty_assertions;
 use graphql_parser::{query as q, schema as s};
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::time::Duration;
 
+use graph::data::graphql::effort::QueryEffort;
 use graph::prelude::{
     o, slog, Logger, Query, QueryExecutionError, QueryResult, Schema, SubgraphDeploymentId,
 };
@@ -558,6 +560,10 @@ fn introspection_query(schema: Schema, query: &str) -> QueryResult {
         resolver: MockResolver,
         deadline: None,
         max_first: std::u32::MAX,
+        effort: Arc::new(QueryEffort::new(
+            Duration::from_millis(0),
+            Duration::from_millis(0),
+        )),
     };
 
     let result = PreparedQuery::new(query, None, 100)

server/http/src/service.rs

@@ -475,6 +475,7 @@ mod tests {
     use hyper::{Body, Method, Request};
     use std::collections::BTreeMap;
 
+    use graph::data::graphql::effort::QueryEffort;
     use graph::prelude::*;
     use graph_mock::{mock_store_with_users_subgraph, MockMetricsRegistry};
     use graphql_parser::query as q;
@@ -513,6 +514,10 @@ mod tests {
         fn run_subscription(&self, _subscription: Subscription) -> SubscriptionResultFuture {
             unreachable!();
         }
+
+        fn effort(&self) -> Arc<QueryEffort> {
+            unimplemented!()
+        }
     }
 
     #[test]

server/http/tests/server.rs

@@ -4,6 +4,7 @@ use hyper::{Body, Client, Request};
 use std::collections::BTreeMap;
 use std::time::Duration;
 
+use graph::data::graphql::effort::QueryEffort;
 use graph::prelude::*;
 
 use graph_server_http::test_utils;
@@ -60,6 +61,10 @@ impl GraphQlRunner for TestGraphQlRunner {
     fn run_subscription(&self, _subscription: Subscription) -> SubscriptionResultFuture {
         unreachable!();
     }
+
+    fn effort(&self) -> Arc<QueryEffort> {
+        unimplemented!()
+    }
 }
 
 #[cfg(test)]

server/index-node/src/service.rs

@@ -109,6 +109,7 @@ where
             .and_then(move |query| {
                 let logger = logger.clone();
                 let graphql_runner = graphql_runner.clone();
+                let effort = graphql_runner.effort();
 
                 // Run the query using the index node resolver
                 tokio::task::block_in_place(|| {
@@ -117,6 +118,7 @@ where
                         resolver: IndexNodeResolver::new(&logger, graphql_runner, store),
                         deadline: None,
                         max_first: std::u32::MAX,
+                        effort,
                     };
                     let result = PreparedQuery::new(query, None, 100)
                         .and_then(|query| execute_query(query, None, None, options));

store/test-store/src/lib.rs

@@ -2,6 +2,7 @@
 extern crate diesel;
 
 use crate::tokio::runtime::{Builder, Runtime};
+use graph::data::graphql::effort::QueryEffort;
 use graph::log;
 use graph::prelude::{Store as _, *};
 use graph_graphql::prelude::{
@@ -383,6 +384,7 @@ fn execute_subgraph_query_internal(
 ) -> QueryResult {
     let logger = Logger::root(slog::Discard, o!());
     let query = return_err!(PreparedQuery::new(query, max_complexity, 100));
+    //let effort =
     let mut values = std::collections::BTreeMap::new();
     let mut errors = Vec::new();
     for (bc, selection_set) in return_err!(query.block_constraint()) {
@@ -402,6 +404,10 @@ fn execute_subgraph_query_internal(
                 resolver,
                 deadline,
                 max_first: std::u32::MAX,
+                effort: Arc::new(QueryEffort::new(
+                    Duration::from_millis(0),
+                    Duration::from_millis(0),
+                )),
             },
         ) {
             Err(errs) => errors.extend(errs),