feat(metrics): per-source ingest error counter (hotblocks_ingest_source_errors_total)

crates/data-client/src/reqwest/client.rs

@@ -226,6 +226,39 @@ impl DataClient for ReqwestDataClient {
     fn is_retryable(&self, err: &anyhow::Error) -> bool {
         self.is_retryable(err)
     }
+
+    fn error_kind(&self, err: &anyhow::Error) -> &'static str {
+        for cause in err.chain() {
+            if cause.downcast_ref::<UnexpectedHttpStatus>().is_some() {
+                return "http";
+            }
+            if let Some(reqwest_error) = cause.downcast_ref::<reqwest::Error>() {
+                if reqwest_error.is_timeout() {
+                    return "timeout";
+                }
+                if reqwest_error.is_connect() {
+                    return "connect";
+                }
+                if reqwest_error.is_status() {
+                    return "http";
+                }
+                if reqwest_error.is_body() || reqwest_error.is_decode() {
+                    return "decode";
+                }
+                if reqwest_error.is_request() {
+                    return "request";
+                }
+            }
+            if cause.downcast_ref::<std::io::Error>().is_some() {
+                return "io";
+            }
+        }
+        "other"
+    }
+
+    fn source_label(&self) -> String {
+        self.url.host_str().unwrap_or("unknown").to_string()
+    }
 }
 
 #[derive(Debug)]

crates/data-client/src/types.rs

@@ -45,4 +45,14 @@ pub trait DataClient: Send + Sync + Debug + Unpin {
     fn get_finalized_head(&self) -> BoxFuture<'static, anyhow::Result<Option<BlockRef>>>;
 
     fn is_retryable(&self, err: &anyhow::Error) -> bool;
+
+    /// Error class for the `ingest_source_errors` metric label.
+    fn error_kind(&self, _err: &anyhow::Error) -> &'static str {
+        "other"
+    }
+
+    /// Source identity (host) for the `ingest_source_errors` metric label.
+    fn source_label(&self) -> String {
+        "unknown".to_string()
+    }
 }

crates/data-source/Cargo.toml

@@ -6,6 +6,7 @@ edition = "2021"
 [dependencies]
 anyhow = { workspace = true }
 futures = { workspace = true }
+prometheus-client = { workspace = true }
 sqd-data-client = { path = "../data-client" }
 sqd-primitives = { path = "../primitives" }
 tokio = { workspace = true }

crates/data-source/src/lib.rs

@@ -1,4 +1,5 @@
 mod map;
+pub mod metrics;
 mod standard;
 mod types;
 

crates/data-source/src/metrics.rs

@@ -0,0 +1,15 @@
+use std::sync::LazyLock;
+
+use prometheus_client::metrics::{counter::Counter, family::Family};
+
+type Labels = Vec<(&'static str, String)>;
+
+/// Upstream data-source ingestion errors by `source` host and `kind`.
+/// Registered in the hotblocks metrics registry.
+pub static INGEST_SOURCE_ERRORS: LazyLock<Family<Labels, Counter>> = LazyLock::new(Default::default);
+
+pub(crate) fn record_ingest_source_error(source: &str, kind: &'static str) {
+    INGEST_SOURCE_ERRORS
+        .get_or_create(&vec![("source", source.to_string()), ("kind", kind.to_string())])
+        .inc();
+}

crates/data-source/src/standard.rs

@@ -221,6 +221,8 @@ impl<C: DataClient> Endpoint<C> {
     }
 
     fn on_error(&mut self, error: anyhow::Error) {
+        crate::metrics::record_ingest_source_error(&self.client.source_label(), self.client.error_kind(&error));
+
         let backoff = [0, 100, 200, 500, 1000, 2000, 5000, 10000];
         let pause = backoff[std::cmp::min(self.error_counter, backoff.len() - 1)];
         if pause > 0 {

crates/hotblocks/src/metrics.rs

@@ -184,6 +184,12 @@ pub fn build_metrics_registry() -> Registry {
         QUERY_ERROR_TOO_MANY_DATA_WAITERS.clone()
     );
 
+    registry.register(
+        "ingest_source_errors",
+        "Upstream data source ingestion errors, by source host and kind (connect/timeout/http/io/...)",
+        sqd_data_source::metrics::INGEST_SOURCE_ERRORS.clone()
+    );
+
     registry.register("http_status", "Number of sent HTTP responses", HTTP_STATUS.clone());
     registry.register(
         "http_seconds_to_first_byte",