feat: add subgraph health endpoint

README.md

@@ -113,6 +113,35 @@ curl -X POST \
 }
 ```
 
+## Subgraph health check
+```bash
+curl http://localhost:7600/subgraphs/health/QmVhiE4nax9i86UBnBmQCYDzvjWuwHShYh7aspGPQhU5Sj
+```
+```json
+{
+    "health": "healthy"
+}
+```
+## Unfound subgraph
+```bash
+curl http://localhost:7600/subgraphs/health/QmacQnSgia4iDPWHpeY6aWxesRFdb8o5DKZUx96zZqEWrB
+```
+```json
+{
+    "error": "Deployment not found"
+}
+```
+## Failed Subgraph
+```bash
+curl http://localhost:7600/subgraphs/health/QmVGSJyvjEjkk5U9EdxyyB78NCXK3EAoFhrzm6LV7SxxAm
+```
+```json
+{
+    "fatalError": "transaction 21e77ed08fbc9df7be81101e9b03c2616494cee7cac2f6ad4f1ee387cf799e0c: error while executing at wasm backtrace:\t    0: 0x5972 - <unknown>!mappings/core/handleSwap: Mapping aborted at mappings/core.ts, line 73, column 16, with message: unexpected null in handler `handleSwap` at block #36654250 (5ab4d80c8e2cd628d5bf03abab4c302fd21d25d734e66afddff7a706b804fe13)",
+    "health": "failed"
+}
+```
+
 # Network queries
 ## Checks for auth and configuration to serve-network-subgraph
 

common/src/indexer_service/http/health.rs

@@ -0,0 +1,133 @@
+// Copyright 2023-, Edge & Node, GraphOps, and Semiotic Labs.
+// SPDX-License-Identifier: Apache-2.0
+
+use axum::{
+    extract::Path,
+    response::{IntoResponse, Response as AxumResponse},
+    Extension, Json,
+};
+use graphql_client::GraphQLQuery;
+use indexer_config::GraphNodeConfig;
+use reqwest::StatusCode;
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use thiserror::Error;
+
+#[derive(Deserialize, Debug)]
+struct Response {
+    data: SubgraphData,
+}
+
+#[derive(Deserialize, Debug)]
+#[allow(non_snake_case)]
+struct SubgraphData {
+    indexingStatuses: Vec<IndexingStatus>,
+}
+
+#[derive(Deserialize, Debug)]
+#[allow(non_snake_case)]
+struct IndexingStatus {
+    health: Health,
+    fatalError: Option<Message>,
+    nonFatalErrors: Vec<Message>,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Message {
+    message: String,
+}
+
+#[derive(GraphQLQuery)]
+#[graphql(
+    schema_path = "../graphql/indexing_status.schema.graphql",
+    query_path = "../graphql/subgraph_health.query.graphql",
+    response_derives = "Debug",
+    variables_derives = "Clone"
+)]
+pub struct HealthQuery;
+
+#[derive(Deserialize, Debug)]
+#[allow(non_camel_case_types)]
+enum Health {
+    healthy,
+    unhealthy,
+    failed,
+}
+
+impl Health {
+    fn as_str(&self) -> &str {
+        match self {
+            Health::healthy => "healthy",
+            Health::unhealthy => "unhealthy",
+            Health::failed => "failed",
+        }
+    }
+}
+
+#[derive(Debug, Error)]
+pub enum CheckHealthError {
+    #[error("Deployment not found")]
+    DeploymentNotFound,
+    #[error("Failed to process query")]
+    QueryForwardingError,
+}
+
+impl IntoResponse for CheckHealthError {
+    fn into_response(self) -> AxumResponse {
+        let (status, error_message) = match &self {
+            CheckHealthError::DeploymentNotFound => (StatusCode::NOT_FOUND, "Deployment not found"),
+            CheckHealthError::QueryForwardingError => {
+                (StatusCode::INTERNAL_SERVER_ERROR, "Failed to process query")
+            }
+        };
+
+        let body = serde_json::json!({
+            "error": error_message,
+        });
+
+        (status, Json(body)).into_response()
+    }
+}
+
+pub async fn health(
+    Path(deployment_id): Path<String>,
+    Extension(graph_node): Extension<GraphNodeConfig>,
+) -> Result<impl IntoResponse, CheckHealthError> {
+    let req_body = HealthQuery::build_query(health_query::Variables {
+        ids: vec![deployment_id],
+    });
+
+    let client = reqwest::Client::new();
+    let response = client
+        .post(graph_node.status_url)
+        .json(&req_body)
+        .send()
+        .await;
+    let res = response.expect("Failed to get response");
+    let response_json: Result<Response, reqwest::Error> = res.json().await;
+
+    match response_json {
+        Ok(res) => {
+            if res.data.indexingStatuses.is_empty() {
+                return Err(CheckHealthError::DeploymentNotFound);
+            };
+            let status = &res.data.indexingStatuses[0];
+            let health_response = match status.health {
+                Health::healthy => json!({ "health": status.health.as_str() }),
+                Health::unhealthy => {
+                    let errors: Vec<&String> = status
+                        .nonFatalErrors
+                        .iter()
+                        .map(|msg| &msg.message)
+                        .collect();
+                    json!({ "health": status.health.as_str(), "nonFatalErrors": errors })
+                }
+                Health::failed => {
+                    json!({ "health": status.health.as_str(), "fatalError": status.fatalError.as_ref().map_or("null", |msg| &msg.message) })
+                }
+            };
+            Ok(Json(health_response))
+        }
+        Err(_) => Err(CheckHealthError::QueryForwardingError),
+    }
+}

common/src/indexer_service/http/indexer_service.rs

@@ -33,8 +33,10 @@ use tower_http::{cors, cors::CorsLayer, normalize_path::NormalizePath, trace::Tr
 use tracing::error;
 use tracing::{info, info_span};
 
+use super::request_handler::request_handler;
 use crate::escrow_accounts::EscrowAccounts;
 use crate::escrow_accounts::EscrowAccountsError;
+use crate::indexer_service::http::health::health;
 use crate::{
     address::public_key,
     indexer_service::http::static_subgraph::static_subgraph_request_handler,
@@ -44,8 +46,6 @@ use crate::{
     },
     tap::IndexerTapContext,
 };
-
-use super::request_handler::request_handler;
 use indexer_config::Config;
 
 pub trait IndexerServiceResponse {
@@ -386,7 +386,7 @@ impl IndexerService {
             .route("/", get("Service is up and running"))
             .route("/version", get(Json(options.release)))
             .route("/info", get(operator_address))
-            .layer(misc_rate_limiter);
+            .layer(misc_rate_limiter.clone());
 
         // Rate limits by allowing bursts of 50 requests and requiring 20ms of
         // time between consecutive requests after that, effectively rate
@@ -401,6 +401,12 @@ impl IndexerService {
             ),
         };
 
+        // Check subgraph Health
+        misc_routes = misc_routes
+            .route("/subgraph/health/:deployment_id", get(health))
+            .route_layer(Extension(options.config.graph_node.clone()))
+            .layer(misc_rate_limiter);
+
         if options.config.service.serve_network_subgraph {
             info!("Serving network subgraph at /network");
 

common/src/indexer_service/http/mod.rs

@@ -1,6 +1,7 @@
 // Copyright 2023-, Edge & Node, GraphOps, and Semiotic Labs.
 // SPDX-License-Identifier: Apache-2.0
 
+mod health;
 mod indexer_service;
 mod request_handler;
 mod static_subgraph;

common/src/subgraph_client/client.rs

@@ -6,6 +6,7 @@ use anyhow::anyhow;
 use axum::body::Bytes;
 use graphql_client::GraphQLQuery;
 use reqwest::{header, Url};
+use serde::Serialize;
 use serde_json::{Map, Value};
 use thegraph_core::DeploymentId;
 use thegraph_graphql_http::{
@@ -15,7 +16,7 @@ use thegraph_graphql_http::{
 use tokio::sync::watch::Receiver;
 use tracing::warn;
 
-#[derive(Clone)]
+#[derive(Clone, Serialize)]
 pub struct Query {
     pub query: Document,
     pub variables: Map<String, Value>,