graphql,node,server,store: Wire up _logs query in execution

fordN · fordN · commit f47e987b32a0 · 2026-04-01T10:44:06.000-07:00
Integrates _logs query into the GraphQL execution pipeline:

Execution layer:
- Execute _logs queries via log_store.query_logs()
- Convert LogEntry results to GraphQL response objects
- Handle log store errors gracefully

Query parsing:
- Recognize _logs as special query field
- Build LogQuery from GraphQL arguments
- Pass log_store to execution context

Service wiring:
- Create log store from configuration in launcher
- Provide log store to GraphQL runner
- Use NoOpLogStore in test environments

This completes the read path from GraphQL query to log storage backend.
diff --git a/graphql/src/execution/execution.rs b/graphql/src/execution/execution.rs
@@ -8,7 +8,7 @@ use graph::{
     },
     futures03::future::TryFutureExt,
     prelude::{s, CheapClone},
-    schema::{is_introspection_field, INTROSPECTION_QUERY_TYPE, META_FIELD_NAME},
+    schema::{is_introspection_field, INTROSPECTION_QUERY_TYPE, LOGS_FIELD_NAME, META_FIELD_NAME},
     util::{herd_cache::HerdCache, lfu_cache::EvictStats, timed_rw_lock::TimedMutex},
 };
 use lazy_static::lazy_static;
@@ -231,6 +231,9 @@ where
 
     /// Whether to include an execution trace in the result
     pub trace: bool,
+
+    /// The log store to use for querying logs.
+    pub log_store: Arc<dyn graph::components::log_store::LogStore>,
 }
 
 pub(crate) fn get_field<'a>(
@@ -264,6 +267,7 @@ where
             // `cache_status` is a dead value for the introspection context.
             cache_status: AtomicCell::new(CacheStatus::Miss),
             trace: ENV_VARS.log_sql_timing(),
+            log_store: self.log_store.cheap_clone(),
         }
     }
 }
@@ -273,18 +277,21 @@ pub(crate) async fn execute_root_selection_set_uncached(
     selection_set: &a::SelectionSet,
     root_type: &sast::ObjectType,
 ) -> Result<(Object, Trace), Vec<QueryExecutionError>> {
-    // Split the top-level fields into introspection fields and
-    // regular data fields
+    // Split the top-level fields into introspection fields,
+    // logs fields, meta fields, and regular data fields
     let mut data_set = a::SelectionSet::empty_from(selection_set);
     let mut intro_set = a::SelectionSet::empty_from(selection_set);
     let mut meta_items = Vec::new();
+    let mut logs_fields = Vec::new();
 
     for field in selection_set.fields_for(root_type)? {
         // See if this is an introspection or data field. We don't worry about
         // non-existent fields; those will cause an error later when we execute
         // the data_set SelectionSet
         if is_introspection_field(&field.name) {
             intro_set.push(field)?
+        } else if field.name == LOGS_FIELD_NAME {
+            logs_fields.push(field)
         } else if field.name == META_FIELD_NAME || field.name == "__typename" {
             meta_items.push(field)
         } else {
@@ -314,6 +321,64 @@ pub(crate) async fn execute_root_selection_set_uncached(
         );
     }
 
+    // Resolve logs fields, if there are any
+    for field in logs_fields {
+        use graph::data::graphql::object;
+
+        // Build log query from field arguments
+        let log_query = crate::store::logs::build_log_query(field, ctx.query.schema.id())
+            .map_err(|e| vec![e])?;
+
+        // Query the log store
+        let log_entries = ctx.log_store.query_logs(log_query).await.map_err(|e| {
+            vec![QueryExecutionError::StoreError(
+                anyhow::Error::from(e).into(),
+            )]
+        })?;
+
+        // Convert log entries to GraphQL values
+        let log_values: Vec<r::Value> = log_entries
+            .into_iter()
+            .map(|entry| {
+                // Convert arguments Vec<(String, String)> to GraphQL objects
+                let arguments: Vec<r::Value> = entry
+                    .arguments
+                    .into_iter()
+                    .map(|(key, value)| {
+                        object! {
+                            key: key,
+                            value: value,
+                            __typename: "_LogArgument_"
+                        }
+                    })
+                    .collect();
+
+                // Convert log level to string
+                let level_str = entry.level.as_str().to_uppercase();
+
+                object! {
+                    id: entry.id,
+                    subgraphId: entry.subgraph_id.to_string(),
+                    timestamp: entry.timestamp,
+                    level: level_str,
+                    text: entry.text,
+                    arguments: arguments,
+                    meta: object! {
+                        module: entry.meta.module,
+                        line: r::Value::Int(entry.meta.line),
+                        column: r::Value::Int(entry.meta.column),
+                        __typename: "_LogMeta_"
+                    },
+                    __typename: "_Log_"
+                }
+            })
+            .collect();
+
+        let response_key = Word::from(field.response_key());
+        let logs_object = Object::from_iter(vec![(response_key, r::Value::List(log_values))]);
+        values.append(logs_object);
+    }
+
     Ok((values, trace))
 }
 
diff --git a/graphql/src/query/mod.rs b/graphql/src/query/mod.rs
@@ -29,6 +29,9 @@ pub struct QueryExecutionOptions<R> {
 
     /// Whether to include an execution trace in the result
     pub trace: bool,
+
+    /// The log store to use for querying logs.
+    pub log_store: Arc<dyn graph::components::log_store::LogStore>,
 }
 
 /// Executes a query and returns a result.
@@ -52,6 +55,7 @@ where
         max_skip: options.max_skip,
         cache_status: Default::default(),
         trace: options.trace,
+        log_store: options.log_store,
     });
 
     let selection_set = selection_set
