Open telemetry tests

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"]

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<String> 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<GlobalExporters> = 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<Option<(SdkTracerProvider, SdkLoggerProvider)>, 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 {