Improve subgraph running count metrics (#4401)

- deployment_count now shows the absolute value of deployed subgraphs for the node
- deployment_running_count now shows the count of running singraphs which decreases when a subgraph is retrying

Author: mangas

core/src/subgraph/context.rs

@@ -26,30 +26,26 @@ use self::instance::SubgraphInstance;
 #[derive(Clone, Debug)]
 pub struct SubgraphKeepAlive {
     alive_map: Arc<RwLock<HashMap<DeploymentId, CancelGuard>>>,
-    manager_metrics: Arc<SubgraphCountMetric>,
+    sg_metrics: Arc<SubgraphCountMetric>,
 }
 
-impl CheapClone for SubgraphKeepAlive {
-    fn cheap_clone(&self) -> Self {
-        self.clone()
-    }
-}
+impl CheapClone for SubgraphKeepAlive {}
 
 impl SubgraphKeepAlive {
-    pub fn new(metrics_registry: Arc<dyn MetricsRegistry>) -> Self {
+    pub fn new(sg_metrics: Arc<SubgraphCountMetric>) -> Self {
         Self {
-            manager_metrics: Arc::new(SubgraphCountMetric::new(metrics_registry)),
+            sg_metrics,
             alive_map: Arc::new(RwLock::new(HashMap::default())),
         }
     }
 
     pub fn remove(&self, deployment_id: &DeploymentId) {
         self.alive_map.write().unwrap().remove(deployment_id);
-        self.manager_metrics.subgraph_count.dec();
+        self.sg_metrics.running_count.dec();
     }
     pub fn insert(&self, deployment_id: DeploymentId, guard: CancelGuard) {
         self.alive_map.write().unwrap().insert(deployment_id, guard);
-        self.manager_metrics.subgraph_count.inc();
+        self.sg_metrics.running_count.inc();
     }
 }
 

core/src/subgraph/instance_manager.rs

@@ -161,6 +161,7 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
         env_vars: Arc<EnvVars>,
         subgraph_store: Arc<S>,
         chains: Arc<BlockchainMap>,
+        sg_metrics: Arc<SubgraphCountMetric>,
         metrics_registry: Arc<dyn MetricsRegistry>,
         link_resolver: Arc<dyn LinkResolver>,
         ipfs_service: IpfsService,
@@ -174,7 +175,7 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
             subgraph_store,
             chains,
             metrics_registry: metrics_registry.cheap_clone(),
-            instances: SubgraphKeepAlive::new(metrics_registry),
+            instances: SubgraphKeepAlive::new(sg_metrics),
             link_resolver,
             ipfs_service,
             static_filters,

core/src/subgraph/provider.rs

@@ -8,9 +8,42 @@ use graph::{
     prelude::{SubgraphAssignmentProvider as SubgraphAssignmentProviderTrait, *},
 };
 
+#[derive(Debug)]
+struct DeploymentRegistry {
+    subgraphs_deployed: Arc<Mutex<HashSet<DeploymentId>>>,
+    subgraph_metrics: Arc<SubgraphCountMetric>,
+}
+
+impl DeploymentRegistry {
+    fn new(subgraph_metrics: Arc<SubgraphCountMetric>) -> Self {
+        Self {
+            subgraphs_deployed: Arc::new(Mutex::new(HashSet::new())),
+            subgraph_metrics,
+        }
+    }
+
+    fn insert(&self, id: DeploymentId) -> bool {
+        if !self.subgraphs_deployed.lock().unwrap().insert(id) {
+            return false;
+        }
+
+        self.subgraph_metrics.deployment_count.inc();
+        true
+    }
+
+    fn remove(&self, id: &DeploymentId) -> bool {
+        if !self.subgraphs_deployed.lock().unwrap().remove(id) {
+            return false;
+        }
+
+        self.subgraph_metrics.deployment_count.dec();
+        true
+    }
+}
+
 pub struct SubgraphAssignmentProvider<I> {
     logger_factory: LoggerFactory,
-    subgraphs_running: Arc<Mutex<HashSet<DeploymentId>>>,
+    deployment_registry: DeploymentRegistry,
     link_resolver: Arc<dyn LinkResolver>,
     instance_manager: Arc<I>,
 }
@@ -20,16 +53,17 @@ impl<I: SubgraphInstanceManager> SubgraphAssignmentProvider<I> {
         logger_factory: &LoggerFactory,
         link_resolver: Arc<dyn LinkResolver>,
         instance_manager: I,
+        subgraph_metrics: Arc<SubgraphCountMetric>,
     ) -> Self {
         let logger = logger_factory.component_logger("SubgraphAssignmentProvider", None);
         let logger_factory = logger_factory.with_parent(logger.clone());
 
         // Create the subgraph provider
         SubgraphAssignmentProvider {
             logger_factory,
-            subgraphs_running: Arc::new(Mutex::new(HashSet::new())),
             link_resolver: link_resolver.with_retries().into(),
             instance_manager: Arc::new(instance_manager),
+            deployment_registry: DeploymentRegistry::new(subgraph_metrics),
         }
     }
 }
@@ -44,7 +78,7 @@ impl<I: SubgraphInstanceManager> SubgraphAssignmentProviderTrait for SubgraphAss
         let logger = self.logger_factory.subgraph_logger(&loc);
 
         // If subgraph ID already in set
-        if !self.subgraphs_running.lock().unwrap().insert(loc.id) {
+        if !self.deployment_registry.insert(loc.id) {
             info!(logger, "Subgraph deployment is already running");
 
             return Err(SubgraphAssignmentProviderError::AlreadyRunning(
@@ -74,12 +108,7 @@ impl<I: SubgraphInstanceManager> SubgraphAssignmentProviderTrait for SubgraphAss
         deployment: DeploymentLocator,
     ) -> Result<(), SubgraphAssignmentProviderError> {
         // If subgraph ID was in set
-        if self
-            .subgraphs_running
-            .lock()
-            .unwrap()
-            .remove(&deployment.id)
-        {
+        if self.deployment_registry.remove(&deployment.id) {
             // Shut down subgraph processing
             self.instance_manager.stop_subgraph(deployment).await;
         }

graph/src/components/metrics/subgraph.rs

@@ -88,19 +88,30 @@ impl SubgraphInstanceMetrics {
 
 #[derive(Debug)]
 pub struct SubgraphCountMetric {
-    pub subgraph_count: Box<Gauge>,
+    pub running_count: Box<Gauge>,
+    pub deployment_count: Box<Gauge>,
 }
 
 impl SubgraphCountMetric {
     pub fn new(registry: Arc<dyn MetricsRegistry>) -> Self {
-        let subgraph_count = registry
+        let running_count = registry
             .new_gauge(
-                "deployment_count",
+                "deployment_running_count",
                 "Counts the number of deployments currently being indexed by the graph-node.",
                 HashMap::new(),
             )
             .expect("failed to create `deployment_count` gauge");
-        Self { subgraph_count }
+        let deployment_count = registry
+            .new_gauge(
+                "deployment_count",
+                "Counts the number of deployments currently deployed to the graph-node.",
+                HashMap::new(),
+            )
+            .expect("failed to create `deployment_count` gauge");
+        Self {
+            running_count,
+            deployment_count,
+        }
     }
 }
 

node/src/main.rs

@@ -440,11 +440,14 @@ async fn main() {
         }
         let static_filters = ENV_VARS.experimental_static_filters;
 
+        let sg_count = Arc::new(SubgraphCountMetric::new(metrics_registry.cheap_clone()));
+
         let subgraph_instance_manager = SubgraphInstanceManager::new(
             &logger_factory,
             env_vars.cheap_clone(),
             network_store.subgraph_store(),
             blockchain_map.cheap_clone(),
+            sg_count.cheap_clone(),
             metrics_registry.clone(),
             link_resolver.clone(),
             ipfs_service,
@@ -456,6 +459,7 @@ async fn main() {
             &logger_factory,
             link_resolver.clone(),
             subgraph_instance_manager,
+            sg_count,
         );
 
         // Check version switching mode environment variable

node/src/manager/commands/run.rs

@@ -21,7 +21,8 @@ use graph::env::EnvVars;
 use graph::firehose::FirehoseEndpoints;
 use graph::prelude::{
     anyhow, tokio, BlockNumber, DeploymentHash, LoggerFactory, NodeId, SubgraphAssignmentProvider,
-    SubgraphName, SubgraphRegistrar, SubgraphStore, SubgraphVersionSwitchingMode, ENV_VARS,
+    SubgraphCountMetric, SubgraphName, SubgraphRegistrar, SubgraphStore,
+    SubgraphVersionSwitchingMode, ENV_VARS,
 };
 use graph::slog::{debug, info, Logger};
 use graph_chain_ethereum as ethereum;
@@ -148,12 +149,15 @@ pub async fn run(
 
     let static_filters = ENV_VARS.experimental_static_filters;
 
+    let sg_metrics = Arc::new(SubgraphCountMetric::new(metrics_registry.clone()));
+
     let blockchain_map = Arc::new(blockchain_map);
     let subgraph_instance_manager = SubgraphInstanceManager::new(
         &logger_factory,
         env_vars.cheap_clone(),
         subgraph_store.clone(),
         blockchain_map.clone(),
+        sg_metrics.cheap_clone(),
         metrics_registry.clone(),
         link_resolver.cheap_clone(),
         ipfs_service,
@@ -165,6 +169,7 @@ pub async fn run(
         &logger_factory,
         link_resolver.cheap_clone(),
         subgraph_instance_manager,
+        sg_metrics,
     ));
 
     let panicking_subscription_manager = Arc::new(PanicSubscriptionManager {});

tests/src/fixture/mod.rs

@@ -26,8 +26,9 @@ use graph::prelude::ethabi::ethereum_types::H256;
 use graph::prelude::serde_json::{self, json};
 use graph::prelude::{
     async_trait, r, ApiVersion, BigInt, BlockNumber, DeploymentHash, GraphQlRunner as _,
-    LoggerFactory, MetricsRegistry, NodeId, QueryError, SubgraphAssignmentProvider, SubgraphName,
-    SubgraphRegistrar, SubgraphStore as _, SubgraphVersionSwitchingMode, TriggerProcessor,
+    LoggerFactory, MetricsRegistry, NodeId, QueryError, SubgraphAssignmentProvider,
+    SubgraphCountMetric, SubgraphName, SubgraphRegistrar, SubgraphStore as _,
+    SubgraphVersionSwitchingMode, TriggerProcessor,
 };
 use graph::slog::crit;
 use graph_core::polling_monitor::ipfs_service;
@@ -354,13 +355,15 @@ pub async fn setup<C: Blockchain>(
         env_vars.mappings.ipfs_timeout,
         env_vars.mappings.ipfs_request_limit,
     );
+    let sg_count = Arc::new(SubgraphCountMetric::new(mock_registry.cheap_clone()));
 
     let blockchain_map = Arc::new(blockchain_map);
     let subgraph_instance_manager = SubgraphInstanceManager::new(
         &logger_factory,
         env_vars.cheap_clone(),
         subgraph_store.clone(),
         blockchain_map.clone(),
+        sg_count.cheap_clone(),
         mock_registry.clone(),
         link_resolver.cheap_clone(),
         ipfs_service,
@@ -391,6 +394,7 @@ pub async fn setup<C: Blockchain>(
         &logger_factory,
         link_resolver.cheap_clone(),
         subgraph_instance_manager.clone(),
+        sg_count,
     ));
 
     let panicking_subscription_manager = Arc::new(PanicSubscriptionManager {});