graph,graphql,node,server,store,tests: Make subgraph logs queryable

Author: fordN

graph/src/components/log_store/elasticsearch.rs

@@ -0,0 +1,210 @@
+use async_trait::async_trait;
+use reqwest::Client;
+use serde::Deserialize;
+use serde_json::json;
+use std::collections::HashMap;
+use std::time::Duration;
+
+use crate::log::elastic::ElasticLoggingConfig;
+use crate::prelude::DeploymentHash;
+
+use super::{LogEntry, LogLevel, LogMeta, LogQuery, LogStore, LogStoreError};
+
+pub struct ElasticsearchLogStore {
+    endpoint: String,
+    username: Option<String>,
+    password: Option<String>,
+    client: Client,
+    index: String,
+}
+
+impl ElasticsearchLogStore {
+    pub fn new(config: ElasticLoggingConfig, index: String) -> Self {
+        Self {
+            endpoint: config.endpoint,
+            username: config.username,
+            password: config.password,
+            client: config.client,
+            index,
+        }
+    }
+
+    fn build_query(&self, query: &LogQuery) -> serde_json::Value {
+        let mut must_clauses = Vec::new();
+
+        // Filter by subgraph ID
+        must_clauses.push(json!({
+            "term": {
+                "subgraphId": query.subgraph_id.to_string()
+            }
+        }));
+
+        // Filter by log level
+        if let Some(level) = &query.level {
+            must_clauses.push(json!({
+                "term": {
+                    "level": level.as_str()
+                }
+            }));
+        }
+
+        // Filter by time range
+        if query.from.is_some() || query.to.is_some() {
+            let mut range = serde_json::Map::new();
+            if let Some(from) = &query.from {
+                range.insert("gte".to_string(), json!(from));
+            }
+            if let Some(to) = &query.to {
+                range.insert("lte".to_string(), json!(to));
+            }
+            must_clauses.push(json!({
+                "range": {
+                    "timestamp": range
+                }
+            }));
+        }
+
+        // Filter by text search
+        if let Some(text) = &query.text {
+            must_clauses.push(json!({
+                "match": {
+                    "text": text
+                }
+            }));
+        }
+
+        json!({
+            "query": {
+                "bool": {
+                    "must": must_clauses
+                }
+            },
+            "from": query.skip,
+            "size": query.first,
+            "sort": [
+                { "timestamp": { "order": "desc" } }
+            ]
+        })
+    }
+
+    async fn execute_search(
+        &self,
+        query_body: serde_json::Value,
+    ) -> Result<Vec<LogEntry>, LogStoreError> {
+        let url = format!("{}/{}/_search", self.endpoint, self.index);
+
+        let mut request = self
+            .client
+            .post(&url)
+            .json(&query_body)
+            .timeout(Duration::from_secs(10));
+
+        // Add basic auth if credentials provided
+        if let (Some(username), Some(password)) = (&self.username, &self.password) {
+            request = request.basic_auth(username, Some(password));
+        }
+
+        let response = request.send().await.map_err(|e| {
+            LogStoreError::QueryFailed(
+                anyhow::Error::from(e).context("Elasticsearch request failed"),
+            )
+        })?;
+
+        if !response.status().is_success() {
+            let status = response.status();
+            // Do not include the response body in the error to avoid leaking
+            // sensitive Elasticsearch internals
+            return Err(LogStoreError::QueryFailed(anyhow::anyhow!(
+                "Elasticsearch query failed with status {}",
+                status
+            )));
+        }
+
+        let response_body: ElasticsearchResponse = response.json().await.map_err(|e| {
+            LogStoreError::QueryFailed(
+                anyhow::Error::from(e).context(
+                    "failed to parse Elasticsearch search response: response format may have changed or be invalid",
+                ),
+            )
+        })?;
+
+        let entries = response_body
+            .hits
+            .hits
+            .into_iter()
+            .filter_map(|hit| self.parse_log_entry(hit.source))
+            .collect();
+
+        Ok(entries)
+    }
+
+    fn parse_log_entry(&self, source: ElasticsearchLogDocument) -> Option<LogEntry> {
+        let level = LogLevel::from_str(&source.level)?;
+        let subgraph_id = DeploymentHash::new(&source.subgraph_id).ok()?;
+
+        // Convert arguments HashMap to Vec<(String, String)>
+        let arguments: Vec<(String, String)> = source.arguments.into_iter().collect();
+
+        Some(LogEntry {
+            id: source.id,
+            subgraph_id,
+            timestamp: source.timestamp,
+            level,
+            text: source.text,
+            arguments,
+            meta: LogMeta {
+                module: source.meta.module,
+                line: source.meta.line,
+                column: source.meta.column,
+            },
+        })
+    }
+}
+
+#[async_trait]
+impl LogStore for ElasticsearchLogStore {
+    async fn query_logs(&self, query: LogQuery) -> Result<Vec<LogEntry>, LogStoreError> {
+        let query_body = self.build_query(&query);
+        self.execute_search(query_body).await
+    }
+
+    fn is_available(&self) -> bool {
+        true
+    }
+}
+
+// Elasticsearch response types
+#[derive(Debug, Deserialize)]
+struct ElasticsearchResponse {
+    hits: ElasticsearchHits,
+}
+
+#[derive(Debug, Deserialize)]
+struct ElasticsearchHits {
+    hits: Vec<ElasticsearchHit>,
+}
+
+#[derive(Debug, Deserialize)]
+struct ElasticsearchHit {
+    #[serde(rename = "_source")]
+    source: ElasticsearchLogDocument,
+}
+
+#[derive(Debug, Deserialize)]
+struct ElasticsearchLogDocument {
+    id: String,
+    #[serde(rename = "subgraphId")]
+    subgraph_id: String,
+    timestamp: String,
+    level: String,
+    text: String,
+    arguments: HashMap<String, String>,
+    meta: ElasticsearchLogMeta,
+}
+
+#[derive(Debug, Deserialize)]
+struct ElasticsearchLogMeta {
+    module: String,
+    line: i64,
+    column: i64,
+}

