diff --git a/Cargo.lock b/Cargo.lock index 26f981c..edb2b13 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1743,9 +1743,9 @@ checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" [[package]] name = "opentelemetry" -version = "0.31.0" +version = "0.33.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b84bcd6ae87133e903af7ef497404dda70c60d0ea14895fc8a5e6722754fc2a0" +checksum = "8c1f1a6ac9c875a4941e4c7c085b61e8f4daa326de03b4701768c3491a6ad9ee" dependencies = [ "futures-core", "futures-sink", @@ -1757,22 +1757,21 @@ dependencies = [ [[package]] name = "opentelemetry-appender-tracing" -version = "0.31.1" +version = "0.33.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ef6a1ac5ca3accf562b8c306fa8483c85f4390f768185ab775f242f7fe8fdcc2" +checksum = "1f01f1c53c46ce80e5fd8cb7a646f13766fd8f315a8a5b75b9bad59a08d2c5c7" dependencies = [ "opentelemetry", "tracing", "tracing-core", - "tracing-opentelemetry", "tracing-subscriber", ] [[package]] name = "opentelemetry-otlp" -version = "0.31.1" +version = "0.33.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1f69cd6acbb9af919df949cd1ec9e5e7fdc2ef15d234b6b795aaa525cc02f71f" +checksum = "0756af24841e367128ef31c831999937b8b5c5197d47914590d8f4e2e88c2e1a" dependencies = [ "http", "opentelemetry", @@ -1782,13 +1781,14 @@ dependencies = [ "thiserror 2.0.18", "tokio", "tonic", + "tonic-types", ] [[package]] name = "opentelemetry-proto" -version = "0.31.0" +version = "0.33.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a7175df06de5eaee9909d4805a3d07e28bb752c34cab57fa9cff549da596b30f" +checksum = "d576f5dddc3299341bcd85a04be17b2564844c0c9563214f02d85f2524d1afae" dependencies = [ "opentelemetry", "opentelemetry_sdk", @@ -1799,15 +1799,16 @@ dependencies = [ [[package]] name = "opentelemetry_sdk" -version = "0.31.0" +version = "0.33.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e14ae4f5991976fd48df6d843de219ca6d31b01daaab2dad5af2badeded372bd" +checksum = "372a912db2974d1990207006fd0e3aa0ef3fd4ad3be43c65cfaadf2206866061" dependencies = [ "futures-channel", "futures-executor", "futures-util", "opentelemetry", "percent-encoding", + "portable-atomic", "rand 0.9.4", "thiserror 2.0.18", ] @@ -1927,6 +1928,12 @@ version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6" +[[package]] +name = "portable-atomic" +version = "1.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05c8b63e8d9609db387f0324918f81d68fe27748f084ef092fb35954d0539a85" + [[package]] name = "potential_utf" version = "0.1.5" @@ -1993,6 +2000,15 @@ dependencies = [ "syn", ] +[[package]] +name = "prost-types" +version = "0.14.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8991c4cbdb8bc5b11f0b074ffe286c30e523de90fee5ba8132f1399f23cb3dd7" +dependencies = [ + "prost", +] + [[package]] name = "pty-process" version = "0.5.3" @@ -3268,6 +3284,17 @@ dependencies = [ "tonic", ] +[[package]] +name = "tonic-types" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a875a902255423d34c1f20838ab374126db8eb41625b7947a1d54113b0b7399" +dependencies = [ + "prost", + "prost-types", + "tonic", +] + [[package]] name = "tower" version = "0.5.3" @@ -3383,9 +3410,9 @@ dependencies = [ [[package]] name = "tracing-opentelemetry" -version = "0.32.1" +version = "0.34.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1ac28f2d093c6c477eaa76b23525478f38de514fa9aeb1285738d4b97a9552fc" +checksum = "0a904802a1b902f43638b677ff2a650847e3b4404101b6c586d648e8c1e3e8fe" dependencies = [ "js-sys", "opentelemetry", diff --git a/docs/fleet-cache.md b/docs/fleet-cache.md index 22604ff..95ac26d 100644 --- a/docs/fleet-cache.md +++ b/docs/fleet-cache.md @@ -17,9 +17,10 @@ and reuse it. The helper checks the key again after building: if inputs changed while database reads/grouping were in flight, it discards the result and retries. Build errors are returned, not cached; cancelling a reader releases the lock. The previous view is not served as a fallback after a rebuild error. -Retries currently have no fixed limit. Sustained mutations faster than rebuilding -can delay the rebuilding reader and readers waiting on its lock; bounding this -behavior is tracked as a reliability follow-up. +After three invalidated builds, the reader stops and releases the lock. Fleet +list/wake requests return HTTP 503 with code `fleet_busy`; clients can retry +later. No stale view or command is used as a fallback. This bounds the number +of rebuilds per reader, not database execution time or total concurrent requests. Wake uses the same view, but command dispatch still checks the current session; a cached route cannot grant access to a disconnected agent. diff --git a/docs/roadmap.md b/docs/roadmap.md index 7259154..5a68529 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -41,6 +41,7 @@ of remembered identifiers based on observation expiry. - [x] Lazy grouped fleet-view cache with store/connection invalidation and stale-build rejection. See [fleet cache](fleet-cache.md). - [x] Separate console, OTLP log, and trace filters using per-layer filtering. Explicit signal config overrides `RUST_LOG`, then CLI verbosity supplies defaults. Unconfigured defaults remain info-level; exporter-failure logs stay local. Live validation pending. +- [x] Upgrade the matched OpenTelemetry stack to 0.33 with exporter-managed backoff/jitter retries. Controlled collector tests cover all-signal recovery, bounded retries, and no retry on authentication rejection; live outage validation remains pending. - [ ] Record hostname-resolution success/failure without exporting hostnames or resolved addresses. Current spans show timing but do not explicitly classify the result. - [ ] Investigate the slow resolver with a reproducible device-side check. Timing isolated the phase, not the cause; do not treat instrumentation as a fix. - [x] Propagate optional W3C trace context for periodic/requested WS snapshots so agent collection and CC persistence can be inspected together. Wire compatibility and context linkage tested; live validation pending. @@ -52,7 +53,7 @@ or current fleet size. Metric labels should remain bounded; no per-device labels ## Validation And Reliability -- [ ] Bound fleet-cache rebuild retries under sustained mutations; current stale-build rejection can keep a request waiting while inputs continuously change. Do not silently serve stale wake routes as a fallback. +- [x] Bound fleet-cache rebuilds to three attempts under sustained mutations; return retryable HTTP 503 `fleet_busy` rather than silently serving stale wake routes. Deterministic limit/recovery tests included. - [ ] Reproduce/check Ctrl-C cleanup on the test device: process groups, PTYs, relay tasks, and telemetry shutdown. Leftovers were suspected, not confirmed. - [ ] Exercise collector unavailability and recovery with running agent/CC instances. Verify normal operations continue and telemetry queues remain bounded; outage losses are expected under the current best-effort policy. - [ ] Validate terminal recovery instrumentation live: generation timing/size, transport attach/resume, and backpressure episodes ending in drain or disconnect. diff --git a/docs/telemetry.md b/docs/telemetry.md index a1fb849..b8c13db 100644 --- a/docs/telemetry.md +++ b/docs/telemetry.md @@ -153,9 +153,20 @@ dashboards are not part of this initial slice. ## Failure policy Trace and log queues are capped at 256 records each, with export batches of 64 -and a 3-second request timeout. Export happens off the application path. On an +and a 3-second request timeout. Export happens off the application path. The +OTLP exporters use their recommended retry policy: up to three retries after +the first attempt, exponential backoff and jitter, within the exporter timeout +budget. Temporary transport/receiver failures can recover; authentication +rejection and partial-success responses are not retried. Requests already in +flight can exceed the retry budget, so it is not a hard wall-clock deadline. +The application does not wrap exports in another retry loop. On an outage, telemetry can be dropped rather than accumulating unbounded router RAM; terminal output follows its existing lossless backpressure policy independently. + +Automated receiver tests reject the first export of each signal and confirm +recovery, check the retry-attempt limit on a persistent outage, and confirm that +bad credentials cause only one attempt. These tests do not establish durable +delivery during a long outage; bounded queues can still drop telemetry. The bound is a record count, not an absolute byte limit on arbitrary log messages. Providers flush on orderly daemon exit (Ctrl-C/SIGTERM), on a blocking worker while Tokio remains alive. Only completed spans are exported; this is not graceful draining diff --git a/wakey-control-plane/src/api/control/fleet/mod.rs b/wakey-control-plane/src/api/control/fleet/mod.rs index c7cea7b..bcf8cc5 100644 --- a/wakey-control-plane/src/api/control/fleet/mod.rs +++ b/wakey-control-plane/src/api/control/fleet/mod.rs @@ -51,6 +51,21 @@ async fn fleet_view(state: &AppState) -> anyhow::Result ApiError { + if err.is::() { + return ApiError::new( + StatusCode::SERVICE_UNAVAILABLE, + "fleet_busy", + err.to_string(), + ); + } + ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + fallback_code, + err.to_string(), + ) +} + #[tracing::instrument(name = "fleet_list", skip_all)] pub async fn list_fleet_devices( State(state): State, @@ -60,11 +75,7 @@ pub async fn list_fleet_devices( Ok(devices) => Ok((StatusCode::OK, Json(devices))), Err(err) => { warn!(error = %err, "failed to list fleet devices"); - Err(ApiError::new( - StatusCode::INTERNAL_SERVER_ERROR, - "list_fleet_devices_failed", - err.to_string(), - )) + Err(fleet_error(err, "list_fleet_devices_failed")) } } } @@ -226,11 +237,7 @@ pub async fn wake_fleet_device( }; let devices = load_fleet_devices(&state, &query).await.map_err(|err| { warn!(error = %err, "failed loading fleet devices for wake"); - ApiError::new( - StatusCode::INTERNAL_SERVER_ERROR, - "load_fleet_devices_failed", - err.to_string(), - ) + fleet_error(err, "load_fleet_devices_failed") })?; let device = devices diff --git a/wakey-control-plane/src/api/control/fleet/tests.rs b/wakey-control-plane/src/api/control/fleet/tests.rs index 1b2686a..412fe3e 100644 --- a/wakey-control-plane/src/api/control/fleet/tests.rs +++ b/wakey-control-plane/src/api/control/fleet/tests.rs @@ -12,6 +12,22 @@ use super::normalize_refresh_agent_ids; use super::types::ListFleetDevicesQuery; use crate::state::{AgentDeviceRow, AgentDeviceWithChildren, DeviceIdentifier, KnownDevice}; +#[test] +fn cache_rebuild_limit_is_retryable_but_db_failures_keep_their_error_code() { + let busy = super::fleet_error( + crate::runtime::view_cache::RebuildLimit.into(), + "list_fleet_devices_failed", + ); + assert_eq!(busy.status, axum::http::StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(busy.code, "fleet_busy"); + let failed = super::fleet_error( + anyhow::anyhow!("database failed"), + "list_fleet_devices_failed", + ); + assert_eq!(failed.status, axum::http::StatusCode::INTERNAL_SERVER_ERROR); + assert_eq!(failed.code, "list_fleet_devices_failed"); +} + #[tokio::test] async fn fleet_cache_tracks_store_edits_and_connections_without_caching_filters() { use crate::runtime::{AgentSession, AppState}; diff --git a/wakey-control-plane/src/runtime/view_cache.rs b/wakey-control-plane/src/runtime/view_cache.rs index a9cba35..f8d3813 100644 --- a/wakey-control-plane/src/runtime/view_cache.rs +++ b/wakey-control-plane/src/runtime/view_cache.rs @@ -5,6 +5,19 @@ use std::sync::Arc; use tokio::sync::Mutex; +const MAX_REBUILDS: usize = 3; + +#[derive(Debug)] +pub(crate) struct RebuildLimit; + +impl std::fmt::Display for RebuildLimit { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("view inputs kept changing during rebuilding; retry the request") + } +} + +impl std::error::Error for RebuildLimit {} + pub(crate) struct ViewCache { entry: Mutex)>>, } @@ -21,19 +34,20 @@ impl ViewCache { /// Reuse a matching view or rebuild it. `key` must reflect every input to /// `build`; rechecking it prevents an in-flight old build being published. /// Errors are not cached, and cancellation releases the rebuild lock. - pub(crate) async fn get_or_try_init( + /// Three invalidated builds return `RebuildLimit` rather than starving readers. + pub(crate) async fn get_or_try_init( &self, key: impl Fn() -> KF, build: impl Fn() -> BF, - ) -> Result, E> + ) -> anyhow::Result> where KF: Future, - BF: Future>, + BF: Future>, { // Keep publication and rebuilding under one lock. Waiting readers then // reuse the completed view rather than launching duplicate DB work. let mut entry = self.entry.lock().await; - loop { + for _ in 0..MAX_REBUILDS { let before = key().await; if let Some((stored_key, value)) = &*entry && *stored_key == before @@ -47,6 +61,7 @@ impl ViewCache { return Ok(value); } } + Err(RebuildLimit.into()) } } @@ -65,7 +80,7 @@ mod tests { let build = || async { builds.fetch_add(1, Ordering::SeqCst); tokio::task::yield_now().await; - Ok::<_, ()>(42) + Ok::<_, anyhow::Error>(42) }; let (a, b) = tokio::join!( cache.get_or_try_init(key, build), @@ -93,7 +108,7 @@ mod tests { started.notify_one(); release.notified().await; } - Ok::<_, ()>(captured) + Ok::<_, anyhow::Error>(captured) }, ); let writer = async { @@ -111,13 +126,16 @@ mod tests { let cache = ViewCache::default(); assert!( cache - .get_or_try_init(|| async { 0 }, || async { Err::("db failed") }) + .get_or_try_init( + || async { 0 }, + || async { Err::(anyhow::anyhow!("db failed")) } + ) .await .is_err() ); assert_eq!( *cache - .get_or_try_init(|| async { 0 }, || async { Ok::<_, &str>(7) }) + .get_or_try_init(|| async { 0 }, || async { Ok::<_, anyhow::Error>(7) }) .await .unwrap(), 7 @@ -137,7 +155,7 @@ mod tests { || async { 0 }, || async { started.notify_one(); - std::future::pending::>().await + std::future::pending::>().await }, ) .await @@ -148,10 +166,41 @@ mod tests { assert!(task.await.unwrap_err().is_cancelled()); assert_eq!( *cache - .get_or_try_init(|| async { 0 }, || async { Ok::<_, ()>(7) }) + .get_or_try_init(|| async { 0 }, || async { Ok::<_, anyhow::Error>(7) }) .await .unwrap(), 7 ); } + + #[tokio::test] + async fn sustained_changes_stop_at_the_limit_and_a_later_read_recovers() { + let cache = ViewCache::default(); + let revision = AtomicU64::new(0); + let key = || async { revision.load(Ordering::SeqCst) }; + cache + .get_or_try_init(key, || async { Ok(0_u8) }) + .await + .unwrap(); + revision.store(1, Ordering::SeqCst); + let builds = AtomicU64::new(0); + let error = cache + .get_or_try_init(key, || async { + builds.fetch_add(1, Ordering::SeqCst); + revision.fetch_add(1, Ordering::SeqCst); + Ok(1) + }) + .await + .unwrap_err(); + assert!(error.is::()); + assert_eq!(builds.load(Ordering::SeqCst), MAX_REBUILDS as u64); + let recovered = cache + .get_or_try_init(key, || async { Ok(2) }) + .await + .unwrap(); + assert_eq!( + *recovered, 2, + "must not fall back to the previous cached value" + ); + } } diff --git a/wakey-core/Cargo.toml b/wakey-core/Cargo.toml index c68e5e1..66cb34e 100644 --- a/wakey-core/Cargo.toml +++ b/wakey-core/Cargo.toml @@ -6,12 +6,12 @@ publish = ["gitea"] [dependencies] anyhow = { workspace = true, optional = true } -opentelemetry = { version = "0.31", optional = true } -opentelemetry-otlp = { version = "0.31", default-features = false, features = ["grpc-tonic", "trace", "logs", "metrics"], optional = true } -opentelemetry_sdk = { version = "0.31", optional = true } -opentelemetry-appender-tracing = { version = "0.31", features = ["experimental_use_tracing_span_context"], optional = true } +opentelemetry = { version = "0.33", optional = true } +opentelemetry-otlp = { version = "0.33", default-features = false, features = ["grpc-tonic", "trace", "logs", "metrics"], optional = true } +opentelemetry_sdk = { version = "0.33", optional = true } +opentelemetry-appender-tracing = { version = "0.33", optional = true } tracing = { workspace = true, optional = true } -tracing-opentelemetry = { version = "0.32", optional = true } +tracing-opentelemetry = { version = "0.34", optional = true } tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt", "json", "time"], optional = true } time = { version = "0.3", features = ["formatting", "local-offset"], optional = true } tonic = { version = "0.14", default-features = false, optional = true } @@ -31,4 +31,4 @@ serde_json.workspace = true tokio = { version = "1", features = ["macros", "rt-multi-thread", "net", "sync", "time"] } tokio-stream = { version = "0.1", features = ["net"] } tonic = "0.14" -opentelemetry-proto = { version = "0.31", default-features = false, features = ["gen-tonic", "trace", "logs", "metrics"] } +opentelemetry-proto = { version = "0.33", default-features = false, features = ["gen-tonic", "trace", "logs", "metrics"] } diff --git a/wakey-core/src/telemetry.rs b/wakey-core/src/telemetry.rs index c883215..a7be4ae 100644 --- a/wakey-core/src/telemetry.rs +++ b/wakey-core/src/telemetry.rs @@ -105,11 +105,14 @@ pub fn init( } let resource = resource.build(); // Small fixed queues are intentional: a disconnected collector cannot grow router RAM. + // Exporters own transient retries/backoff; authentication failures are + // not retried. The timeout remains their budget, not a durable outbox. let trace_exporter = opentelemetry_otlp::SpanExporter::builder() .with_tonic() .with_metadata(metadata.clone()) .with_endpoint(endpoint) .with_timeout(Duration::from_secs(3)) + .with_retry_policy(opentelemetry_otlp::RetryPolicy::recommended()) .build() .context("OTLP trace exporter")?; let trace_processor = opentelemetry_sdk::trace::BatchSpanProcessor::builder(trace_exporter) @@ -129,6 +132,7 @@ pub fn init( .with_metadata(metadata.clone()) .with_endpoint(endpoint) .with_timeout(Duration::from_secs(3)) + .with_retry_policy(opentelemetry_otlp::RetryPolicy::recommended()) .build() .context("OTLP log exporter")?; let log_processor = opentelemetry_sdk::logs::BatchLogProcessor::builder(log_exporter) @@ -148,6 +152,7 @@ pub fn init( .with_metadata(metadata) .with_endpoint(endpoint) .with_timeout(Duration::from_secs(3)) + .with_retry_policy(opentelemetry_otlp::RetryPolicy::recommended()) .build() .context("OTLP metric exporter")?; let reader = opentelemetry_sdk::metrics::PeriodicReader::builder(metric_exporter) diff --git a/wakey-core/tests/telemetry.rs b/wakey-core/tests/telemetry.rs index 61619b5..9602515 100644 --- a/wakey-core/tests/telemetry.rs +++ b/wakey-core/tests/telemetry.rs @@ -3,6 +3,7 @@ use opentelemetry_proto::tonic::collector::{ logs::v1 as logs, metrics::v1 as metrics, trace::v1 as traces, }; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use tonic::{Request, Response, Status}; use wakey_core::telemetry::{self, TelemetryConfig}; @@ -12,6 +13,8 @@ struct Collector { traces: Arc>>, logs: Arc>>, metrics: Arc>>, + fail_until: usize, + attempts: [Arc; 3], } fn check_auth(request: &Request) -> Result<(), Status> { @@ -28,13 +31,78 @@ fn check_auth(request: &Request) -> Result<(), Status> { Ok(()) } +/// Use a real receiver and counted RPCs, not sleep/elapsed-time expectations. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn exporter_retries_are_bounded_and_bad_credentials_are_not_retried() { + use opentelemetry_otlp::{RetryPolicy, WithExportConfig, WithTonicConfig}; + use opentelemetry_sdk::trace::SpanExporter as _; + use std::time::Duration; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("http://{}", listener.local_addr().unwrap()); + let collector = Collector { + fail_until: usize::MAX, + ..Default::default() + }; + let (stop_tx, stop_rx) = tokio::sync::oneshot::channel(); + let server = tokio::spawn( + tonic::transport::Server::builder() + .add_service(traces::trace_service_server::TraceServiceServer::new( + collector.clone(), + )) + .serve_with_incoming_shutdown( + tokio_stream::wrappers::TcpListenerStream::new(listener), + async { + let _ = stop_rx.await; + }, + ), + ); + let mut metadata = tonic::metadata::MetadataMap::new(); + metadata.insert("authorization", "Basic dGVzdDp0ZXN0".parse().unwrap()); + let policy = RetryPolicy::recommended() + .with_max_retries(2) + .with_initial_delay(Duration::from_millis(1)) + .with_max_delay(Duration::from_millis(4)) + .with_max_jitter(Duration::ZERO); + let exporter = opentelemetry_otlp::SpanExporter::builder() + .with_tonic() + .with_endpoint(&endpoint) + .with_metadata(metadata) + .with_timeout(Duration::from_secs(3)) + .with_retry_policy(policy.clone()) + .build() + .unwrap(); + assert!(exporter.export(vec![]).await.is_err()); + assert_eq!(collector.attempts[0].load(Ordering::SeqCst), 3); + + let unauthorized = opentelemetry_otlp::SpanExporter::builder() + .with_tonic() + .with_endpoint(endpoint) + .with_timeout(Duration::from_secs(3)) + .with_retry_policy(policy) + .build() + .unwrap(); + assert!(unauthorized.export(vec![]).await.is_err()); + assert_eq!( + collector.attempts[0].load(Ordering::SeqCst), + 4, + "authentication failure must add only one attempt" + ); + let _ = stop_tx.send(()); + server.await.unwrap().unwrap(); +} + #[tonic::async_trait] impl traces::trace_service_server::TraceService for Collector { async fn export( &self, request: Request, ) -> Result, Status> { + let attempt = self.attempts[0].fetch_add(1, Ordering::SeqCst); check_auth(&request)?; + if attempt < self.fail_until { + return Err(Status::unavailable("temporary receiver outage")); + } self.traces.lock().unwrap().push(request.into_inner()); Ok(Response::new(Default::default())) } @@ -45,7 +113,11 @@ impl logs::logs_service_server::LogsService for Collector { &self, request: Request, ) -> Result, Status> { + let attempt = self.attempts[1].fetch_add(1, Ordering::SeqCst); check_auth(&request)?; + if attempt < self.fail_until { + return Err(Status::unavailable("temporary receiver outage")); + } self.logs.lock().unwrap().push(request.into_inner()); Ok(Response::new(Default::default())) } @@ -56,7 +128,11 @@ impl metrics::metrics_service_server::MetricsService for Collector { &self, request: Request, ) -> Result, Status> { + let attempt = self.attempts[2].fetch_add(1, Ordering::SeqCst); check_auth(&request)?; + if attempt < self.fail_until { + return Err(Status::unavailable("temporary receiver outage")); + } self.metrics.lock().unwrap().push(request.into_inner()); Ok(Response::new(Default::default())) } @@ -68,33 +144,40 @@ async fn exports_all_signals_and_preserves_remote_command_parent() { // Exercise the real global subscriber without mutating this process's environment // or inheriting a developer's RUST_LOG=warn/off setting. if std::env::var_os("WAKEY_OTLP_TEST_CHILD").is_none() { - let output = std::process::Command::new(std::env::current_exe().unwrap()) - .args([ - "--exact", - "exports_all_signals_and_preserves_remote_command_parent", - "--nocapture", - ]) - .env("WAKEY_OTLP_TEST_CHILD", "1") - .env("RUST_LOG", "info") - .env("WAKEY_OTLP_TEST_AUTH", "Basic dGVzdDp0ZXN0") - .output() - .unwrap(); - assert!( - output.status.success(), - "{}", - String::from_utf8_lossy(&output.stderr) - ); - for console in [&output.stdout, &output.stderr] { + for retry in ["0", "1"] { + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "exports_all_signals_and_preserves_remote_command_parent", + "--nocapture", + ]) + .env("WAKEY_OTLP_TEST_CHILD", "1") + .env("WAKEY_OTLP_TEST_RETRY", retry) + .env("RUST_LOG", "info") + .env("WAKEY_OTLP_TEST_AUTH", "Basic dGVzdDp0ZXN0") + .output() + .unwrap(); assert!( - !String::from_utf8_lossy(console) - .contains("debug log exported without console output") + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) ); + for console in [&output.stdout, &output.stderr] { + assert!( + !String::from_utf8_lossy(console) + .contains("debug log exported without console output") + ); + } } return; } let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let endpoint = format!("http://{}", listener.local_addr().unwrap()); - let collector = Collector::default(); + let retry = std::env::var("WAKEY_OTLP_TEST_RETRY").as_deref() == Ok("1"); + let collector = Collector { + fail_until: usize::from(retry), + ..Default::default() + }; let (stop_tx, stop_rx) = tokio::sync::oneshot::channel(); let server = tonic::transport::Server::builder() .add_service(traces::trace_service_server::TraceServiceServer::new( @@ -159,6 +242,13 @@ async fn exports_all_signals_and_preserves_remote_command_parent() { let _ = stop_tx.send(()); server.await.unwrap().unwrap(); + for attempts in &collector.attempts { + assert!( + attempts.load(Ordering::SeqCst) > usize::from(retry), + "each signal must recover from its first failed export" + ); + } + let traces = collector.traces.lock().unwrap(); let spans: Vec<_> = traces .iter()