diff --git a/Cargo.lock b/Cargo.lock index e66775b..6336d25 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -84,29 +84,6 @@ version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" -[[package]] -name = "aws-lc-rs" -version = "1.18.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b281d307588d634de920874890732659e2e7672f72b5e10e81badc1a8a83621e" -dependencies = [ - "aws-lc-sys", - "zeroize", -] - -[[package]] -name = "aws-lc-sys" -version = "0.45.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9bff6c3b54fad79a2e60b8102caf565819711497c1f5f092f49508e2f5c31b27" -dependencies = [ - "cc", - "cmake", - "dunce", - "fs_extra", - "pkg-config", -] - [[package]] name = "axum" version = "0.8.9" @@ -250,8 +227,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f74872d07caf508b30a21f6836e7d7016a2eaf7d9ff4f48deaa58cd8a0407630" dependencies = [ "find-msvc-tools", - "jobserver", - "libc", "shlex", ] @@ -261,12 +236,6 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600" -[[package]] -name = "cfg_aliases" -version = "0.2.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" - [[package]] name = "chacha20" version = "0.10.2" @@ -290,15 +259,6 @@ dependencies = [ "windows-link", ] -[[package]] -name = "cmake" -version = "0.1.58" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c0f78a02292a74a88ac736019ab962ece0bc380e3f977bf72e376c5d78ff0678" -dependencies = [ - "cc", -] - [[package]] name = "cmov" version = "0.5.4" @@ -451,12 +411,6 @@ version = "0.15.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" -[[package]] -name = "dunce" -version = "1.0.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" - [[package]] name = "either" version = "1.19.0" @@ -575,12 +529,6 @@ dependencies = [ "num-bigint", ] -[[package]] -name = "fs_extra" -version = "1.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" - [[package]] name = "futures-channel" version = "0.3.34" @@ -681,10 +629,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" dependencies = [ "cfg-if", - "js-sys", "libc", "wasi", - "wasm-bindgen", ] [[package]] @@ -708,11 +654,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" dependencies = [ "cfg-if", - "js-sys", "libc", "r-efi 6.0.0", "rand_core 0.10.1", - "wasm-bindgen", ] [[package]] @@ -1097,16 +1041,6 @@ dependencies = [ "syn 2.0.119", ] -[[package]] -name = "jobserver" -version = "0.1.35" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1c00acbd29eabad4a2392fa0e921c874934dbbf4194312ad20f04a0ed67a3cb3" -dependencies = [ - "getrandom 0.4.3", - "libc", -] - [[package]] name = "js-sys" version = "0.3.106" @@ -1214,12 +1148,6 @@ version = "0.4.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" -[[package]] -name = "lru-slab" -version = "0.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4050469837a6ff301cd14c1f8f24f88549e6d548f24f64e2148eb0f72cebc51f" - [[package]] name = "matchit" version = "0.8.4" @@ -1279,6 +1207,7 @@ dependencies = [ "opentelemetry-otlp", "opentelemetry_sdk", "reqwest", + "rustls", "serde", "serde_json", "sha2 0.10.9", @@ -1294,19 +1223,25 @@ dependencies = [ [[package]] name = "nddev-device-sync-telemetry" -version = "0.0.1-alpha.2" +version = "0.0.1-alpha.3" dependencies = [ "android_log-sys", "chrono", "jsonschema", "opentelemetry", + "opentelemetry-otlp", + "opentelemetry-proto", "opentelemetry_sdk", + "prost", + "reqwest", + "rustls", "serde", "serde_json", "thiserror", "tracing", "tracing-opentelemetry", "tracing-subscriber", + "url", "uuid", ] @@ -1579,63 +1514,6 @@ dependencies = [ "syn 2.0.119", ] -[[package]] -name = "quinn" -version = "0.11.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4051e23e9185c255a7e33ef59cdbca87a22d359052eecd22fc6b901fb37d9d11" -dependencies = [ - "bytes", - "cfg_aliases", - "pin-project-lite", - "quinn-proto", - "quinn-udp", - "rustc-hash", - "rustls", - "socket2", - "thiserror", - "tokio", - "tracing", - "web-time", -] - -[[package]] -name = "quinn-proto" -version = "0.11.19" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0e750cca55fe4f0439a15d0bb529da9651e79993e8e72c61a899a36d462befbe" -dependencies = [ - "aws-lc-rs", - "bytes", - "getrandom 0.4.3", - "lru-slab", - "rand 0.10.3", - "rand_pcg", - "ring", - "rustc-hash", - "rustls", - "rustls-pki-types", - "slab", - "thiserror", - "tinyvec", - "tracing", - "web-time", -] - -[[package]] -name = "quinn-udp" -version = "0.5.16" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "af66907df18639dcf4db56ca65490cabc4b27a97dbadd96f2926cca73298f016" -dependencies = [ - "cfg_aliases", - "libc", - "once_cell", - "socket2", - "tracing", - "windows-sys 0.61.2", -] - [[package]] name = "quote" version = "1.0.47" @@ -1703,15 +1581,6 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" -[[package]] -name = "rand_pcg" -version = "0.10.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "caa0f4137e1c0a72f4c651489402276c8e8e1cf081f3b0ba156d2cbeef09e86a" -dependencies = [ - "rand_core 0.10.1", -] - [[package]] name = "redox_syscall" version = "0.5.18" @@ -1808,7 +1677,6 @@ dependencies = [ "log", "percent-encoding", "pin-project-lite", - "quinn", "rustls", "rustls-pki-types", "rustls-platform-verifier", @@ -1840,12 +1708,6 @@ dependencies = [ "windows-sys 0.52.0", ] -[[package]] -name = "rustc-hash" -version = "2.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" - [[package]] name = "rustc_version" version = "0.4.1" @@ -1861,7 +1723,6 @@ version = "0.23.45" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d41d731c7d2f962d1ccc364cec258de3c0e93b38c2fb3ba97ac74513048d634" dependencies = [ - "aws-lc-rs", "once_cell", "ring", "rustls-pki-types", @@ -1888,7 +1749,6 @@ version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" dependencies = [ - "web-time", "zeroize", ] @@ -1925,7 +1785,6 @@ version = "0.103.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" dependencies = [ - "aws-lc-rs", "ring", "rustls-pki-types", "untrusted", diff --git a/Cargo.toml b/Cargo.toml index 5604333..2b13a55 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -20,8 +20,9 @@ hyper-util = { version = "0.1", features = ["tokio", "service"] } chrono = { version = "0.4", default-features = false, features = ["clock", "serde"] } opentelemetry = { version = "0.33.1", default-features = false, features = ["trace", "metrics"] } opentelemetry_sdk = { version = "0.33.1", default-features = false, features = ["trace", "metrics"] } -opentelemetry-otlp = { version = "0.33.1", default-features = false, features = ["trace", "metrics", "http-proto", "reqwest-blocking-client", "reqwest-rustls"] } -reqwest = { version = "0.13", default-features = false, features = ["json", "rustls"] } +opentelemetry-otlp = { version = "0.33.1", default-features = false, features = ["trace", "metrics", "http-proto", "reqwest-blocking-client"] } +reqwest = { version = "0.13", default-features = false, features = ["json", "rustls-no-provider"] } +rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12"] } serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" sha2 = "0.10" diff --git a/crates/sdk/Cargo.toml b/crates/sdk/Cargo.toml index 2077cb3..fd2735c 100644 --- a/crates/sdk/Cargo.toml +++ b/crates/sdk/Cargo.toml @@ -1,11 +1,19 @@ [package] name = "nddev-device-sync-telemetry" -version = "0.0.1-alpha.2" +version = "0.0.1-alpha.3" edition = "2024" rust-version = "1.99" license = "AGPL-3.0-only" +[features] +default = [] +otlp = ["dep:opentelemetry-otlp", "dep:url", "dep:reqwest", "dep:rustls"] + [dependencies] +opentelemetry-otlp = { version = "0.33.1", optional = true, default-features = false, features = ["trace", "http-proto", "reqwest-blocking-client"] } +reqwest = { version = "0.13", optional = true, default-features = false, features = ["blocking", "rustls-no-provider"] } +rustls = { version = "0.23", optional = true, default-features = false, features = ["ring", "std", "tls12"] } +url = { version = "2.5", optional = true } chrono = { version = "0.4", default-features = false, features = ["clock"] } jsonschema = { version = "0.58.6", default-features = false } opentelemetry = { version = "0.33.1", default-features = false, features = ["trace"] } @@ -20,3 +28,7 @@ uuid = { version = "1.27", features = ["v4"] } [target.'cfg(target_os = "android")'.dependencies] android_log-sys = "=0.3.2" + +[dev-dependencies] +prost = "0.14" +opentelemetry-proto = { version = "0.33.1", default-features = false, features = ["gen-tonic-messages", "trace"] } diff --git a/crates/sdk/src/lib.rs b/crates/sdk/src/lib.rs index 3c6d78c..42a07e3 100644 --- a/crates/sdk/src/lib.rs +++ b/crates/sdk/src/lib.rs @@ -20,6 +20,9 @@ use tracing::{ span::{Attributes, Id, Record}, }; use tracing_opentelemetry::OpenTelemetrySpanExt; + +#[cfg(feature = "otlp")] +pub mod otlp; use tracing_subscriber::{ Layer, layer::{Context as LayerContext, SubscriberExt}, diff --git a/crates/sdk/src/otlp.rs b/crates/sdk/src/otlp.rs new file mode 100644 index 0000000..46f21ae --- /dev/null +++ b/crates/sdk/src/otlp.rs @@ -0,0 +1,231 @@ +//! Optional native OTLP/HTTP egress for trusted process adapters. No collector +//! account is embedded. Mobile ingress authentication belongs to its transport. +use crate::{Config, Error}; +use opentelemetry::{Context, KeyValue, trace::TracerProvider as _}; +use opentelemetry_otlp::{Protocol, WithExportConfig, WithHttpConfig}; +use opentelemetry_sdk::{ + Resource, + error::OTelSdkResult, + trace::{ + BatchConfigBuilder, BatchSpanProcessor, Sampler, SdkTracerProvider, Span, SpanData, + SpanExporter, SpanProcessor, Tracer, + }, +}; +use std::{ + fmt, + sync::{ + Arc, Mutex, + atomic::{AtomicBool, AtomicU64, Ordering}, + }, + time::Duration, +}; + +const CAPACITY: u64 = 128; + +#[derive(Default, Debug)] +pub struct Delivery { + pub successes: AtomicU64, + pub failures: AtomicU64, + pub pending: AtomicU64, + pub dropped: AtomicU64, + pub last_success_ms: AtomicU64, + closing: AtomicBool, +} + +struct ObservedExporter { + inner: opentelemetry_otlp::SpanExporter, + delivery: Arc, +} +impl fmt::Debug for ObservedExporter { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("ObservedExporter") + } +} +impl SpanExporter for ObservedExporter { + async fn export(&self, batch: Vec) -> OTelSdkResult { + let count = batch.len() as u64; + // The maintained exporter owns bounded retry/backoff inside its 2s + // budget. Never layer another independent retry loop on top of it. + let outcome = self.inner.export(batch).await; + if outcome.is_ok() { + self.delivery.successes.fetch_add(1, Ordering::Relaxed); + self.delivery.last_success_ms.store( + chrono::Utc::now().timestamp_millis().max(0) as u64, + Ordering::Relaxed, + ); + } else { + self.delivery.failures.fetch_add(1, Ordering::Relaxed); + self.delivery.dropped.fetch_add(count, Ordering::Relaxed); + } + self.delivery.pending.fetch_sub(count, Ordering::AcqRel); + outcome + } + fn set_resource(&mut self, resource: &Resource) { + self.inner.set_resource(resource); + } + fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult { + self.inner.shutdown_with_timeout(timeout) + } +} + +#[derive(Debug)] +struct BoundedProcessor { + inner: BatchSpanProcessor, + delivery: Arc, + admission: Mutex<()>, +} +impl SpanProcessor for BoundedProcessor { + fn on_start(&self, span: &mut Span, context: &Context) { + self.inner.on_start(span, context); + } + fn on_end(&self, mut span: SpanData) { + if !span.span_context.is_sampled() { + return; + } + let _admission = self.admission.lock().unwrap_or_else(|e| e.into_inner()); + if self.delivery.closing.load(Ordering::Acquire) + || self + .delivery + .pending + .try_update(Ordering::AcqRel, Ordering::Acquire, |pending| { + (pending < CAPACITY).then_some(pending + 1) + }) + .is_err() + { + self.delivery.dropped.fetch_add(1, Ordering::Relaxed); + return; + } + // An explicit projection applies to native spans too. Raw events, + // exception prose, baggage, links and identity labels are never exported. + span.events.events.clear(); + span.links.links.clear(); + if !matches!(span.name.as_ref(), "http.request" | "module.operation") { + span.name = "module.operation".into(); + } + span.attributes.retain(|attribute| { + matches!( + attribute.key.as_str(), + "module" + | "scope" + | "http.request.method" + | "http.route" + | "http.response.status_code" + ) + }); + if matches!(span.status, opentelemetry::trace::Status::Error { .. }) { + span.status = opentelemetry::trace::Status::error(""); + } + self.inner.on_end(span); + } + fn force_flush(&self) -> OTelSdkResult { + self.inner.force_flush() + } + fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult { + let admission = self.admission.lock().unwrap_or_else(|e| e.into_inner()); + self.delivery.closing.store(true, Ordering::Release); + drop(admission); + self.inner.shutdown_with_timeout(timeout) + } + fn set_resource(&mut self, resource: &Resource) { + self.inner.set_resource(resource); + } +} + +pub struct NativeTraces { + provider: SdkTracerProvider, + pub delivery: Arc, +} +impl NativeTraces { + /// Call outside an async runtime. HTTP is for an operator-owned private + /// network/loopback collector; remote clients must use authenticated HTTPS. + /// No environment-provided endpoint can override this explicit origin. + pub fn new(config: &Config, origin: &str) -> Result { + crate::Policy::new(config)?; + let endpoint = url::Url::parse(origin).map_err(|_| Error)?; + if origin.len() > 2048 + || !matches!(endpoint.scheme(), "http" | "https") + || endpoint.host().is_none() + || !endpoint.username().is_empty() + || endpoint.password().is_some() + || endpoint.query().is_some() + || endpoint.fragment().is_some() + || endpoint.path() != "/" + { + return Err(Error); + } + let _ = rustls::crypto::ring::default_provider().install_default(); + let client = reqwest::blocking::Client::builder() + .no_proxy() + .redirect(reqwest::redirect::Policy::none()) + .connect_timeout(Duration::from_secs(1)) + .timeout(Duration::from_secs(2)) + .build() + .map_err(|_| Error)?; + let exporter = opentelemetry_otlp::SpanExporter::builder() + .with_http() + .with_http_client(client) + .with_protocol(Protocol::HttpBinary) + .with_endpoint(format!( + "{}/v1/traces", + endpoint.as_str().trim_end_matches('/') + )) + .with_timeout(Duration::from_secs(2)) + .with_retry_policy( + opentelemetry_otlp::RetryPolicy::recommended() + .with_max_retries(2) + .with_initial_delay(Duration::from_millis(50)) + .with_max_delay(Duration::from_millis(200)), + ) + .build() + .map_err(|_| Error)?; + let delivery = Arc::new(Delivery::default()); + let processor = BoundedProcessor { + admission: Mutex::new(()), + inner: BatchSpanProcessor::builder(ObservedExporter { + inner: exporter, + delivery: delivery.clone(), + }) + .with_batch_config( + BatchConfigBuilder::default() + .with_max_queue_size(256) + .with_max_export_batch_size(64) + .with_scheduled_delay(Duration::from_millis(200)) + .build(), + ) + .build(), + delivery: delivery.clone(), + }; + let resource = Resource::builder_empty() + .with_attributes([ + KeyValue::new("service.name", config.service.clone()), + KeyValue::new("service.version", config.version.clone()), + KeyValue::new("source.repository", config.repository.clone()), + KeyValue::new("source.commit", config.commit.clone()), + KeyValue::new("deployment.environment.name", config.environment.clone()), + ]) + .build(); + let provider = SdkTracerProvider::builder() + .with_resource(resource) + .with_sampler(Sampler::ParentBased(Box::new(Sampler::AlwaysOn))) + .with_span_processor(processor) + .with_max_events_per_span(0) + .with_max_links_per_span(0) + .with_max_attributes_per_span(16) + .build(); + Ok(Self { provider, delivery }) + } + pub fn tracer(&self) -> Tracer { + self.provider.tracer("nds-telemetry") + } + pub fn flush(&self) -> OTelSdkResult { + self.provider.force_flush() + } + pub fn shutdown(&self) { + let _ = self.provider.shutdown_with_timeout(Duration::from_secs(3)); + } +} +impl Drop for NativeTraces { + fn drop(&mut self) { + self.shutdown(); + } +} diff --git a/crates/sdk/tests/native_otlp.rs b/crates/sdk/tests/native_otlp.rs new file mode 100644 index 0000000..32bbdeb --- /dev/null +++ b/crates/sdk/tests/native_otlp.rs @@ -0,0 +1,127 @@ +#![cfg(feature = "otlp")] +use nddev_device_sync_telemetry::{ + Config, finish_http_server_span, http_server_span, otlp::NativeTraces, +}; +use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest; +use prost::Message; +use std::{ + io::{Read, Write}, + net::TcpListener, + sync::atomic::Ordering, + time::Duration, +}; + +fn config() -> Config { + Config { + service: "nds-otlp-acceptance".into(), + version: env!("CARGO_PKG_VERSION").into(), + repository: "NDDev-OpenNetwork/nddev-device-sync-observability".into(), + commit: "unknown".into(), + channel: "alpha".into(), + standards: "v0.0.1-alpha.9".into(), + environment: "test".into(), + target_prefixes: vec!["nds_acceptance".into()], + default_module: "process".into(), + debug: None, + } +} + +#[test] +fn actual_http_otlp_retries_and_exports_only_projected_native_spans() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let endpoint = format!("http://{}/", listener.local_addr().unwrap()); + let receiver = std::thread::spawn(move || { + let mut bodies = Vec::new(); + for response in ["503 Service Unavailable", "200 OK"] { + let (mut socket, _) = listener.accept().unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut bytes = Vec::new(); + let (header_size, length) = loop { + let mut chunk = [0; 4096]; + let count = socket.read(&mut chunk).unwrap(); + assert!(count > 0 && bytes.len() + count < 128 * 1024); + bytes.extend_from_slice(&chunk[..count]); + if let Some(end) = bytes.windows(4).position(|value| value == b"\r\n\r\n") { + let header = std::str::from_utf8(&bytes[..end]).unwrap(); + assert!(header.starts_with("POST /v1/traces HTTP/1.1")); + let length = header + .lines() + .find_map(|line| { + line.to_ascii_lowercase() + .strip_prefix("content-length: ") + .map(str::to_owned) + }) + .unwrap() + .parse::() + .unwrap(); + break (end + 4, length); + } + }; + while bytes.len() < header_size + length { + let mut chunk = [0; 4096]; + let count = socket.read(&mut chunk).unwrap(); + assert!(count > 0 && bytes.len() + count < 128 * 1024); + bytes.extend_from_slice(&chunk[..count]); + } + bodies.push(bytes[header_size..header_size + length].to_vec()); + write!( + socket, + "HTTP/1.1 {response}\r\ncontent-length: 0\r\nconnection: close\r\n\r\n" + ) + .unwrap(); + } + assert_eq!(bodies[0], bodies[1]); + ExportTraceServiceRequest::decode(bodies[1].as_slice()).unwrap() + }); + let native = NativeTraces::new(&config(), &endpoint).unwrap(); + let (subscriber, _) = + nddev_device_sync_telemetry::subscriber(config(), std::io::sink(), native.tracer()) + .unwrap(); + tracing::subscriber::with_default(subscriber, || { + let span = http_server_span("GET", "/v1/health").unwrap(); + span.in_scope(|| { + tracing::event!(target:"nds_acceptance",tracing::Level::INFO,{event.name="acceptance.completed",outcome="ok"}); + finish_http_server_span(&span, 200).unwrap(); + }); + }); + native.flush().unwrap(); + let received = receiver.join().unwrap(); + let spans = &received.resource_spans[0].scope_spans[0].spans; + assert_eq!(spans.len(), 1); + assert_eq!(spans[0].name, "http.request"); + assert_eq!(spans[0].kind, 2); // Native SERVER span, not a reconstructed log. + assert!(spans[0].end_time_unix_nano > spans[0].start_time_unix_nano); + assert!(spans[0].events.is_empty() && spans[0].links.is_empty()); + assert!(spans[0].attributes.iter().all(|attribute| matches!( + attribute.key.as_str(), + "module" | "scope" | "http.request.method" | "http.route" | "http.response.status_code" + ))); + assert_eq!(native.delivery.successes.load(Ordering::Relaxed), 1); + assert_eq!(native.delivery.failures.load(Ordering::Relaxed), 0); + assert_eq!(native.delivery.pending.load(Ordering::Relaxed), 0); + assert!(native.delivery.last_success_ms.load(Ordering::Relaxed) > 0); +} + +#[test] +fn actual_unresponsive_collector_cannot_grow_the_queue_or_hide_loss() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let endpoint = format!("http://{}/", listener.local_addr().unwrap()); + let native = NativeTraces::new(&config(), &endpoint).unwrap(); + let (subscriber, _) = + nddev_device_sync_telemetry::subscriber(config(), std::io::sink(), native.tracer()) + .unwrap(); + tracing::subscriber::with_default(subscriber, || { + for _ in 0..1000 { + let _span = http_server_span("GET", "/v1/health").unwrap(); + } + }); + assert!(native.delivery.pending.load(Ordering::Relaxed) <= 128); + assert!(native.delivery.dropped.load(Ordering::Relaxed) >= 872); + drop(listener); + native.flush().unwrap(); + assert_eq!(native.delivery.pending.load(Ordering::Relaxed), 0); + assert!(native.delivery.failures.load(Ordering::Relaxed) > 0); + assert_eq!(native.delivery.dropped.load(Ordering::Relaxed), 1000); +} diff --git a/justfile b/justfile index c605045..9d12b38 100644 --- a/justfile +++ b/justfile @@ -5,8 +5,8 @@ default: check: cargo fmt --all -- --check - cargo test --workspace --locked - cargo clippy --workspace --all-targets --locked -- -D warnings + cargo test --workspace --all-features --locked --jobs 4 + cargo clippy --workspace --all-targets --all-features --locked --jobs 4 -- -D warnings integration: python3 scripts/check-compose.py diff --git a/src/signals.rs b/src/signals.rs index d793268..a1545a4 100644 --- a/src/signals.rs +++ b/src/signals.rs @@ -177,6 +177,7 @@ pub struct Signals { impl Signals { /// Construct outside Tokio: the OTLP SDK owns bounded blocking-export threads. pub fn new(config: &Config) -> Result { + let _ = rustls::crypto::ring::default_provider().install_default(); let metrics_delivery = Arc::new(Delivery::default()); let traces_delivery = Arc::new(Delivery::default()); let pending = Arc::new(AtomicU64::new(0));