diff --git a/graphql/src/runner.rs b/graphql/src/runner.rs
@@ -26,6 +26,7 @@ pub struct GraphQlRunner<S> {
     store: Arc<S>,
     load_manager: Arc<LoadManager>,
     graphql_metrics: Arc<GraphQLMetrics>,
+    log_store: Arc<dyn graph::components::log_store::LogStore>,
 }
 
 #[cfg(debug_assertions)]
@@ -44,6 +45,7 @@ where
         store: Arc<S>,
         load_manager: Arc<LoadManager>,
         registry: Arc<MetricsRegistry>,
+        log_store: Arc<dyn graph::components::log_store::LogStore>,
     ) -> Self {
         let logger = logger.new(o!("component" => "GraphQlRunner"));
         let graphql_metrics = Arc::new(GraphQLMetrics::new(registry));
@@ -52,6 +54,7 @@ where
             store,
             load_manager,
             graphql_metrics,
+            log_store,
         }
     }
 
@@ -186,6 +189,7 @@ where
                     max_first: max_first.unwrap_or(ENV_VARS.graphql.max_first),
                     max_skip: max_skip.unwrap_or(ENV_VARS.graphql.max_skip),
                     trace: do_trace,
+                    log_store: self.log_store.cheap_clone(),
                 },
             ));
         }
diff --git a/node/src/launcher.rs b/node/src/launcher.rs
@@ -37,6 +37,7 @@ use tokio_util::sync::CancellationToken;
 
 use crate::config::Config;
 use crate::helpers::watch_subgraph_updates;
+use crate::log_config_provider::{LogStoreConfigProvider, LogStoreConfigSources};
 use crate::network_setup::Networks;
 use crate::opt::Opt;
 use crate::store_builder::StoreBuilder;
