feat: add new route to check subgraph health

Author: shiyasmohd

common/src/indexer_service/http/health.rs

@@ -0,0 +1,114 @@
+use crate::subgraph_client::Query;
+use axum::{
+    extract::Path,
+    response::{IntoResponse, Response as AxumResponse},
+    Extension, Json,
+};
+use reqwest::StatusCode;
+use serde::Deserialize;
+use serde_json::json;
+use thiserror::Error;
+
+use super::GraphNodeConfig;
+
+#[derive(Deserialize, Debug)]
+struct Response {
+    data: SubgraphData,
+}
+
+#[derive(Deserialize, Debug)]
+#[allow(non_snake_case)]
+struct SubgraphData {
+    indexingStatuses: Vec<IndexingStatus>,
+}
+
+#[derive(Deserialize, Debug)]
+struct IndexingStatus {
+    health: Health,
+}
+
+#[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("Graph node config not found")]
+    GraphNodeConfigNotFound,
+    #[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::GraphNodeConfigNotFound => {
+                (StatusCode::NOT_FOUND, "Graph node config not found")
+            }
+            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<Option<GraphNodeConfig>>,
+) -> Result<impl IntoResponse, CheckHealthError> {
+    let url = if let Some(graph_node) = graph_node {
+        graph_node.status_url
+    } else {
+        return Err(CheckHealthError::GraphNodeConfigNotFound);
+    };
+
+    let body = Query::new_with_variables(
+        r#"
+            query indexingStatuses($ids: [String!]!) {
+                indexingStatuses(subgraphs: $ids) {
+                    health
+                }
+            }
+        "#,
+        [("ids", json!([deployment_id]))],
+    );
+
+    let client = reqwest::Client::new();
+    let response = client.post(url).json(&body).send().await;
+    let response = response.expect("Failed to get response");
+    let response_json: Result<Response, reqwest::Error> = response.json().await;
+
+    match response_json {
+        Ok(res) => {
+            if res.data.indexingStatuses.len() == 0 {
+                return Err(CheckHealthError::DeploymentNotFound);
+            };
+            let health_status = res.data.indexingStatuses[0].health.as_str();
+            Ok(Json(json!({ "health": health_status })))
+        }
+        Err(_) => return Err(CheckHealthError::QueryForwardingError),
+    }
+}

common/src/indexer_service/http/indexer_service.rs

@@ -36,6 +36,7 @@ use tracing::{info, info_span};
 
 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,
@@ -352,6 +353,11 @@ impl IndexerService {
             .route("/info", get(operator_address))
             .layer(misc_rate_limiter);
 
+        // Check subgraph Health
+        misc_routes = misc_routes
+            .route("/subgraph/health/:deployment_id", get(health))
+            .route_layer(Extension(options.config.graph_node));
+
         // Rate limits by allowing bursts of 50 requests and requiring 20ms of
         // time between consecutive requests after that, effectively rate
         // limiting to 50 req/s.

common/src/indexer_service/http/mod.rs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 mod config;
+mod health;
 mod indexer_service;
 mod request_handler;
 mod static_subgraph;

common/src/subgraph_client/client.rs

@@ -7,6 +7,7 @@ use axum::body::Bytes;
 use eventuals::Eventual;
 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 tracing::warn;
 
-#[derive(Clone)]
+#[derive(Clone, Serialize)]
 pub struct Query {
     pub query: Document,
     pub variables: Map<String, Value>,