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/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/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/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 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/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" 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; +}