@@ -367,6 +368,7 @@ fn build_graphql_server(
     metrics_registry: Arc<MetricsRegistry>,
     network_store: &Arc<Store>,
     logger_factory: &LoggerFactory,
+    log_store: Arc<dyn graph::components::log_store::LogStore>,
 ) -> GraphQLQueryServer<GraphQlRunner<Store>> {
     let shards: Vec<_> = config.stores.keys().cloned().collect();
     let load_manager = Arc::new(LoadManager::new(
@@ -380,6 +382,7 @@ fn build_graphql_server(
         network_store.clone(),
         load_manager,
         metrics_registry,
+        log_store,
     ));
 
     GraphQLQueryServer::new(logger_factory, graphql_runner.clone())
@@ -443,20 +446,121 @@ pub async fn run(
 
     info!(logger, "Starting up"; "node_id" => &node_id);
 
-    // Optionally, identify the Elasticsearch logging configuration
-    let elastic_config = opt
-        .elasticsearch_url
-        .clone()
-        .map(|endpoint| ElasticLoggingConfig {
-            endpoint,
-            username: opt.elasticsearch_user.clone(),
-            password: opt.elasticsearch_password.clone(),
-            client: reqwest::Client::new(),
-        });
+    // Create log store configuration provider
+    // Build LogStoreConfig from CLI args with backward compatibility
+    let cli_config = if let Some(backend) = opt.log_store_backend.as_ref() {
+        // New generic CLI args used
+        match backend.to_lowercase().as_str() {
+            "elasticsearch" | "elastic" | "es" => {
+                let url = opt
+                    .log_store_elasticsearch_url
+                    .clone()
+                    .or_else(|| {
+                        if opt.elasticsearch_url.is_some() {
+                            warn!(
+                                logger,
+                                "Using deprecated --elasticsearch-url, use --log-store-elasticsearch-url instead"
+                            );
+                        }
+                        opt.elasticsearch_url.clone()
+                    });
+
+                url.map(|endpoint| {
+                    let index = opt
+                        .log_store_elasticsearch_index
+                        .clone()
+                        .or_else(|| std::env::var("GRAPH_LOG_STORE_ELASTICSEARCH_INDEX").ok())
+                        .or_else(|| std::env::var("GRAPH_ELASTIC_SEARCH_INDEX").ok())
+                        .unwrap_or_else(|| "subgraph".to_string());
+
+                    let timeout_secs = std::env::var("GRAPH_LOG_STORE_ELASTICSEARCH_TIMEOUT")
+                        .or_else(|_| std::env::var("GRAPH_ELASTICSEARCH_TIMEOUT"))
+                        .ok()
+                        .and_then(|s| s.parse::<u64>().ok())
+                        .unwrap_or(10);
+
+                    graph::components::log_store::LogStoreConfig::Elasticsearch {
+                        endpoint,
+                        username: opt
+                            .log_store_elasticsearch_user
+                            .clone()
+                            .or_else(|| opt.elasticsearch_user.clone()),
+                        password: opt
+                            .log_store_elasticsearch_password
+                            .clone()
+                            .or_else(|| opt.elasticsearch_password.clone()),
+                        index,
+                        timeout_secs,
+                    }
+                })
+            }
+
+            "loki" => opt.log_store_loki_url.clone().map(|endpoint| {
+                graph::components::log_store::LogStoreConfig::Loki {
+                    endpoint,
+                    tenant_id: opt.log_store_loki_tenant_id.clone(),
+                }
+            }),
+
+            "file" | "files" => opt.log_store_file_dir.clone().map(|directory| {
+                graph::components::log_store::LogStoreConfig::File {
+                    directory: std::path::PathBuf::from(directory),
+                    max_file_size: opt.log_store_file_max_size.unwrap_or(100 * 1024 * 1024),
+                    retention_days: opt.log_store_file_retention_days.unwrap_or(30),
+                }
+            }),
+
+            "disabled" | "none" => None,
+
+            other => {
+                warn!(logger, "Invalid log store backend: {}", other);
+                None
+            }
+        }
+    } else if opt.elasticsearch_url.is_some() {
+        // Old Elasticsearch-specific CLI args used (backward compatibility)
+        warn!(
+            logger,
+            "Using deprecated --elasticsearch-url CLI argument, \
+             please use --log-store-backend elasticsearch --log-store-elasticsearch-url instead"
+        );
+
+        let index = opt
+            .log_store_elasticsearch_index
+            .clone()
+            .or_else(|| std::env::var("GRAPH_LOG_STORE_ELASTICSEARCH_INDEX").ok())
+            .or_else(|| std::env::var("GRAPH_ELASTIC_SEARCH_INDEX").ok())
+            .unwrap_or_else(|| "subgraph".to_string());
+
+        let timeout_secs = std::env::var("GRAPH_LOG_STORE_ELASTICSEARCH_TIMEOUT")
+            .or_else(|_| std::env::var("GRAPH_ELASTICSEARCH_TIMEOUT"))
+            .ok()
+            .and_then(|s| s.parse::<u64>().ok())
+            .unwrap_or(10);
+
+        Some(
+            graph::components::log_store::LogStoreConfig::Elasticsearch {
+                endpoint: opt.elasticsearch_url.clone().unwrap(),
+                username: opt.elasticsearch_user.clone(),
+                password: opt.elasticsearch_password.clone(),
+                index,
+                timeout_secs,
+            },
+        )
+    } else {
+        // No CLI config provided
+        None
+    };
+
+    let log_config_provider = LogStoreConfigProvider::new(LogStoreConfigSources { cli_config });
+
+    // Resolve log store (for querying) and config (for drains)
+    // Priority: GRAPH_LOG_STORE env var → CLI config → NoOp/None
+    let (log_store, log_store_config) = log_config_provider.resolve(&logger);
 
     // Create a component and subgraph logger factory
     let logger_factory =
-        LoggerFactory::new(logger.clone(), elastic_config, metrics_registry.clone());
+        LoggerFactory::new(logger.clone(), log_store_config, metrics_registry.clone());
 
     let arweave_resolver = Arc::new(ArweaveClient::new(
         logger.cheap_clone(),
@@ -573,6 +677,7 @@ pub async fn run(
             metrics_registry.clone(),
             &network_store,
             &logger_factory,
+            log_store.clone(),
         );
 
         let index_node_server = IndexNodeServer::new(
diff --git a/server/index-node/src/service.rs b/server/index-node/src/service.rs
@@ -153,6 +153,7 @@ where
                 max_first: u32::MAX,
                 max_skip: u32::MAX,
                 trace: false,
+                log_store: Arc::new(graph::components::log_store::NoOpLogStore),
             };
             let (result, _) = execute_query(query_clone.cheap_clone(), None, None, options).await;
             query_clone.log_execution(0);
diff --git a/store/test-store/src/store.rs b/store/test-store/src/store.rs
@@ -570,7 +570,7 @@ async fn execute_subgraph_query_internal(
                 error_policy,
                 query.schema.id().clone(),
                 graphql_metrics(),
-                LOAD_MANAGER.clone()
+                LOAD_MANAGER.clone(),
             )
             .await
         );
@@ -584,6 +584,7 @@ async fn execute_subgraph_query_internal(
                 max_first: u32::MAX,
                 max_skip: u32::MAX,
                 trace,
+                log_store: std::sync::Arc::new(graph::components::log_store::NoOpLogStore),
             },
         )
         .await;