graph/src/components/log_store/mod.rs

@@ -0,0 +1,97 @@
+pub mod elasticsearch;
+
+use async_trait::async_trait;
+use thiserror::Error;
+
+use crate::prelude::DeploymentHash;
+
+#[derive(Error, Debug)]
+pub enum LogStoreError {
+    #[error("log store query failed: {0}")]
+    QueryFailed(#[from] anyhow::Error),
+
+    #[error("log store is unavailable")]
+    Unavailable,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum LogLevel {
+    Critical,
+    Error,
+    Warning,
+    Info,
+    Debug,
+}
+
+impl LogLevel {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            LogLevel::Critical => "critical",
+            LogLevel::Error => "error",
+            LogLevel::Warning => "warning",
+            LogLevel::Info => "info",
+            LogLevel::Debug => "debug",
+        }
+    }
+
+    pub fn from_str(s: &str) -> Option<Self> {
+        match s.trim().to_lowercase().as_str() {
+            "critical" => Some(LogLevel::Critical),
+            "error" => Some(LogLevel::Error),
+            "warning" => Some(LogLevel::Warning),
+            "info" => Some(LogLevel::Info),
+            "debug" => Some(LogLevel::Debug),
+            _ => None,
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct LogMeta {
+    pub module: String,
+    pub line: i64,
+    pub column: i64,
+}
+
+#[derive(Debug, Clone)]
+pub struct LogEntry {
+    pub id: String,
+    pub subgraph_id: DeploymentHash,
+    pub timestamp: String,
+    pub level: LogLevel,
+    pub text: String,
+    pub arguments: Vec<(String, String)>,
+    pub meta: LogMeta,
+}
+
+#[derive(Debug, Clone)]
+pub struct LogQuery {
+    pub subgraph_id: DeploymentHash,
+    pub level: Option<LogLevel>,
+    pub from: Option<String>,
+    pub to: Option<String>,
+    pub text: Option<String>,
+    pub first: u32,
+    pub skip: u32,
+}
+
+#[async_trait]
+pub trait LogStore: Send + Sync + 'static {
+    async fn query_logs(&self, query: LogQuery) -> Result<Vec<LogEntry>, LogStoreError>;
+    fn is_available(&self) -> bool;
+}
+
+/// A no-op LogStore that returns empty results
+/// Used when Elasticsearch is not configured
+pub struct NoOpLogStore;
+
+#[async_trait]
+impl LogStore for NoOpLogStore {
+    async fn query_logs(&self, _query: LogQuery) -> Result<Vec<LogEntry>, LogStoreError> {
+        Ok(vec![])
+    }
+
+    fn is_available(&self) -> bool {
+        false
+    }
+}

graph/src/components/mod.rs

@@ -50,6 +50,9 @@ pub mod server;
 /// Components dealing with storing entities.
 pub mod store;
 
+/// Components dealing with log storage.
+pub mod log_store;
+
 pub mod link_resolver;
 
 pub mod trigger_processor;