From 4ec31b0523d0b4a55495090d11c043f29dccee23 Mon Sep 17 00:00:00 2001 From: Joshua Date: Sat, 10 Oct 2026 23:23:20 +0300 Subject: [PATCH 1/3] feat(telemetry): guest tracing to X-Ray and metrics to CloudWatch A new host interface, enclave:telemetry (wit/telemetry), that a guest imports to report what it is doing beyond its stdout: - tracing: events, and a `span` resource that nests by the order a guest opens spans and closes on `end` or on drop. Each invocation (request, task run, stream message) is one X-Ray trace, its root annotated with the tenant and the invocation kind, spans as subsegments. - metrics: counter-add, gauge-set, histogram-record, aggregated over a 30 s window (sum, last value, values/counts distribution) and published with PutMetricData. No tenant dimension; CloudWatch bills per series. No function takes a tenant id: the serve layer puts a TelemetryContext on the store, and the trace ships when the store drops, so every exit (timeout and trap included) flushes and no call site has to remember to. Host calls are synchronous and never touch the network. Shipping is two forwarders shaped like the guest-log one: bounded queue, batches, two backed-off retries, drops counted and reported once a minute. What CloudWatch or X-Ray would refuse (an out-of-range value, a repeated dimension, a 64 KiB+ document) is dropped or shed when recorded rather than failing a whole request. With telemetry off the calls still succeed, and events and spans are mirrored to the console either way. Hand-wired like tasks/notify/streams (no bindgen), including the span resource. tests/telemetry.rs serves the example guest's /telemetry route through the real linker with in-memory X-Ray and CloudWatch, so the ABI names and types are checked against what wit-bindgen generates. aws-sdk-xray 1.109 and aws-sdk-cloudwatch 1.127 are the releases that match the locked smithy runtime; no existing crate moves. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 137 +++ docs/TELEMETRY.md | 142 +++ examples/guest-http/src/main.rs | 32 + examples/guest-http/wit/app.wit | 1 + .../wit/deps/telemetry/telemetry.wit | 109 +++ runtime/Cargo.toml | 3 + runtime/src/lib.rs | 1 + runtime/src/linker.rs | 1 + runtime/src/main.rs | 133 ++- runtime/src/serve/http.rs | 37 +- runtime/src/state.rs | 15 + runtime/src/telemetry/metrics.rs | 669 +++++++++++++ runtime/src/telemetry/mod.rs | 484 ++++++++++ runtime/src/telemetry/xray.rs | 899 ++++++++++++++++++ runtime/src/testing/mod.rs | 11 + runtime/tests/serve_auth.rs | 1 + runtime/tests/serve_grpc_wire.rs | 1 + runtime/tests/serve_h2.rs | 1 + runtime/tests/serve_tls.rs | 1 + runtime/tests/telemetry.rs | 191 ++++ wit/telemetry/telemetry.wit | 109 +++ 21 files changed, 2964 insertions(+), 14 deletions(-) create mode 100644 docs/TELEMETRY.md create mode 100644 examples/guest-http/wit/deps/telemetry/telemetry.wit create mode 100644 runtime/src/telemetry/metrics.rs create mode 100644 runtime/src/telemetry/mod.rs create mode 100644 runtime/src/telemetry/xray.rs create mode 100644 runtime/tests/telemetry.rs create mode 100644 wit/telemetry/telemetry.wit diff --git a/Cargo.lock b/Cargo.lock index 208f31f..1fc9518 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11,6 +11,12 @@ dependencies = [ "gimli", ] +[[package]] +name = "adler2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" + [[package]] name = "aho-corasick" version = "1.1.4" @@ -436,6 +442,36 @@ dependencies = [ "uuid", ] +[[package]] +name = "aws-sdk-cloudwatch" +version = "1.127.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac606233394937c6014a46fc776214f5eed0b6bde3bdd38c8fa1aa28e0f3ad09" +dependencies = [ + "arc-swap", + "aws-credential-types", + "aws-runtime", + "aws-smithy-async", + "aws-smithy-cbor", + "aws-smithy-compression", + "aws-smithy-http 0.64.0", + "aws-smithy-json 0.63.0", + "aws-smithy-observability 0.3.0", + "aws-smithy-runtime", + "aws-smithy-runtime-api", + "aws-smithy-schema", + "aws-smithy-types", + "aws-types", + "fastrand", + "flate2", + "http 0.2.12", + "http 1.4.0", + "http-body 1.0.1", + "http-body-util", + "regex-lite", + "tracing", +] + [[package]] name = "aws-sdk-cloudwatchlogs" version = "1.149.0" @@ -571,6 +607,32 @@ dependencies = [ "tracing", ] +[[package]] +name = "aws-sdk-xray" +version = "1.109.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aaa8a863929c975e71cece4092363b01140da9afeabd2a54d8c66e7cf9fce56b" +dependencies = [ + "arc-swap", + "aws-credential-types", + "aws-runtime", + "aws-smithy-async", + "aws-smithy-http 0.64.0", + "aws-smithy-json 0.63.0", + "aws-smithy-observability 0.3.0", + "aws-smithy-runtime", + "aws-smithy-runtime-api", + "aws-smithy-schema", + "aws-smithy-types", + "aws-types", + "bytes", + "fastrand", + "http 0.2.12", + "http 1.4.0", + "regex-lite", + "tracing", +] + [[package]] name = "aws-sigv4" version = "1.5.1" @@ -605,6 +667,19 @@ dependencies = [ "tokio", ] +[[package]] +name = "aws-smithy-cbor" +version = "0.62.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f069169abdc367533ea89b869b7dfaadd4134a9bc287c13dbe5ff364ff13228" +dependencies = [ + "aws-smithy-runtime-api", + "aws-smithy-schema", + "aws-smithy-types", + "minicbor", + "num-bigint", +] + [[package]] name = "aws-smithy-checksums" version = "0.64.7" @@ -626,6 +701,24 @@ dependencies = [ "tracing", ] +[[package]] +name = "aws-smithy-compression" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "016ef334a13c53bd75184bb8f89b0fb2aa4d56cb3cfc478ab3f05f13654b60a1" +dependencies = [ + "aws-smithy-runtime-api", + "aws-smithy-types", + "bytes", + "flate2", + "futures-util", + "http 1.4.0", + "http-body 1.0.1", + "http-body-util", + "pin-project-lite", + "tracing", +] + [[package]] name = "aws-smithy-eventstream" version = "0.60.20" @@ -1990,10 +2083,12 @@ dependencies = [ "aws-config", "aws-credential-types", "aws-lc-rs", + "aws-sdk-cloudwatch", "aws-sdk-cloudwatchlogs", "aws-sdk-kms", "aws-sdk-s3", "aws-sdk-ssm", + "aws-sdk-xray", "aws-sigv4", "aws-smithy-http-client", "aws-smithy-runtime-api", @@ -2165,6 +2260,17 @@ version = "0.4.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b7ac824320a75a52197e8f2d787f6a38b6718bb6897a35142d749af3c0e8f4fe" +[[package]] +name = "flate2" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e634e2e0ebac1ee034020da1ca582e17ffe4e0f5e985823721e168928136dcb" +dependencies = [ + "crc32fast", + "miniz_oxide", + "zlib-rs", +] + [[package]] name = "fnv" version = "1.0.7" @@ -3162,12 +3268,31 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" +[[package]] +name = "minicbor" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c12b4033ffaa92fbf9df03df38d19324f52bad130dd223f811734a8006dd2d69" +dependencies = [ + "half", +] + [[package]] name = "minimal-lexical" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" +[[package]] +name = "miniz_oxide" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b63fbc4a50860e98e7b2aa7804ded1db5cbc3aff9193adaff57a6931bf7c4b4c" +dependencies = [ + "adler2", + "simd-adler32", +] + [[package]] name = "mio" version = "1.2.0" @@ -4434,6 +4559,12 @@ dependencies = [ "libc", ] +[[package]] +name = "simd-adler32" +version = "0.3.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a219298ac11a56ea9a6d2120044824d6f01aeb034955e7af7bc16858527deea" + [[package]] name = "siphasher" version = "1.0.2" @@ -6425,6 +6556,12 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "zlib-rs" +version = "0.6.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b268e58e7c693d7c271f93ffc4ba3b380412554231c85bf61ca7af91042a4112" + [[package]] name = "zmij" version = "1.0.21" diff --git a/docs/TELEMETRY.md b/docs/TELEMETRY.md new file mode 100644 index 0000000..fd01a55 --- /dev/null +++ b/docs/TELEMETRY.md @@ -0,0 +1,142 @@ +# Telemetry + +A guest can report what it is doing beyond its stdout: **spans and events to AWS +X-Ray**, and **metrics to CloudWatch**. The ABI is +[`wit/telemetry/telemetry.wit`](../wit/telemetry/telemetry.wit); the runtime's +side is [`runtime/src/telemetry`](../runtime/src/telemetry). + +As with every interface the runtime adds, nothing in it takes a tenant id. The +runtime already knows which tenant an invocation belongs to, and attributes what +the guest reports to it. + +## From a guest + +Include the world, and call it: + +```wit +world app { + include enclave:telemetry/observed@0.1.0; +} +``` + +```rust +use bindings::enclave::telemetry::{metrics, tracing::{self, Field, Level, Span}}; + +let span = Span::new(Level::Info, "sign", &[]); // a subsegment of the invocation +span.set_field("session", &session_id); +tracing::event(Level::Warn, "retrying", &[]); // lands on the innermost open span +drop(span); // or span.end() + +metrics::counter_add("signatures", 1.0, &[]); +metrics::histogram_record("sign-latency", ms, metrics::Unit::Milliseconds, &[]); +``` + +Every call returns nothing and never fails. When X-Ray or metrics are switched +off, the calls succeed and do less, so a guest never branches on whether +telemetry is on. Each event and closed span is also mirrored to the runtime's +console under the `guest` target, so an emulator with neither still shows what a +guest reported. + +## What happens to it + +**One trace per invocation.** An inbound request, a task run and a stream +message are each one X-Ray trace. Its root segment is named `enclave-guest` and +carries two *annotations*, the `tenant` (hex) and the `invocation` kind, which +X-Ray indexes, so "every trace for this tenant" is a filter expression. Each span +a guest opens is a subsegment of whichever span was open when it was created; +its level is an annotation. Events and fields ride in the span's `metadata`. + +The trace is shipped when the invocation's store is dropped. That is the end of +the invocation however it ends, a timeout or a trap included. Spans the guest +never closed are closed then. + +**Metrics are aggregated, not queued.** Over each 30-second window a counter is +summed, a gauge keeps its last value, and a histogram keeps each distinct value +with its count. That is CloudWatch's own `values`/`counts` distribution, so +percentiles survive aggregation. Each window becomes one `PutMetricData` call +per 1000 datums, under the deployment's namespace. The runtime adds no tenant +dimension: CloudWatch bills per unique combination, and that is the runtime's +cost to decide, not a guest's. + +## What leaves the enclave + +What a guest puts in a span name, a message, a field or a metric is its own +business, exactly as its stdout already is. The runtime adds only what it knows: +the tenant id, the invocation kind, the image's PCR0, the region and its own +version. All of it goes to the operator's own AWS account, the one guest logs +already go to. + +Whether an enclave exports traces or metrics at all is part of its image (below), +so it is measured into PCR0. A client can tell from an attestation whether its +guest's traces leave the box. + +## Limits + +A guest that reports too much loses detail. The guest's calls never fail. + +| | Limit | Past it | +|---|---|---| +| Spans per invocation | 256 | counted (`dropped_spans`), not recorded | +| Events per span | 64 | counted (`dropped_events`) | +| Fields per span or event | 32 | the rest ignored | +| Field key / message or value | 64 / 1024 characters | clipped | +| Span name | 200 characters of what X-Ray accepts | clipped; other characters become `_` | +| Trace document | 64 KiB (X-Ray's limit) | events and fields shed, then subsegments; marked `trimmed` | +| Metric name | 1-255 printable ASCII | dropped and counted | +| Dimensions | ≤10, each name and value 1-64 of `[A-Za-z0-9_.-]`, no name twice | dropped and counted | +| Metric value | finite, and 0 or within CloudWatch's range | dropped and counted | +| Series per window | 1000 | new series dropped; existing ones still count | +| Distinct histogram values | 150 per datum | sealed into another datum, not lost | + +The metric checks run when a value is recorded, not when it is sent. CloudWatch +refuses a whole request over one bad datum, so letting one through would lose +every other metric in the same call. + +## Delivery + +Best effort, like guest logs, and for the same reason: nothing a guest calls +waits on the network. A finished trace goes to a bounded queue (1024) by +`try_send`. A forwarder ships batches of up to 50 documents or 1 MiB, after a +one-second linger. A failure that may heal (throttling, a 5xx, an expired session +token) is retried twice, after 0.5 s and 2 s. One that cannot (access denied, a +malformed request) is dropped. Drops are counted and reported at most once a +minute. At shutdown the queue and the last metrics window are flushed within five +seconds. + +## Turning it on + +In the image, [`deploy/nix/deployment.nix`](../deploy/nix/deployment.nix): + +```nix +xray = true; # each invocation as an X-Ray trace +metricsNamespace = "Merlin/Guest"; # or "" for no metrics; never under AWS/ +``` + +The parent's role needs the grants. In [`deploy/tofu`](../deploy/tofu): + +```hcl +telemetry = { xray = true, metrics_namespace = "Merlin/Guest" } +``` + +That allows `xray:PutTraceSegments`, which X-Ray cannot scope to a resource, and +`cloudwatch:PutMetricData` for that namespace only. Without the grants the +runtime still serves, and logs at most once a minute that X-Ray or CloudWatch +refused it. + +The flags behind these are `ENCLAVE_XRAY` and `ENCLAVE_METRICS_NAMESPACE`. A +malformed namespace stops the boot. A client that cannot be built at startup +does not: that half of telemetry stays on the console. For an emulator, +`ENCLAVE_XRAY_ENDPOINT` and `ENCLAVE_METRICS_ENDPOINT` point the clients at a +stand-in such as LocalStack. + +## Testing + +`runtime/tests/telemetry.rs` serves the example guest's `/telemetry` route +through the real linker, with X-Ray and CloudWatch replaced by in-memory +destinations. It checks the trace (attribution, nesting, fields, events, timings) +and the metrics (aggregation, units, no tenant dimension): + +```console +$ scripts/build-guest.sh http +$ cargo test -p enclave-runtime --features testing --test telemetry -- --ignored +``` diff --git a/examples/guest-http/src/main.rs b/examples/guest-http/src/main.rs index 798b373..3aa031a 100644 --- a/examples/guest-http/src/main.rs +++ b/examples/guest-http/src/main.rs @@ -199,6 +199,10 @@ async fn main(mut req: Request) -> Result, Error> { }) } ("GET", "/env") => Ok(text(StatusCode::OK, environment())), + // Known spans, events and metrics, for the runtime's telemetry tests: a span with a + // field and an event, a child opened inside it, a free-standing event that must land on + // the innermost open span, and one of each kind of metric. + ("GET", "/telemetry") => Ok(text(StatusCode::OK, telemetry())), // Whatever the runtime says about the caller. A guest can only ever // read this header, never write it, and the runtime overwrites it on // every request — so what arrives here is the runtime's word, not the @@ -405,6 +409,34 @@ fn counter() -> Result { Ok(next) } +fn telemetry() -> String { + use bindings::enclave::telemetry::metrics::{self, Dimension, Unit}; + use bindings::enclave::telemetry::tracing::{self, Field, Level, Span}; + + let field = |key: &str, value: &str| Field { + key: key.to_string(), + value: value.to_string(), + }; + let outer = Span::new(Level::Info, "handle", &[field("route", "/telemetry")]); + outer.set_field("phase", "start"); + outer.add_event(Level::Info, "began", &[]); + { + // Ended by being dropped, not by `end`. + let _inner = Span::new(Level::Debug, "inner", &[]); + tracing::event(Level::Warn, "inside inner", &[field("n", "1")]); + } + outer.end(); + + let route = [Dimension { + name: "route".to_string(), + value: "telemetry".to_string(), + }]; + metrics::counter_add("requests", 1.0, &route); + metrics::gauge_set("queue-depth", 3.0, Unit::Count, &[]); + metrics::histogram_record("latency", 12.5, Unit::Milliseconds, &route); + "reported\n".to_string() +} + fn environment() -> String { let mut vars: Vec<_> = std::env::vars().collect(); vars.sort(); diff --git a/examples/guest-http/wit/app.wit b/examples/guest-http/wit/app.wit index 8c2ed24..3d49b9e 100644 --- a/examples/guest-http/wit/app.wit +++ b/examples/guest-http/wit/app.wit @@ -10,4 +10,5 @@ package guest:http@0.1.0; world app { include enclave:tasks/background@0.1.0; import enclave:notify/notify@0.1.0; + include enclave:telemetry/observed@0.1.0; } diff --git a/examples/guest-http/wit/deps/telemetry/telemetry.wit b/examples/guest-http/wit/deps/telemetry/telemetry.wit new file mode 100644 index 0000000..145d3d4 --- /dev/null +++ b/examples/guest-http/wit/deps/telemetry/telemetry.wit @@ -0,0 +1,109 @@ +package enclave:telemetry@0.1.0; + +/// Structured tracing, exported to AWS X-Ray. +/// +/// The runtime roots one trace per guest invocation — an inbound request, a +/// task run, or a stream message — and ships it when that invocation ends. A +/// guest's events and spans attach to that trace; the runtime binds the tenant, +/// so **no function here accepts a tenant id** (the same rule as enclave:notify +/// and enclave:tasks). Calls made in a build with X-Ray switched off are +/// accepted and discarded, so a guest never has to branch on whether it is on. +/// +/// Spans and events carry no secret by construction on the runtime's side: the +/// text and fields are the guest's own, and the runtime adds only what it +/// already knows (the tenant, the image, the region). What a guest puts in a +/// message or field is the guest's responsibility, exactly as with its stdout. +interface tracing { + /// Severity. Mapped onto the runtime's own `tracing` levels for the console, + /// and recorded on the X-Ray (sub)segment. + enum level { + trace, + debug, + info, + warn, + error, + } + + /// A key/value annotation on an event or span. The value is a string the + /// guest has already formatted; the runtime does not parse it. Keys are + /// 1-64 characters of [A-Za-z0-9_.-]. + record field { + key: string, + value: string, + } + + /// Record a point-in-time event on the span that is currently open (or on + /// the invocation's root span if the guest has opened none). Cheap: there is + /// no handle to keep. + event: func(level: level, message: string, fields: list); + + /// A unit of work with a duration, exported as an X-Ray subsegment of + /// whichever span is open when it is created (the invocation's root span if + /// none) — so spans nest by the order a guest opens them. The subsegment's + /// end time is set when the span is closed: by `end`, or by dropping the + /// resource, whichever comes first. + resource span { + /// Open a span beneath the currently-open span. + constructor(level: level, name: string, fields: list); + + /// Annotate the span after it has opened. + set-field: func(key: string, value: string); + + /// Record a point-in-time event on this span. + add-event: func(level: level, message: string, fields: list); + + /// Close the span now. Dropping the resource without calling this closes + /// it at the drop instead; calling it more than once does nothing. + end: func(); + } +} + +/// Metrics, published to AWS CloudWatch through the dedicated `PutMetricData` +/// API under the runtime's configured namespace. +/// +/// As with tracing, the runtime binds the tenant and a guest cannot name it: +/// CloudWatch bills per unique dimension combination, so a tenant dimension is +/// the runtime's decision to make, not a guest's. Dimensions a guest does give +/// must be low cardinality. Calls made in a build with metrics switched off are +/// accepted and discarded. +/// +/// A metric name is 1-255 printable ASCII characters. A call CloudWatch would +/// refuse — a bad name or dimension, a value that is not finite or is outside +/// CloudWatch's range — is dropped and counted rather than sent, because +/// CloudWatch refuses a whole request over one bad datum. +interface metrics { + /// The unit a value is in, as CloudWatch names them. `none` leaves it unset. + enum unit { + none, + count, + seconds, + milliseconds, + bytes, + kilobytes, + percent, + } + + /// A metric dimension: low cardinality only. 1-64 characters of + /// [A-Za-z0-9_.-] each, and at most 10 per call. + record dimension { + name: string, + value: string, + } + + /// Add `value` to a counter (published as a Sum aggregation). + counter-add: func(name: string, value: f64, dimensions: list); + + /// Report a gauge's current value (published as-is). + gauge-set: func(name: string, value: f64, %unit: unit, dimensions: list); + + /// Record one observation into a distribution (published through + /// CloudWatch's values/counts distribution). + histogram-record: func(name: string, value: f64, %unit: unit, dimensions: list); +} + +/// What a guest includes to get both. Both are imports: the guest calls the +/// runtime, and there is nothing here for it to export. +world observed { + import tracing; + import metrics; +} diff --git a/runtime/Cargo.toml b/runtime/Cargo.toml index fc8d1c4..1c68c79 100644 --- a/runtime/Cargo.toml +++ b/runtime/Cargo.toml @@ -176,6 +176,9 @@ aws-sdk-ssm = { version = "1.56.0", default-features = false, features = ["rt-to # own still calls AWS over gvproxy. It brings no new crypto stack; kms and ssm # already pull `rustls`, which is what the ban above is actually about. aws-sdk-cloudwatchlogs = { version = "1.131.0", default-features = false, features = ["rt-tokio", "rustls", "behavior-version-latest"] } +# Telemetry: guest spans to X-Ray and guest metrics to CloudWatch (see src/telemetry). +aws-sdk-xray = { version = "1.109.0", default-features = false, features = ["rt-tokio", "rustls", "behavior-version-latest"] } +aws-sdk-cloudwatch = { version = "1.127.0", default-features = false, features = ["rt-tokio", "rustls", "behavior-version-latest"] } # The default credential chain, which resolves through gvproxy to the parent # instance's IMDS. Named explicitly rather than relied on as a fallback, so it # is obvious *which* identity guest logs are written with. diff --git a/runtime/src/lib.rs b/runtime/src/lib.rs index 58abc60..21e24ad 100644 --- a/runtime/src/lib.rs +++ b/runtime/src/lib.rs @@ -40,6 +40,7 @@ pub mod state; pub mod store; pub mod stream; pub mod tasks; +pub mod telemetry; pub mod tenant; pub mod zfs; diff --git a/runtime/src/linker.rs b/runtime/src/linker.rs index 562b272..957cf82 100644 --- a/runtime/src/linker.rs +++ b/runtime/src/linker.rs @@ -19,6 +19,7 @@ pub fn build_linker(engine: &wasmtime::Engine) -> Result> { crate::tasks::add_to_linker(&mut linker)?; crate::stream::add_to_linker(&mut linker)?; crate::notify::add_to_linker(&mut linker)?; + crate::telemetry::add_to_linker(&mut linker)?; Ok(linker) } diff --git a/runtime/src/main.rs b/runtime/src/main.rs index 83f85ee..d1f2bac 100644 --- a/runtime/src/main.rs +++ b/runtime/src/main.rs @@ -69,6 +69,87 @@ fn nonempty(value: &Option) -> Option { .map(str::to_string) } +/// Start shipping guest spans to X-Ray and guest metrics to CloudWatch, as the flags ask. +/// +/// A malformed namespace fails the boot: it is a deployment mistake that would otherwise surface +/// only as every metric refused. A client that cannot be built in time does not: that guest +/// telemetry stays on the console, and the enclave serves. +async fn start_telemetry( + cli: &Cli, + image: Option, +) -> Result { + use enclave_runtime::telemetry::{ + CloudWatchMetrics, Metrics, MetricsConfig, Telemetry, Tracer, XrayConfig, XrayDestination, + }; + + let tracer = if cli.xray { + let config = XrayConfig { + region: cli.region.clone(), + endpoint: cli.testing.xray_endpoint.clone(), + }; + match tokio::time::timeout( + enclave_runtime::STARTUP_PROBE_TIMEOUT, + XrayDestination::connect(&config), + ) + .await + { + Ok(destination) => { + tracing::info!("guest spans are going to X-Ray"); + Some(Tracer::start( + std::sync::Arc::new(destination), + image, + cli.region.clone(), + )) + } + Err(_) => { + tracing::error!( + timeout = ?enclave_runtime::STARTUP_PROBE_TIMEOUT, + "could not build an X-Ray client in time; guest spans stay on the console" + ); + None + } + } + } else { + None + }; + + let metrics = match nonempty(&cli.metrics_namespace) { + None => None, + Some(namespace) => { + anyhow::ensure!( + MetricsConfig::valid_namespace(&namespace), + "--metrics-namespace must be 1-255 printable ASCII characters, and not under AWS/" + ); + let config = MetricsConfig { + namespace: namespace.clone(), + region: cli.region.clone(), + endpoint: cli.testing.metrics_endpoint.clone(), + }; + match tokio::time::timeout( + enclave_runtime::STARTUP_PROBE_TIMEOUT, + CloudWatchMetrics::connect(&config), + ) + .await + { + Ok(destination) => { + tracing::info!(%namespace, "guest metrics are going to CloudWatch"); + Some(Metrics::start(std::sync::Arc::new(destination), namespace)) + } + Err(_) => { + tracing::error!( + timeout = ?enclave_runtime::STARTUP_PROBE_TIMEOUT, + %namespace, + "could not build a CloudWatch client in time; guest metrics are dropped" + ); + None + } + } + } + }; + + Ok(Telemetry::new(tracer, metrics)) +} + #[derive(Parser, Debug)] #[command( version, @@ -189,6 +270,16 @@ struct Cli { #[arg(long, env = "ENCLAVE_GUEST_LOG_GROUP")] guest_log_group: Option, + /// Export each guest invocation as an AWS X-Ray trace: its spans, its events, and the tenant + /// it ran for. The instance role needs `xray:PutTraceSegments`. + #[arg(long, env = "ENCLAVE_XRAY", value_parser = parse_bool_flag, num_args = 0..=1, default_value_t = false, default_missing_value = "true")] + xray: bool, + + /// CloudWatch namespace guest metrics are published under with `PutMetricData`. Empty or + /// unset means none. The instance role needs `cloudwatch:PutMetricData`. + #[arg(long, env = "ENCLAVE_METRICS_NAMESPACE")] + metrics_namespace: Option, + /// Report on the clock and entropy source and exit, without mounting or /// running anything. For diagnosing a deployment. #[arg(long, alias = "clock-check")] @@ -288,6 +379,14 @@ struct Testing { #[arg(long, env = "ENCLAVE_GUEST_LOG_ENDPOINT")] guest_log_endpoint: Option, + /// Point the X-Ray client somewhere else, such as a local stand-in for the service. + #[arg(long, env = "ENCLAVE_XRAY_ENDPOINT")] + xray_endpoint: Option, + + /// Point the CloudWatch metrics client somewhere else. + #[arg(long, env = "ENCLAVE_METRICS_ENDPOINT")] + metrics_endpoint: Option, + /// Point the push client at a stub: an `http://` endpoint hands wake signals to /// whatever is listening, and requests to it are signed with a placeholder /// rather than the instance's role. @@ -333,6 +432,8 @@ impl Testing { acme_ca: None, acme_contacts: Vec::new(), guest_log_endpoint: None, + xray_endpoint: None, + metrics_endpoint: None, push_endpoint: None, cosign_attestations: false, } @@ -706,20 +807,20 @@ async fn run() -> Result { vec![std::sync::Arc::new(enclave_runtime::TracingLogSink)]; let mut log_forwarder = None; - if let Some(config) = cli.guest_log_config() { - // Which image is writing to this stream. A fixed stream name is - // shared by every boot and every image, so without this nothing - // in it says which enclave produced which line. - let image = attestation - .as_ref() - .and_then(|nsm| match nsm.describe_pcr(0) { - Ok(pcr) => Some(hex::encode(pcr.value)), - Err(e) => { - tracing::warn!(error = %e, "could not read PCR0 for the guest log stream"); - None - } - }); + // Which image is running, for the guest log stream and on every trace. A fixed stream name is + // shared by every boot and every image, so without this nothing in it says which enclave + // produced which line. + let image = attestation + .as_ref() + .and_then(|nsm| match nsm.describe_pcr(0) { + Ok(pcr) => Some(hex::encode(pcr.value)), + Err(e) => { + tracing::warn!(error = %e, "could not read PCR0 for guest logs and traces"); + None + } + }); + if let Some(config) = cli.guest_log_config() { // Bounded, like everything else on this path. Building the // client resolves a credential provider, and that reaches IMDS // through gvproxy — a path this deployment has not verified. A @@ -797,6 +898,11 @@ async fn run() -> Result { } } + // Guest spans and metrics. Bounded like the log client, for the same reason: each client + // resolves a credential provider through gvproxy, and telemetry that cannot be set up must + // leave the guest's calls working and the listener binding. + let telemetry = start_telemetry(&cli, image.clone()).await?; + // Decided before the listener binds, so a setting that cannot work is a boot failure rather // than a surprise on the first wake. Whether the application can deliver is the probe's to // find out once serving starts. @@ -843,6 +949,7 @@ async fn run() -> Result { guest, ServeConfig { notify, + telemetry, // Whether tasks run is the guest's to say, by exporting `run-task`. background_tasks: Some(Default::default()), addr: HTTP_LISTEN, diff --git a/runtime/src/serve/http.rs b/runtime/src/serve/http.rs index a9c33b4..e18ee0d 100644 --- a/runtime/src/serve/http.rs +++ b/runtime/src/serve/http.rs @@ -47,6 +47,9 @@ pub struct ServeConfig { /// here rather than by the caller, because both need the mounted /// filesystem and the trusted clock, and this is where those exist. pub notify: Option, + /// Where guest spans and metrics go. The default sends nowhere, and the guest's calls still + /// succeed — see [`crate::telemetry`]. + pub telemetry: crate::telemetry::Telemetry, /// Address to accept on. pub addr: SocketAddr, /// Where TLS connections take their serving identity. `None` serves @@ -107,6 +110,7 @@ impl Default for ServeConfig { ServeConfig { background_tasks: None, notify: None, + telemetry: Default::default(), addr: ([0, 0, 0, 0], 8080).into(), certificate: None, acme: None, @@ -177,6 +181,9 @@ pub struct ServeHandle { /// Connections this runtime holds for guests — see [`crate::stream`]. streams: Option>, notify: Option>, + /// Where each invocation's trace and metrics go. Every invocation gets a context, so a guest's + /// telemetry calls always resolve. + telemetry: crate::telemetry::Telemetry, guest: Arc, /// One request at a time for callers with no resolved tenant. /// @@ -284,6 +291,7 @@ impl ServeHandle { tasks: None, streams: None, notify: None, + telemetry: Default::default(), pre, guest: Arc::new(guest), anonymous: Arc::new(tokio::sync::Mutex::new(())), @@ -343,6 +351,13 @@ impl ServeHandle { Ok(self) } + /// Ship guest spans and metrics. No tenancy requirement: an anonymous request is attributed + /// to the anonymous tenant, as its filesystem is. + pub fn with_telemetry(mut self, telemetry: crate::telemetry::Telemetry) -> Self { + self.telemetry = telemetry; + self + } + /// An internal component export, never an HTTP route or a fabricated token. /// One message from a held connection, as one guest invocation. /// @@ -386,6 +401,10 @@ impl ServeHandle { tenant, interactive: false, }); + store.data_mut().telemetry = Some( + self.telemetry + .context(tenant, crate::telemetry::Invocation::Message), + ); let run = instance .get_typed_func::<(String, String, Vec), (std::result::Result, String>,)>( &mut store, @@ -465,6 +484,10 @@ impl ServeHandle { tenant: task.tenant, interactive: false, }); + store.data_mut().telemetry = Some( + self.telemetry + .context(task.tenant, crate::telemetry::Invocation::Task), + ); let run = instance .get_typed_func::<(String, Vec), (std::result::Result, String>,)>( &mut store, "run-task", @@ -649,6 +672,10 @@ impl ServeHandle { tenant: tenant_id, interactive: true, }); + instance.store.data_mut().telemetry = Some( + self.telemetry + .context(tenant_id, crate::telemetry::Invocation::Request), + ); let progress = instance.progress.clone(); let (sender, receiver) = tokio::sync::oneshot::channel(); // Wrapped before it becomes a guest resource, which is the last point @@ -791,6 +818,10 @@ impl ServeHandle { // unauthenticated guest `/runtime` and every tenant. let (dir, _) = self.guest.zfs().tenant_dir(ANONYMOUS).await?; let mut store = Store::new(self.pre.engine(), self.guest.new_state(Some(&dir))?); + store.data_mut().telemetry = Some( + self.telemetry + .context(ANONYMOUS, crate::telemetry::Invocation::Request), + ); let progress = Arc::new(StreamProgress::new()); store.set_epoch_deadline(self.watchdog()); self.arm_watchdog(&mut store, progress.clone()); @@ -1623,9 +1654,11 @@ pub async fn serve_component( // wasmtime 44 enables async at the engine level when the `async` feature // is on; `Config::async_support` is a no-op. Same as `run_component`. let engine = ServeHandle::engine_with_watchdog()?; + let telemetry = std::mem::take(&mut config.telemetry); let mut handle = ServeHandle::new(&engine, component_bytes, guest)? .with_timeout(config.request_timeout) - .with_max_interaction(config.max_interaction); + .with_max_interaction(config.max_interaction) + .with_telemetry(telemetry.clone()); if let Some(tenancy) = &config.tenancy { handle = handle.with_tenancy(tenancy.clone()); } @@ -1795,6 +1828,8 @@ pub async fn serve_component( if let Some(forwarder) = notify_forwarder { forwarder.shutdown().await; } + // The same for the last invocations' traces and the metrics still aggregating. + telemetry.shutdown().await; served } diff --git a/runtime/src/state.rs b/runtime/src/state.rs index c397657..d33b48d 100644 --- a/runtime/src/state.rs +++ b/runtime/src/state.rs @@ -11,6 +11,9 @@ pub struct State { /// The connections the runtime holds for this tenant — see [`crate::stream`]. pub(crate) streams: Option, pub(crate) notify: Option, + /// This invocation's trace and where its metrics go — see [`crate::telemetry`]. Its trace is + /// shipped when this is dropped, which is when the store is. + pub(crate) telemetry: Option, wasi: WasiCtx, table: ResourceTable, http: wasmtime_wasi_http::WasiHttpCtx, @@ -25,12 +28,24 @@ impl State { tasks: None, streams: None, notify: None, + telemetry: None, wasi, table: ResourceTable::new(), http: wasmtime_wasi_http::WasiHttpCtx::new(), egress: crate::serve::GuestEgress, } } + + /// The telemetry context and the resource table at once. A guest's span is an entry in the + /// table that points into the context's trace, so every span call needs both halves. + pub(crate) fn telemetry_and_table( + &mut self, + ) -> ( + Option<&mut crate::telemetry::TelemetryContext>, + &mut ResourceTable, + ) { + (self.telemetry.as_mut(), &mut self.table) + } } impl WasiView for State { diff --git a/runtime/src/telemetry/metrics.rs b/runtime/src/telemetry/metrics.rs new file mode 100644 index 0000000..d90301b --- /dev/null +++ b/runtime/src/telemetry/metrics.rs @@ -0,0 +1,669 @@ +//! Guest metrics, aggregated in memory and published to CloudWatch with `PutMetricData`. +//! +//! Aggregated, not queued: a counter a guest bumps a thousand times between flushes is one datum, +//! and CloudWatch bills per request and per datum. Counters sum, gauges keep their last value, and +//! histograms keep each distinct value with its count — CloudWatch's own `values`/`counts` +//! distribution, so percentiles survive aggregation. +//! +//! A metric is checked when it is recorded, not when it is sent. CloudWatch refuses a whole +//! request over one bad datum — a value out of its range, a repeated dimension — so letting one +//! through would take every other guest metric in the same call with it. A refused metric is +//! dropped and counted, as the WIT promises. +//! +//! Series are capped per window, which bounds memory and bills alike: a guest that mints a new +//! dimension value per request gets its first [`MAX_SERIES`] and a count of the rest. + +use std::collections::hash_map::Entry; +use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant, SystemTime}; + +use async_trait::async_trait; +use tokio::sync::Notify; + +use super::{classify, warn_due, Dimension, SendError, Unit, FLUSH_DEADLINE, RETRY_BACKOFF}; + +/// How often aggregated metrics go to CloudWatch. Its standard resolution is a minute. +const FLUSH_INTERVAL: Duration = Duration::from_secs(30); +/// Distinct series (name, kind, unit and dimensions) per window. +const MAX_SERIES: usize = 1000; +/// Datums per `PutMetricData` call. +const MAX_DATUMS_PER_CALL: usize = 1000; +/// Distinct values in one histogram datum — CloudWatch's limit on `values`. +const MAX_DISTINCT_VALUES: usize = 150; +/// Dimensions per metric, as the WIT says. +const MAX_DIMENSIONS: usize = 10; +/// The magnitudes CloudWatch accepts; one value outside them fails the whole request. +const MAX_MAGNITUDE: f64 = 8.515920e+109; +const MIN_MAGNITUDE: f64 = 8.515920e-109; + +/// Where the CloudWatch client sends, and under which namespace. +#[derive(Clone, Debug)] +pub struct MetricsConfig { + pub namespace: String, + pub region: String, + /// A stand-in for the service, for an emulator. `None` is the real one. + pub endpoint: Option, +} + +impl MetricsConfig { + /// A namespace CloudWatch accepts for custom metrics: 1-255 printable ASCII characters, and + /// not under `AWS/`, which is reserved for AWS's own. + pub fn valid_namespace(namespace: &str) -> bool { + (1..=255).contains(&namespace.len()) + && namespace.bytes().all(|b| b.is_ascii_graphic() || b == b' ') + && !namespace.starts_with("AWS/") + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub(super) enum Kind { + Counter, + Gauge, + Histogram, +} + +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +struct Series { + name: String, + kind: Kind, + unit: Unit, + /// Sorted by name, so the same dimensions in any order are one series. + dimensions: Vec<(String, String)>, +} + +enum Aggregate { + Sum(f64), + Last(f64), + Spread(Vec<(f64, f64)>), +} + +/// One CloudWatch datum, before it becomes the SDK's type. Public as the destination's seam. +#[derive(Clone, Debug, PartialEq)] +pub struct Datum { + pub name: String, + pub unit: Unit, + pub dimensions: Vec<(String, String)>, + pub value: DatumValue, +} + +#[derive(Clone, Debug, PartialEq)] +pub enum DatumValue { + Single(f64), + Spread { values: Vec, counts: Vec }, +} + +fn datum(series: &Series, aggregate: Aggregate) -> Datum { + Datum { + name: series.name.clone(), + unit: series.unit, + dimensions: series.dimensions.clone(), + value: match aggregate { + Aggregate::Sum(v) | Aggregate::Last(v) => DatumValue::Single(v), + Aggregate::Spread(points) => { + let (values, counts) = points.into_iter().unzip(); + DatumValue::Spread { values, counts } + } + }, + } +} + +#[derive(Default)] +struct Window { + series: HashMap, + /// Histogram datums sealed early because they hit [`MAX_DISTINCT_VALUES`]. + sealed: Vec, +} + +impl Window { + /// Add one observation. `false` when the series is new and the window already holds + /// [`MAX_SERIES`]. + fn add(&mut self, series: Series, value: f64) -> bool { + if self.series.len() >= MAX_SERIES && !self.series.contains_key(&series) { + return false; + } + match self.series.entry(series) { + Entry::Vacant(entry) => { + let aggregate = match entry.key().kind { + Kind::Counter => Aggregate::Sum(value), + Kind::Gauge => Aggregate::Last(value), + Kind::Histogram => Aggregate::Spread(vec![(value, 1.0)]), + }; + entry.insert(aggregate); + } + Entry::Occupied(mut entry) => { + let full = match entry.get_mut() { + Aggregate::Sum(sum) => { + *sum += value; + None + } + Aggregate::Last(last) => { + *last = value; + None + } + Aggregate::Spread(points) => { + if let Some(point) = points.iter_mut().find(|p| p.0 == value) { + point.1 += 1.0; + None + } else if points.len() < MAX_DISTINCT_VALUES { + points.push((value, 1.0)); + None + } else { + Some(std::mem::replace(points, vec![(value, 1.0)])) + } + } + }; + if let Some(points) = full { + self.sealed + .push(datum(entry.key(), Aggregate::Spread(points))); + } + } + } + true + } + + fn take(&mut self) -> Vec { + let mut data = std::mem::take(&mut self.sealed); + data.extend( + self.series + .drain() + .map(|(series, aggregate)| datum(&series, aggregate)), + ); + data + } +} + +/// Where flushed metrics go. A trait so aggregation and retries can be tested without HTTP. +#[async_trait] +pub trait MetricDestination: Send + Sync { + async fn put( + &self, + namespace: &str, + data: Vec, + timestamp: SystemTime, + ) -> Result<(), SendError>; +} + +struct Shared { + window: Mutex, + /// Metrics refused when recorded, past the series cap, or lost in delivery. + dropped: AtomicU64, +} + +impl Shared { + fn window(&self) -> std::sync::MutexGuard<'_, Window> { + self.window + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } +} + +/// Aggregates guest metrics and publishes them, on a task of its own. +pub struct Metrics { + shared: Arc, + stop: Arc, + task: Mutex>>, +} + +impl Metrics { + /// Start publishing to `destination` under `namespace`. + pub fn start(destination: Arc, namespace: String) -> Arc { + let shared = Arc::new(Shared { + window: Mutex::new(Window::default()), + dropped: AtomicU64::new(0), + }); + let stop = Arc::new(Notify::new()); + let task = tokio::spawn(publish_loop( + destination, + namespace, + shared.clone(), + stop.clone(), + )); + Arc::new(Metrics { + shared, + stop, + task: Mutex::new(Some(task)), + }) + } + + /// Record one observation, or drop and count it if CloudWatch would refuse it. + pub(super) fn record( + &self, + kind: Kind, + name: String, + value: f64, + unit: Unit, + dimensions: Vec, + ) { + let added = match series(kind, name, unit, dimensions, value) { + Some(series) => self.shared.window().add(series, value), + None => false, + }; + if !added { + self.shared.dropped.fetch_add(1, Ordering::Relaxed); + } + } + + /// Metrics dropped so far, refused or lost. + pub fn dropped(&self) -> u64 { + self.shared.dropped.load(Ordering::Relaxed) + } + + /// Publish what has been aggregated and stop, within [`FLUSH_DEADLINE`]. + pub async fn shutdown(&self) { + self.stop.notify_one(); + let task = self + .task + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .take(); + if let Some(task) = task { + let _ = tokio::time::timeout(FLUSH_DEADLINE, task).await; + } + } +} + +/// The series an observation belongs to, or `None` if CloudWatch would refuse it. +fn series( + kind: Kind, + name: String, + unit: Unit, + dimensions: Vec, + value: f64, +) -> Option { + if !valid_metric_name(&name) || !in_range(value) || dimensions.len() > MAX_DIMENSIONS { + return None; + } + let mut dimensions: Vec<(String, String)> = + dimensions.into_iter().map(|d| (d.name, d.value)).collect(); + if !dimensions + .iter() + .all(|(n, v)| valid_dimension_part(n) && valid_dimension_part(v)) + { + return None; + } + dimensions.sort(); + // CloudWatch refuses a datum that names one dimension twice. + if dimensions.windows(2).any(|pair| pair[0].0 == pair[1].0) { + return None; + } + Some(Series { + name, + kind, + unit, + dimensions, + }) +} + +/// 1-255 printable ASCII characters, not all spaces. +fn valid_metric_name(name: &str) -> bool { + (1..=255).contains(&name.len()) + && name.bytes().all(|b| b.is_ascii_graphic() || b == b' ') + && name.bytes().any(|b| b != b' ') +} + +/// What the WIT promises for a dimension's name and value: 1-64 of `[A-Za-z0-9_.-]`. +fn valid_dimension_part(part: &str) -> bool { + (1..=64).contains(&part.len()) + && part + .bytes() + .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'_' | b'.' | b'-')) +} + +fn in_range(value: f64) -> bool { + value.is_finite() && (value == 0.0 || (MIN_MAGNITUDE..=MAX_MAGNITUDE).contains(&value.abs())) +} + +async fn publish_loop( + destination: Arc, + namespace: String, + shared: Arc, + stop: Arc, +) { + let mut tick = tokio::time::interval(FLUSH_INTERVAL); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + // The first tick is immediate, and there is nothing to send yet. + tick.tick().await; + let mut warned = None; + let mut reported = 0; + loop { + let last = tokio::select! { + _ = tick.tick() => false, + _ = stop.notified() => true, + }; + let data = shared.window().take(); + publish( + &*destination, + &namespace, + data, + &shared.dropped, + &mut warned, + ) + .await; + let dropped = shared.dropped.load(Ordering::Relaxed); + if dropped > reported && warn_due(&mut warned) { + tracing::warn!( + total = dropped, + "guest metrics dropped: refused as CloudWatch would refuse them, past the series \ + cap, or not delivered" + ); + reported = dropped; + } + if last { + return; + } + } +} + +async fn publish( + destination: &dyn MetricDestination, + namespace: &str, + data: Vec, + dropped: &AtomicU64, + warned: &mut Option, +) { + let timestamp = SystemTime::now(); + for chunk in data.chunks(MAX_DATUMS_PER_CALL) { + let mut tries = RETRY_BACKOFF.iter(); + loop { + match destination.put(namespace, chunk.to_vec(), timestamp).await { + Ok(()) => break, + Err(SendError::Definitive(e)) => { + dropped.fetch_add(chunk.len() as u64, Ordering::Relaxed); + if warn_due(warned) { + tracing::warn!( + error = %e, + "CloudWatch refused guest metrics; the instance role needs \ + cloudwatch:PutMetricData" + ); + } + break; + } + Err(SendError::Transient(e)) => match tries.next() { + Some(wait) => tokio::time::sleep(*wait).await, + None => { + dropped.fetch_add(chunk.len() as u64, Ordering::Relaxed); + if warn_due(warned) { + tracing::warn!(error = %e, "could not reach CloudWatch; guest metrics dropped"); + } + break; + } + }, + } + } + } +} + +/// The real service, through the SDK. +pub struct CloudWatchMetrics { + client: aws_sdk_cloudwatch::Client, +} + +impl CloudWatchMetrics { + /// Built as [`super::XrayDestination::connect`] is, and for the same reasons. + pub async fn connect(config: &MetricsConfig) -> Self { + use aws_sdk_cloudwatch::config::{ + retry::RetryConfig, timeout::TimeoutConfig, BehaviorVersion, Region, + }; + let mut builder = aws_sdk_cloudwatch::Config::builder() + .behavior_version(BehaviorVersion::latest()) + .region(Region::new(config.region.clone())) + .credentials_provider( + aws_config::default_provider::credentials::DefaultCredentialsChain::builder() + .region(Region::new(config.region.clone())) + .build() + .await, + ) + .retry_config(RetryConfig::disabled()) + .timeout_config( + TimeoutConfig::builder() + .operation_timeout(Duration::from_secs(3)) + .connect_timeout(Duration::from_secs(3)) + .build(), + ); + if let Some(endpoint) = &config.endpoint { + builder = builder.endpoint_url(endpoint); + } + CloudWatchMetrics { + client: aws_sdk_cloudwatch::Client::from_conf(builder.build()), + } + } +} + +#[async_trait] +impl MetricDestination for CloudWatchMetrics { + async fn put( + &self, + namespace: &str, + data: Vec, + timestamp: SystemTime, + ) -> Result<(), SendError> { + use aws_sdk_cloudwatch::primitives::DateTime; + use aws_sdk_cloudwatch::types::{Dimension, MetricDatum, StandardUnit}; + let at = DateTime::from(timestamp); + let data = data + .into_iter() + .map(|d| { + let unit = match d.unit { + Unit::None => StandardUnit::None, + Unit::Count => StandardUnit::Count, + Unit::Seconds => StandardUnit::Seconds, + Unit::Milliseconds => StandardUnit::Milliseconds, + Unit::Bytes => StandardUnit::Bytes, + Unit::Kilobytes => StandardUnit::Kilobytes, + Unit::Percent => StandardUnit::Percent, + }; + let mut datum = MetricDatum::builder() + .metric_name(d.name) + .unit(unit) + .timestamp(at); + for (name, value) in d.dimensions { + datum = datum.dimensions(Dimension::builder().name(name).value(value).build()); + } + match d.value { + DatumValue::Single(value) => datum.value(value), + DatumValue::Spread { values, counts } => { + datum.set_values(Some(values)).set_counts(Some(counts)) + } + } + .build() + }) + .collect(); + self.client + .put_metric_data() + .namespace(namespace) + .set_metric_data(Some(data)) + .send() + .await + .map_err(classify)?; + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn dim(name: &str, value: &str) -> Dimension { + Dimension { + name: name.into(), + value: value.into(), + } + } + + fn take(metrics: &Metrics) -> Vec { + let mut data = metrics.shared.window().take(); + data.sort_by(|a, b| a.name.cmp(&b.name)); + data + } + + /// Records every call; fails transiently `fail` times first. + struct Recorder { + calls: Mutex)>>, + fail: AtomicU64, + } + + #[async_trait] + impl MetricDestination for Recorder { + async fn put( + &self, + namespace: &str, + data: Vec, + _: SystemTime, + ) -> Result<(), SendError> { + self.calls + .lock() + .unwrap() + .push((namespace.to_string(), data)); + if self.fail.load(Ordering::SeqCst) > 0 { + self.fail.fetch_sub(1, Ordering::SeqCst); + return Err(SendError::Transient("later".into())); + } + Ok(()) + } + } + + fn recorder(fail: u64) -> Arc { + Arc::new(Recorder { + calls: Mutex::new(Vec::new()), + fail: AtomicU64::new(fail), + }) + } + + #[tokio::test] + async fn counters_sum_gauges_keep_the_last_value_and_histograms_count_each_value() { + let metrics = Metrics::start(recorder(0), "ns".into()); + metrics.record(Kind::Counter, "c".into(), 2.0, Unit::Count, vec![]); + metrics.record(Kind::Counter, "c".into(), 3.0, Unit::Count, vec![]); + metrics.record(Kind::Gauge, "g".into(), 1.0, Unit::Percent, vec![]); + metrics.record(Kind::Gauge, "g".into(), 7.0, Unit::Percent, vec![]); + for v in [5.0, 5.0, 9.0] { + metrics.record(Kind::Histogram, "h".into(), v, Unit::Milliseconds, vec![]); + } + let data = take(&metrics); + assert_eq!(data[0].value, DatumValue::Single(5.0)); + assert_eq!(data[1].value, DatumValue::Single(7.0)); + assert_eq!(data[1].unit, Unit::Percent); + let DatumValue::Spread { values, counts } = &data[2].value else { + panic!("a histogram is a distribution") + }; + let mut points: Vec<(f64, f64)> = + values.iter().copied().zip(counts.iter().copied()).collect(); + points.sort_by(|a, b| a.0.total_cmp(&b.0)); + assert_eq!(points, [(5.0, 2.0), (9.0, 1.0)]); + assert_eq!(metrics.dropped(), 0); + } + + #[tokio::test] + async fn the_same_dimensions_in_any_order_are_one_series() { + let metrics = Metrics::start(recorder(0), "ns".into()); + metrics.record( + Kind::Counter, + "c".into(), + 1.0, + Unit::Count, + vec![dim("a", "1"), dim("b", "2")], + ); + metrics.record( + Kind::Counter, + "c".into(), + 1.0, + Unit::Count, + vec![dim("b", "2"), dim("a", "1")], + ); + let data = take(&metrics); + assert_eq!(data.len(), 1); + assert_eq!(data[0].value, DatumValue::Single(2.0)); + } + + #[tokio::test] + async fn what_cloudwatch_would_refuse_is_dropped_and_counted() { + let metrics = Metrics::start(recorder(0), "ns".into()); + let refused: Vec<(String, f64, Vec)> = vec![ + ("".into(), 1.0, vec![]), + ("ok".into(), f64::NAN, vec![]), + ("ok".into(), f64::INFINITY, vec![]), + ("ok".into(), 1e200, vec![]), + ("ok".into(), 1e-200, vec![]), + ("ok".into(), 1.0, vec![dim("a", "1"), dim("a", "2")]), + ("ok".into(), 1.0, vec![dim("has space", "1")]), + ( + "ok".into(), + 1.0, + (0..11).map(|i| dim(&format!("d{i}"), "v")).collect(), + ), + ]; + let count = refused.len() as u64; + for (name, value, dimensions) in refused { + metrics.record(Kind::Gauge, name, value, Unit::None, dimensions); + } + metrics.record(Kind::Gauge, "zero is fine".into(), 0.0, Unit::None, vec![]); + assert_eq!(metrics.dropped(), count); + assert_eq!(take(&metrics).len(), 1); + } + + #[tokio::test] + async fn a_histogram_past_the_distinct_value_limit_is_sealed_not_lost() { + let metrics = Metrics::start(recorder(0), "ns".into()); + for v in 0..(MAX_DISTINCT_VALUES + 10) { + metrics.record(Kind::Histogram, "h".into(), v as f64, Unit::None, vec![]); + } + let data = take(&metrics); + assert_eq!(data.len(), 2); + let observations: usize = data + .iter() + .map(|d| match &d.value { + DatumValue::Spread { values, .. } => values.len(), + DatumValue::Single(_) => 0, + }) + .sum(); + assert_eq!(observations, MAX_DISTINCT_VALUES + 10); + } + + #[tokio::test] + async fn new_series_past_the_cap_are_dropped_while_existing_ones_still_count() { + let metrics = Metrics::start(recorder(0), "ns".into()); + for i in 0..MAX_SERIES { + metrics.record(Kind::Counter, format!("c{i}"), 1.0, Unit::Count, vec![]); + } + metrics.record( + Kind::Counter, + "one too many".into(), + 1.0, + Unit::Count, + vec![], + ); + metrics.record(Kind::Counter, "c0".into(), 1.0, Unit::Count, vec![]); + assert_eq!(metrics.dropped(), 1); + let data = take(&metrics); + assert_eq!(data.len(), MAX_SERIES); + } + + #[tokio::test(start_paused = true)] + async fn metrics_are_published_on_the_interval_and_at_shutdown_with_a_retry() { + let destination = recorder(1); + let metrics = Metrics::start(destination.clone(), "Merlin/Guest".into()); + metrics.record(Kind::Counter, "c".into(), 1.0, Unit::Count, vec![]); + tokio::time::sleep(FLUSH_INTERVAL + Duration::from_secs(5)).await; + { + let calls = destination.calls.lock().unwrap(); + assert_eq!(calls.len(), 2, "one transient failure, then the retry"); + assert_eq!(calls[1].0, "Merlin/Guest"); + assert_eq!(calls[1].1.len(), 1); + } + metrics.record(Kind::Counter, "c".into(), 4.0, Unit::Count, vec![]); + metrics.shutdown().await; + let calls = destination.calls.lock().unwrap(); + assert_eq!(calls.last().unwrap().1[0].value, DatumValue::Single(4.0)); + assert_eq!(metrics.dropped(), 0); + } + + #[test] + fn namespaces_cloudwatch_reserves_or_refuses_are_rejected() { + assert!(MetricsConfig::valid_namespace("Merlin/Guest")); + assert!(!MetricsConfig::valid_namespace("")); + assert!(!MetricsConfig::valid_namespace("AWS/EC2")); + assert!(!MetricsConfig::valid_namespace("tab\there")); + } +} diff --git a/runtime/src/telemetry/mod.rs b/runtime/src/telemetry/mod.rs new file mode 100644 index 0000000..1338ce7 --- /dev/null +++ b/runtime/src/telemetry/mod.rs @@ -0,0 +1,484 @@ +//! Telemetry a guest reports: spans and events to AWS X-Ray, metrics to CloudWatch. +//! +//! The ABI is `wit/telemetry/telemetry.wit`. Like every interface the runtime adds, nothing in it +//! takes a tenant id: the serve layer puts a [`TelemetryContext`] on the store before the guest +//! runs, carrying the tenant it has already authenticated. +//! +//! # One trace per invocation, shipped when the store goes +//! +//! A context opens a trace when it is made and hands it to X-Ray when it is dropped — which is +//! when the store holding it is, at the end of the invocation however that ends, a timeout or a +//! trap included. No call site has to remember to flush, so none can forget. +//! +//! # Never in the guest's way +//! +//! Every host call here is synchronous and does no I/O. A span is a node in an in-memory tree, a +//! metric is an entry in a map, and a finished trace goes to a bounded queue by `try_send`. The +//! network belongs to the forwarders ([`xray`], [`metrics`]) on tasks of their own. Telemetry that +//! cannot keep up is dropped and counted, never waited for — the guest-log pipeline's rule (see +//! [`crate::guest_io`]), for the same reason. +//! +//! With X-Ray or metrics off the calls still succeed; they do less. Every event and closed span is +//! also mirrored to the runtime's console, so an emulator with neither still shows what a guest +//! reported. +//! +//! # What leaves the enclave +//! +//! What a guest puts in a span name, a message, a field or a metric is its own business, exactly +//! as its stdout is. The runtime adds only what it already knows: the tenant (an X-Ray annotation, +//! never a metric dimension — CloudWatch bills per unique combination), the kind of invocation, +//! the image's PCR0, the region and its own version. + +mod metrics; +mod xray; + +pub use metrics::{ + CloudWatchMetrics, Datum, DatumValue, MetricDestination, Metrics, MetricsConfig, +}; +pub use xray::{TraceDestination, Tracer, XrayConfig, XrayDestination}; + +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use wasmtime::component::{ComponentType, Lift, Linker, Resource, ResourceType}; + +use crate::state::State; + +/// Waits between tries for a failure that may heal, shared by both forwarders. Telemetry is best +/// effort: after these it is dropped and counted rather than held. +const RETRY_BACKOFF: [Duration; 2] = [Duration::from_millis(500), Duration::from_secs(2)]; + +/// At most one warning per this, however much is being dropped or refused. +const WARN_INTERVAL: Duration = Duration::from_secs(60); + +/// How long shutdown waits for the last batch to go. +const FLUSH_DEADLINE: Duration = Duration::from_secs(5); + +/// The console target guest telemetry is mirrored under — the guest-log one, so one filter +/// governs everything a guest says. +const GUEST_TARGET: &str = "guest"; + +// --- The ABI's types --------------------------------------------------------------------------- + +/// The WIT `level`. +#[derive(ComponentType, Lift, Clone, Copy, Debug, PartialEq, Eq)] +#[component(enum)] +#[repr(u8)] +pub enum Level { + #[component(name = "trace")] + Trace, + #[component(name = "debug")] + Debug, + #[component(name = "info")] + Info, + #[component(name = "warn")] + Warn, + #[component(name = "error")] + Error, +} + +impl Level { + pub fn as_str(self) -> &'static str { + match self { + Level::Trace => "trace", + Level::Debug => "debug", + Level::Info => "info", + Level::Warn => "warn", + Level::Error => "error", + } + } +} + +/// The WIT `field`. +#[derive(ComponentType, Lift, Clone, Debug, PartialEq, Eq)] +#[component(record)] +pub struct Field { + #[component(name = "key")] + pub key: String, + #[component(name = "value")] + pub value: String, +} + +/// The WIT `unit`. +#[derive(ComponentType, Lift, Clone, Copy, Debug, PartialEq, Eq, Hash)] +#[component(enum)] +#[repr(u8)] +pub enum Unit { + #[component(name = "none")] + None, + #[component(name = "count")] + Count, + #[component(name = "seconds")] + Seconds, + #[component(name = "milliseconds")] + Milliseconds, + #[component(name = "bytes")] + Bytes, + #[component(name = "kilobytes")] + Kilobytes, + #[component(name = "percent")] + Percent, +} + +/// The WIT `dimension`. +#[derive(ComponentType, Lift, Clone, Debug, PartialEq, Eq)] +#[component(record)] +pub struct Dimension { + #[component(name = "name")] + pub name: String, + #[component(name = "value")] + pub value: String, +} + +/// What a guest's `span` handle refers to: a node in its invocation's trace, or nothing when the +/// span was not recorded (no context, or the trace was already full). Every method on a span that +/// refers to nothing is accepted and does nothing. +pub struct SpanHandle { + node: Option, +} + +// --- The process-wide half --------------------------------------------------------------------- + +/// What started an invocation, recorded on its trace. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Invocation { + /// An inbound request. + Request, + /// A background task run. + Task, + /// A message on a held connection. + Message, +} + +impl Invocation { + fn as_str(self) -> &'static str { + match self { + Invocation::Request => "request", + Invocation::Task => "task", + Invocation::Message => "message", + } + } +} + +/// Where traces and metrics go, for the whole process. Cheap to clone; the default sends nowhere. +#[derive(Clone, Default)] +pub struct Telemetry { + tracer: Option>, + metrics: Option>, +} + +impl std::fmt::Debug for Telemetry { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Telemetry") + .field("xray", &self.tracer.is_some()) + .field("metrics", &self.metrics.is_some()) + .finish() + } +} + +impl Telemetry { + pub fn new(tracer: Option>, metrics: Option>) -> Self { + Telemetry { tracer, metrics } + } + + /// The context for one invocation, to put on its store before the guest runs. + pub fn context(&self, tenant: [u8; 16], invocation: Invocation) -> TelemetryContext { + let tenant = hex::encode(tenant); + TelemetryContext { + trace: xray::Trace::new(tenant.clone(), invocation), + tenant, + tracer: self.tracer.clone(), + metrics: self.metrics.clone(), + } + } + + /// Flush what is queued and stop. Each part is bounded by its own deadline, so telemetry that + /// will not go can never be what keeps an enclave from stopping. + pub async fn shutdown(&self) { + if let Some(tracer) = &self.tracer { + tracer.shutdown().await; + } + if let Some(metrics) = &self.metrics { + metrics.shutdown().await; + } + } +} + +// --- The per-invocation half ------------------------------------------------------------------- + +/// One invocation's telemetry: its trace, and where that trace and its metrics go. +pub struct TelemetryContext { + tenant: String, + trace: xray::Trace, + tracer: Option>, + metrics: Option>, +} + +impl TelemetryContext { + fn event(&mut self, on: Option, level: Level, message: String, fields: Vec) { + console( + level, + &self.tenant, + "guest event", + None, + None, + &message, + &fields, + ); + self.trace.event(on, level, message, fields); + } + + fn open(&mut self, level: Level, name: String, fields: Vec) -> Option { + self.trace.open(level, name, fields) + } + + fn set_field(&mut self, node: usize, key: String, value: String) { + self.trace.set_field(node, key, value); + } + + fn close(&mut self, node: usize) { + if let Some((level, name, seconds)) = self.trace.close(node) { + console( + level, + &self.tenant, + "guest span", + Some(&name), + Some(seconds), + "", + &[], + ); + } + } + + fn record( + &self, + kind: metrics::Kind, + name: String, + value: f64, + unit: Unit, + dimensions: Vec, + ) { + if let Some(metrics) = &self.metrics { + metrics.record(kind, name, value, unit, dimensions); + } + } +} + +impl Drop for TelemetryContext { + /// The end of the invocation: ship its trace. Never blocks — see [`Tracer::submit`]. + fn drop(&mut self) { + if let Some(tracer) = &self.tracer { + let document = self.trace.finish(tracer.image(), tracer.region()); + tracer.submit(document); + } + } +} + +/// Mirror one report to the runtime's console. +/// +/// The guest's text is only ever a quoted **value**, never the message — the rule +/// [`crate::guest_io::TracingLogSink`] follows, for the same reason: a guest that can choose the +/// message can forge a line that looks like the runtime's own. +fn console( + level: Level, + tenant: &str, + what: &'static str, + span: Option<&str>, + seconds: Option, + message: &str, + fields: &[Field], +) { + let fields: Vec<(&str, &str)> = fields + .iter() + .map(|f| (f.key.as_str(), f.value.as_str())) + .collect(); + let duration_ms = seconds.map(|s| (s * 1000.0).round() as u64); + macro_rules! at { + ($level:ident) => { + tracing::$level!( + target: GUEST_TARGET, + tenant, + span = ?span, + duration_ms = ?duration_ms, + guest_message = ?message, + guest_fields = ?fields, + "{what}" + ) + }; + } + match level { + Level::Trace => at!(trace), + Level::Debug => at!(debug), + Level::Info => at!(info), + Level::Warn => at!(warn), + Level::Error => at!(error), + } +} + +// --- Delivery, shared by both forwarders ------------------------------------------------------- + +/// Why a call to X-Ray or CloudWatch failed, and whether trying again could help. +#[derive(Debug)] +pub enum SendError { + /// Throttling, a 5xx, a broken connection, an expired session token: worth another go. + Transient(String), + /// Refused for good: the role may not write here, or the request itself is malformed. + Definitive(String), +} + +impl std::fmt::Display for SendError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + SendError::Transient(e) | SendError::Definitive(e) => f.write_str(e), + } + } +} + +/// Sort an SDK failure, the way [`crate::guest_io::cloudwatch`] does for logs. +/// +/// Credential errors are deliberately transient: an `UnrecognizedClientException` is usually an +/// expired session token, which heals the moment the SDK refreshes it from IMDS. +fn classify(error: aws_smithy_runtime_api::client::result::SdkError) -> SendError +where + E: aws_smithy_types::error::metadata::ProvideErrorMetadata + std::error::Error + 'static, + R: std::fmt::Debug, +{ + use aws_smithy_types::error::metadata::ProvideErrorMetadata as _; + let code = error.code().map(str::to_string); + let message = format!( + "{}: {}", + code.as_deref().unwrap_or("unknown"), + aws_smithy_types::error::display::DisplayErrorContext(&error) + ); + match code.as_deref() { + Some( + "AccessDeniedException" + | "AccessDenied" + | "InvalidParameterValue" + | "InvalidParameterCombination" + | "MissingParameter" + | "InvalidRequestException", + ) => SendError::Definitive(message), + _ => SendError::Transient(message), + } +} + +/// Whether a rate-limited warning may go out now. +fn warn_due(last: &mut Option) -> bool { + let now = Instant::now(); + if last.is_some_and(|t| now.duration_since(t) < WARN_INTERVAL) { + return false; + } + *last = Some(now); + true +} + +// --- The linker -------------------------------------------------------------------------------- + +/// The ABI is defined in wit/telemetry/telemetry.wit. No tenant id is accepted. +/// +/// Synchronous host functions: nothing here waits on anything, so there is no future to box. +pub fn add_to_linker(linker: &mut Linker) -> wasmtime::Result<()> { + let mut tracing = linker.instance("enclave:telemetry/tracing@0.1.0")?; + + // Dropping the handle closes the span, if `end` has not already. + tracing.resource( + "span", + ResourceType::host::(), + |mut store, rep| { + let (ctx, table) = store.data_mut().telemetry_and_table(); + let handle = table.delete(Resource::::new_own(rep))?; + if let (Some(ctx), Some(node)) = (ctx, handle.node) { + ctx.close(node); + } + Ok(()) + }, + )?; + + tracing.func_wrap( + "event", + |mut store, (level, message, fields): (Level, String, Vec)| { + if let Some(ctx) = store.data_mut().telemetry.as_mut() { + ctx.event(None, level, message, fields); + } + Ok(()) + }, + )?; + + tracing.func_wrap( + "[constructor]span", + |mut store, (level, name, fields): (Level, String, Vec)| { + let (ctx, table) = store.data_mut().telemetry_and_table(); + let node = ctx.and_then(|ctx| ctx.open(level, name, fields)); + Ok((table.push(SpanHandle { node })?,)) + }, + )?; + + tracing.func_wrap( + "[method]span.set-field", + |mut store, (span, key, value): (Resource, String, String)| { + let (ctx, table) = store.data_mut().telemetry_and_table(); + if let (Some(ctx), Some(node)) = (ctx, table.get(&span)?.node) { + ctx.set_field(node, key, value); + } + Ok(()) + }, + )?; + + tracing.func_wrap( + "[method]span.add-event", + |mut store, + (span, level, message, fields): (Resource, Level, String, Vec)| { + let (ctx, table) = store.data_mut().telemetry_and_table(); + if let (Some(ctx), Some(node)) = (ctx, table.get(&span)?.node) { + ctx.event(Some(node), level, message, fields); + } + Ok(()) + }, + )?; + + tracing.func_wrap( + "[method]span.end", + |mut store, (span,): (Resource,)| { + let (ctx, table) = store.data_mut().telemetry_and_table(); + if let (Some(ctx), Some(node)) = (ctx, table.get(&span)?.node) { + ctx.close(node); + } + Ok(()) + }, + )?; + + let mut metrics = linker.instance("enclave:telemetry/metrics@0.1.0")?; + + metrics.func_wrap( + "counter-add", + |store, (name, value, dimensions): (String, f64, Vec)| { + if let Some(ctx) = store.data().telemetry.as_ref() { + ctx.record(metrics::Kind::Counter, name, value, Unit::Count, dimensions); + } + Ok(()) + }, + )?; + + metrics.func_wrap( + "gauge-set", + |store, (name, value, unit, dimensions): (String, f64, Unit, Vec)| { + if let Some(ctx) = store.data().telemetry.as_ref() { + ctx.record(metrics::Kind::Gauge, name, value, unit, dimensions); + } + Ok(()) + }, + )?; + + metrics.func_wrap( + "histogram-record", + |store, (name, value, unit, dimensions): (String, f64, Unit, Vec)| { + if let Some(ctx) = store.data().telemetry.as_ref() { + ctx.record(metrics::Kind::Histogram, name, value, unit, dimensions); + } + Ok(()) + }, + )?; + + Ok(()) +} diff --git a/runtime/src/telemetry/xray.rs b/runtime/src/telemetry/xray.rs new file mode 100644 index 0000000..7cfdd2e --- /dev/null +++ b/runtime/src/telemetry/xray.rs @@ -0,0 +1,899 @@ +//! Guest spans as AWS X-Ray traces: the segment tree for one invocation, and the forwarder that +//! ships finished ones with `PutTraceSegments`. +//! +//! The forwarder is the guest-log forwarder in miniature (see [`crate::guest_io::cloudwatch`]): a +//! bounded queue a guest call never waits on, batches sealed by size or by a linger timer, a few +//! backed-off retries for what may heal, and drops counted and reported in aggregate rather than +//! one line each. +//! +//! # One document per invocation +//! +//! The invocation is the root segment; every span a guest opens is a subsegment of whichever span +//! was open when it was created, so spans nest by the order a guest opens them. Events have no +//! type of their own in X-Ray and ride in the (sub)segment's `metadata`, with the span's fields. +//! The tenant, the kind of invocation and a span's level are **annotations**, which X-Ray indexes — +//! so "every trace for this tenant" or "every error span" is a filter expression, not a scan. +//! +//! X-Ray caps a document at 64 KiB. The caps below keep a trace near that by construction, and +//! [`Trace::finish`] sheds detail — events and fields first, then the subsegments — rather than +//! send something X-Ray would refuse whole. + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use async_trait::async_trait; +use serde_json::{json, Map, Value}; +use tokio::sync::{mpsc, Notify}; + +use super::{ + classify, warn_due, Field, Invocation, Level, SendError, FLUSH_DEADLINE, RETRY_BACKOFF, +}; + +/// X-Ray's ceiling on one segment document. +const MAX_DOCUMENT_BYTES: usize = 64 * 1024; +/// Spans one invocation may open; past this they are counted, not recorded. +const MAX_SPANS: usize = 256; +/// Events one span (or the invocation itself) may carry. +const MAX_EVENTS: usize = 64; +/// Fields one span or event may carry. +const MAX_FIELDS: usize = 32; +/// A field key, in characters. +const MAX_KEY_CHARS: usize = 64; +/// A message or field value, in characters. +const MAX_TEXT_CHARS: usize = 1024; +/// X-Ray's limit on a (sub)segment name. +const MAX_NAME_CHARS: usize = 200; + +/// Every root segment's name, so X-Ray's service map shows the guest as one service. What kind of +/// invocation a trace is lives in an annotation. +const SERVICE_NAME: &str = "enclave-guest"; + +/// Finished traces waiting to be shipped. Past this, new ones are dropped and counted. +const QUEUE_CAPACITY: usize = 1024; +/// Documents per `PutTraceSegments` call: the batch the X-Ray daemon itself sends. +const MAX_BATCH_DOCUMENTS: usize = 50; +/// Bytes per call, well inside the request limit. +const MAX_BATCH_BYTES: usize = 1024 * 1024; +/// How long a partial batch waits for company. +const BATCH_LINGER: Duration = Duration::from_secs(1); + +/// Where the X-Ray client sends. +#[derive(Clone, Debug)] +pub struct XrayConfig { + pub region: String, + /// A stand-in for the service, for an emulator. `None` is the real one. + pub endpoint: Option, +} + +// --- The trace --------------------------------------------------------------------------------- + +/// How much of a trace to render; [`Trace::finish`] steps down until it fits. +#[derive(Clone, Copy, PartialEq, Eq)] +enum Detail { + /// Everything. + Full, + /// The span tree and its timings, without events or fields. + Structure, + /// The invocation alone. + Root, +} + +struct Node { + id: String, + name: String, + level: Level, + /// Index of the enclosing node; the root's is itself and never read. + parent: usize, + /// Epoch seconds, as X-Ray wants them. + start: f64, + end: Option, + fields: Vec<(String, String)>, + events: Vec, +} + +struct Event { + at: f64, + level: Level, + message: String, + fields: Vec<(String, String)>, +} + +/// One invocation's spans, as an X-Ray segment tree. +pub(super) struct Trace { + trace_id: String, + tenant: String, + invocation: Invocation, + /// The wall clock is read once, when the trace opens; every later time is that plus a + /// monotonic offset, so a clock step mid-invocation cannot put a span's end before its start. + opened_at: f64, + opened: Instant, + /// `nodes[0]` is the invocation's root segment. + nodes: Vec, + /// Spans still open, innermost last. Never holds the root. + open: Vec, + dropped_spans: u32, + dropped_events: u32, +} + +impl Trace { + pub(super) fn new(tenant: String, invocation: Invocation) -> Self { + let opened_at = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs_f64()) + .unwrap_or(0.0); + let root = Node { + id: random_hex(8), + name: SERVICE_NAME.to_string(), + level: Level::Info, + parent: 0, + start: opened_at, + end: None, + fields: Vec::new(), + events: Vec::new(), + }; + Trace { + // X-Ray's format: version, the start in epoch seconds as 8 hex digits, 96 random bits. + trace_id: format!("1-{:08x}-{}", opened_at as u64 as u32, random_hex(12)), + tenant, + invocation, + opened_at, + opened: Instant::now(), + nodes: vec![root], + open: Vec::new(), + dropped_spans: 0, + dropped_events: 0, + } + } + + fn now(&self) -> f64 { + self.opened_at + self.opened.elapsed().as_secs_f64() + } + + /// Open a span beneath the innermost open one. `None` when the trace is full. + pub(super) fn open(&mut self, level: Level, name: String, fields: Vec) -> Option { + if self.nodes.len() > MAX_SPANS { + self.dropped_spans += 1; + return None; + } + let node = Node { + id: random_hex(8), + name: span_name(&name), + level, + parent: self.open.last().copied().unwrap_or(0), + start: self.now(), + end: None, + fields: clip_fields(fields), + events: Vec::new(), + }; + self.nodes.push(node); + let index = self.nodes.len() - 1; + self.open.push(index); + Some(index) + } + + pub(super) fn set_field(&mut self, node: usize, key: String, value: String) { + if let Some(node) = self.nodes.get_mut(node) { + if node.fields.len() < MAX_FIELDS { + node.fields + .push((clip(key, MAX_KEY_CHARS), clip(value, MAX_TEXT_CHARS))); + } + } + } + + /// An event on `on`, or on the innermost open span, or on the invocation itself. + pub(super) fn event( + &mut self, + on: Option, + level: Level, + message: String, + fields: Vec, + ) { + let at = self.now(); + let target = on.or(self.open.last().copied()).unwrap_or(0); + let full = self + .nodes + .get(target) + .is_none_or(|n| n.events.len() >= MAX_EVENTS); + if full { + self.dropped_events += 1; + return; + } + self.nodes[target].events.push(Event { + at, + level, + message: clip(message, MAX_TEXT_CHARS), + fields: clip_fields(fields), + }); + } + + /// Close a span. `Some` the first time only, with its level, name and length in seconds. + pub(super) fn close(&mut self, node: usize) -> Option<(Level, String, f64)> { + if node == 0 { + return None; + } + let now = self.now(); + let span = self.nodes.get_mut(node)?; + if span.end.is_some() { + return None; + } + span.end = Some(now); + let closed = (span.level, span.name.clone(), now - span.start); + self.open.retain(|&i| i != node); + Some(closed) + } + + /// Close whatever is still open and render the trace as one segment document, shedding detail + /// until it fits under X-Ray's limit. + pub(super) fn finish(&mut self, image: Option<&str>, region: &str) -> String { + let now = self.now(); + for node in &mut self.nodes { + node.end.get_or_insert(now); + } + self.open.clear(); + let mut document = String::new(); + for detail in [Detail::Full, Detail::Structure, Detail::Root] { + document = self.render(image, region, detail).to_string(); + if document.len() <= MAX_DOCUMENT_BYTES { + break; + } + } + document + } + + fn render(&self, image: Option<&str>, region: &str, detail: Detail) -> Value { + let mut children: Vec> = vec![Vec::new(); self.nodes.len()]; + for (index, node) in self.nodes.iter().enumerate().skip(1) { + children[node.parent].push(index); + } + + let root = &self.nodes[0]; + let mut guest = Map::new(); + if detail == Detail::Full && !root.events.is_empty() { + guest.insert("events".into(), events_json(&root.events)); + } + if self.dropped_spans > 0 { + guest.insert("dropped_spans".into(), self.dropped_spans.into()); + } + if self.dropped_events > 0 { + guest.insert("dropped_events".into(), self.dropped_events.into()); + } + if detail != Detail::Full { + guest.insert("trimmed".into(), true.into()); + } + + let mut document = json!({ + "name": root.name, + "id": root.id, + "trace_id": self.trace_id, + "start_time": root.start, + "end_time": root.end.unwrap_or(root.start), + "annotations": { + "tenant": self.tenant, + "invocation": self.invocation.as_str(), + }, + "metadata": { + "enclave": { + "image": image, + "region": region, + "version": env!("CARGO_PKG_VERSION"), + }, + "guest": guest, + }, + }); + if detail != Detail::Root && !children[0].is_empty() { + document["subsegments"] = children[0] + .iter() + .map(|&child| self.subsegment(child, &children, detail)) + .collect(); + } + document + } + + fn subsegment(&self, index: usize, children: &[Vec], detail: Detail) -> Value { + let node = &self.nodes[index]; + let mut subsegment = json!({ + "id": node.id, + "name": node.name, + "start_time": node.start, + "end_time": node.end.unwrap_or(node.start), + "annotations": { "level": node.level.as_str() }, + }); + if detail == Detail::Full { + let mut guest = Map::new(); + if !node.fields.is_empty() { + guest.insert("fields".into(), fields_json(&node.fields)); + } + if !node.events.is_empty() { + guest.insert("events".into(), events_json(&node.events)); + } + if !guest.is_empty() { + subsegment["metadata"] = json!({ "guest": guest }); + } + } + if !children[index].is_empty() { + subsegment["subsegments"] = children[index] + .iter() + .map(|&child| self.subsegment(child, children, detail)) + .collect(); + } + subsegment + } +} + +fn fields_json(fields: &[(String, String)]) -> Value { + Value::Object( + fields + .iter() + .map(|(k, v)| (k.clone(), Value::String(v.clone()))) + .collect(), + ) +} + +fn events_json(events: &[Event]) -> Value { + events + .iter() + .map(|e| { + json!({ + "at": e.at, + "level": e.level.as_str(), + "message": e.message, + "fields": fields_json(&e.fields), + }) + }) + .collect() +} + +/// A name X-Ray accepts: letters, digits, whitespace and `_ . : / % & # = + \ - @`, at most 200. +fn span_name(name: &str) -> String { + let name: String = name + .chars() + .take(MAX_NAME_CHARS) + .map(|c| { + if c.is_alphanumeric() || c.is_whitespace() || "_.:/%&#=+\\-@".contains(c) { + c + } else { + '_' + } + }) + .collect(); + if name.trim().is_empty() { + "span".to_string() + } else { + name + } +} + +fn clip(mut text: String, chars: usize) -> String { + if let Some((cut, _)) = text.char_indices().nth(chars) { + text.truncate(cut); + } + text +} + +fn clip_fields(fields: Vec) -> Vec<(String, String)> { + fields + .into_iter() + .take(MAX_FIELDS) + .map(|f| (clip(f.key, MAX_KEY_CHARS), clip(f.value, MAX_TEXT_CHARS))) + .collect() +} + +/// An id X-Ray only needs to be unique within a trace — not a secret. +fn random_hex(bytes: usize) -> String { + let mut buf = vec![0u8; bytes]; + if getrandom::fill(&mut buf).is_err() { + static NEXT: AtomicU64 = AtomicU64::new(1); + let n = NEXT.fetch_add(1, Ordering::Relaxed).to_le_bytes(); + for (i, b) in buf.iter_mut().enumerate() { + *b = n[i % 8] ^ i as u8; + } + } + hex::encode(buf) +} + +// --- Shipping ---------------------------------------------------------------------------------- + +/// Where finished segment documents go. A trait so batching and retries can be tested without +/// HTTP, and so the SDK client is reachable from one place. +#[async_trait] +pub trait TraceDestination: Send + Sync { + /// Ship the documents. `Ok(n)` is how many X-Ray accepted the call but refused. + async fn put(&self, documents: Vec) -> Result; +} + +/// Ships finished traces, on a task of its own. +pub struct Tracer { + image: Option, + region: String, + queue: mpsc::Sender, + dropped: Arc, + stop: Arc, + task: Mutex>>, +} + +impl Tracer { + /// Start shipping to `destination`. `image` (PCR0, hex) and `region` go on every trace. + pub fn start( + destination: Arc, + image: Option, + region: String, + ) -> Arc { + let (queue, received) = mpsc::channel(QUEUE_CAPACITY); + let dropped = Arc::new(AtomicU64::new(0)); + let stop = Arc::new(Notify::new()); + let task = tokio::spawn(forward( + destination, + received, + stop.clone(), + dropped.clone(), + )); + Arc::new(Tracer { + image, + region, + queue, + dropped, + stop, + task: Mutex::new(Some(task)), + }) + } + + pub(super) fn image(&self) -> Option<&str> { + self.image.as_deref() + } + + pub(super) fn region(&self) -> &str { + &self.region + } + + /// Hand a finished document to the forwarder. `try_send`, never `send`: this runs as an + /// invocation's store is dropped, and waiting here is exactly the stall this design refuses. + pub(super) fn submit(&self, document: String) { + if self.queue.try_send(document).is_err() { + self.dropped.fetch_add(1, Ordering::Relaxed); + } + } + + /// Traces dropped so far, whether the queue was full or delivery failed. + pub fn dropped(&self) -> u64 { + self.dropped.load(Ordering::Relaxed) + } + + /// Ship what is queued and stop, within [`FLUSH_DEADLINE`]. + pub async fn shutdown(&self) { + self.stop.notify_one(); + let task = self + .task + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .take(); + if let Some(task) = task { + let _ = tokio::time::timeout(FLUSH_DEADLINE, task).await; + } + } +} + +#[derive(Default)] +struct Batch { + documents: Vec, + bytes: usize, +} + +impl Batch { + fn fits(&self, document: &str) -> bool { + self.documents.len() < MAX_BATCH_DOCUMENTS && self.bytes + document.len() <= MAX_BATCH_BYTES + } + + fn push(&mut self, document: String) { + self.bytes += document.len(); + self.documents.push(document); + } + + fn take(&mut self) -> Vec { + self.bytes = 0; + std::mem::take(&mut self.documents) + } +} + +async fn forward( + destination: Arc, + mut received: mpsc::Receiver, + stop: Arc, + dropped: Arc, +) { + let mut batch = Batch::default(); + let mut linger: Option = None; + let mut warned = None; + let mut reported = 0; + loop { + let expiry = async { + match linger { + Some(at) => tokio::time::sleep_until(at).await, + None => std::future::pending().await, + } + }; + tokio::select! { + document = received.recv() => { + // Every sender gone: nothing more can arrive. + let Some(document) = document else { break }; + if !batch.fits(&document) { + ship(&*destination, batch.take(), &dropped, &mut warned).await; + linger = None; + } + linger.get_or_insert_with(|| tokio::time::Instant::now() + BATCH_LINGER); + batch.push(document); + } + _ = expiry => { + ship(&*destination, batch.take(), &dropped, &mut warned).await; + linger = None; + } + _ = stop.notified() => { + // What is queued now is the last of it. + received.close(); + while let Ok(document) = received.try_recv() { + if !batch.fits(&document) { + ship(&*destination, batch.take(), &dropped, &mut warned).await; + } + batch.push(document); + } + break; + } + } + report(&dropped, &mut reported, &mut warned); + } + ship(&*destination, batch.take(), &dropped, &mut warned).await; + report(&dropped, &mut reported, &mut warned); +} + +/// One batch, with the retries [`RETRY_BACKOFF`] allows, then dropped and counted. +async fn ship( + destination: &dyn TraceDestination, + documents: Vec, + dropped: &AtomicU64, + warned: &mut Option, +) { + if documents.is_empty() { + return; + } + let count = documents.len() as u64; + let mut tries = RETRY_BACKOFF.iter(); + loop { + match destination.put(documents.clone()).await { + Ok(0) => return, + Ok(refused) => { + dropped.fetch_add(refused as u64, Ordering::Relaxed); + if warn_due(warned) { + tracing::warn!(refused, "X-Ray refused some guest trace segments"); + } + return; + } + Err(SendError::Definitive(e)) => { + dropped.fetch_add(count, Ordering::Relaxed); + if warn_due(warned) { + tracing::warn!( + error = %e, + "X-Ray refused guest traces; the instance role needs xray:PutTraceSegments" + ); + } + return; + } + Err(SendError::Transient(e)) => match tries.next() { + Some(wait) => tokio::time::sleep(*wait).await, + None => { + dropped.fetch_add(count, Ordering::Relaxed); + if warn_due(warned) { + tracing::warn!(error = %e, "could not reach X-Ray; guest traces dropped"); + } + return; + } + }, + } + } +} + +/// Say, at most once per interval, how many traces have been lost so far. +fn report(dropped: &AtomicU64, reported: &mut u64, warned: &mut Option) { + let now = dropped.load(Ordering::Relaxed); + if now > *reported && warn_due(warned) { + tracing::warn!(total = now, "guest traces dropped"); + *reported = now; + } +} + +/// The real service, through the SDK. +pub struct XrayDestination { + client: aws_sdk_xray::Client, +} + +impl XrayDestination { + /// Built the way the guest-log client is (see + /// [`crate::guest_io::cloudwatch::CloudWatchDestination::connect`]): the default credential + /// chain through gvproxy to the parent's role, the region given to the chain as well as the + /// client, the SDK's own retries off in favour of the forwarder's, and short timeouts. + pub async fn connect(config: &XrayConfig) -> Self { + use aws_sdk_xray::config::{ + retry::RetryConfig, timeout::TimeoutConfig, BehaviorVersion, Region, + }; + let mut builder = aws_sdk_xray::Config::builder() + .behavior_version(BehaviorVersion::latest()) + .region(Region::new(config.region.clone())) + .credentials_provider( + aws_config::default_provider::credentials::DefaultCredentialsChain::builder() + .region(Region::new(config.region.clone())) + .build() + .await, + ) + .retry_config(RetryConfig::disabled()) + .timeout_config( + TimeoutConfig::builder() + .operation_timeout(Duration::from_secs(3)) + .connect_timeout(Duration::from_secs(3)) + .build(), + ); + if let Some(endpoint) = &config.endpoint { + builder = builder.endpoint_url(endpoint); + } + XrayDestination { + client: aws_sdk_xray::Client::from_conf(builder.build()), + } + } +} + +#[async_trait] +impl TraceDestination for XrayDestination { + async fn put(&self, documents: Vec) -> Result { + let output = self + .client + .put_trace_segments() + .set_trace_segment_documents(Some(documents)) + .send() + .await + .map_err(classify)?; + Ok(output.unprocessed_trace_segments().len()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn field(key: &str, value: &str) -> Field { + Field { + key: key.into(), + value: value.into(), + } + } + + fn parse(document: &str) -> Value { + serde_json::from_str(document).expect("a segment document is JSON") + } + + #[test] + fn spans_nest_by_the_order_they_are_opened() { + let mut trace = Trace::new("aa".into(), Invocation::Request); + let outer = trace.open(Level::Info, "outer".into(), vec![]).unwrap(); + let inner = trace + .open(Level::Debug, "inner".into(), vec![field("k", "v")]) + .unwrap(); + trace.close(inner); + let sibling = trace.open(Level::Info, "sibling".into(), vec![]).unwrap(); + trace.close(sibling); + trace.close(outer); + + let doc = parse(&trace.finish(Some("pcr0"), "us-east-1")); + assert_eq!(doc["name"], SERVICE_NAME); + assert_eq!(doc["annotations"]["tenant"], "aa"); + assert_eq!(doc["annotations"]["invocation"], "request"); + assert_eq!(doc["metadata"]["enclave"]["image"], "pcr0"); + let outer = &doc["subsegments"][0]; + assert_eq!(outer["name"], "outer"); + assert_eq!(outer["subsegments"][0]["name"], "inner"); + assert_eq!(outer["subsegments"][0]["annotations"]["level"], "debug"); + assert_eq!( + outer["subsegments"][0]["metadata"]["guest"]["fields"]["k"], + "v" + ); + assert_eq!(outer["subsegments"][1]["name"], "sibling"); + assert!(doc["trace_id"].as_str().unwrap().starts_with("1-")); + } + + #[test] + fn an_event_lands_on_the_innermost_open_span_or_the_one_it_names() { + let mut trace = Trace::new("aa".into(), Invocation::Task); + trace.event(None, Level::Info, "on the root".into(), vec![]); + let outer = trace.open(Level::Info, "outer".into(), vec![]).unwrap(); + let inner = trace.open(Level::Info, "inner".into(), vec![]).unwrap(); + trace.event(None, Level::Warn, "on inner".into(), vec![]); + trace.event( + Some(outer), + Level::Error, + "on outer".into(), + vec![field("a", "b")], + ); + trace.close(inner); + trace.close(outer); + + let doc = parse(&trace.finish(None, "r")); + assert_eq!( + doc["metadata"]["guest"]["events"][0]["message"], + "on the root" + ); + let outer = &doc["subsegments"][0]; + assert_eq!( + outer["metadata"]["guest"]["events"][0]["message"], + "on outer" + ); + assert_eq!(outer["metadata"]["guest"]["events"][0]["level"], "error"); + assert_eq!( + outer["subsegments"][0]["metadata"]["guest"]["events"][0]["message"], + "on inner" + ); + } + + #[test] + fn closing_twice_is_once_and_finishing_closes_what_was_left_open() { + let mut trace = Trace::new("aa".into(), Invocation::Message); + let span = trace.open(Level::Info, "s".into(), vec![]).unwrap(); + assert!(trace.close(span).is_some()); + assert!(trace.close(span).is_none()); + let left = trace.open(Level::Info, "left open".into(), vec![]).unwrap(); + let _ = left; + let doc = parse(&trace.finish(None, "r")); + let end = doc["subsegments"][1]["end_time"].as_f64().unwrap(); + let start = doc["subsegments"][1]["start_time"].as_f64().unwrap(); + assert!(end >= start); + assert!(doc["end_time"].as_f64().unwrap() >= end); + } + + #[test] + fn a_full_trace_counts_what_it_could_not_keep() { + let mut trace = Trace::new("aa".into(), Invocation::Request); + for i in 0..MAX_SPANS + 10 { + if let Some(span) = trace.open(Level::Info, format!("s{i}"), vec![]) { + trace.close(span); + } + } + for _ in 0..MAX_EVENTS + 3 { + trace.event(None, Level::Info, "e".into(), vec![]); + } + let doc = parse(&trace.finish(None, "r")); + assert!(doc["metadata"]["guest"]["dropped_spans"].as_u64().unwrap() > 0); + assert_eq!(doc["metadata"]["guest"]["dropped_events"], 3); + } + + #[test] + fn a_trace_too_big_for_x_ray_sheds_detail_rather_than_being_refused() { + let mut trace = Trace::new("aa".into(), Invocation::Request); + for _ in 0..200 { + let span = trace.open(Level::Info, "s".into(), vec![]).unwrap(); + for _ in 0..MAX_EVENTS { + trace.event(Some(span), Level::Info, "x".repeat(MAX_TEXT_CHARS), vec![]); + } + trace.close(span); + } + let document = trace.finish(None, "r"); + assert!( + document.len() <= MAX_DOCUMENT_BYTES, + "{} bytes", + document.len() + ); + assert_eq!(parse(&document)["metadata"]["guest"]["trimmed"], true); + } + + #[test] + fn guest_text_is_clipped_and_span_names_are_made_acceptable() { + let mut trace = Trace::new("aa".into(), Invocation::Request); + let span = trace + .open( + Level::Info, + "bad{}".into(), + vec![field(&"k".repeat(200), "v")], + ) + .unwrap(); + trace.set_field(span, "x".into(), "é".repeat(MAX_TEXT_CHARS + 5)); + trace.close(span); + let empty = trace.open(Level::Info, " ".into(), vec![]).unwrap(); + trace.close(empty); + let doc = parse(&trace.finish(None, "r")); + assert_eq!(doc["subsegments"][0]["name"], "bad_name___"); + assert_eq!(doc["subsegments"][1]["name"], "span"); + let fields = doc["subsegments"][0]["metadata"]["guest"]["fields"] + .as_object() + .unwrap(); + assert!(fields.keys().all(|k| k.chars().count() <= MAX_KEY_CHARS)); + assert_eq!( + fields["x"].as_str().unwrap().chars().count(), + MAX_TEXT_CHARS + ); + } + + /// Records what it is given; fails transiently `fail` times first. + struct Recorder { + calls: Mutex>>, + fail: AtomicU64, + definitive: bool, + } + + impl Recorder { + fn new(fail: u64, definitive: bool) -> Arc { + Arc::new(Recorder { + calls: Mutex::new(Vec::new()), + fail: AtomicU64::new(fail), + definitive, + }) + } + fn documents(&self) -> Vec { + self.calls + .lock() + .unwrap() + .iter() + .flatten() + .cloned() + .collect() + } + } + + #[async_trait] + impl TraceDestination for Recorder { + async fn put(&self, documents: Vec) -> Result { + self.calls.lock().unwrap().push(documents); + if self.fail.load(Ordering::SeqCst) > 0 { + self.fail.fetch_sub(1, Ordering::SeqCst); + return Err(if self.definitive { + SendError::Definitive("no".into()) + } else { + SendError::Transient("later".into()) + }); + } + Ok(0) + } + } + + #[tokio::test(start_paused = true)] + async fn a_trace_is_shipped_after_the_linger_and_everything_queued_goes_at_shutdown() { + let recorder = Recorder::new(0, false); + let tracer = Tracer::start(recorder.clone(), None, "r".into()); + tracer.submit("one".into()); + tokio::time::sleep(BATCH_LINGER * 2).await; + assert_eq!(recorder.documents(), vec!["one".to_string()]); + + tracer.submit("two".into()); + tracer.submit("three".into()); + tracer.shutdown().await; + assert_eq!(recorder.documents(), ["one", "two", "three"]); + assert_eq!(tracer.dropped(), 0); + } + + #[tokio::test(start_paused = true)] + async fn a_transient_failure_is_retried_and_a_definitive_one_is_dropped() { + let flaky = Recorder::new(1, false); + let tracer = Tracer::start(flaky.clone(), None, "r".into()); + tracer.submit("d".into()); + tracer.shutdown().await; + assert_eq!(flaky.calls.lock().unwrap().len(), 2, "retried once"); + assert_eq!(tracer.dropped(), 0); + + let refusing = Recorder::new(u64::MAX, true); + let tracer = Tracer::start(refusing.clone(), None, "r".into()); + tracer.submit("d".into()); + tracer.shutdown().await; + assert_eq!(refusing.calls.lock().unwrap().len(), 1, "never retried"); + assert_eq!(tracer.dropped(), 1); + } + + #[tokio::test(start_paused = true)] + async fn a_batch_never_exceeds_the_document_limit() { + let recorder = Recorder::new(0, false); + let tracer = Tracer::start(recorder.clone(), None, "r".into()); + for i in 0..MAX_BATCH_DOCUMENTS + 5 { + tracer.submit(format!("{i}")); + } + tracer.shutdown().await; + let calls = recorder.calls.lock().unwrap(); + assert!(calls.iter().all(|c| c.len() <= MAX_BATCH_DOCUMENTS)); + assert_eq!( + calls.iter().map(Vec::len).sum::(), + MAX_BATCH_DOCUMENTS + 5 + ); + } +} diff --git a/runtime/src/testing/mod.rs b/runtime/src/testing/mod.rs index 6382993..59f9dec 100644 --- a/runtime/src/testing/mod.rs +++ b/runtime/src/testing/mod.rs @@ -99,6 +99,7 @@ pub struct EnclaveBuilder { guest: Vec, background_tasks: bool, notify: bool, + telemetry: crate::telemetry::Telemetry, } impl EnclaveBuilder { @@ -119,6 +120,14 @@ impl EnclaveBuilder { self } + /// Ship guest spans and metrics through `telemetry` — typically a [`crate::telemetry::Tracer`] + /// and [`crate::telemetry::Metrics`] over in-memory destinations, so a test can see what a + /// guest reported without AWS. + pub fn telemetry(mut self, telemetry: crate::telemetry::Telemetry) -> Self { + self.telemetry = telemetry; + self + } + pub async fn start(self) -> Result { Enclave::start(self).await } @@ -138,6 +147,7 @@ impl Enclave { guest: guest.into(), background_tasks: false, notify: false, + telemetry: Default::default(), } } @@ -205,6 +215,7 @@ impl Enclave { guest_env, ServeConfig { notify, + telemetry: builder.telemetry.clone(), background_tasks: builder.background_tasks.then(Default::default), addr, certificate: Some(crate::CertificateSlot::fixed(Arc::new(tls))), diff --git a/runtime/tests/serve_auth.rs b/runtime/tests/serve_auth.rs index 073ceb1..7633d0c 100644 --- a/runtime/tests/serve_auth.rs +++ b/runtime/tests/serve_auth.rs @@ -192,6 +192,7 @@ async fn start_with_background(background: bool) -> Harness { ServeConfig { background_tasks: background.then(Default::default), notify: None, + telemetry: Default::default(), addr, certificate: Some(enclave_runtime::CertificateSlot::fixed(Arc::new(tls))), acme: None, diff --git a/runtime/tests/serve_grpc_wire.rs b/runtime/tests/serve_grpc_wire.rs index ec6bc3a..4f749f0 100644 --- a/runtime/tests/serve_grpc_wire.rs +++ b/runtime/tests/serve_grpc_wire.rs @@ -179,6 +179,7 @@ async fn start() -> Harness { ServeConfig { background_tasks: None, notify: None, + telemetry: Default::default(), addr, certificate: Some(enclave_runtime::CertificateSlot::fixed(identity)), acme: None, diff --git a/runtime/tests/serve_h2.rs b/runtime/tests/serve_h2.rs index 596d563..1067541 100644 --- a/runtime/tests/serve_h2.rs +++ b/runtime/tests/serve_h2.rs @@ -113,6 +113,7 @@ async fn start() -> std::net::SocketAddr { ServeConfig { background_tasks: None, notify: None, + telemetry: Default::default(), addr, certificate: Some(enclave_runtime::CertificateSlot::fixed(identity)), acme: None, diff --git a/runtime/tests/serve_tls.rs b/runtime/tests/serve_tls.rs index 235732d..4193e6a 100644 --- a/runtime/tests/serve_tls.rs +++ b/runtime/tests/serve_tls.rs @@ -163,6 +163,7 @@ async fn start_inner(slot: Option) -> Harness ServeConfig { background_tasks: None, notify: None, + telemetry: Default::default(), addr, certificate, acme: None, diff --git a/runtime/tests/telemetry.rs b/runtime/tests/telemetry.rs new file mode 100644 index 0000000..c50fea5 --- /dev/null +++ b/runtime/tests/telemetry.rs @@ -0,0 +1,191 @@ +//! Guest telemetry, end to end. +//! +//! A real component calls `enclave:telemetry` through the runtime's own linker — so the +//! hand-wired names and types in `telemetry::add_to_linker` are checked against what wit-bindgen +//! actually generates — and what would reach X-Ray and CloudWatch is inspected. Both services are +//! replaced by in-memory destinations, so nothing here needs AWS. +//! +//! ```console +//! $ scripts/build-guest.sh http +//! $ cargo test -p enclave-runtime --test telemetry -- --ignored +//! ``` + +use std::sync::{Arc, Mutex}; +use std::time::{Duration, SystemTime}; + +use async_trait::async_trait; +use enclave_runtime::telemetry::{ + Datum, DatumValue, MetricDestination, Metrics, SendError, Telemetry, TraceDestination, Tracer, + Unit, +}; +use enclave_runtime::testing::Enclave; +use serde_json::Value; + +fn guest() -> Vec { + let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../examples/guest-http/target/wasm32-wasip2/release/guest-http.wasm"); + std::fs::read(path).expect("build examples/guest-http for wasm32-wasip2") +} + +#[derive(Default)] +struct Traces(Mutex>); + +#[async_trait] +impl TraceDestination for Traces { + async fn put(&self, documents: Vec) -> Result { + self.0.lock().unwrap().extend(documents); + Ok(0) + } +} + +#[derive(Default)] +struct Data(Mutex>); + +#[async_trait] +impl MetricDestination for Data { + async fn put(&self, namespace: &str, data: Vec, _: SystemTime) -> Result<(), SendError> { + let mut seen = self.0.lock().unwrap(); + seen.extend(data.into_iter().map(|d| (namespace.to_string(), d))); + Ok(()) + } +} + +/// The trace whose invocation opened a span named `name`, once it has been shipped. A trace goes +/// when its invocation's store is dropped, a little after the response, and then waits out the +/// forwarder's linger — so this polls rather than assuming. +async fn trace_with_span(traces: &Traces, name: &str) -> Value { + for _ in 0..100 { + let found = traces + .0 + .lock() + .unwrap() + .iter() + .map(|d| serde_json::from_str::(d).expect("a segment document is JSON")) + .find(|d| d["subsegments"][0]["name"] == name); + if let Some(trace) = found { + return trace; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + panic!("no trace with a `{name}` span arrived"); +} + +#[tokio::test(flavor = "multi_thread")] +#[ignore = "needs examples/guest-http built for wasm32-wasip2"] +async fn a_guests_spans_and_metrics_reach_x_ray_and_cloudwatch_attributed_to_its_tenant() { + let traces = Arc::new(Traces::default()); + let data = Arc::new(Data::default()); + let tracer = Tracer::start( + traces.clone(), + Some("pcr0-of-this-image".into()), + "us-east-1".into(), + ); + let metrics = Metrics::start(data.clone(), "Test/Guest".into()); + let enclave = Enclave::builder(guest()) + .telemetry(Telemetry::new(Some(tracer.clone()), Some(metrics.clone()))) + .start() + .await + .unwrap(); + + let alice = enclave.enrol().await.unwrap(); + let (status, body) = enclave + .signed(&alice, "GET", "/telemetry", "") + .await + .unwrap(); + assert_eq!(status, 200, "{body}"); + + // --- The trace ----------------------------------------------------------------------------- + let trace = trace_with_span(&traces, "handle").await; + assert_eq!( + trace["annotations"]["tenant"], + alice.tenant(), + "attributed to its tenant" + ); + assert_eq!(trace["annotations"]["invocation"], "request"); + assert_eq!(trace["metadata"]["enclave"]["image"], "pcr0-of-this-image"); + assert_eq!(trace["metadata"]["enclave"]["region"], "us-east-1"); + + let handle = &trace["subsegments"][0]; + assert_eq!(handle["annotations"]["level"], "info"); + assert_eq!(handle["metadata"]["guest"]["fields"]["route"], "/telemetry"); + assert_eq!( + handle["metadata"]["guest"]["fields"]["phase"], "start", + "set-field" + ); + assert_eq!( + handle["metadata"]["guest"]["events"][0]["message"], "began", + "add-event" + ); + + let inner = &handle["subsegments"][0]; + assert_eq!( + inner["name"], "inner", + "opened inside `handle`, so its child" + ); + assert_eq!(inner["annotations"]["level"], "debug"); + let event = &inner["metadata"]["guest"]["events"][0]; + assert_eq!( + event["message"], "inside inner", + "a free event lands on the innermost span" + ); + assert_eq!(event["level"], "warn"); + assert_eq!(event["fields"]["n"], "1"); + + // `inner` was closed by dropping its handle, `handle` by `end`; both have a real duration. + for span in [handle, inner] { + assert!(span["end_time"].as_f64().unwrap() >= span["start_time"].as_f64().unwrap()); + } + assert!( + inner["end_time"].as_f64().unwrap() <= handle["end_time"].as_f64().unwrap(), + "the child closed before its parent" + ); + + // --- The metrics --------------------------------------------------------------------------- + // Aggregated in memory until a flush; shutdown is one. + metrics.shutdown().await; + let data = data.0.lock().unwrap().clone(); + let find = |name: &str| { + data.iter() + .find(|(_, d)| d.name == name) + .unwrap_or_else(|| panic!("no `{name}` datum in {data:?}")) + }; + let route = vec![("route".to_string(), "telemetry".to_string())]; + + let (namespace, requests) = find("requests"); + assert_eq!(namespace, "Test/Guest"); + assert_eq!(requests.value, DatumValue::Single(1.0)); + assert_eq!(requests.unit, Unit::Count); + assert_eq!(requests.dimensions, route); + + let (_, depth) = find("queue-depth"); + assert_eq!(depth.value, DatumValue::Single(3.0)); + assert!(depth.dimensions.is_empty(), "no tenant dimension is added"); + + let (_, latency) = find("latency"); + assert_eq!(latency.unit, Unit::Milliseconds); + assert_eq!( + latency.value, + DatumValue::Spread { + values: vec![12.5], + counts: vec![1.0] + } + ); + + tracer.shutdown().await; + assert_eq!(tracer.dropped(), 0); + assert_eq!(metrics.dropped(), 0); +} + +#[tokio::test(flavor = "multi_thread")] +#[ignore = "needs examples/guest-http built for wasm32-wasip2"] +async fn with_telemetry_off_a_guests_calls_still_succeed() { + // The default: no X-Ray, no metrics. The guest is the same one, and it must not fail. + let enclave = Enclave::builder(guest()).start().await.unwrap(); + let alice = enclave.enrol().await.unwrap(); + let (status, body) = enclave + .signed(&alice, "GET", "/telemetry", "") + .await + .unwrap(); + assert_eq!(status, 200, "{body}"); + assert_eq!(body.trim(), "reported"); +} diff --git a/wit/telemetry/telemetry.wit b/wit/telemetry/telemetry.wit new file mode 100644 index 0000000..145d3d4 --- /dev/null +++ b/wit/telemetry/telemetry.wit @@ -0,0 +1,109 @@ +package enclave:telemetry@0.1.0; + +/// Structured tracing, exported to AWS X-Ray. +/// +/// The runtime roots one trace per guest invocation — an inbound request, a +/// task run, or a stream message — and ships it when that invocation ends. A +/// guest's events and spans attach to that trace; the runtime binds the tenant, +/// so **no function here accepts a tenant id** (the same rule as enclave:notify +/// and enclave:tasks). Calls made in a build with X-Ray switched off are +/// accepted and discarded, so a guest never has to branch on whether it is on. +/// +/// Spans and events carry no secret by construction on the runtime's side: the +/// text and fields are the guest's own, and the runtime adds only what it +/// already knows (the tenant, the image, the region). What a guest puts in a +/// message or field is the guest's responsibility, exactly as with its stdout. +interface tracing { + /// Severity. Mapped onto the runtime's own `tracing` levels for the console, + /// and recorded on the X-Ray (sub)segment. + enum level { + trace, + debug, + info, + warn, + error, + } + + /// A key/value annotation on an event or span. The value is a string the + /// guest has already formatted; the runtime does not parse it. Keys are + /// 1-64 characters of [A-Za-z0-9_.-]. + record field { + key: string, + value: string, + } + + /// Record a point-in-time event on the span that is currently open (or on + /// the invocation's root span if the guest has opened none). Cheap: there is + /// no handle to keep. + event: func(level: level, message: string, fields: list); + + /// A unit of work with a duration, exported as an X-Ray subsegment of + /// whichever span is open when it is created (the invocation's root span if + /// none) — so spans nest by the order a guest opens them. The subsegment's + /// end time is set when the span is closed: by `end`, or by dropping the + /// resource, whichever comes first. + resource span { + /// Open a span beneath the currently-open span. + constructor(level: level, name: string, fields: list); + + /// Annotate the span after it has opened. + set-field: func(key: string, value: string); + + /// Record a point-in-time event on this span. + add-event: func(level: level, message: string, fields: list); + + /// Close the span now. Dropping the resource without calling this closes + /// it at the drop instead; calling it more than once does nothing. + end: func(); + } +} + +/// Metrics, published to AWS CloudWatch through the dedicated `PutMetricData` +/// API under the runtime's configured namespace. +/// +/// As with tracing, the runtime binds the tenant and a guest cannot name it: +/// CloudWatch bills per unique dimension combination, so a tenant dimension is +/// the runtime's decision to make, not a guest's. Dimensions a guest does give +/// must be low cardinality. Calls made in a build with metrics switched off are +/// accepted and discarded. +/// +/// A metric name is 1-255 printable ASCII characters. A call CloudWatch would +/// refuse — a bad name or dimension, a value that is not finite or is outside +/// CloudWatch's range — is dropped and counted rather than sent, because +/// CloudWatch refuses a whole request over one bad datum. +interface metrics { + /// The unit a value is in, as CloudWatch names them. `none` leaves it unset. + enum unit { + none, + count, + seconds, + milliseconds, + bytes, + kilobytes, + percent, + } + + /// A metric dimension: low cardinality only. 1-64 characters of + /// [A-Za-z0-9_.-] each, and at most 10 per call. + record dimension { + name: string, + value: string, + } + + /// Add `value` to a counter (published as a Sum aggregation). + counter-add: func(name: string, value: f64, dimensions: list); + + /// Report a gauge's current value (published as-is). + gauge-set: func(name: string, value: f64, %unit: unit, dimensions: list); + + /// Record one observation into a distribution (published through + /// CloudWatch's values/counts distribution). + histogram-record: func(name: string, value: f64, %unit: unit, dimensions: list); +} + +/// What a guest includes to get both. Both are imports: the guest calls the +/// runtime, and there is nothing here for it to export. +world observed { + import tracing; + import metrics; +} From a2d87824fd8314974fdc0b97a20180a51d528cef Mon Sep 17 00:00:00 2001 From: Joshua Date: Sat, 10 Oct 2026 23:23:20 +0300 Subject: [PATCH 2/3] deploy: switch guest telemetry on per image, and grant it - flake.nix: ENCLAVE_XRAY and ENCLAVE_METRICS_NAMESPACE from deployment `xray` and `metricsNamespace`, defaulted with `or`, so a deployment.nix written before telemetry still builds, to the same derivation, with it off. Measured into PCR0 like every other image setting. - deploy/tofu: a `telemetry` variable granting the parent's role exactly what the runtime calls: xray:PutTraceSegments (X-Ray cannot be scoped to a resource) and cloudwatch:PutMetricData held to the one namespace by condition. Nothing is granted when telemetry is off. Co-Authored-By: Claude Opus 5.5 --- deploy/nix/deployment.nix | 13 ++++++++++++ deploy/tofu/main.tf | 44 +++++++++++++++++++++++++++++++++++++++ deploy/tofu/variables.tf | 9 ++++++++ flake.nix | 9 ++++++++ 4 files changed, 75 insertions(+) diff --git a/deploy/nix/deployment.nix b/deploy/nix/deployment.nix index 26166f8..cef0834 100644 --- a/deploy/nix/deployment.nix +++ b/deploy/nix/deployment.nix @@ -80,6 +80,19 @@ # which names it "/${name_prefix}-${environment}/guest". guestLogGroup = "/CHANGE-ME-production/guest"; + # Guest telemetry: what a guest reports through wit/telemetry. `xray` + # exports each guest invocation as one X-Ray trace, attributed to its + # tenant; `metricsNamespace` publishes guest metrics to CloudWatch under that + # namespace (not under `AWS/`), or "" for none. Each needs the instance + # role's grant — `deploy/tofu` gives it when `telemetry` is set there. + # + # Measured by PCR0 like everything here, so a client can tell from an + # attestation whether an enclave ships its traces off the box. What a trace + # carries is the guest's own text plus the tenant id; see + # runtime/src/telemetry. + xray = false; + metricsNamespace = ""; + # The guest is not here either. The runtime fetches it from # `guest/guest.wasm` in `rootsBucket` (`bucketPrefix` is not applied) and # measures it into PCR16 before it asks KMS for a key, so changing the guest diff --git a/deploy/tofu/main.tf b/deploy/tofu/main.tf index 08a289c..e85ef4c 100644 --- a/deploy/tofu/main.tf +++ b/deploy/tofu/main.tf @@ -278,6 +278,50 @@ resource "aws_iam_role_policy" "push" { policy = data.aws_iam_policy_document.push[0].json } +# Guest telemetry, as the image names it (`xray` and `metricsNamespace` in +# deploy/nix/deployment.nix), and only what the runtime calls. X-Ray takes no +# resource-level scoping, so trace segments are account-wide; metrics are held to the +# image's one namespace by condition. Nothing when telemetry is off. +locals { + telemetry_on = var.telemetry.xray || var.telemetry.metrics_namespace != "" +} + +data "aws_iam_policy_document" "telemetry" { + count = local.telemetry_on ? 1 : 0 + + dynamic "statement" { + for_each = var.telemetry.xray ? [1] : [] + content { + sid = "PutGuestTraces" + effect = "Allow" + actions = ["xray:PutTraceSegments"] + resources = ["*"] + } + } + + dynamic "statement" { + for_each = var.telemetry.metrics_namespace != "" ? [1] : [] + content { + sid = "PutGuestMetrics" + effect = "Allow" + actions = ["cloudwatch:PutMetricData"] + resources = ["*"] + condition { + test = "StringEquals" + variable = "cloudwatch:namespace" + values = [var.telemetry.metrics_namespace] + } + } + } +} + +resource "aws_iam_role_policy" "telemetry" { + count = local.telemetry_on ? 1 : 0 + name = "${local.name}-telemetry" + role = aws_iam_role.parent.id + policy = data.aws_iam_policy_document.telemetry[0].json +} + # For a shell without opening port 22. resource "aws_iam_role_policy_attachment" "ssm" { role = aws_iam_role.parent.name diff --git a/deploy/tofu/variables.tf b/deploy/tofu/variables.tf index 537ca64..203a66d 100644 --- a/deploy/tofu/variables.tf +++ b/deploy/tofu/variables.tf @@ -91,6 +91,15 @@ variable "push_app_id" { default = "" } +variable "telemetry" { + description = "Guest telemetry, as the image names it (xray and metricsNamespace in deployment.nix). With xray the parent's role may put X-Ray trace segments; with a metrics_namespace it may publish CloudWatch metrics to that namespace and no other." + type = object({ + xray = optional(bool, false) + metrics_namespace = optional(string, "") + }) + default = {} +} + variable "pool_size_gib" { type = number default = 32 diff --git a/flake.nix b/flake.nix index 4e0371f..85ce57f 100644 --- a/flake.nix +++ b/flake.nix @@ -363,6 +363,15 @@ # them. ENCLAVE_GUEST_LOG_GROUP = deployment.guestLogGroup; + # Guest telemetry (wit/telemetry). With `xray`, each guest invocation + # goes to X-Ray as one trace, attributed to its tenant; with a + # `metricsNamespace`, guest metrics go to CloudWatch under it through + # PutMetricData. Called from inside, like the log group, so each needs + # its IAM grant (deploy/tofu `telemetry`). Defaulted, so a + # deployment.nix written before telemetry still builds, with it off. + ENCLAVE_XRAY = lib.boolToString (deployment.xray or false); + ENCLAVE_METRICS_NAMESPACE = deployment.metricsNamespace or ""; + # Every request that could reach the guest needs a fresh WebAuthn # assertion bound to exactly that request. Without an RP id the # runtime serves the guest to anyone who can open a connection and From 4942d57e2899d177c9946d9ae478d968e6239a49 Mon Sep 17 00:00:00 2001 From: Joshua Date: Sun, 11 Oct 2026 19:03:19 +0300 Subject: [PATCH 3/3] ci(bench): start each run without Criterion's local baseline The benchmarks job failed whenever the cache restored a partial target/criterion. Criterion found a `base/` without its `sample.json`, printed "Criterion.rs ERROR: Failed to access file ..." onto each result's line, and github-action-benchmark could not parse bench.txt. The numbers themselves were fine. It failed this way on main on Oct 4 and Oct 6, and on this branch. History is the tracker's (gh-pages), not Criterion's, so the script now removes target/criterion before benchmarking. Checked by planting a half-restored state and running scripts/ci-bench.sh: three bencher lines, no errors. Co-Authored-By: Claude Opus 5.5 --- scripts/ci-bench.sh | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/scripts/ci-bench.sh b/scripts/ci-bench.sh index c6bcea4..9b02a51 100755 --- a/scripts/ci-bench.sh +++ b/scripts/ci-bench.sh @@ -28,6 +28,13 @@ say "benchmarks" # that understands it. # # lib.sh sets `pipefail`, so a failing cargo still fails the script despite tee. +# +# Criterion compares each run with the last one it finds under target/criterion, and in CI that +# is whatever the cache restored — sometimes a `base/` without its `sample.json`. Criterion then +# prints an error onto each result's line, which the tracker cannot parse, and the job fails on +# the cache's state rather than on a number. The history is the tracker's (gh-pages), not +# Criterion's, so every run starts without it. +rm -rf "$REPO/target/criterion" : > "$REPO/bench.txt" cargo bench -p enclave-runtime --bench instance_cost -- --output-format bencher \ | tee -a "$REPO/bench.txt"