all: Report moving average of total query time to Prometheus

Author: lutter

graph/src/data/graphql/effort.rs

@@ -9,6 +9,7 @@ use std::str::FromStr;
 use std::sync::{Arc, RwLock};
 use std::time::{Duration, Instant};
 
+use crate::components::metrics::{Gauge, MetricsRegistry};
 use crate::components::store::PoolWaitStats;
 use crate::data::graphql::shape_hash::shape_hash;
 use crate::prelude::{info, o, warn, Logger};
@@ -78,9 +79,10 @@ impl QueryEffort {
         }
     }
 
-    pub fn add(&self, shape_hash: u64, duration: Duration) {
+    pub fn add(&self, shape_hash: u64, duration: Duration, gauge: &Box<Gauge>) {
         let mut inner = self.inner.write().unwrap();
         inner.add(shape_hash, duration);
+        gauge.set(inner.total.average().unwrap_or(*ZERO_DURATION).as_millis() as f64);
     }
 
     /// Return what we know right now about the effort for the query
@@ -195,13 +197,15 @@ pub struct LoadManager {
     /// restarting the process
     jailed_queries: RwLock<HashSet<u64>>,
     kill_state: RwLock<KillState>,
+    effort_gauge: Box<Gauge>,
 }
 
 impl LoadManager {
     pub fn new(
         logger: &Logger,
         store_wait_stats: PoolWaitStats,
         blocked_queries: Vec<Arc<q::Document>>,
+        registry: Arc<dyn MetricsRegistry>,
     ) -> Self {
         let logger = logger.new(o!("component" => "LoadManager"));
         let blocked_queries = blocked_queries
@@ -212,19 +216,27 @@ impl LoadManager {
             logger,
             "Creating LoadManager with disabled={}", *LOAD_MANAGEMENT_DISABLED,
         );
+        let effort_gauge = registry
+            .new_gauge(
+                String::from("query_effort_ms"),
+                String::from("Moving average of time spent running queries"),
+                HashMap::new(),
+            )
+            .expect("failed to create `query_effort_ms` counter");
         Self {
             logger,
             effort: QueryEffort::default(),
             store_wait_stats,
             blocked_queries,
             jailed_queries: RwLock::new(HashSet::new()),
             kill_state: RwLock::new(KillState::new()),
+            effort_gauge,
         }
     }
 
     pub fn add_query(&self, shape_hash: u64, duration: Duration) {
         if !*LOAD_MANAGEMENT_DISABLED {
-            self.effort.add(shape_hash, duration);
+            self.effort.add(shape_hash, duration, &self.effort_gauge);
         }
     }
 

graphql/tests/query.rs

@@ -5,21 +5,20 @@ use graphql_parser::{query as q, Pos};
 use lazy_static::lazy_static;
 use std::collections::HashMap;
 use std::iter::FromIterator;
-use std::sync::{Arc, RwLock};
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 
-use graph::data::graphql::effort::LoadManager;
 use graph::prelude::{
     futures03::stream::StreamExt, futures03::FutureExt, futures03::TryFutureExt, o, slog, tokio,
-    Entity, EntityKey, EntityOperation, EthereumBlockPointer, FutureExtension, Logger, MovingStats,
-    Query, QueryError, QueryExecutionError, QueryResult, QueryVariables, Schema, Store,
+    Entity, EntityKey, EntityOperation, EthereumBlockPointer, FutureExtension, Logger, Query,
+    QueryError, QueryExecutionError, QueryResult, QueryVariables, Schema, Store,
     SubgraphDeploymentEntity, SubgraphDeploymentId, SubgraphDeploymentStore, SubgraphManifest,
     Subscription, SubscriptionError, Value,
 };
 use graph_graphql::prelude::*;
 use test_store::{
     execute_subgraph_query_with_complexity, execute_subgraph_query_with_deadline, return_err,
-    transact_entity_operations, BLOCK_ONE, GENESIS_PTR, LOGGER, STORE,
+    transact_entity_operations, BLOCK_ONE, GENESIS_PTR, LOAD_MANAGER, LOGGER, STORE,
 };
 
 lazy_static! {
@@ -231,9 +230,7 @@ fn execute_query_document_with_variables(
     query: q::Document,
     variables: Option<QueryVariables>,
 ) -> QueryResult {
-    let stats = Arc::new(RwLock::new(MovingStats::default()));
-    let load_manager = Arc::new(LoadManager::new(&*LOGGER, stats, vec![]));
-    let runner = GraphQlRunner::new(&*LOGGER, STORE.clone(), load_manager);
+    let runner = GraphQlRunner::new(&*LOGGER, STORE.clone(), LOAD_MANAGER.clone());
     let query = Query::new(Arc::new(api_test_schema()), query, variables);
 
     return_err!(runner.execute(query, None, None, None))

node/src/main.rs

@@ -516,6 +516,7 @@ async fn main() {
     let connection_pool_registry = metrics_registry.clone();
     let stores_metrics_registry = metrics_registry.clone();
     let graphql_metrics_registry = metrics_registry.clone();
+
     let stores_logger = logger.clone();
     let stores_error_logger = logger.clone();
     let stores_eth_adapters = eth_adapters.clone();
@@ -593,6 +594,7 @@ async fn main() {
                 &logger,
                 wait_stats.clone(),
                 expensive_queries,
+                metrics_registry.clone(),
             ));
             let graphql_runner = Arc::new(GraphQlRunner::new(
                 &logger,

store/test-store/src/lib.rs

@@ -39,7 +39,11 @@ lazy_static! {
 
     pub static ref POOL_WAIT_STATS: PoolWaitStats = Arc::new(RwLock::new(MovingStats::default()));
 
-    pub static ref LOAD_MANAGER: Arc<LoadManager> = Arc::new(LoadManager::new(&*LOGGER, POOL_WAIT_STATS.clone(), Vec::new()));
+    pub static ref LOAD_MANAGER: Arc<LoadManager> = Arc::new(
+        LoadManager::new(&*LOGGER,
+                         POOL_WAIT_STATS.clone(),
+                         Vec::new(),
+                         Arc::new(MockMetricsRegistry::new())));
 
     // Create Store instance once for use with each of the tests.
     pub static ref STORE: Arc<Store> = {