diff --git a/Cargo.lock b/Cargo.lock index db494695bd..453505974c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5034,6 +5034,25 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "mock-collector" +version = "0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb5f0b06ce724a9050cf5cd5038580123ed1fcfc3d8e3803feccf68bbd66039f" +dependencies = [ + "axum", + "opentelemetry-otlp 0.31.1", + "opentelemetry-proto 0.31.0", + "prost", + "serde", + "serde_json", + "thiserror 2.0.18", + "tokio", + "tokio-stream", + "tonic", + "tower-http", +] + [[package]] name = "moka" version = "0.12.15" @@ -5395,6 +5414,20 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" +[[package]] +name = "opentelemetry" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b84bcd6ae87133e903af7ef497404dda70c60d0ea14895fc8a5e6722754fc2a0" +dependencies = [ + "futures-core", + "futures-sink", + "js-sys", + "pin-project-lite", + "thiserror 2.0.18", + "tracing", +] + [[package]] name = "opentelemetry" version = "0.32.0" @@ -5415,12 +5448,24 @@ version = "0.32.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2c0080f0dc1d7c786f467cd85a4e395fcab11ee852004f39a29a18ab7c25d837" dependencies = [ - "opentelemetry", + "opentelemetry 0.32.0", "tracing", "tracing-core", "tracing-subscriber", ] +[[package]] +name = "opentelemetry-http" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7a6d09a73194e6b66df7c8f1b680f156d916a1a942abf2de06823dd02b7855d" +dependencies = [ + "async-trait", + "bytes", + "http", + "opentelemetry 0.31.0", +] + [[package]] name = "opentelemetry-http" version = "0.32.0" @@ -5430,10 +5475,25 @@ dependencies = [ "async-trait", "bytes", "http", - "opentelemetry", + "opentelemetry 0.32.0", "reqwest 0.13.4", ] +[[package]] +name = "opentelemetry-otlp" +version = "0.31.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f69cd6acbb9af919df949cd1ec9e5e7fdc2ef15d234b6b795aaa525cc02f71f" +dependencies = [ + "http", + "opentelemetry 0.31.0", + "opentelemetry-http 0.31.0", + "opentelemetry-proto 0.31.0", + "opentelemetry_sdk 0.31.0", + "prost", + "thiserror 2.0.18", +] + [[package]] name = "opentelemetry-otlp" version = "0.32.0" @@ -5441,10 +5501,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9966929966d17620d7c316c643ba62631826e10021409357772d5eea84f62c35" dependencies = [ "http", - "opentelemetry", - "opentelemetry-http", - "opentelemetry-proto", - "opentelemetry_sdk", + "opentelemetry 0.32.0", + "opentelemetry-http 0.32.0", + "opentelemetry-proto 0.32.0", + "opentelemetry_sdk 0.32.1", "prost", "reqwest 0.13.4", "serde_json", @@ -5454,6 +5514,23 @@ dependencies = [ "tonic-types", ] +[[package]] +name = "opentelemetry-proto" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7175df06de5eaee9909d4805a3d07e28bb752c34cab57fa9cff549da596b30f" +dependencies = [ + "base64", + "const-hex", + "opentelemetry 0.31.0", + "opentelemetry_sdk 0.31.0", + "prost", + "serde", + "serde_json", + "tonic", + "tonic-prost", +] + [[package]] name = "opentelemetry-proto" version = "0.32.0" @@ -5462,8 +5539,8 @@ checksum = "56d658ba1faf63f7b9c492cfbe6e0ec365440a16132d3270c1065f7b33f1b638" dependencies = [ "base64", "const-hex", - "opentelemetry", - "opentelemetry_sdk", + "opentelemetry 0.32.0", + "opentelemetry_sdk 0.32.1", "prost", "serde", "tonic", @@ -5477,8 +5554,23 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a1b1c6a247d79091f0062a5f4bd058589525cf987a8d4c169440d9c1be72f0ad" dependencies = [ "chrono", - "opentelemetry", - "opentelemetry_sdk", + "opentelemetry 0.32.0", + "opentelemetry_sdk 0.32.1", +] + +[[package]] +name = "opentelemetry_sdk" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e14ae4f5991976fd48df6d843de219ca6d31b01daaab2dad5af2badeded372bd" +dependencies = [ + "futures-channel", + "futures-executor", + "futures-util", + "opentelemetry 0.31.0", + "percent-encoding", + "rand 0.9.4", + "thiserror 2.0.18", ] [[package]] @@ -5490,7 +5582,7 @@ dependencies = [ "futures-channel", "futures-executor", "futures-util", - "opentelemetry", + "opentelemetry 0.32.0", "percent-encoding", "portable-atomic", "rand 0.9.4", @@ -6733,12 +6825,13 @@ dependencies = [ "itertools 0.14.0", "jsonwebtoken", "minijinja", + "mock-collector", "once_cell", - "opentelemetry", + "opentelemetry 0.32.0", "opentelemetry-appender-tracing", - "opentelemetry-otlp", + "opentelemetry-otlp 0.32.0", "opentelemetry-stdout", - "opentelemetry_sdk", + "opentelemetry_sdk 0.32.1", "parking_lot", "parquet", "poem", @@ -6748,6 +6841,7 @@ dependencies = [ "quick_cache", "raphtory", "raphtory-api", + "raphtory-graphql", "raphtory-storage", "rayon", "reqwest 0.13.4", @@ -8561,8 +8655,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef" dependencies = [ "async-trait", + "axum", "base64", "bytes", + "flate2", + "h2", "http", "http-body", "http-body-util", @@ -8571,6 +8668,7 @@ dependencies = [ "hyper-util", "percent-encoding", "pin-project", + "socket2", "sync_wrapper", "tokio", "tokio-rustls", @@ -8579,6 +8677,7 @@ dependencies = [ "tower-layer", "tower-service", "tracing", + "zstd", ] [[package]] @@ -8708,7 +8807,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "adbc64cba7137545b8044cb1fe9814f7aacf3c6b5f9b45be8bb5db538befdb26" dependencies = [ "js-sys", - "opentelemetry", + "opentelemetry 0.32.0", "smallvec", "tracing", "tracing-core", diff --git a/raphtory-graphql/Cargo.toml b/raphtory-graphql/Cargo.toml index 6764ec9c1c..6c4b9ab4e8 100644 --- a/raphtory-graphql/Cargo.toml +++ b/raphtory-graphql/Cargo.toml @@ -70,10 +70,15 @@ tempfile = { workspace = true } pretty_assertions = { workspace = true } arrow = { workspace = true } parquet = { workspace = true } +# Self-reference lets cargo test compile this crate with the `integration-test` feature enabled. +raphtory-graphql = { workspace = true, default-features = false, features = ["integration-test"] } +opentelemetry_sdk = { version = "0.32.1", features = ["rt-tokio", "testing"] } +mock-collector = "0.2" [features] default = [] vectors = ["raphtory/vectors"] +integration-test = [] python = ["dep:pyo3", "dep:pythonize", "raphtory/python"] search = ["raphtory/search"] panic-on-drop = ["raphtory/panic-on-drop"] diff --git a/raphtory-graphql/src/config/otlp_config.rs b/raphtory-graphql/src/config/otlp_config.rs index aff0d0d063..0b4b49e7c5 100644 --- a/raphtory-graphql/src/config/otlp_config.rs +++ b/raphtory-graphql/src/config/otlp_config.rs @@ -7,6 +7,8 @@ use opentelemetry_appender_tracing::layer::{ OpenTelemetryTracingBridge, OpenTelemetryTracingBridgeBuilder, }; use opentelemetry_otlp::{LogExporter, Protocol, SpanExporter, WithExportConfig, WithHttpConfig}; +#[cfg(feature = "integration-test")] +use opentelemetry_sdk::{logs::InMemoryLogExporter, trace::InMemorySpanExporter}; use opentelemetry_sdk::{ logs::SdkLoggerProvider, trace::{Sampler, SdkTracerProvider}, @@ -15,7 +17,9 @@ use opentelemetry_sdk::{ use raphtory_api::core::storage::arc_str::OptionAsStr; use reqwest::{blocking::ClientBuilder, Certificate}; use serde::Deserialize; -use std::{collections::HashMap, env, fs::File, io::Read, path::PathBuf, time::Duration}; +use std::{ + collections::HashMap, env, fs::File, io::Read, path::PathBuf, sync::LazyLock, time::Duration, +}; use strum::IntoEnumIterator; use strum_macros::{Display, EnumIter, EnumString, IntoStaticStr}; @@ -80,6 +84,8 @@ pub enum TracingProtocol { TONIC, HTTP, STDOUT, + #[cfg(feature = "integration-test")] + IN_MEMORY, } impl TryFrom for TracingProtocol { @@ -109,6 +115,28 @@ pub const DEFAULT_OTLP_TRANSPORT_PROTOCOL: TracingProtocol = TracingProtocol::TO pub const DEFAULT_OTLP_AGENT_PORT_TONIC: u16 = 4317; pub const DEFAULT_OTLP_TRACING_SERVICE_NAME: &'static str = "Raphtory"; +#[cfg(feature = "integration-test")] +// in-memory exporters to retrieve spans and logs in tests. +#[derive(Clone)] +pub struct GlobalExporters { + pub span: InMemorySpanExporter, + pub log: InMemoryLogExporter, +} + +#[cfg(feature = "integration-test")] +/* GraphServer registers span and log exporters + across the entire process, which can conflict + when starting up servers with their own exporters. + Making in-memory exporters global allows them to be + initialized once and reused across multiple tests + allowing the tests to retrieve spans and logs + without conflicts. +*/ +pub static GLOBAL_EXPORTERS: LazyLock = LazyLock::new(|| GlobalExporters { + span: InMemorySpanExporter::default(), + log: InMemoryLogExporter::default(), +}); + #[derive(Clone, Deserialize, Debug, PartialEq, serde::Serialize, FieldName)] pub struct TracingConfig { pub enabled: bool, @@ -163,6 +191,34 @@ impl TracingConfig { (tracer, logger) } + #[cfg(feature = "integration-test")] + fn with_simple_exporter< + E: opentelemetry_sdk::trace::SpanExporter + 'static, + L: opentelemetry_sdk::logs::LogExporter + 'static, + >( + &self, + span_exporter: E, + log_exporter: L, + ) -> (SdkTracerProvider, SdkLoggerProvider) { + let resource = Resource::builder() + .with_attributes(vec![KeyValue::new( + "service.name", + self.service_name.clone(), + )]) + .build(); + let tracer = SdkTracerProvider::builder() + .with_simple_exporter(span_exporter) + .with_sampler(Sampler::AlwaysOn) + .with_resource(resource.clone()) + .build(); + + let logger = SdkLoggerProvider::builder() + .with_simple_exporter(log_exporter) + .with_resource(resource) + .build(); + (tracer, logger) + } + pub async fn tracer_provider( &self, ) -> Result, ServerError> { @@ -265,6 +321,17 @@ impl TracingConfig { opentelemetry_stdout::LogExporter::default(), )) } + #[cfg(feature = "integration-test")] + TracingProtocol::IN_MEMORY => { + eprintln!( + "Sending traces to in-memory exporter with tracing level `{}`", + self.level + ); + Ok(self.with_simple_exporter( + GLOBAL_EXPORTERS.span.clone(), + GLOBAL_EXPORTERS.log.clone(), + )) + } }; match providers { diff --git a/raphtory-graphql/src/server.rs b/raphtory-graphql/src/server.rs index 2af27384eb..19e7893169 100644 --- a/raphtory-graphql/src/server.rs +++ b/raphtory-graphql/src/server.rs @@ -19,7 +19,7 @@ use opentelemetry::trace::TracerProvider; use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge; use opentelemetry_sdk::{ logs::SdkLoggerProvider, - trace::{SdkTracerProvider as TP, SdkTracerProvider, Tracer}, + trace::{SdkTracerProvider, Tracer}, }; use poem::{ get, @@ -56,7 +56,10 @@ use tokio::{ }; use tracing::{debug, error, info, warn}; use tracing_subscriber::{ - fmt, fmt::format::FmtSpan, layer::SubscriberExt, util::SubscriberInitExt, Registry, + fmt::{self, format::FmtSpan}, + layer::SubscriberExt, + util::SubscriberInitExt, + Registry, }; use url::ParseError; @@ -477,6 +480,10 @@ async fn server_termination( match tp { None => {} Some((tp, lp)) => { + /* Avoid shutting down global tracing exporters on server shutdown during integration tests + since they are reused across multiple tests. + */ + #[cfg(not(feature = "integration-test"))] task::spawn_blocking(move || { let res = tp.shutdown(); if let Err(e) = res { @@ -537,7 +544,6 @@ mod server_tests { running.stop().await } - #[tokio::test] async fn test_server_start_stop() { global_info_logger(); diff --git a/raphtory-graphql/tests/open_telemetry.rs b/raphtory-graphql/tests/open_telemetry.rs new file mode 100644 index 0000000000..8395a8d111 --- /dev/null +++ b/raphtory-graphql/tests/open_telemetry.rs @@ -0,0 +1,178 @@ +// OpenTelemetry Tests must be separated into their own binary to prevent polluting other tests since the span and log exporters are set globally. +use raphtory::db::api::storage::storage::Config; +use raphtory::prelude::{Graph, StableEncode}; +use raphtory_graphql::client::raphtory_client::RaphtoryGraphQLClient; +use raphtory_graphql::config::{ + app_config::AppConfigBuilder, + otlp_config::{TracingLevel, TracingProtocol, GLOBAL_EXPORTERS}, +}; +use raphtory_graphql::server::{GraphServer, RunningGraphServer}; +use std::collections::{HashMap, HashSet}; +use tempfile::{tempdir, TempDir}; +use url::Url; + +use opentelemetry_sdk::{logs::InMemoryLogExporter, trace::InMemorySpanExporter}; + +const OPEN_TELEMETRY_QUERY: &str = "query { + updateGraph(path: \"g\") { + addNode(time: 1, name: 1, properties: [{ key: \"seed\", value: { str: \"yes\" } }], nodeType: \"seed\", layer: \"main\") { + success + node { id } + } + addEdge(time: 5, src: 1, dst: 2, properties: [{ key: \"weight\", value: { f64: 1.5 } }], layer: \"main\") { + success + } + graph { + countNodes + hasNode(name: 1) + } + flush + } +}"; + +async fn setup_for_span_tests( + tracing_level: TracingLevel, +) -> ( + RaphtoryGraphQLClient, + RunningGraphServer, + InMemorySpanExporter, + InMemoryLogExporter, + TempDir, +) { + let span_exporter = GLOBAL_EXPORTERS.span.clone(); + let log_exporter = GLOBAL_EXPORTERS.log.clone(); + // reset logs and spans for next test + span_exporter.reset(); + log_exporter.reset(); + let tmp_dir = tempdir().unwrap(); + let graph = Graph::new(); + graph.encode(tmp_dir.path().join("g")).unwrap(); + + let app_config = AppConfigBuilder::new() + .with_tracing(true) + .with_tracing_level(tracing_level) + .with_otlp_transport_protocol(TracingProtocol::IN_MEMORY) + .build(); + + let server = GraphServer::new( + tmp_dir.path().to_path_buf(), + Some(app_config.clone()), + Config::default(), + ) + .await + .unwrap(); + let handler = server.start_with_port(0).await.unwrap(); + + let endpoint = Url::parse(&format!("http://localhost:{}/", handler.port())).unwrap(); + let client = RaphtoryGraphQLClient::new(endpoint, None); + (client, handler, span_exporter, log_exporter, tmp_dir) +} + +async fn test_open_telemetry_spans_complete() { + let (client, handler, span_exporter, log_exporter, _tmp_dir) = + setup_for_span_tests(TracingLevel::COMPLETE).await; + let _ = client + .query(OPEN_TELEMETRY_QUERY, HashMap::new()) + .await + .unwrap(); + handler.stop().await; + + let finished_spans = span_exporter.get_finished_spans().unwrap(); + let all_spans: HashSet = finished_spans + .iter() + .map(|span| span.name.to_string()) + .collect(); + assert_eq!( + all_spans, + HashSet::from([ + "addEdge".to_string(), + "addNode".to_string(), + "countNodes".to_string(), + "execute".to_string(), + "flush".to_string(), + "graph".to_string(), + "hasNode".to_string(), + "id".to_string(), + "node".to_string(), + "parse".to_string(), + "request".to_string(), + "success".to_string(), + "updateGraph".to_string(), + "validation".to_string(), + ]) + ); + + let emitted_logs = log_exporter.get_emitted_logs().unwrap(); + assert!(!emitted_logs.is_empty()); + handler.wait().await.unwrap(); +} + +async fn test_open_telemetry_spans_essential() { + let (client, handler, span_exporter, log_exporter, _tmp_dir) = + setup_for_span_tests(TracingLevel::ESSENTIAL).await; + let _ = client + .query(OPEN_TELEMETRY_QUERY, HashMap::new()) + .await + .unwrap(); + handler.stop().await; + + let finished_spans = span_exporter.get_finished_spans().unwrap(); + let all_spans: HashSet = finished_spans + .iter() + .map(|span| span.name.to_string()) + .collect(); + assert_eq!( + all_spans, + HashSet::from([ + "addEdge".to_string(), + "addNode".to_string(), + "execute".to_string(), + "graph".to_string(), + "node".to_string(), + "parse".to_string(), + "request".to_string(), + "updateGraph".to_string(), + "validation".to_string(), + ]) + ); + + let emitted_logs = log_exporter.get_emitted_logs().unwrap(); + assert!(!emitted_logs.is_empty()); + handler.wait().await.unwrap(); +} + +async fn test_open_telemetry_spans_minimal() { + let (client, handler, span_exporter, log_exporter, _tmp_dir) = + setup_for_span_tests(TracingLevel::MINIMAL).await; + let _ = client + .query(OPEN_TELEMETRY_QUERY, HashMap::new()) + .await + .unwrap(); + handler.stop().await; + + let finished_spans = span_exporter.get_finished_spans().unwrap(); + let all_spans: HashSet = finished_spans + .iter() + .map(|span| span.name.to_string()) + .collect(); + assert_eq!( + all_spans, + HashSet::from([ + "execute".to_string(), + "parse".to_string(), + "request".to_string(), + "validation".to_string(), + ]) + ); + + let emitted_logs = log_exporter.get_emitted_logs().unwrap(); + assert!(!emitted_logs.is_empty()); +} + +#[tokio::test] +async fn test_open_telemetry_spans() { + // The following tests share the same global in-memory exporters and hence need to be run sequentially to prevent spans and logs from getting mangled. + test_open_telemetry_spans_complete().await; + test_open_telemetry_spans_essential().await; + test_open_telemetry_spans_minimal().await; +} diff --git a/raphtory-graphql/tests/open_telemetry_e2e.rs b/raphtory-graphql/tests/open_telemetry_e2e.rs new file mode 100644 index 0000000000..d290f1bc11 --- /dev/null +++ b/raphtory-graphql/tests/open_telemetry_e2e.rs @@ -0,0 +1,103 @@ +use mock_collector::{MockServer, Protocol}; +use raphtory::db::api::storage::storage::Config; +use raphtory::prelude::{Graph, StableEncode}; +use raphtory_graphql::client::raphtory_client::RaphtoryGraphQLClient; +use raphtory_graphql::config::{ + app_config::AppConfigBuilder, + otlp_config::{TracingLevel, TracingProtocol}, +}; +use raphtory_graphql::server::GraphServer; +use std::collections::{HashMap, HashSet}; +use std::time::Duration; +use tempfile::tempdir; +use url::Url; + +#[tokio::test] +async fn test_open_telemetry_http_tracing_server() { + let tracing_server = MockServer::builder() + .protocol(Protocol::HttpBinary) + .start() + .await + .unwrap(); + + let work_dir = tempdir().unwrap(); + let graph = Graph::new(); + graph.encode(work_dir.path().join("g")).unwrap(); + + let app_config = AppConfigBuilder::new() + .with_tracing(true) + .with_tracing_level(TracingLevel::ESSENTIAL) + .with_otlp_agent_host(Some(format!("http://{}", tracing_server.addr()))) + .with_otlp_transport_protocol(TracingProtocol::HTTP) + .build(); + + let server = GraphServer::new( + work_dir.path().to_path_buf(), + Some(app_config), + Config::default(), + ) + .await + .unwrap(); + let handler = server.start_with_port(0).await.unwrap(); + + let endpoint = Url::parse(&format!("http://localhost:{}/", handler.port())).unwrap(); + let client = RaphtoryGraphQLClient::new(endpoint, None); + let open_telemetry_query = "query { + updateGraph(path: \"g\") { + addNode(time: 1, name: 1, properties: [{ key: \"seed\", value: { str: \"yes\" } }], nodeType: \"seed\", layer: \"main\") { + success + node { id } + } + addEdge(time: 5, src: 1, dst: 2, properties: [{ key: \"weight\", value: { f64: 1.5 } }], layer: \"main\") { + success + } + graph { + countNodes + hasNode(name: 1) + } + flush + } +}".to_string(); + let _ = client + .query(&open_telemetry_query, HashMap::new()) + .await + .unwrap(); + + handler.stop().await; + + tracing_server + .wait_for_spans(1, Duration::from_secs(5)) + .await + .unwrap(); + tracing_server + .wait_for_logs(1, Duration::from_secs(5)) + .await + .unwrap(); + + tracing_server + .with_collector(|collector| { + let all_spans: HashSet = collector + .spans() + .iter() + .map(|span| span.span().name.clone()) + .collect(); + assert_eq!( + all_spans, + HashSet::from([ + "addEdge".to_string(), + "addNode".to_string(), + "execute".to_string(), + "graph".to_string(), + "node".to_string(), + "parse".to_string(), + "request".to_string(), + "updateGraph".to_string(), + "validation".to_string(), + ]) + ); + assert!(collector.log_count() > 0); + }) + .await; + + tracing_server.shutdown().await.unwrap(); +}