graph: Add endpoint metrics

chain/ethereum/examples/firehose.rs

@@ -1,7 +1,9 @@
 use anyhow::Error;
 use graph::{
+    endpoint::EndpointMetrics,
     env::env_var,
     firehose::SubgraphLimit,
+    log::logger,
     prelude::{prost, tokio, tonic},
     {firehose, firehose::FirehoseEndpoint},
 };
@@ -20,13 +22,18 @@ async fn main() -> Result<(), Error> {
         token = Some(token_env);
     }
 
+    let logger = logger(false);
+    let host = "https://api.streamingfast.io:443".to_string();
+    let metrics = Arc::new(EndpointMetrics::new(logger, &[host.clone()]));
+
     let firehose = Arc::new(FirehoseEndpoint::new(
         "firehose",
-        "https://api.streamingfast.io:443",
+        &host,
         token,
         false,
         false,
         SubgraphLimit::Unlimited,
+        metrics,
     ));
 
     loop {

chain/ethereum/src/chain.rs

@@ -399,7 +399,7 @@ impl Blockchain for Chain {
     ) -> Result<BlockPtr, IngestorError> {
         match self.client.as_ref() {
             ChainClient::Firehose(endpoints) => endpoints
-                .random()?
+                .endpoint()?
                 .block_ptr_for_number::<HeaderOnlyBlock>(logger, number)
                 .await
                 .map_err(IngestorError::Unknown),

chain/substreams/examples/substreams.rs

@@ -1,6 +1,7 @@
 use anyhow::{format_err, Context, Error};
 use graph::blockchain::block_stream::BlockStreamEvent;
 use graph::blockchain::substreams_block_stream::SubstreamsBlockStream;
+use graph::endpoint::EndpointMetrics;
 use graph::firehose::SubgraphLimit;
 use graph::prelude::{info, tokio, DeploymentHash, Registry};
 use graph::tokio_stream::StreamExt;
@@ -41,13 +42,16 @@ async fn main() -> Result<(), Error> {
         prometheus_registry.clone(),
     ));
 
+    let endpoint_metrics = EndpointMetrics::new(logger.clone(), &[endpoint.clone()]);
+
     let firehose = Arc::new(FirehoseEndpoint::new(
         "substreams",
         &endpoint,
         token,
         false,
         false,
         SubgraphLimit::Unlimited,
+        Arc::new(endpoint_metrics),
     ));
 
     let mut stream: SubstreamsBlockStream<graph_chain_substreams::Chain> =

graph/Cargo.toml

@@ -17,7 +17,7 @@ chrono = "0.4.23"
 envconfig = "0.10.0"
 Inflector = "0.11.3"
 isatty = "0.1.9"
-reqwest = { version = "0.11.2", features = ["json", "stream", "multipart"] }
+reqwest = { version = "0.11.14", features = ["json", "stream", "multipart"] }
 ethabi = "17.2"
 hex = "0.4.3"
 http = "0.2.3"

graph/src/blockchain/client.rs

@@ -25,7 +25,7 @@ impl<C: Blockchain> ChainClient<C> {
         // adapter limits in the configuration can effectively disable firehose
         // by setting a limit to 0.
         // In this case we should fallback to an rpc client.
-        let firehose_available = firehose_endpoints.random().is_ok();
+        let firehose_available = firehose_endpoints.endpoint().is_ok();
 
         match firehose_available {
             true => Self::Firehose(firehose_endpoints),
@@ -42,7 +42,7 @@ impl<C: Blockchain> ChainClient<C> {
 
     pub fn firehose_endpoint(&self) -> anyhow::Result<Arc<FirehoseEndpoint>> {
         match self {
-            ChainClient::Firehose(endpoints) => endpoints.random(),
+            ChainClient::Firehose(endpoints) => endpoints.endpoint(),
             _ => Err(anyhow!("firehose endpoint requested on rpc chain client")),
         }
     }

graph/src/blockchain/firehose_block_stream.rs

@@ -226,6 +226,7 @@ fn stream_blocks<C: Blockchain, F: FirehoseMapper<C>>(
                 "start_block" => start_block_num,
                 "subgraph" => &deployment,
                 "cursor" => latest_cursor.to_string(),
+                "provider_err_count" => endpoint.current_error_count(),
             );
 
             // We just reconnected, assume that we want to back off on errors