-
Notifications
You must be signed in to change notification settings - Fork 980
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
core, graph: track gas metrics in csv
- Loading branch information
1 parent
b1bf41c
commit 256520d
Showing
9 changed files
with
224 additions
and
57 deletions.
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,64 +1,112 @@ | ||
| use super::MetricsRegistry; | ||
| use crate::prelude::DeploymentHash; | ||
| use prometheus::CounterVec; | ||
| use std::sync::Arc; | ||
| use crate::blockchain::BlockPtr; | ||
| use crate::components::store::DeploymentId; | ||
| use crate::env::ENV_VARS; | ||
| use crate::spawn; | ||
| use anyhow::Result; | ||
| use cloud_storage::{Bucket, Object}; | ||
| use csv::Writer; | ||
| use slog::{error, info, Logger}; | ||
| use std::sync::RwLock; | ||
| use std::{collections::HashMap, sync::Arc}; | ||
|
|
||
| #[derive(Clone)] | ||
| pub struct GasMetrics { | ||
| pub gas_counter: CounterVec, | ||
| pub op_counter: CounterVec, | ||
| pub gas_counter_map: Arc<RwLock<HashMap<String, u64>>>, | ||
| pub op_counter_map: Arc<RwLock<HashMap<String, u64>>>, | ||
| } | ||
|
|
||
| impl GasMetrics { | ||
| pub fn new(subgraph_id: DeploymentHash, registry: Arc<MetricsRegistry>) -> Self { | ||
| let gas_counter = registry | ||
| .global_deployment_counter_vec( | ||
| "deployment_gas", | ||
| "total gas used", | ||
| subgraph_id.as_str(), | ||
| &["method"], | ||
| ) | ||
| .unwrap_or_else(|err| { | ||
| panic!( | ||
| "Failed to register deployment_gas prometheus counter for {}: {}", | ||
| subgraph_id, err | ||
| ) | ||
| }); | ||
|
|
||
| let op_counter = registry | ||
| .global_deployment_counter_vec( | ||
| "deployment_op_count", | ||
| "total number of operations", | ||
| subgraph_id.as_str(), | ||
| &["method"], | ||
| ) | ||
| .unwrap_or_else(|err| { | ||
| panic!( | ||
| "Failed to register deployment_op_count prometheus counter for {}: {}", | ||
| subgraph_id, err | ||
| ) | ||
| }); | ||
| pub fn new() -> Self { | ||
| let gas_counter_map = Arc::new(RwLock::new(HashMap::new())); | ||
| let op_counter_map = Arc::new(RwLock::new(HashMap::new())); | ||
|
|
||
| GasMetrics { | ||
| gas_counter, | ||
| op_counter, | ||
| gas_counter_map, | ||
| op_counter_map, | ||
| } | ||
| } | ||
|
|
||
| pub fn mock() -> Self { | ||
| let subgraph_id = DeploymentHash::default(); | ||
| Self::new(subgraph_id, Arc::new(MetricsRegistry::mock())) | ||
| // Converts the map to CSV and returns it as a String | ||
| fn map_to_csv(data: &HashMap<String, u64>) -> Result<String> { | ||
| let mut wtr = Writer::from_writer(vec![]); | ||
| for (key, value) in data { | ||
| wtr.serialize((key, value))?; | ||
| } | ||
| wtr.flush()?; | ||
| Ok(String::from_utf8(wtr.into_inner()?)?) | ||
| } | ||
|
|
||
| async fn write_csv_to_gcs(bucket_name: &str, path: &str, data: String) -> Result<()> { | ||
| let bucket = Bucket::read(bucket_name).await?; | ||
|
|
||
| let data_bytes = data.into_bytes(); | ||
|
|
||
| let _ = Object::create(&bucket.name, data_bytes, path, "text/csv").await?; | ||
|
|
||
| Ok(()) | ||
| } | ||
|
|
||
| /// Flushes gas and op metrics to GCS asynchronously, clearing metrics maps afterward. | ||
| /// | ||
| /// Serializes metrics to CSV and spawns tasks for GCS upload, logging successes or errors. | ||
| /// Metrics are organized by block number and subgraph ID in the GCS bucket. | ||
| /// Returns `Ok(())` to indicate the upload process has started. | ||
| pub fn flush_metrics_to_gcs( | ||
| &self, | ||
| logger: &Logger, | ||
| block_ptr: BlockPtr, | ||
| subgraph_id: DeploymentId, | ||
| ) -> Result<()> { | ||
| let logger = logger.clone(); | ||
| let gas_data = Self::map_to_csv(&self.gas_counter_map.read().unwrap())?; | ||
| let op_data = Self::map_to_csv(&self.op_counter_map.read().unwrap())?; | ||
|
|
||
| spawn(async move { | ||
| let gas_file = format!("{}/gas/{}.csv", subgraph_id, block_ptr.number); | ||
| let op_file = format!("{}/op/{}.csv", subgraph_id, block_ptr.number); | ||
|
|
||
| let bucket = &ENV_VARS.gas_metrics_gcs_bucket; | ||
|
|
||
| match Self::write_csv_to_gcs(bucket, &gas_file, gas_data).await { | ||
| Ok(_) => { | ||
| info!( | ||
| logger, | ||
| "Wrote gas metrics to GCS for block {}", block_ptr.number | ||
| ); | ||
| } | ||
| Err(e) => error!(logger, "Error writing gas metrics to GCS: {}", e), | ||
| } | ||
|
|
||
| match Self::write_csv_to_gcs(bucket, &op_file, op_data).await { | ||
| Ok(_) => { | ||
| info!( | ||
| logger, | ||
| "Wrote op metrics to GCS for block {}", block_ptr.number | ||
| ); | ||
| } | ||
| Err(e) => error!(logger, "Error writing op metrics to GCS: {}", e), | ||
| } | ||
| }); | ||
|
|
||
| // Clear the maps | ||
| self.gas_counter_map.write().unwrap().clear(); | ||
| self.op_counter_map.write().unwrap().clear(); | ||
|
|
||
| Ok(()) | ||
| } | ||
|
|
||
| pub fn mock() -> Arc<Self> { | ||
| Arc::new(Self::new()) | ||
| } | ||
|
|
||
| pub fn track_gas(&self, method: &str, gas_used: u64) { | ||
| self.gas_counter | ||
| .with_label_values(&[method]) | ||
| .inc_by(gas_used as f64); | ||
| let mut map = self.gas_counter_map.write().unwrap(); // | ||
| let counter = map.entry(method.to_string()).or_insert(0); | ||
| *counter += gas_used; | ||
| } | ||
|
|
||
| pub fn track_operations(&self, method: &str, op_count: u64) { | ||
| self.op_counter | ||
| .with_label_values(&[method]) | ||
| .inc_by(op_count as f64); | ||
| let mut map = self.op_counter_map.write().unwrap(); | ||
| let counter = map.entry(method.to_string()).or_insert(0); | ||
| *counter += op_count; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.