diff --git a/demos/envoy-rust-dynamic-module/.gitignore b/demos/envoy-rust-dynamic-module/.gitignore new file mode 100644 index 0000000000..b1fba44031 --- /dev/null +++ b/demos/envoy-rust-dynamic-module/.gitignore @@ -0,0 +1,4 @@ +# Build outputs. Rebuilt by bench/build.sh. +bin/ +rust-module/target/ +bench/results/ diff --git a/demos/envoy-rust-dynamic-module/DESIGN.md b/demos/envoy-rust-dynamic-module/DESIGN.md new file mode 100644 index 0000000000..7dd171c035 --- /dev/null +++ b/demos/envoy-rust-dynamic-module/DESIGN.md @@ -0,0 +1,236 @@ +# Co-existence design: Rust dynamic module in front of ext_proc + +A minimal, reversible way to put the Rust dynamic module into the ingress path +**without changing the ext_proc contract, the resumer, parking, or anything the +Go router does today.** + +Implemented and measured in this directory as arm C. + +## The principle + +> The module is a **cache**. ext_proc stays the **only** thing that resolves an +> actor, and the only thing that talks to ate-apiserver. + +The module never calls the control plane, never holds a client certificate, +never decides that an actor exists. It answers repeat requests from a binding +that **ext_proc itself already produced**, and gets out of the way for +everything else. + +That single constraint removes most of the risk. The module cannot route +anywhere ext_proc has not already routed; the worst a stale entry can do is send +a request to a worker that ext_proc chose recently. + +## Request flow + +``` + ┌─ hit ──→ publish original_dst metadata + │ set x-ate-route-resolved: 1 + request → module ──┤ clear_route_cache() ───→ route "resolved_by_module" + │ (ext_proc disabled) ──→ worker + │ + └─ miss ─→ (does nothing) ──→ ext_proc ──→ ResumeActor ──→ worker + │ + on_response_headers: read the metadata + ext_proc published, store it in the cache +``` + +**Miss** is byte-for-byte today's path. The module contributes one map lookup +and, on the way back, one map insert. + +**Hit** skips the ext_proc filter entirely — proven in this demo: 23 requests +produced 2 `ext_proc.streams_started`. + +## How a hit skips ext_proc + +ext_proc has no "enabled by metadata" switch — only `ExtProcPerRoute.disabled`, +which is per **route**. So the module selects a different route: + +1. The module sets `x-ate-route-resolved: 1`. +2. It calls `clear_route_cache()`, forcing Envoy to re-run route matching. +3. The route config gains one route, matched on that header, carrying + `typed_per_filter_config: {envoy.filters.http.ext_proc: {disabled: true}}`. +4. When the ext_proc filter runs, it reads that per-route override and skips + itself. + +The header is set by the module, never trusted from the client — a client that +sends `x-ate-route-resolved: 1` itself would select the ext_proc-disabled route +with **no** metadata published, and the ORIGINAL_DST cluster would fail the +request rather than route it anywhere. Still, the module should overwrite the +header unconditionally on every request (set to `1` on a hit, remove it on a +miss). See "Open items". + +## The change to substrate + +Three additive edits, all in +[`xds.go`](../../cmd/atenet/internal/router/xds.go): + +**1. `buildRoutes()` — add one route above the catch-all** (guarded by the flag): + +```go +// Requests the dynamic module already resolved skip the ext_proc call. +{ + Name: "resolved_by_module", + Match: &routev3.RouteMatch{ + PathSpecifier: &routev3.RouteMatch_Prefix{Prefix: "/"}, + Headers: []*routev3.HeaderMatcher{{ + Name: ingress.RouteResolvedHeader, + HeaderMatchSpecifier: &routev3.HeaderMatcher_StringMatch{ + StringMatch: &matcherv3.StringMatcher{ + MatchPattern: &matcherv3.StringMatcher_Exact{Exact: "1"}, + }, + }, + }}, + }, + Action: /* same RouteAction as the catch-all */, + TypedPerFilterConfig: map[string]*anypb.Any{ + "envoy.filters.http.ext_proc": newAny(&extprocv3filter.ExtProcPerRoute{ + Override: &extprocv3filter.ExtProcPerRoute_Disabled{Disabled: true}, + }), + }, +}, +``` + +**2. `buildHcm()` — insert the module filter immediately before ext_proc**, only +when the flag is on. It goes *after* `authorityFilterStateFilter()` so the +module can read `dev.ate.authority`. + +**3. `cmd.go` — one flag**, defaulting off: + +``` +--ingress-route-cache-ttl duration + How long the dataplane's in-process route cache may serve an actor->worker + binding that ext_proc previously resolved, skipping the ext_proc call for + repeat requests. 0 (the default) disables the cache and the dataplane + filter entirely. +``` + +No change to `ingress/`, `extproc/`, `resumer.go`, or `parking.go`. + +## Why this is safe to land + +* **Default off.** `--ingress-route-cache-ttl=0` emits neither the filter nor + the route. The generated xDS is identical to today's, byte for byte. +* **Rollback is a flag flip**, applied over xDS with no restart and no image + change. +* **Fail-open by construction.** Every path the module does not understand — an + authority it cannot parse, a missing filter state, an empty cache, an expired + entry — returns `Continue` and lets ext_proc handle it, including producing + the exact 404/503 bodies clients see today. +* **The error contract is untouched**, because the module never generates + errors. All of `ingress/errors.go` still runs on every miss. +* **Parking, singleflight and the resume metrics keep working**, because every + cold actor is a miss and every miss is ext_proc. +* **Blast radius is bounded by TTL.** A few seconds of stale routing for actors + that moved, versus a permanent per-request RPC. + +## Measured effect + +Arm C vs arm A, same host, same load, 50 hot actors, concurrency 8, 20s +(reproduced twice, within 2%): + +| | A: ext_proc today | C: module + ext_proc | change | +|---|---:|---:|---:| +| throughput | 3,735 rps | 42,766 rps | **11.5×** | +| p50 | 2.11 ms | 0.17 ms | **−92%** | +| p95 | 2.53 ms | 0.30 ms | **−88%** | +| p99 | 3.01 ms | 0.44 ms | **−85%** | +| ResumeActor RPCs | 87,231 | 262 | **−99.7%** | +| atenet-router CPU | ~47% | ~1% | **−98%** | +| CPU per request | ~0.23 ms | ~0.037 ms | **~6×** | + +Arm C reaches **96% of the throughput of arm B2**, the full replacement that +drops ext_proc and calls the control plane itself. Nearly the entire win comes +from not making the call — not from removing the Go router. That is the whole +argument for this design: take the win, keep the router. + +## Rollout + +1. **Land the module and the flag, default off.** Nothing changes in production. +2. **Enable with a 1s TTL in one cell.** Compare `ate_router.cache_hit` / + `cache_miss` against the existing route-duration histogram, and watch the + 404/503 rate for any divergence. +3. **Raise the TTL** toward — but well under — the idle-suspend timeout, watching + for requests routed to workers that no longer host the actor. +4. **Then, and only then**, consider whether the module should resolve misses + itself (arm B2). It buys ~4% more throughput and costs the dataplane a client + certificate, an HTTP resolve endpoint on ate-apiserver, and its own + coalescing. On these numbers that trade is not obviously worth making. + +## The correctness point that matters most + +**A cache hit skips `ResumeActor`, and `ResumeActor` is what wakes a suspended +actor.** Today, a request for an actor that has been suspended *causes* it to +resume. With a cache in front, a stale hit routes to the old worker and fails +instead. + +So eviction must **re-run the slow path inside the same request**, never +evict-and-fail. Concretely: on the stale signal the module must clear the entry, +remove `x-ate-route-resolved`, `clear_route_cache()`, and let the request retry +through ext_proc — not return the error to the client. Without that, the TTL +becomes a user-visible error window on exactly the request that should have +triggered a cold resume. + +## The stale signal already exists, and nothing reads it + +`internal/atunnel/ingress.go:46` defines `StaleAssignmentHeader = +"X-Ate-Assignment-Stale"`, and `reject()` (`ingress.go:524-527`) sets it with a +**421** whenever a worker is asked for an actor it no longer hosts. There are no +consumers anywhere in the repo: the router cannot even see it, because +`xds.go:1016` sets `ResponseHeaderMode: SKIP`. + +That is precisely the negative-feedback signal this design needs, and it is +**free to a dynamic module** — `on_response_headers` is a local function call — +where it would cost Go ext_proc a second gRPC round trip per response to obtain. + +Three eviction triggers are required, not one: + +1. `421` + `x-ate-assignment-stale: true` → evict, then re-run the slow path. +2. **Upstream connect failure / reset.** If the worker pod is gone there is no + 421 at all — the ORIGINAL_DST cluster produces a local 503. Evict on stream + reset too, or a vanished worker burns every request until the TTL expires. +3. Hard TTL expiry. + +**Prerequisite, one line of Go:** the header is currently forgeable. atunnel's +`ReverseProxy` (`ingress.go:130-150`) has no `ModifyResponse`, so an actor can +emit `421 + X-Ate-Assignment-Stale: true` itself and force a `ResumeActor` per +response — amplification onto exactly the load the cache removes. Strip the +header from proxied responses before the module trusts it. + +## How stale can a binding actually get + +Exhaustive grep for `Status.WorkerAssignment` writes gives **six** invalidation +events: suspend, pause, crash, worker-pod delete, actor delete, and re-resume. +Of those, only worker-pod deletion is spontaneous — and **there is no +idle-suspend controller in the tree at all** (`docs/roadmap.md:114` still lists +TTL-based GC as a future idea), so suspension is externally driven. Worker pods +also carry a 3600s termination grace period +(`workerpool_apply.go:39`) and keep hosting their actor while draining. + +Bindings are considerably more stable than the request rate, which is what makes +even a multi-second TTL reasonable. Start at 1s anyway, and raise it only once +the eviction loop is proven. + +## Known gap this widens + +`atunnel.authorize()` compares actor **ref** (atespace+name), not **uid**. A +delete-and-recreate of the same name landing on the same worker will not produce +a 421. No cross-tenant exposure — same atespace, same name — but the cache +widens that cross-*incarnation* window from milliseconds to the TTL. The worker +already pins `ExpectedActorUID` (`internal/atunnel/credential.go:44`), so the +fix is threading it into the comparison; no proto change. + +## Open items before this ships + +* **Strip a client-supplied `x-ate-route-resolved`.** Have the module remove the + header on every miss, or use dynamic metadata plus a metadata-matched route + instead of a header, so client input can never select the route. +* **Coalesce misses**, so a cold actor with many simultaneous requests does not + produce a burst of ext_proc streams. Today they all fall through, which is the + same as current behaviour — ext_proc's own singleflight collapses the + `ResumeActor` calls behind them — so this is a refinement, not a blocker. +* **Cache bound.** The `DashMap` grows with the actor working set and is never + evicted except on expiry lookup. It needs a size cap with LRU, or a sweeper on + a `new_scheduler` timer. +* **Crash blast radius.** A panic in the module takes Envoy down, where a Go + router crash fails only ext_proc streams. The SDK catches unwinds at the ABI + boundary, but this deserves a soak test before it fronts real traffic. diff --git a/demos/envoy-rust-dynamic-module/README.md b/demos/envoy-rust-dynamic-module/README.md new file mode 100644 index 0000000000..2e9251f494 --- /dev/null +++ b/demos/envoy-rust-dynamic-module/README.md @@ -0,0 +1,184 @@ +# Envoy Rust dynamic module for the atenet ingress path + +A working, measurable prototype that replaces the ingress **ext_proc gRPC hop** +with an **Envoy Rust dynamic module** running in-process inside the dataplane. + +Everything here runs locally with `docker compose`. Arm A runs substrate's real +`atenet router` binary and its real `xds.go` control plane, unmodified, so the +baseline is the actual system rather than an approximation of it. + +## Why this path + +Every request through the ingress gateway does this today: + +1. Envoy pauses the filter chain and sends the full header set to the Go router + over an ext_proc bidi gRPC stream + ([extproc.go](../../cmd/atenet/internal/router/extproc/extproc.go)). +2. The router flattens every header into a map, resolves the actor, and calls + `ResumeActor` on ate-apiserver over gRPC + ([ingress.go:126](../../cmd/atenet/internal/router/ingress/ingress.go#L126)). +3. It returns dynamic metadata that an `ORIGINAL_DST` cluster routes on + ([xds.go:745](../../cmd/atenet/internal/router/xds.go#L745)). + +Step 2 happens on **every request**, including requests to an actor that is +already `RUNNING` on a known worker. `singleflight` only collapses *concurrent* +callers, never sequential ones, and +[resumer.go](../../cmd/atenet/internal/router/ingress/resumer.go) says so +outright: *"the accepted cost of one control-plane RPC per hot actor"*. + +For agent workloads — a working set of warm actors receiving many requests each +— that is one control-plane round trip per request that buys nothing. + +## Measured results + +4-CPU Colima VM, Envoy v1.39.1 (the version substrate pins), 50 hot actors, 20s +after a 5s warmup, load generated from inside the container network. `fakeate` +charges 1ms of simulated control-plane work per resolve. Reproduced twice, +within 2%. + +| Arm | RPS | p50 | p95 | p99 | control-plane calls | +|---|---:|---:|---:|---:|---:| +| **A** — ext_proc → Go router (today) | 3,735 | 2.11 ms | 2.53 ms | 3.01 ms | 87,231 | +| **B1** — Rust module, cache **off** | 4,170 | 1.89 ms | 2.26 ms | 2.59 ms | 105,423 | +| **B2** — Rust module, cache **on**, no ext_proc | 44,817 | 0.16 ms | 0.28 ms | 0.42 ms | 361 | +| **C** — Rust module **+ ext_proc together** | 42,766 | 0.17 ms | 0.30 ms | 0.44 ms | 262 | + +Three separate effects, deliberately measured apart: + +* **Deleting the ext_proc hop** (A → B1, cache off, one resolve per request + either way): **+12% throughput**, p95 −11%. Real, but small. +* **Not making the call at all** (B1 → B2): **12× throughput**, p95 −88%. + Almost the entire win is the per-request `ResumeActor` RPC, not the gRPC hop. +* **Co-existence costs almost nothing** (B2 → C): arm C keeps ext_proc in the + chain and still reaches **96% of the full replacement's throughput**. It gets + there without the dataplane ever talking to ate-apiserver. + +**CPU per request** (`docker stats`, steady state): + +| Arm | dataplane | atenet-router | RPS | CPU-ms / request | +|---|---:|---:|---:|---:| +| A | ~39% | ~47% | 3,735 | ~0.23 | +| C | ~155% | **~1%** | 42,766 | ~0.037 | + +About **6× less CPU per request**, with the Go router's own CPU down ~98%. + +> Demo numbers on a laptop VM, not production numbers. `fakeate`'s 1ms stands in +> for real ate-apiserver work and the backends do nothing, so the absolute +> figures will not transfer. What transfers is the shape: the per-request +> control-plane RPC dominates, and removing it is where the win is. + +**Arm C is the one to read.** See [DESIGN.md](DESIGN.md) for how it works and +what it would take to land. + +## What the module does + +[`rust-module/src/lib.rs`](rust-module/src/lib.rs) is a like-for-like +replacement for `ingress.Handler.HandleRequestHeaders`: + +| Go handler | Rust module | +|---|---| +| reads `filter_state['dev.ate.authority']` via ext_proc `request_attributes` | `get_filter_state_bytes(b"dev.ate.authority")` | +| `resources.ParseActorDNSName` | `parse_actor_ref` | +| `ResumeActor` gRPC, every request | TTL cache; `send_http_callout` only on miss | +| returns `structpb` dynamic metadata | `set_dynamic_metadata_string(...)` | +| `HeaderMutation` for the atunnel port | `set_request_header(...)` | +| 404 / 503 via `ImmediateResponse` | `send_response(...)` | +| OTel histogram with string attributes | Envoy-native counters via `define_counter` | + +The cache is a process-global `DashMap` behind a `OnceLock`, shared across every +Envoy worker thread (the `.so` is `dlopen`ed once). It holds only the +actor→worker binding, for a short TTL. + +## Running it + +Requires Docker (or Colima) and Go. **No Rust toolchain on the host** — the +module is built inside a container. + +```bash +./bench/build.sh && docker compose up -d && ./bench/run.sh +``` + +Knobs: `DURATION`, `WARMUP`, `CONCURRENCY`, `ACTORS`, `RESUME_LATENCY`. + +Ports: `21080` arm A, `21082` arm B1 (no cache), `21081` arm B2 (cached), +`21083` arm C (co-existence), `21088` fakeate stats, +`21900`/`21901`/`21903` Envoy admin. + +```bash +curl -H "Host: actor-001.demo.actors.resources.substrate.ate.dev" http://127.0.0.1:21081/ +curl -s http://127.0.0.1:21901/stats | grep ate_router # cache hit/miss counters +``` + +## Honest limitations + +These are the gaps between this prototype and something shippable. + +* **The callout is plaintext HTTP/JSON, not mTLS gRPC.** The real router + authenticates to ate-apiserver with a client certificate + ([ateapiauth](../../internal/ateapiauth/)). `send_http_callout` targets an + Envoy *cluster*, so the TLS context is Envoy's — workable, but it means the + dataplane needs the router's identity, which is a real security design + decision, not a detail. ate-apiserver would also need an HTTP/JSON resolve + endpoint, or the module would need to frame gRPC by hand. +* **No request coalescing.** Concurrent misses for the same actor each issue + their own callout; the Go path collapses them with `singleflight`. On a cold + actor with a resume storm this is strictly worse until it is added. +* **Cache invalidation is TTL-only.** There is no watch API on ate-apiserver to + subscribe to, so a suspended or migrated actor is served stale until the TTL + expires. A routing failure should evict the entry — this prototype does not + yet do that. The TTL must stay well under the idle-suspend timeout. +* **Parking is not implemented.** The Go path parks resume-gated requests with a + bounded lot and a retry budget + ([parking.go](../../cmd/atenet/internal/router/ingress/parking.go)). The + module fails fast instead. A `new_scheduler` timer could do it, but it is not + here. +* **Egress is untouched.** This is ingress only. +* **No sandbox.** A panic in the module takes Envoy down with it, unlike a + crash in the Go router, which fails one ext_proc stream. This is the single + biggest operational difference and it is not a small one. +* The demo skips atunnel mTLS on the worker leg + (`--upstream-credential-bundle=`), so the backends speak plaintext on :443. + +## Where else this applies + +Findings from a survey of the tree at commit `69828945`, each checked against +the source: + +1. **Egress CONNECT is the same bug, worse.** `egress.go:168` calls `GetActor` + on *every* CONNECT with no cache and not even singleflight — and carries its + own TODO saying so (`egress.go:167`). Each CONNECT also re-parses and + re-verifies a certificate Envoy already validated (~31 µs of XFCC parsing + + 62-99 µs of chain verification). It fires per TCP connection rather than per + request, so ingress is still the bigger total win. +2. **A `cert_validator` dynamic module** (new in 1.39) gets the raw DER chain at + handshake time, which would delete the XFCC round trip entirely — that header + exists only because CEL attributes cannot express substrate's custom + `ActorIdentity` extension (`egress.go:60-63`). +3. **Requests inside a CONNECT tunnel resume the actor again**, per request, via + `main_internal` — the same filter fixes it. +4. **Free wins needing no Rust at all**: drop + `--component-log-level ...:debug` from the shipped router manifest + (`atenet-router.yaml:275`); set `ext_proc forward_rules.allowed_headers` + (1407 B → 329 B per request); use `credbundle.Loader` for atunnel's + `GetCertificate` (67-79 µs → ~2.7 µs per worker handshake). + +One claim in that analysis is **refuted by this demo**: it argues a module must +replace ext_proc rather than sit in front of it, because "a module in front of +ext_proc cannot suppress it." Arm C shows it can — `clear_route_cache()` plus a +route carrying `ExtProcPerRoute.disabled` — measured at 23 requests producing 2 +`ext_proc.streams_started`. + +## What should stay in Go + +Not everything belongs in the dataplane: + +* the **xDS control plane** ([xds.go](../../cmd/atenet/internal/router/xds.go)), +* the **ActorTemplate controller** and anything needing a Kubernetes watch, +* **egress actor-certificate authentication**, which needs the actor-identity CA + and a custom X.509 extension parse + ([egress.go](../../cmd/atenet/internal/router/egress/egress.go)), +* anything that has to survive a dataplane crash. + +The module is a **cache and a fast path**, not a replacement for the router. +That is exactly what arm C is, and why it is the recommended shape: see +[DESIGN.md](DESIGN.md). diff --git a/demos/envoy-rust-dynamic-module/actorbackend/main.go b/demos/envoy-rust-dynamic-module/actorbackend/main.go new file mode 100644 index 0000000000..78f5f4d0dc --- /dev/null +++ b/demos/envoy-rust-dynamic-module/actorbackend/main.go @@ -0,0 +1,57 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// actorbackend stands in for the worker pod hosting an actor. +// +// It listens on the port the router routes to (atunnel's ingress port, 443) +// and echoes back the actor DNS name it was addressed as, so the load generator +// can assert that a request actually reached the worker its actor is pinned to. +// It does as little work as possible on purpose: the demo measures the routing +// decision, so anything the backend spends would only dilute the difference +// between the two arms. +package main + +import ( + "flag" + "fmt" + "log" + "net/http" + "time" +) + +var ( + addr = flag.String("addr", ":443", "listen address; matches the atunnel ingress port the router targets") + name = flag.String("name", "worker", "worker name reported in responses") + delay = flag.Duration("delay", 0, "artificial per-request handling delay") +) + +func main() { + flag.Parse() + + srv := &http.Server{ + Addr: *addr, + Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if *delay > 0 { + time.Sleep(*delay) + } + w.Header().Set("x-ate-worker", *name) + // Host is the actor's own DNS name: the router deliberately leaves + // :authority untouched so the worker can authorize on it. + fmt.Fprintf(w, "worker=%s actor=%s port=%s\n", *name, r.Host, r.Header.Get("x-ate-target-port")) + }), + ReadHeaderTimeout: 5 * time.Second, + } + log.Printf("actorbackend %s listening on %s", *name, *addr) + log.Fatal(srv.ListenAndServe()) +} diff --git a/demos/envoy-rust-dynamic-module/actortemplates.yaml b/demos/envoy-rust-dynamic-module/actortemplates.yaml new file mode 100644 index 0000000000..c641f27f31 --- /dev/null +++ b/demos/envoy-rust-dynamic-module/actortemplates.yaml @@ -0,0 +1,12 @@ +# Offline ActorTemplate list, so `atenet router` runs with no Kubernetes API at +# all (--actor-templates-file switches the store from the k8s client to a file +# and skips creating the cluster clients entirely; see router.go NewRouterServer). +apiVersion: api.substrate.ate.dev/v1alpha1 +kind: ActorTemplateList +items: +- apiVersion: api.substrate.ate.dev/v1alpha1 + kind: ActorTemplate + metadata: + name: echo + namespace: demo + spec: {} diff --git a/demos/envoy-rust-dynamic-module/bench/build.sh b/demos/envoy-rust-dynamic-module/bench/build.sh new file mode 100755 index 0000000000..fbc57e786c --- /dev/null +++ b/demos/envoy-rust-dynamic-module/bench/build.sh @@ -0,0 +1,50 @@ +#!/usr/bin/env bash +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# Builds everything the demo needs: the Go binaries (including substrate's own +# atenet) for the container architecture, and the Rust dynamic module. +set -euo pipefail +cd "$(dirname "$0")/.." +REPO_ROOT="$(cd ../.. && pwd)" + +ARCH="$(docker info --format '{{.Architecture}}')" +case "$ARCH" in + x86_64|amd64) GOARCH=amd64; RUST_IMAGE=rust:1-slim-bullseye ;; + aarch64|arm64) GOARCH=arm64; RUST_IMAGE=rust:1-slim-bullseye ;; + *) echo "unsupported docker architecture: $ARCH" >&2; exit 1 ;; +esac +echo "building for linux/$GOARCH" + +mkdir -p bin +for b in fakeate actorbackend loadgen; do + ( cd "$REPO_ROOT" && CGO_ENABLED=0 GOOS=linux GOARCH="$GOARCH" \ + go build -o "demos/envoy-rust-dynamic-module/bin/$b" "./demos/envoy-rust-dynamic-module/$b/" ) +done +# substrate's real router binary: arm A runs it unmodified. +( cd "$REPO_ROOT" && CGO_ENABLED=0 GOOS=linux GOARCH="$GOARCH" \ + go build -o "demos/envoy-rust-dynamic-module/bin/atenet" ./cmd/atenet/ ) + +# The module is built inside a container so no Rust toolchain is needed on the +# host. bullseye (glibc 2.31) is deliberately older than the Envoy image's +# glibc 2.35, so the .so loads there; building on a newer glibc would not. +docker run --rm \ + -v ate-cargo:/usr/local/cargo/registry \ + -v ate-cargo-git:/usr/local/cargo/git \ + -v "$(pwd)/rust-module:/src" -w /src "$RUST_IMAGE" sh -c ' + apt-get update -qq >/dev/null 2>&1 + apt-get install -y -qq git ca-certificates libclang-dev clang >/dev/null 2>&1 + cargo build --release' + +ls -la bin/ rust-module/target/release/*.so diff --git a/demos/envoy-rust-dynamic-module/bench/run.sh b/demos/envoy-rust-dynamic-module/bench/run.sh new file mode 100755 index 0000000000..bfb471d9a1 --- /dev/null +++ b/demos/envoy-rust-dynamic-module/bench/run.sh @@ -0,0 +1,48 @@ +#!/usr/bin/env bash +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# Runs the three arms back to back and reports latency plus the number of +# control-plane calls each one caused. +# +# The load generator runs as a container on the demo's own network rather than +# on the host: the host port forward would add the same overhead to every arm +# and compress the differences being measured. +set -euo pipefail +cd "$(dirname "$0")/.." + +DURATION="${DURATION:-20s}" +WARMUP="${WARMUP:-5s}" +CONCURRENCY="${CONCURRENCY:-32}" +ACTORS="${ACTORS:-50}" +NETWORK="ate-rust-dynmod_atenet" +STATS="http://127.0.0.1:21088" + +run_arm() { + local label="$1" target="$2" + curl -s -X POST "$STATS/stats/reset" >/dev/null + docker run --rm --network "$NETWORK" -v "$(pwd)/bin:/bin/ate:ro" \ + debian:bookworm-slim \ + /bin/ate/loadgen --target "http://$target" --label "$label" \ + --actors "$ACTORS" --concurrency "$CONCURRENCY" \ + --duration "$DURATION" --warmup "$WARMUP" + printf ' control-plane calls: %s\n\n' "$(curl -s "$STATS/stats")" +} + +echo "duration=$DURATION warmup=$WARMUP concurrency=$CONCURRENCY hot actors=$ACTORS" +echo +run_arm "A ext_proc (Go)" "envoy-baseline:8080" +run_arm "B1 rust, no cache" "envoy-dynmod:8082" +run_arm "B2 rust, cached" "envoy-dynmod:8080" +run_arm "C rust + ext_proc" "envoy-coexist:8080" diff --git a/demos/envoy-rust-dynamic-module/docker-compose.yaml b/demos/envoy-rust-dynamic-module/docker-compose.yaml new file mode 100644 index 0000000000..5da5278167 --- /dev/null +++ b/demos/envoy-rust-dynamic-module/docker-compose.yaml @@ -0,0 +1,144 @@ +# Two gateways, one workload, one control plane. +# +# arm A localhost:10000 -> envoy-baseline -> ext_proc gRPC -> atenet-router -> fakeate (gRPC/mTLS) +# arm B localhost:10001 -> envoy-dynmod -> Rust dynamic module in-process -> fakeate (HTTP, on miss only) +# +# Both Envoys are the same v1.39 image substrate pins, both route to the same +# four worker backends through an ORIGINAL_DST cluster, and both resolve actors +# through the same fakeate with the same simulated control-plane cost. The only +# difference is where the routing decision is made. +name: ate-rust-dynmod + +networks: + atenet: + ipam: + config: + - subnet: 172.28.0.0/16 + +volumes: + shared: + +services: + # Stands in for ate-apiserver. Mints the demo CA on startup and writes it, + # plus the router's client bundle, to the shared volume. + fakeate: + image: debian:bookworm-slim + networks: [atenet] + volumes: + - shared:/shared + - ./bin:/bin/ate:ro + command: + - /bin/ate/fakeate + - --grpc-addr=:9443 + - --http-addr=:8081 + - --workers=172.28.0.11,172.28.0.12,172.28.0.13,172.28.0.14 + - --resume-latency=${RESUME_LATENCY:-1ms} + ports: + # Exposed so the bench script can read and reset the control-plane call + # counters between arms. + - "21088:8081" + + # The real substrate router, unmodified, running with no Kubernetes at all. + atenet-router: + image: debian:bookworm-slim + networks: { atenet: { ipv4_address: 172.28.0.20 } } + depends_on: [fakeate] + volumes: + - shared:/shared + - ./bin:/bin/ate:ro + - ./actortemplates.yaml:/etc/ate/actortemplates.yaml:ro + # Wait for fakeate to publish the CA before dialing it. + entrypoint: ["/bin/sh", "-c"] + command: + - | + while [ ! -f /shared/client-bundle.pem ]; do sleep 0.2; done + exec /bin/ate/atenet router \ + --mode=ingress \ + --actor-templates-file=/etc/ate/actortemplates.yaml \ + --ateapi-address=dns:///fakeate:9443 \ + --ateapi-ca-file=/shared/ca.pem \ + --ateapi-client-cert=/shared/client-bundle.pem \ + --ateapi-server-name=fakeate \ + --extproc-address=172.28.0.20 \ + --port-xds=18000 \ + --port-extproc=50051 \ + --port-http=8080 \ + --port-https=0 \ + --port-connect-tls=0 \ + --upstream-credential-bundle= \ + --upstream-trust-bundle= \ + --otlp-collector-address= \ + --envoy-admin-address=envoy-baseline:9901 \ + --status-port=4040 \ + --health-interval=1h \ + --log-level=${ROUTER_LOG_LEVEL:-warn} + + # Arm A: Envoy configured entirely over ADS by the router above. + envoy-baseline: + image: envoyproxy/envoy:v1.39-latest + networks: [atenet] + depends_on: [atenet-router] + user: root + volumes: + - ./envoy/baseline-bootstrap.yaml:/etc/envoy/envoy.yaml:ro + command: ["/usr/local/bin/envoy", "-c", "/etc/envoy/envoy.yaml", "--log-level", "warn"] + ports: + - "21080:8080" + - "21900:9901" + + # Arm B: Envoy with the Rust dynamic module and no Go router in the path. + envoy-dynmod: + image: envoyproxy/envoy:v1.39-latest + networks: [atenet] + depends_on: [fakeate] + user: root + environment: + # Envoy dlopens "${ENVOY_DYNAMIC_MODULES_SEARCH_PATH}/lib.so", + # where is dynamic_module_config.name in the filter config. + ENVOY_DYNAMIC_MODULES_SEARCH_PATH: /modules + volumes: + - ./envoy/dynmod.yaml:/etc/envoy/envoy.yaml:ro + - ./rust-module/target/release:/modules:ro + command: ["/usr/local/bin/envoy", "-c", "/etc/envoy/envoy.yaml", "--log-level", "warn"] + ports: + - "21081:8080" + - "21082:8082" + - "21901:9901" + + actor-1: + image: debian:bookworm-slim + networks: { atenet: { ipv4_address: 172.28.0.11 } } + volumes: [./bin:/bin/ate:ro] + command: [/bin/ate/actorbackend, --name=worker-1, --addr=:443] + actor-2: + image: debian:bookworm-slim + networks: { atenet: { ipv4_address: 172.28.0.12 } } + volumes: [./bin:/bin/ate:ro] + command: [/bin/ate/actorbackend, --name=worker-2, --addr=:443] + actor-3: + image: debian:bookworm-slim + networks: { atenet: { ipv4_address: 172.28.0.13 } } + volumes: [./bin:/bin/ate:ro] + command: [/bin/ate/actorbackend, --name=worker-3, --addr=:443] + actor-4: + image: debian:bookworm-slim + networks: { atenet: { ipv4_address: 172.28.0.14 } } + volumes: [./bin:/bin/ate:ro] + command: [/bin/ate/actorbackend, --name=worker-4, --addr=:443] + + # Arm C: the Rust module and ext_proc co-existing in one chain. This is the + # shape a real migration would take. + envoy-coexist: + image: envoyproxy/envoy:v1.39-latest + networks: [atenet] + depends_on: [atenet-router] + user: root + environment: + ENVOY_DYNAMIC_MODULES_SEARCH_PATH: /modules + volumes: + - ./envoy/coexist.yaml:/etc/envoy/envoy.yaml:ro + - ./rust-module/target/release:/modules:ro + command: ["/usr/local/bin/envoy", "-c", "/etc/envoy/envoy.yaml", "--log-level", "warn"] + ports: + - "21083:8080" + - "21903:9901" diff --git a/demos/envoy-rust-dynamic-module/envoy/baseline-bootstrap.yaml b/demos/envoy-rust-dynamic-module/envoy/baseline-bootstrap.yaml new file mode 100644 index 0000000000..ccb417c72a --- /dev/null +++ b/demos/envoy-rust-dynamic-module/envoy/baseline-bootstrap.yaml @@ -0,0 +1,60 @@ +# Arm A: today's substrate architecture, unmodified. +# +# This is the Envoy bootstrap from manifests/ate-install/atenet-router.yaml with +# only the xds_cluster address changed from 127.0.0.1 (same pod) to the router's +# container name. Everything Envoy actually runs -- listeners, the ext_proc +# filter, the ORIGINAL_DST cluster -- is served over ADS by the real +# `atenet router` process, so this arm exercises substrate's real xds.go and its +# real ext_proc handler rather than a hand-written approximation of them. +admin: + address: + socket_address: + address: "::" + ipv4_compat: true + port_value: 9901 + +node: + id: substrate-envoy-node + cluster: substrate-router-cluster + +# Required for main_internal (see xds.go's buildMainInternalListener): Envoy +# rejects any Listener.internal_listener without this registered. +bootstrap_extensions: +- name: envoy.bootstrap.internal_listener + typed_config: + "@type": type.googleapis.com/envoy.extensions.bootstrap.internal_listener.v3.InternalListener + +dynamic_resources: + lds_config: + resource_api_version: V3 + ads: {} + cds_config: + resource_api_version: V3 + ads: {} + ads_config: + api_type: GRPC + transport_api_version: V3 + grpc_services: + - envoy_grpc: + cluster_name: xds_cluster + +static_resources: + clusters: + - name: xds_cluster + connect_timeout: 0.25s + type: STRICT_DNS + lb_policy: ROUND_ROBIN + typed_extension_protocol_options: + envoy.extensions.upstreams.http.v3.HttpProtocolOptions: + "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions + explicit_http_config: + http2_protocol_options: {} + load_assignment: + cluster_name: xds_cluster + endpoints: + - lb_endpoints: + - endpoint: + address: + socket_address: + address: atenet-router + port_value: 18000 diff --git a/demos/envoy-rust-dynamic-module/envoy/coexist.yaml b/demos/envoy-rust-dynamic-module/envoy/coexist.yaml new file mode 100644 index 0000000000..cf02aa888b --- /dev/null +++ b/demos/envoy-rust-dynamic-module/envoy/coexist.yaml @@ -0,0 +1,144 @@ +# Arm C: the migration shape -- the Rust module and ext_proc in the SAME chain. +# +# The module runs first as a pure cache. On a hit it publishes the routing +# decision and marks the request so route selection lands on a route where +# ext_proc is disabled. On a miss it does nothing, and the request takes exactly +# today's path through the real Go router; the module then learns the answer +# from the dynamic metadata ext_proc published. +# +# ext_proc's own config is byte-identical to what xds.go generates today. The +# only additions are the module filter and the second route. +admin: + address: + socket_address: { address: 0.0.0.0, port_value: 9901 } + +node: + id: substrate-envoy-node + cluster: substrate-router-cluster + +static_resources: + listeners: + - name: ingress_http_listener + address: + socket_address: { address: 0.0.0.0, port_value: 8080 } + filter_chains: + - filters: + - name: envoy.filters.network.http_connection_manager + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager + stat_prefix: ingress_http + generate_request_id: true + route_config: + name: substrate_routes + virtual_hosts: + - name: local_service + domains: ["*"] + routes: + # Cache hit: the module already resolved this request, so skip the + # ext_proc call. This route is the entire opt-in. + - name: resolved_by_module + match: + prefix: "/" + headers: + - name: x-ate-route-resolved + string_match: { exact: "1" } + route: + cluster: actor_original_dst + timeout: 10s + typed_per_filter_config: + envoy.filters.http.ext_proc: + "@type": type.googleapis.com/envoy.extensions.filters.http.ext_proc.v3.ExtProcPerRoute + disabled: true + # Everything else: today's path, unchanged. + - name: catch_all + match: { prefix: "/" } + route: + cluster: actor_original_dst + timeout: 10s + request_headers_to_add: + - header: + key: x-ate-target-port + value: "%DYNAMIC_METADATA(envoy.filters.listener.original_dst:port)%" + append_action: OVERWRITE_IF_EXISTS_OR_ADD + http_filters: + - name: envoy.filters.http.set_filter_state + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.set_filter_state.v3.Config + on_request_headers: + - object_key: dev.ate.authority + factory_key: envoy.string + format_string: + text_format_source: + inline_string: "%REQ(:AUTHORITY)%" + shared_with_upstream: ONCE + # The new filter. Delete these 16 lines and the chain is today's. + - name: envoy.filters.http.dynamic_modules + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.dynamic_modules.v3.DynamicModuleFilter + dynamic_module_config: + name: ate_router_module + filter_name: actor_router_cache + filter_config: + "@type": type.googleapis.com/google.protobuf.StringValue + value: | + { + "ateapi_cluster": "unused-in-coexistence-mode", + "cache_ttl_seconds": 5, + "actor_dns_suffix": "actors.resources.substrate.ate.dev", + "atunnel_port": 443, + "target_port_header": "x-ate-target-port", + "cache_enabled": true + } + # Unchanged: same config xds.go emits today. + - name: envoy.filters.http.ext_proc + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.ext_proc.v3.ExternalProcessor + grpc_service: + envoy_grpc: { cluster_name: ate-cluster } + timeout: 10s + message_timeout: 10s + processing_mode: + request_header_mode: SEND + response_header_mode: SKIP + request_trailer_mode: SKIP + response_trailer_mode: SKIP + request_attributes: + - "filter_state['dev.ate.authority']" + mutation_rules: { allow_all_routing: true } + metadata_options: + forwarding_namespaces: + untyped: ["envoy.filters.listener.original_dst"] + receiving_namespaces: + untyped: ["envoy.filters.listener.original_dst"] + - name: envoy.filters.http.router + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router + + clusters: + - name: actor_original_dst + connect_timeout: 5s + type: ORIGINAL_DST + lb_policy: CLUSTER_PROVIDED + original_dst_lb_config: + metadata_key: + key: envoy.filters.listener.original_dst + path: + - key: local + + # The real atenet router's ext_proc server. + - name: ate-cluster + connect_timeout: 0.25s + type: STATIC + lb_policy: ROUND_ROBIN + typed_extension_protocol_options: + envoy.extensions.upstreams.http.v3.HttpProtocolOptions: + "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions + explicit_http_config: + http2_protocol_options: {} + load_assignment: + cluster_name: ate-cluster + endpoints: + - lb_endpoints: + - endpoint: + address: + socket_address: { address: 172.28.0.20, port_value: 50051 } diff --git a/demos/envoy-rust-dynamic-module/envoy/dynmod.yaml b/demos/envoy-rust-dynamic-module/envoy/dynmod.yaml new file mode 100644 index 0000000000..bd49598e6c --- /dev/null +++ b/demos/envoy-rust-dynamic-module/envoy/dynmod.yaml @@ -0,0 +1,185 @@ +# Arm B: the Rust dynamic module does the ingress routing decision in-process. +# +# This mirrors what xds.go builds for the ingress listener, with exactly one +# substitution: the envoy.filters.http.ext_proc filter (which calls out to the +# Go atenet router over gRPC for every request) is replaced by +# envoy.filters.http.dynamic_modules. Everything else -- the set_filter_state +# filter that captures :authority, the ORIGINAL_DST cluster keyed on dynamic +# metadata, the route -- is the same, so the comparison isolates the hop. +admin: + address: + socket_address: { address: 0.0.0.0, port_value: 9901 } + +node: + id: substrate-envoy-node + cluster: substrate-router-cluster + +static_resources: + listeners: + - name: ingress_http_listener + address: + socket_address: { address: 0.0.0.0, port_value: 8080 } + filter_chains: + - filters: + - name: envoy.filters.network.http_connection_manager + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager + stat_prefix: ingress_http + generate_request_id: true + route_config: + name: substrate_routes + virtual_hosts: + - name: local_service + domains: ["*"] + routes: + - match: { prefix: "/" } + route: + cluster: actor_original_dst + timeout: 10s + # Same declarative port propagation xds.go uses: atunnel cannot + # read dynamic metadata, so the port travels as a header. + request_headers_to_add: + - header: + key: x-ate-target-port + value: "%DYNAMIC_METADATA(envoy.filters.listener.original_dst:port)%" + append_action: OVERWRITE_IF_EXISTS_OR_ADD + http_filters: + # Captures :authority into filter state, exactly as + # xds.go's authorityFilterStateFilter does. The module reads it back + # from there, so it behaves identically for CONNECT-tunneled requests + # whose own :authority is not the actor's. + - name: envoy.filters.http.set_filter_state + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.set_filter_state.v3.Config + on_request_headers: + - object_key: dev.ate.authority + # dev.ate.authority is a custom (non-well-known) key, so the + # generic string factory has to be named explicitly -- without + # it Envoy rejects the config with "does not have an object + # factory". Same as xds.go's authorityFilterStateFilter. + factory_key: envoy.string + format_string: + text_format_source: + inline_string: "%REQ(:AUTHORITY)%" + shared_with_upstream: ONCE + # The replacement for ext_proc. + - name: envoy.filters.http.dynamic_modules + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.dynamic_modules.v3.DynamicModuleFilter + dynamic_module_config: + name: ate_router_module + filter_name: actor_router + filter_config: + "@type": type.googleapis.com/google.protobuf.StringValue + value: | + { + "ateapi_cluster": "ateapi", + "cache_ttl_seconds": 5, + "callout_timeout_ms": 5000, + "actor_dns_suffix": "actors.resources.substrate.ate.dev", + "atunnel_port": 443, + "target_port_header": "x-ate-target-port", + "cache_enabled": true + } + - name: envoy.filters.http.router + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router + + # Arm B2: the same module with the cache switched off, so every request pays + # a resolve callout. Isolates the saving that comes from deleting the ext_proc + # hop alone, separately from the saving that comes from caching. + - name: ingress_http_listener_nocache + address: + socket_address: { address: 0.0.0.0, port_value: 8082 } + filter_chains: + - filters: + - name: envoy.filters.network.http_connection_manager + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager + stat_prefix: ingress_http_nocache + generate_request_id: true + route_config: + name: substrate_routes + virtual_hosts: + - name: local_service + domains: ["*"] + routes: + - match: { prefix: "/" } + route: + cluster: actor_original_dst + timeout: 10s + # Same declarative port propagation xds.go uses: atunnel cannot + # read dynamic metadata, so the port travels as a header. + request_headers_to_add: + - header: + key: x-ate-target-port + value: "%DYNAMIC_METADATA(envoy.filters.listener.original_dst:port)%" + append_action: OVERWRITE_IF_EXISTS_OR_ADD + http_filters: + # Captures :authority into filter state, exactly as + # xds.go's authorityFilterStateFilter does. The module reads it back + # from there, so it behaves identically for CONNECT-tunneled requests + # whose own :authority is not the actor's. + - name: envoy.filters.http.set_filter_state + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.set_filter_state.v3.Config + on_request_headers: + - object_key: dev.ate.authority + # dev.ate.authority is a custom (non-well-known) key, so the + # generic string factory has to be named explicitly -- without + # it Envoy rejects the config with "does not have an object + # factory". Same as xds.go's authorityFilterStateFilter. + factory_key: envoy.string + format_string: + text_format_source: + inline_string: "%REQ(:AUTHORITY)%" + shared_with_upstream: ONCE + # The replacement for ext_proc. + - name: envoy.filters.http.dynamic_modules + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.dynamic_modules.v3.DynamicModuleFilter + dynamic_module_config: + name: ate_router_module + filter_name: actor_router + filter_config: + "@type": type.googleapis.com/google.protobuf.StringValue + value: | + { + "ateapi_cluster": "ateapi", + "cache_ttl_seconds": 5, + "callout_timeout_ms": 5000, + "actor_dns_suffix": "actors.resources.substrate.ate.dev", + "atunnel_port": 443, + "target_port_header": "x-ate-target-port", + "cache_enabled": false + } + - name: envoy.filters.http.router + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router + + clusters: + # Mirrors buildOriginalDstCluster in xds.go: the upstream host comes from the + # dynamic metadata the routing decision published, not from :authority. + - name: actor_original_dst + connect_timeout: 5s + type: ORIGINAL_DST + lb_policy: CLUSTER_PROVIDED + original_dst_lb_config: + metadata_key: + key: envoy.filters.listener.original_dst + path: + - key: local + + # The cluster the module calls out to on a cache miss. In production this + # would be ate-apiserver itself; see the README for what that requires. + - name: ateapi + connect_timeout: 1s + type: STRICT_DNS + lb_policy: ROUND_ROBIN + load_assignment: + cluster_name: ateapi + endpoints: + - lb_endpoints: + - endpoint: + address: + socket_address: { address: fakeate, port_value: 8081 } diff --git a/demos/envoy-rust-dynamic-module/fakeate/main.go b/demos/envoy-rust-dynamic-module/fakeate/main.go new file mode 100644 index 0000000000..1477cde4a8 --- /dev/null +++ b/demos/envoy-rust-dynamic-module/fakeate/main.go @@ -0,0 +1,311 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// fakeate is a stand-in ate-apiserver for the Envoy Rust dynamic module demo. +// +// It serves the real ateapipb.Control gRPC service over TLS, exactly as the +// atenet router expects to find it, plus a small JSON endpoint the Rust +// dynamic module calls out to. Both entry points resolve an actor to a worker +// IP through the same code path and the same simulated control-plane cost, so +// the two demo arms are charged identically for a cache miss and the only +// difference measured is where the resolution happens. +// +// It deliberately implements no scheduling: every actor is already RUNNING and +// pinned to a worker by a stable hash. That models the hot-actor case, which is +// where the per-request ResumeActor RPC is pure overhead. +package main + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/json" + "encoding/pem" + "flag" + "fmt" + "hash/fnv" + "log" + "math/big" + "net" + "net/http" + "os" + "strings" + "sync/atomic" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/status" + + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" +) + +var ( + grpcAddr = flag.String("grpc-addr", ":9443", "TLS gRPC listen address for the ateapipb.Control service") + httpAddr = flag.String("http-addr", ":8081", "plaintext HTTP listen address for module callouts and /stats") + caOut = flag.String("ca-out", "/shared/ca.pem", "path to write the generated CA certificate to") + clientOut = flag.String("client-bundle-out", "/shared/client-bundle.pem", "path to write the router's client credential bundle (cert+key PEM) to") + certDNS = flag.String("cert-dns", "fakeate", "DNS SAN to put on the serving certificate") + workersFlag = flag.String("workers", "", "comma-separated worker pod IPs actors are pinned to") + resumeLatency = flag.Duration("resume-latency", time.Millisecond, "simulated control-plane cost of one ResumeActor call") +) + +// stats counts how much control-plane work each arm of the demo actually caused. +// grpcResumes is what the Go ext_proc path drives; httpResumes is what the Rust +// module drives on a cache miss. The gap between them is the headline number. +type stats struct { + grpcResumes atomic.Int64 + httpResumes atomic.Int64 +} + +var counters stats + +// resolver pins each actor to a worker by a stable hash of its name, so a given +// actor always resolves to the same worker for the life of the demo. +type resolver struct{ workers []string } + +func (r *resolver) workerFor(atespace, actor string) string { + h := fnv.New32a() + fmt.Fprintf(h, "%s/%s", atespace, actor) + return r.workers[int(h.Sum32())%len(r.workers)] +} + +type controlServer struct { + ateapipb.UnimplementedControlServer + res *resolver +} + +func (s *controlServer) ResumeActor(ctx context.Context, req *ateapipb.ResumeActorRequest) (*ateapipb.ResumeActorResponse, error) { + counters.grpcResumes.Add(1) + time.Sleep(*resumeLatency) + + ref := req.GetActor() + if ref.GetName() == "" { + return nil, status.Error(codes.InvalidArgument, "actor name is required") + } + return &ateapipb.ResumeActorResponse{ + Actor: s.actor(ref.GetAtespace(), ref.GetName()), + Resumed: false, // the actor was already running: the hot path this demo measures + }, nil +} + +func (s *controlServer) GetActor(ctx context.Context, req *ateapipb.GetActorRequest) (*ateapipb.Actor, error) { + ref := req.GetActor() + return s.actor(ref.GetAtespace(), ref.GetName()), nil +} + +// ListActors exists only because the router's health checker calls it once per +// health interval to decide whether ateapi is reachable. +func (s *controlServer) ListActors(ctx context.Context, req *ateapipb.ListActorsRequest) (*ateapipb.ListActorsResponse, error) { + return &ateapipb.ListActorsResponse{}, nil +} + +func (s *controlServer) actor(atespace, name string) *ateapipb.Actor { + return &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: atespace, Name: name, Uid: atespace + "/" + name, Version: 1}, + ActorTemplateNamespace: "demo", + ActorTemplateName: "echo", + Status: &ateapipb.ActorStatus{ + State: ateapipb.ActorState_ACTOR_STATE_RUNNING, + WorkerAssignment: &ateapipb.WorkerAssignment{ + Worker: &ateapipb.ObjectRef{Name: "worker-1"}, + WorkerNamespace: "ate-system", + WorkerPool: "demo-pool", + WorkerPodIp: s.res.workerFor(atespace, name), + }, + }, + } +} + +func main() { + flag.Parse() + if *workersFlag == "" { + log.Fatal("--workers is required") + } + res := &resolver{workers: strings.Split(*workersFlag, ",")} + + caPEM, srvCert, clientBundle, caPool, err := generateCerts(*certDNS) + if err != nil { + log.Fatalf("generating certificates: %v", err) + } + if err := os.WriteFile(*caOut, caPEM, 0o644); err != nil { + log.Fatalf("writing CA to %s: %v", *caOut, err) + } + if err := os.WriteFile(*clientOut, clientBundle, 0o600); err != nil { + log.Fatalf("writing client bundle to %s: %v", *clientOut, err) + } + log.Printf("wrote CA to %s and client bundle to %s; pinning actors across workers %v", *caOut, *clientOut, res.workers) + + go serveHTTP(res) + + lis, err := net.Listen("tcp", *grpcAddr) + if err != nil { + log.Fatalf("listening on %s: %v", *grpcAddr, err) + } + // Require a client certificate, as the real ate-apiserver does: the router + // authenticates to it with the podidentity credential bundle. Keeping mTLS + // in the demo matters because it is one of the things a dynamic module + // cannot do for itself — see the README's "what stays in Go". + srv := grpc.NewServer(grpc.Creds(credentials.NewTLS(&tls.Config{ + Certificates: []tls.Certificate{srvCert}, + MinVersion: tls.VersionTLS13, + ClientAuth: tls.RequireAndVerifyClientCert, + ClientCAs: caPool, + }))) + ateapipb.RegisterControlServer(srv, &controlServer{res: res}) + log.Printf("fakeate gRPC (TLS) on %s, HTTP on %s, resume latency %s", *grpcAddr, *httpAddr, *resumeLatency) + if err := srv.Serve(lis); err != nil { + log.Fatalf("serving gRPC: %v", err) + } +} + +// serveHTTP exposes the resolve endpoint the Rust dynamic module calls out to, +// plus the counters the benchmark reads. The endpoint is plaintext HTTP because +// the demo's point is the routing path, not the transport; see the README for +// what production would need instead. +func serveHTTP(res *resolver) { + mux := http.NewServeMux() + + // GET /v1/resume?atespace=&actor= -> {"worker_ip":"..."} + // This is the module's cache-miss path. It is charged the same simulated + // control-plane latency as the gRPC ResumeActor above. + mux.HandleFunc("/v1/resume", func(w http.ResponseWriter, r *http.Request) { + counters.httpResumes.Add(1) + time.Sleep(*resumeLatency) + + atespace := r.URL.Query().Get("atespace") + actor := r.URL.Query().Get("actor") + if atespace == "" || actor == "" { + http.Error(w, `{"error":"atespace and actor are required"}`, http.StatusBadRequest) + return + } + w.Header().Set("content-type", "application/json") + json.NewEncoder(w).Encode(map[string]string{"worker_ip": res.workerFor(atespace, actor)}) + }) + + mux.HandleFunc("/stats", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("content-type", "application/json") + json.NewEncoder(w).Encode(map[string]int64{ + "grpc_resume_calls": counters.grpcResumes.Load(), + "http_resume_calls": counters.httpResumes.Load(), + }) + }) + + mux.HandleFunc("/stats/reset", func(w http.ResponseWriter, r *http.Request) { + counters.grpcResumes.Store(0) + counters.httpResumes.Store(0) + w.WriteHeader(http.StatusNoContent) + }) + + if err := http.ListenAndServe(*httpAddr, mux); err != nil { + log.Fatalf("serving HTTP: %v", err) + } +} + +// generateCerts mints a throwaway CA and a serving certificate for it. The +// router verifies the server against the CA PEM this returns, which is the same +// trust model as the real deployment, just with a CA that lives for one demo run. +func generateCerts(dnsName string) (caPEM []byte, srvCert tls.Certificate, clientBundle []byte, caPool *x509.CertPool, err error) { + caKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + return nil, tls.Certificate{}, nil, nil, err + } + caTmpl := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "fakeate-demo-ca"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(24 * time.Hour), + IsCA: true, + KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature, + BasicConstraintsValid: true, + } + caDER, err := x509.CreateCertificate(rand.Reader, caTmpl, caTmpl, &caKey.PublicKey, caKey) + if err != nil { + return nil, tls.Certificate{}, nil, nil, err + } + caCert, err := x509.ParseCertificate(caDER) + if err != nil { + return nil, tls.Certificate{}, nil, nil, err + } + + srvKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + return nil, tls.Certificate{}, nil, nil, err + } + srvTmpl := &x509.Certificate{ + SerialNumber: big.NewInt(2), + Subject: pkix.Name{CommonName: dnsName}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(24 * time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + DNSNames: []string{dnsName, "localhost"}, + IPAddresses: []net.IP{net.ParseIP("127.0.0.1")}, + } + srvDER, err := x509.CreateCertificate(rand.Reader, srvTmpl, caCert, &srvKey.PublicKey, caKey) + if err != nil { + return nil, tls.Certificate{}, nil, nil, err + } + srvKeyDER, err := x509.MarshalPKCS8PrivateKey(srvKey) + if err != nil { + return nil, tls.Certificate{}, nil, nil, err + } + + // The router's client identity, standing in for the podidentity bundle. + cliKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + return nil, tls.Certificate{}, nil, nil, err + } + cliTmpl := &x509.Certificate{ + SerialNumber: big.NewInt(3), + Subject: pkix.Name{CommonName: "atenet-router"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(24 * time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth}, + } + cliDER, err := x509.CreateCertificate(rand.Reader, cliTmpl, caCert, &cliKey.PublicKey, caKey) + if err != nil { + return nil, tls.Certificate{}, nil, nil, err + } + cliKeyDER, err := x509.MarshalPKCS8PrivateKey(cliKey) + if err != nil { + return nil, tls.Certificate{}, nil, nil, err + } + + caPEM = pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: caDER}) + srvCert, err = tls.X509KeyPair( + pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: srvDER}), + pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: srvKeyDER}), + ) + if err != nil { + return nil, tls.Certificate{}, nil, nil, err + } + + // credbundle expects the certificate chain and the PKCS#8 key in one file. + clientBundle = append( + pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: cliDER}), + pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: cliKeyDER})..., + ) + + caPool = x509.NewCertPool() + caPool.AddCert(caCert) + + return caPEM, srvCert, clientBundle, caPool, nil +} diff --git a/demos/envoy-rust-dynamic-module/loadgen/main.go b/demos/envoy-rust-dynamic-module/loadgen/main.go new file mode 100644 index 0000000000..14d2d778ef --- /dev/null +++ b/demos/envoy-rust-dynamic-module/loadgen/main.go @@ -0,0 +1,193 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// loadgen drives one arm of the demo and reports the end-to-end latency +// distribution. +// +// It spreads requests over a fixed set of actors, addressing each by the actor +// DNS name the router parses, which is the shape of traffic the comparison is +// about: a working set of already-running actors, hit repeatedly. That is +// exactly the case the current router spends one ResumeActor RPC on per +// request and the Rust module serves from cache. +package main + +import ( + "context" + "encoding/json" + "flag" + "fmt" + "io" + "log" + "math/rand" + "net/http" + "os" + "sort" + "sync" + "sync/atomic" + "time" +) + +var ( + target = flag.String("target", "http://localhost:10000", "base URL of the gateway under test") + actors = flag.Int("actors", 50, "size of the hot actor working set") + atespace = flag.String("atespace", "demo", "atespace the actors live in") + suffix = flag.String("suffix", "actors.resources.substrate.ate.dev", "actor DNS suffix") + concurrency = flag.Int("concurrency", 32, "number of concurrent connections") + duration = flag.Duration("duration", 20*time.Second, "measurement window") + warmup = flag.Duration("warmup", 3*time.Second, "warmup window, excluded from the reported numbers") + label = flag.String("label", "arm", "label for this run") + jsonOut = flag.String("json-out", "", "if set, write the result as JSON to this path") +) + +type result struct { + Label string `json:"label"` + Requests int64 `json:"requests"` + Errors int64 `json:"errors"` + NonOK int64 `json:"non_2xx"` + RPS float64 `json:"rps"` + P50ms float64 `json:"p50_ms"` + P90ms float64 `json:"p90_ms"` + P95ms float64 `json:"p95_ms"` + P99ms float64 `json:"p99_ms"` + MaxMs float64 `json:"max_ms"` + MeanMs float64 `json:"mean_ms"` + Concurrency int `json:"concurrency"` + Actors int `json:"actors"` +} + +func main() { + flag.Parse() + + // One transport shared by every worker, with a connection pool at least as + // large as the concurrency: otherwise the client, not the gateway, becomes + // the bottleneck and both arms measure the same thing. + transport := &http.Transport{ + MaxIdleConns: *concurrency * 2, + MaxIdleConnsPerHost: *concurrency * 2, + MaxConnsPerHost: *concurrency * 2, + IdleConnTimeout: 90 * time.Second, + } + client := &http.Client{Transport: transport, Timeout: 30 * time.Second} + + var ( + mu sync.Mutex + latencies []time.Duration + errs atomic.Int64 + nonOK atomic.Int64 + measuring atomic.Bool + ) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + var wg sync.WaitGroup + for i := 0; i < *concurrency; i++ { + wg.Add(1) + go func(seed int) { + defer wg.Done() + rng := rand.New(rand.NewSource(int64(seed))) + local := make([]time.Duration, 0, 4096) + for ctx.Err() == nil { + actor := fmt.Sprintf("actor-%03d", rng.Intn(*actors)) + host := fmt.Sprintf("%s.%s.%s", actor, *atespace, *suffix) + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, *target+"/", nil) + if err != nil { + continue + } + req.Host = host + + start := time.Now() + resp, err := client.Do(req) + elapsed := time.Since(start) + if err != nil { + if ctx.Err() == nil { + errs.Add(1) + } + continue + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + if resp.StatusCode/100 != 2 { + nonOK.Add(1) + } + if measuring.Load() { + local = append(local, elapsed) + } + } + mu.Lock() + latencies = append(latencies, local...) + mu.Unlock() + }(i) + } + + time.Sleep(*warmup) + measuring.Store(true) + measureStart := time.Now() + time.Sleep(*duration) + measured := time.Since(measureStart) + measuring.Store(false) + cancel() + wg.Wait() + + if len(latencies) == 0 { + log.Fatalf("%s: no successful requests were recorded", *label) + } + sort.Slice(latencies, func(i, j int) bool { return latencies[i] < latencies[j] }) + + ms := func(d time.Duration) float64 { return float64(d.Nanoseconds()) / 1e6 } + pct := func(p float64) float64 { + idx := int(p / 100 * float64(len(latencies))) + if idx >= len(latencies) { + idx = len(latencies) - 1 + } + return ms(latencies[idx]) + } + var total time.Duration + for _, d := range latencies { + total += d + } + + r := result{ + Label: *label, + Requests: int64(len(latencies)), + Errors: errs.Load(), + NonOK: nonOK.Load(), + RPS: float64(len(latencies)) / measured.Seconds(), + P50ms: pct(50), + P90ms: pct(90), + P95ms: pct(95), + P99ms: pct(99), + MaxMs: ms(latencies[len(latencies)-1]), + MeanMs: ms(total / time.Duration(len(latencies))), + Concurrency: *concurrency, + Actors: *actors, + } + + fmt.Printf("%-22s reqs=%-8d rps=%-9.0f p50=%-7.2f p95=%-7.2f p99=%-7.2f max=%-8.2f err=%d non2xx=%d\n", + r.Label, r.Requests, r.RPS, r.P50ms, r.P95ms, r.P99ms, r.MaxMs, r.Errors, r.NonOK) + + if *jsonOut != "" { + f, err := os.Create(*jsonOut) + if err != nil { + log.Fatalf("creating %s: %v", *jsonOut, err) + } + defer f.Close() + enc := json.NewEncoder(f) + enc.SetIndent("", " ") + if err := enc.Encode(r); err != nil { + log.Fatalf("writing result: %v", err) + } + } +} diff --git a/demos/envoy-rust-dynamic-module/rust-module/Cargo.lock b/demos/envoy-rust-dynamic-module/rust-module/Cargo.lock new file mode 100644 index 0000000000..98ca82875d --- /dev/null +++ b/demos/envoy-rust-dynamic-module/rust-module/Cargo.lock @@ -0,0 +1,466 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "aho-corasick" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba" +dependencies = [ + "memchr", +] + +[[package]] +name = "anstyle" +version = "1.0.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" + +[[package]] +name = "ate-router-module" +version = "0.1.0" +dependencies = [ + "dashmap", + "envoy-proxy-dynamic-modules-rust-sdk", + "serde", + "serde_json", +] + +[[package]] +name = "bindgen" +version = "0.70.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f49d8fed880d473ea71efb9bf597651e77201bdd4893efe54c9e5d65ae04ce6f" +dependencies = [ + "bitflags", + "cexpr", + "clang-sys", + "itertools", + "log", + "prettyplease", + "proc-macro2", + "quote", + "regex", + "rustc-hash", + "shlex", + "syn 2.0.119", +] + +[[package]] +name = "bitflags" +version = "2.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da" + +[[package]] +name = "cexpr" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766" +dependencies = [ + "nom", +] + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "clang-sys" +version = "1.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "157a8ba7b480713b56f4c09fd13fc3e0a22a5dfab8097ba61cbc5feef950788a" +dependencies = [ + "glob", + "libc", + "libloading", +] + +[[package]] +name = "crossbeam-utils" +version = "0.8.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61803da095bee82a81bb1a452ecc25d3b2f1416d1897eb86430c6159ef717c17" + +[[package]] +name = "dashmap" +version = "6.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6361d5c062261c78a176addb82d4c821ae42bed6089de0e12603cd25de2059c" +dependencies = [ + "cfg-if", + "crossbeam-utils", + "hashbrown", + "lock_api", + "once_cell", + "parking_lot_core", +] + +[[package]] +name = "downcast" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1435fa1053d8b2fbbe9be7e97eca7f33d37b28409959813daefc1446a14247f1" + +[[package]] +name = "either" +version = "1.18.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "252afb9ae5eaa683babdc6a068b3f5726eb19e05070c731f9b2a23a7c3e8ed34" + +[[package]] +name = "envoy-proxy-dynamic-modules-rust-sdk" +version = "0.1.0" +source = "git+https://github.com/envoyproxy/envoy?rev=b579d07d3ad7ee11d32b105e91a5a39ad24718d7#b579d07d3ad7ee11d32b105e91a5a39ad24718d7" +dependencies = [ + "bindgen", + "mockall", +] + +[[package]] +name = "fragile" +version = "2.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8878864ba14bb86e818a412bfd6f18f9eabd4ec0f008a28e8f7eb61db532fcf9" +dependencies = [ + "futures-core", +] + +[[package]] +name = "futures-core" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" + +[[package]] +name = "glob" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e4eba85ea1d0a966a983acd07deee566e67395d2d96b6fb39e62b5a833f1eb0b" + +[[package]] +name = "hashbrown" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" + +[[package]] +name = "itertools" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186" +dependencies = [ + "either", +] + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "libc" +version = "0.2.189" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" + +[[package]] +name = "libloading" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7c4b02199fee7c5d21a5ae7d8cfa79a6ef5bb2fc834d6e9058e89c825efdc55" +dependencies = [ + "cfg-if", + "windows-link", +] + +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + +[[package]] +name = "log" +version = "0.4.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "minimal-lexical" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" + +[[package]] +name = "mockall" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39a6bfcc6c8c7eed5ee98b9c3e33adc726054389233e201c95dab2d41a3839d2" +dependencies = [ + "cfg-if", + "downcast", + "fragile", + "mockall_derive", + "predicates", + "predicates-tree", +] + +[[package]] +name = "mockall_derive" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "25ca3004c2efe9011bd4e461bd8256445052b9615405b4f7ea43fc8ca5c20898" +dependencies = [ + "cfg-if", + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "nom" +version = "7.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a" +dependencies = [ + "memchr", + "minimal-lexical", +] + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-link", +] + +[[package]] +name = "predicates" +version = "3.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ada8f2932f28a27ee7b70dd6c1c39ea0675c55a36879ab92f3a715eaa1e63cfe" +dependencies = [ + "anstyle", + "predicates-core", +] + +[[package]] +name = "predicates-core" +version = "1.0.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cad38746f3166b4031b1a0d39ad9f954dd291e7854fcc0eed52ee41a0b50d144" + +[[package]] +name = "predicates-tree" +version = "1.0.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0de1b847b39c8131db0467e9df1ff60e6d0562ab8e9a16e568ad0fdb372e2f2" +dependencies = [ + "predicates-core", + "termtree", +] + +[[package]] +name = "prettyplease" +version = "0.2.37" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" +dependencies = [ + "proc-macro2", + "syn 2.0.119", +] + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", +] + +[[package]] +name = "regex" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f020237b6c8eed93db2e2cb53c00c60a8e1bc73da7d073199a1180401450218d" +dependencies = [ + "aho-corasick", + "memchr", + "regex-automata", + "regex-syntax", +] + +[[package]] +name = "regex-automata" +version = "0.4.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ad8553b9b26413251cbf30e620595c7a41b3887f03da04579c0e6b0d6a06b4b2" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax", +] + +[[package]] +name = "regex-syntax" +version = "0.8.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" + +[[package]] +name = "rustc-hash" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08d43f7aa6b08d49f382cde6a7982047c3426db949b1424bc4b7ec9ae12c6ce2" + +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + +[[package]] +name = "serde_json" +version = "1.0.151" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "shlex" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" + +[[package]] +name = "smallvec" +version = "1.15.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" + +[[package]] +name = "syn" +version = "2.0.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "syn" +version = "3.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6275cddf4610d1775e6d1fe9469b2e77d0f39fd98fb7450901b821e0c53649f" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "termtree" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f50febec83f5ee1df3015341d8bd429f2d1cc62bcba7ea2076759d315084683" + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/demos/envoy-rust-dynamic-module/rust-module/Cargo.toml b/demos/envoy-rust-dynamic-module/rust-module/Cargo.toml new file mode 100644 index 0000000000..ef63e296b0 --- /dev/null +++ b/demos/envoy-rust-dynamic-module/rust-module/Cargo.toml @@ -0,0 +1,24 @@ +[package] +name = "ate-router-module" +version = "0.1.0" +edition = "2021" +license = "Apache-2.0" + +[dependencies] +# The SDK revision MUST match the Envoy binary it is loaded into. This rev is the +# commit of the envoyproxy/envoy release tagged v1.39.1, which is the version +# reported by `envoy --version` in envoyproxy/envoy:v1.39-latest. +envoy-proxy-dynamic-modules-rust-sdk = { git = "https://github.com/envoyproxy/envoy", rev = "b579d07d3ad7ee11d32b105e91a5a39ad24718d7" } +serde = { version = "1.0", features = ["derive"] } +serde_json = "1.0" +dashmap = "6.1" + +[lib] +name = "ate_router_module" +path = "src/lib.rs" +crate-type = ["cdylib", "rlib"] + +[profile.release] +opt-level = 3 +lto = true +codegen-units = 1 diff --git a/demos/envoy-rust-dynamic-module/rust-module/DESIGN.md b/demos/envoy-rust-dynamic-module/rust-module/DESIGN.md new file mode 100644 index 0000000000..bef87577e7 --- /dev/null +++ b/demos/envoy-rust-dynamic-module/rust-module/DESIGN.md @@ -0,0 +1,202 @@ +# Rust module design + +Internal design of `ate_router_module`. The system-level case for it — why a +cache in front of ext_proc, what changes in `xds.go`, how to roll it out — is in +[../DESIGN.md](../DESIGN.md). This document is about the module itself. + +## What it is + +Envoy's ext_proc client has **no cache and no way to grow one**: the filter's +only cache-shaped fields (`disable_clear_route_cache`, `route_cache_action`) +concern Envoy's route cache, not memoizing a `ProcessingResponse`. Adding one +means patching Envoy C++. + +This module is that cache, assembled from two supported extension points: + +* `envoy.filters.http.dynamic_modules` — an HTTP filter running in-process, + ahead of ext_proc in the chain; +* `ExtProcPerRoute.disabled` — the per-route override the module steers into to + suppress the ext_proc call on a hit. + +Nothing else in the chain changes. + +## Compatibility + +The SDK is a git dependency on the Envoy repo, pinned to **the commit of the +Envoy binary it loads into** — `b579d07d3ad7ee11d32b105e91a5a39ad24718d7` +(v1.39.1), matching the `envoyproxy/envoy:v1.39-latest` substrate pins. Envoy +guarantees a module built for X.Y works on X.Y and X.(Y+1) only, so **the pin is +a hard coupling**: bumping the Envoy image is a module rebuild, and that +constraint belongs in whatever bumps the image. + +The `.so` must also be built against a glibc **no newer** than the Envoy image's +(2.35 on Ubuntu 22.04). The build container is `rust:1-slim-bullseye` +(glibc 2.31) for exactly this reason. + +## Layout + +``` +declare_init_functions!(init, new_http_filter_config_fn) lib.rs:40 + │ + ├─ init() once per process, at dlopen + └─ new_http_filter_config_fn() once per filter-chain config + │ parses JSON config, defines Envoy-native counters + └─ FilterConfig ────────────────── lib.rs:160 + │ new_http_filter() per HTTP request + ├─ Filter (mode: replace) lib.rs:207 + └─ CacheFilter (mode: co-exist) lib.rs:414 + ▲ + both read/write one process-global cache + │ + cache(): &'static DashMap lib.rs:153 +``` + +`filter_name` in the Envoy config selects the mode: `actor_router` replaces +ext_proc outright, `actor_router_cache` sits in front of it. **`actor_router_cache` +is the mode intended for production**; `actor_router` exists to measure the +ceiling. + +## State and the thread model + +Envoy `dlopen`s the `.so` once and runs the filter chain on N worker threads. +Filter instances are per-request and single-threaded; anything shared is not. + +The cache is therefore process-global, behind a `OnceLock`: + +```rust +fn cache() -> &'static DashMap { + static CACHE: OnceLock> = OnceLock::new(); + CACHE.get_or_init(DashMap::new) +} +``` + +`DashMap` shards its locks, so worker threads don't serialize on a single mutex +the way they would behind `RwLock`. Process-global also means the cache +**survives an ECDS config redelivery** that re-runs `new_http_filter_config_fn` — +a config push does not cold-start the cache. + +Key: `format!("{}/{}", atespace, name)`, matching `resources.ActorRef`. It must +be atespace-qualified; an actor name is only unique within its atespace. + +Value: + +```rust +struct Binding { + worker_ip: String, + expires_at: Instant, +} +``` + +Only the worker IP is cached. The port is configuration, never a parsed value, +so a malformed control-plane answer cannot redirect traffic to another port. + +## Request path (`actor_router_cache`) + +**`on_request_headers`** + +1. Read the authority: `get_filter_state_bytes(b"dev.ate.authority")`, falling + back to `:authority`. Filter state is what the Go handler reads too, and it + is the only correct source — a reinjected CONNECT tunnel's own `:authority` + has nothing to do with the actor. +2. Parse `..[:port]`. +3. Cache lookup. + * **Hit, live** → publish `envoy.filters.listener.original_dst` dynamic + metadata (`local` = `ip:443`, `port` = actor port), set the target-port + header, set `x-ate-route-resolved: 1`, `clear_route_cache()`, `Continue`. + Route re-selection now lands on the route carrying + `ExtProcPerRoute.disabled`, and ext_proc never runs. + * **Hit, expired** → remove, fall through. + * **Miss** → record the key in `learn_key`, `Continue`. ext_proc runs exactly + as today. + +**`on_response_headers`** — if this request missed, read back the +`original_dst`/`local` metadata **ext_proc published**, split off the port, and +insert the binding. Reaching response headers means the request was actually +routed, so the module only ever caches a decision that demonstrably worked. + +The module never calls the control plane in this mode. It cannot route anywhere +ext_proc has not already routed. + +## Failure behaviour: everything unknown is a `Continue` + +A helper filter must never turn a request away. Missing filter state, an +unparseable authority, an empty cache, an expired entry, a malformed metadata +value — all return `Continue` and let ext_proc handle it, including producing the +exact 404/503 bodies clients see today. `ingress/errors.go` still runs on every +miss. The module has **no error responses of its own** in co-existence mode. + +That is what makes removal a no-op: delete the filter, and every path it touches +was already the ext_proc path. + +## Config + +JSON, via `filter_config` as a `google.protobuf.StringValue`: + +```json +{ + "ateapi_cluster": "ateapi", + "cache_ttl_seconds": 5, + "callout_timeout_ms": 5000, + "actor_dns_suffix": "actors.resources.substrate.ate.dev", + "atunnel_port": 443, + "target_port_header": "x-ate-target-port", + "cache_enabled": true +} +``` + +Every field except `ateapi_cluster` has a serde default. `ateapi_cluster` is +unused in co-existence mode. A config that fails to parse returns `None` from +the factory, which makes Envoy **reject the config** rather than start with a +filter that silently does nothing. + +## Observability + +Counters are Envoy-native, defined once per filter config via `define_counter` +and incremented by id — `ate_router.cache_hit`, `ate_router.cache_miss`. They +appear in Envoy's own `/stats` (`dynamicmodulescustom.ate_router.*`) next to +every other dataplane counter, with no per-request attribute allocation, unlike +the OTel histogram with four string attributes the Go path records per request. + +Hit rate is the operational signal: it should approach `1 - M/(N·T)`. A hit rate +that falls without a matching change in the actor working set means bindings are +churning, and the TTL is too long. + +## Replacement mode (`actor_router`), and why it is not the recommendation + +`Filter` resolves misses itself: `send_http_callout` to a configured cluster, +`StopIteration`, then `on_http_callout_done` parses the JSON, validates the IP, +caches, routes, and `continue_decoding()`. Measured 4% faster than co-existence. + +It is not the recommendation because it costs: an ate-apiserver client identity +in the dataplane (callouts are HTTP to an Envoy cluster, not authenticated gRPC), +a JSON resolve endpoint that does not exist today, and Rust reimplementations of +parking, singleflight and the error contract. It exists in the tree to show what +the remaining 4% is worth, which is: not much. + +## Not yet implemented + +The module as written is TTL-only. Before it fronts traffic it needs, in +`on_response_headers` and on stream reset: + +1. **Evict on `421` + `x-ate-assignment-stale`**, then **re-run the slow path in + the same request** — clear the header, `clear_route_cache()`, retry through + ext_proc. A cache hit skips `ResumeActor`, which is what wakes a suspended + actor, so evict-and-fail would turn the TTL into a user-visible error window. + This is the one place the design can regress behaviour. +2. **Evict on upstream reset / local 503**, since a vanished worker produces no + 421 at all. +3. **A bounded cache.** The `DashMap` grows with the actor working set and is + only pruned on lookup. It needs a size cap with LRU, or a sweeper on a + `new_scheduler` timer. +4. **Miss coalescing**, so a cold actor with simultaneous requests does not + produce a burst of ext_proc streams. Not a blocker — ext_proc's own + singleflight collapses the `ResumeActor` calls behind them — but it gives up + an easy win. + +## Risk that does not go away + +The module is not sandboxed. A panic or a bad pointer takes Envoy down with it, +where a Go router crash fails only ext_proc streams and Envoy keeps serving. The +SDK catches unwinds at the ABI boundary, and this module holds no unsafe code and +no raw pointers, but "in-process" is a different operational class from "separate +container" and it should be soaked before it fronts real traffic. diff --git a/demos/envoy-rust-dynamic-module/rust-module/src/lib.rs b/demos/envoy-rust-dynamic-module/rust-module/src/lib.rs new file mode 100644 index 0000000000..d929194c81 --- /dev/null +++ b/demos/envoy-rust-dynamic-module/rust-module/src/lib.rs @@ -0,0 +1,504 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! An Envoy dynamic module that does the atenet router's ingress routing +//! in-process, instead of over an ext_proc gRPC hop to the Go router. +//! +//! It is a like-for-like replacement for +//! `cmd/atenet/internal/router/ingress.Handler.HandleRequestHeaders`: it reads +//! the same authority from the same filter-state key, parses the same actor DNS +//! name, and publishes the same `envoy.filters.listener.original_dst` dynamic +//! metadata that the ORIGINAL_DST cluster in `xds.go` routes on. The difference +//! is where the work happens and how often the control plane is consulted: +//! +//! * The Go path calls `ResumeActor` on ate-apiserver for **every** request, +//! including requests to an actor that is already running on a known worker. +//! * This module answers those from a TTL cache shared across Envoy's worker +//! threads, and only calls the control plane on a miss. +//! +//! The cache is deliberately conservative: it holds only the actor -> worker +//! binding, for a short TTL, and a routing failure evicts the entry so the next +//! request re-resolves. See the README for why that is safe and where it is not. + +use dashmap::DashMap; +use envoy_proxy_dynamic_modules_rust_sdk::*; +use serde::Deserialize; +use std::sync::OnceLock; +use std::time::{Duration, Instant}; + +declare_init_functions!(init, new_http_filter_config_fn); + +fn init() -> bool { + true +} + +fn new_http_filter_config_fn( + _envoy_filter_config: &mut EC, + filter_name: &str, + filter_config: &[u8], +) -> Option>> { + match filter_name { + // Two modes, sharing one cache: + // + // "actor_router" full replacement: resolves misses itself with an + // HTTP callout, and ext_proc is not in the chain. + // "actor_router_cache" co-existence: a cache in FRONT of the existing + // ext_proc filter. It answers hits and lets every + // miss fall through to ext_proc untouched, then + // learns the binding from what ext_proc decided. + // Nothing about the Go path changes. + "actor_router" | "actor_router_cache" => { + let raw = std::str::from_utf8(filter_config).ok()?; + let config: Config = serde_json::from_str(raw) + .map_err(|e| { + envoy_log_error!("actor_router: invalid filter config: {}", e); + }) + .ok()?; + envoy_log_info!( + "actor_router: cluster={} ttl={}s suffix={}", + config.ateapi_cluster, + config.cache_ttl_seconds, + config.actor_dns_suffix + ); + // Counters are Envoy-native stats, defined once per filter config and + // incremented by id. They show up in Envoy's own /stats alongside every + // other dataplane counter, with no per-request attribute allocation -- + // unlike the OTel histogram the Go path records per request. + Some(Box::new(FilterConfig { + cache_hit: _envoy_filter_config.define_counter("ate_router.cache_hit").ok(), + cache_miss: _envoy_filter_config.define_counter("ate_router.cache_miss").ok(), + coexist: filter_name == "actor_router_cache", + settings: config, + })) + } + other => { + envoy_log_error!("actor_router: unknown filter name {}", other); + None + } + } +} + +/// Config is the filter's `filter_config`, supplied as JSON in the Envoy +/// listener config. FilterConfig below pairs it with the Envoy counter ids; the +/// filter itself keeps a clone of these settings. +#[derive(Deserialize, Clone)] +struct Config { + /// Envoy cluster name to send the resolve callout to on a cache miss. Must + /// be defined in the Envoy config; the module cannot invent clusters. + ateapi_cluster: String, + /// How long an actor -> worker binding may be served from cache. Must stay + /// well below the control plane's idle-suspend timeout: see README. + #[serde(default = "default_ttl")] + cache_ttl_seconds: u64, + /// Callout timeout on a cache miss. + #[serde(default = "default_timeout")] + callout_timeout_ms: u64, + /// The DNS suffix every actor authority ends with. + #[serde(default = "default_suffix")] + actor_dns_suffix: String, + /// The port atunnel listens on at the worker. The Go handler hardcodes 443. + #[serde(default = "default_atunnel_port")] + atunnel_port: u16, + /// Header carrying the actor's target port to atunnel, which cannot read + /// dynamic metadata. Mirrors atunnel.TargetPortHeader. + #[serde(default = "default_port_header")] + target_port_header: String, + /// Set false to measure the module with caching off, which isolates the + /// saving from removing the ext_proc hop alone. + #[serde(default = "default_cache_enabled")] + cache_enabled: bool, +} + +fn default_ttl() -> u64 { 5 } +fn default_timeout() -> u64 { 5000 } +fn default_suffix() -> String { "actors.resources.substrate.ate.dev".to_string() } +fn default_atunnel_port() -> u16 { 443 } +fn default_port_header() -> String { "x-ate-target-port".to_string() } +fn default_cache_enabled() -> bool { true } + +/// Filter-state key the dataplane publishes the real authority under. Ingress +/// reads this rather than :authority because a reinjected CONNECT tunnel's own +/// authority has nothing to do with the actor. +/// Mirrors ingress.AuthorityFilterStateKey. +const AUTHORITY_FILTER_STATE_KEY: &[u8] = b"dev.ate.authority"; + +/// Dynamic-metadata namespace and keys the ORIGINAL_DST cluster reads. +/// Mirrors ingress.OriginalDstMetadataKey / OriginalDstAddressKey / OriginalDstPortKey. +const ORIGINAL_DST_NAMESPACE: &str = "envoy.filters.listener.original_dst"; +const ORIGINAL_DST_ADDRESS_KEY: &str = "local"; +const ORIGINAL_DST_PORT_KEY: &str = "port"; + +/// A resolved actor -> worker binding. +#[derive(Clone)] +struct Binding { + worker_ip: String, + expires_at: Instant, +} + +/// The cache is process-global and shared by every Envoy worker thread: the +/// module is dlopen'd once, so one map serves all of them. DashMap gives +/// sharded locking, so worker threads do not contend on a single mutex the way +/// they would behind a global RwLock. +fn cache() -> &'static DashMap { + static CACHE: OnceLock> = OnceLock::new(); + CACHE.get_or_init(DashMap::new) +} + +/// FilterConfig is the per-filter-chain configuration: the settings parsed from +/// JSON, plus the Envoy counter ids defined once at config load. +struct FilterConfig { + settings: Config, + cache_hit: Option, + cache_miss: Option, + coexist: bool, +} + +impl HttpFilterConfig for FilterConfig { + fn new_http_filter(&self, _envoy: &mut EHF) -> Box> { + if self.coexist { + return Box::new(CacheFilter { + config: self.settings.clone(), + cache_hit: self.cache_hit, + cache_miss: self.cache_miss, + learn_key: None, + }); + } + Box::new(Filter { + config: self.settings.clone(), + cache_hit: self.cache_hit, + cache_miss: self.cache_miss, + pending: None, + }) + } +} + +/// An actor reference parsed out of a request authority. +struct ActorRef { + atespace: String, + name: String, + /// The port named in the authority, or the actor's default port. + target_port: u16, +} + +impl ActorRef { + fn cache_key(&self) -> String { + format!("{}/{}", self.atespace, self.name) + } +} + +/// State carried from on_request_headers to on_http_callout_done for a request +/// that missed the cache. +struct Pending { + actor: ActorRef, + callout_id: u64, +} + +struct Filter { + config: Config, + cache_hit: Option, + cache_miss: Option, + pending: Option, +} + +/// Helpers shared by both filter modes. They depend only on the parsed config, +/// so they live here rather than on either filter -- in particular so the +/// co-existence filter never has to name the callout filter's type. +impl Config { + /// Publishes the routing decision the same way the Go ingress handler does: + /// dynamic metadata for the ORIGINAL_DST cluster, plus the target-port + /// header for atunnel. :authority is deliberately left untouched so atunnel + /// still authorizes by the actor's own DNS name. + fn route_to(&self, envoy_filter: &mut EHF, actor: &ActorRef, worker_ip: &str) { + let target = format!("{}:{}", worker_ip, self.atunnel_port); + envoy_filter.set_dynamic_metadata_string(ORIGINAL_DST_NAMESPACE, ORIGINAL_DST_ADDRESS_KEY, &target); + envoy_filter.set_dynamic_metadata_string( + ORIGINAL_DST_NAMESPACE, + ORIGINAL_DST_PORT_KEY, + &actor.target_port.to_string(), + ); + envoy_filter.set_request_header( + &self.target_port_header, + actor.target_port.to_string().as_bytes(), + ); + } + + /// Reads the authority the dataplane resolved for this request, preferring + /// the filter-state key the ingress listener publishes and falling back to + /// :authority when the listener does not set it. + fn authority(&self, envoy_filter: &EHF) -> Option { + if let Some(buf) = envoy_filter.get_filter_state_bytes(AUTHORITY_FILTER_STATE_KEY) { + if !buf.as_slice().is_empty() { + return std::str::from_utf8(buf.as_slice()).ok().map(str::to_string); + } + } + envoy_filter + .get_request_header_value(":authority") + .and_then(|b| std::str::from_utf8(b.as_slice()).ok().map(str::to_string)) + } + + /// Parses ".." with an optional ":port". + /// Mirrors resources.ParseActorDNSName + ingress.parseActorRef. + fn parse_actor_ref(&self, authority: &str) -> Option { + let (host, target_port) = match authority.rsplit_once(':') { + Some((h, p)) => (h, p.parse::().ok()?), + None => (authority, 80), + }; + let labels = host.strip_suffix(&self.actor_dns_suffix)?.strip_suffix('.')?; + let (name, atespace) = labels.split_once('.')?; + if name.is_empty() || atespace.is_empty() || atespace.contains('.') { + return None; + } + Some(ActorRef { + atespace: atespace.to_string(), + name: name.to_string(), + target_port, + }) + } +} + +impl HttpFilter for Filter { + fn on_request_headers( + &mut self, + envoy_filter: &mut EHF, + _end_of_stream: bool, + ) -> abi::envoy_dynamic_module_type_on_http_filter_request_headers_status { + let Some(authority) = self.config.authority(envoy_filter) else { + envoy_filter.send_response(404, &[], Some(b"no authority on request"), None); + return abi::envoy_dynamic_module_type_on_http_filter_request_headers_status::StopIteration; + }; + + let Some(actor) = self.config.parse_actor_ref(&authority) else { + // Same disposition as ingress.invalidHostErr: an authority that is + // not an actor DNS name is a 404, not a 500. + envoy_filter.send_response(404, &[], Some(b"invalid actor host"), None); + return abi::envoy_dynamic_module_type_on_http_filter_request_headers_status::StopIteration; + }; + + // Fast path: a live binding answers without touching the control plane. + // This is the request the Go path spends a full ResumeActor RPC on. + if self.config.cache_enabled { + let key = actor.cache_key(); + if let Some(entry) = cache().get(&key) { + if entry.expires_at > Instant::now() { + let worker_ip = entry.worker_ip.clone(); + drop(entry); + if let Some(id) = self.cache_hit { + let _ = envoy_filter.increment_counter(id, 1); + } + self.config.route_to(envoy_filter, &actor, &worker_ip); + return abi::envoy_dynamic_module_type_on_http_filter_request_headers_status::Continue; + } + // Expired: drop it so a concurrent request cannot serve it either. + drop(entry); + cache().remove(&key); + } + } + + // Slow path: resolve through the control plane, holding the request. + if let Some(id) = self.cache_miss { + let _ = envoy_filter.increment_counter(id, 1); + } + let path = format!( + "/v1/resume?atespace={}&actor={}", + actor.atespace, actor.name + ); + let (result, callout_id) = envoy_filter.send_http_callout( + &self.config.ateapi_cluster, + &[ + (":method", b"GET"), + (":path", path.as_bytes()), + (":authority", self.config.ateapi_cluster.as_bytes()), + ], + None, + self.config.callout_timeout_ms, + ); + if result != abi::envoy_dynamic_module_type_http_callout_init_result::Success { + envoy_log_error!("actor_router: callout init failed: {:?}", result); + envoy_filter.send_response(503, &[], Some(b"control plane unreachable"), None); + return abi::envoy_dynamic_module_type_on_http_filter_request_headers_status::StopIteration; + } + + self.pending = Some(Pending { actor, callout_id }); + abi::envoy_dynamic_module_type_on_http_filter_request_headers_status::StopIteration + } + + fn on_http_callout_done( + &mut self, + envoy_filter: &mut EHF, + callout_id: u64, + result: abi::envoy_dynamic_module_type_http_callout_result, + _response_headers: Option<&[(EnvoyBuffer, EnvoyBuffer)]>, + response_body: Option<&[EnvoyBuffer]>, + ) { + let Some(pending) = self.pending.take() else { + return; + }; + if pending.callout_id != callout_id { + return; + } + + if result != abi::envoy_dynamic_module_type_http_callout_result::Success { + envoy_log_error!("actor_router: resume callout failed: {:?}", result); + envoy_filter.send_response(503, &[], Some(b"actor resume failed"), None); + return; + } + + let mut body = Vec::new(); + if let Some(chunks) = response_body { + for chunk in chunks { + body.extend_from_slice(chunk.as_slice()); + } + } + + #[derive(Deserialize)] + struct ResumeResponse { + worker_ip: String, + } + + let Ok(resp) = serde_json::from_slice::(&body) else { + envoy_log_error!("actor_router: unparsable resume response"); + envoy_filter.send_response(503, &[], Some(b"actor resume failed"), None); + return; + }; + if resp.worker_ip.parse::().is_err() { + // Mirrors the net.ParseIP guard in ingress.go: a non-IP answer is a + // control-plane bug, and routing to it would fail confusingly later. + envoy_log_error!("actor_router: resume returned non-IP {}", resp.worker_ip); + envoy_filter.send_response(500, &[], Some(b"actor routing failed"), None); + return; + } + + if self.config.cache_enabled { + cache().insert( + pending.actor.cache_key(), + Binding { + worker_ip: resp.worker_ip.clone(), + expires_at: Instant::now() + Duration::from_secs(self.config.cache_ttl_seconds), + }, + ); + } + + self.config.route_to(envoy_filter, &pending.actor, &resp.worker_ip); + envoy_filter.continue_decoding(); + } +} + +/// Header the co-existence filter sets on a cache hit. The route config matches +/// it to select a route that has ext_proc disabled via typed_per_filter_config, +/// which is how a hit skips the gRPC hop. A request that does not carry it +/// takes the normal route and the normal ext_proc path. +const ROUTE_RESOLVED_HEADER: &str = "x-ate-route-resolved"; + +/// CacheFilter sits in front of the existing ext_proc filter and changes +/// nothing about it. +/// +/// * On a hit it publishes the same dynamic metadata ext_proc would have +/// published, marks the request so route selection skips ext_proc, and +/// continues. +/// * On a miss it does nothing at all: ext_proc runs exactly as it does today, +/// with its resumer, singleflight, parking and metrics intact. Once the +/// response comes back the filter reads the decision ext_proc published and +/// caches it for next time. +/// +/// So the module never talks to ate-apiserver, needs no client certificate, and +/// cannot invent a route ext_proc would not have chosen. Removing the filter +/// from the chain restores today's behaviour exactly. +struct CacheFilter { + config: Config, + cache_hit: Option, + cache_miss: Option, + /// Set when this request missed, naming the entry to fill in on the way back. + learn_key: Option, +} + +impl HttpFilter for CacheFilter { + fn on_request_headers( + &mut self, + envoy_filter: &mut EHF, + _end_of_stream: bool, + ) -> abi::envoy_dynamic_module_type_on_http_filter_request_headers_status { + // A helper filter must never turn a request away: anything it cannot + // understand is simply handed to ext_proc, which owns the error + // responses and their exact wording. + let Some(authority) = self.config.authority(envoy_filter) else { + return abi::envoy_dynamic_module_type_on_http_filter_request_headers_status::Continue; + }; + let Some(actor) = self.config.parse_actor_ref(&authority) else { + return abi::envoy_dynamic_module_type_on_http_filter_request_headers_status::Continue; + }; + + let key = actor.cache_key(); + if self.config.cache_enabled { + if let Some(entry) = cache().get(&key) { + if entry.expires_at > Instant::now() { + let worker_ip = entry.worker_ip.clone(); + drop(entry); + if let Some(id) = self.cache_hit { + let _ = envoy_filter.increment_counter(id, 1); + } + self.config.route_to(envoy_filter, &actor, &worker_ip); + // Mark the request and force route re-selection, so ext_proc + // picks up the per-route "disabled" override. + envoy_filter.set_request_header(ROUTE_RESOLVED_HEADER, b"1"); + envoy_filter.clear_route_cache(); + return abi::envoy_dynamic_module_type_on_http_filter_request_headers_status::Continue; + } + drop(entry); + cache().remove(&key); + } + } + + if let Some(id) = self.cache_miss { + let _ = envoy_filter.increment_counter(id, 1); + } + // Fall through to ext_proc and learn from whatever it decides. + self.learn_key = Some(key); + abi::envoy_dynamic_module_type_on_http_filter_request_headers_status::Continue + } + + fn on_response_headers( + &mut self, + envoy_filter: &mut EHF, + _end_of_stream: bool, + ) -> abi::envoy_dynamic_module_type_on_http_filter_response_headers_status { + // Only a request that missed has anything to learn. Reaching response + // headers at all means the request was routed to a worker, so the + // metadata ext_proc published is a decision that actually worked. + if let (Some(key), true) = (self.learn_key.take(), self.config.cache_enabled) { + if let Some(buf) = envoy_filter.get_metadata_string( + abi::envoy_dynamic_module_type_metadata_source::Dynamic, + ORIGINAL_DST_NAMESPACE, + ORIGINAL_DST_ADDRESS_KEY, + ) { + if let Ok(addr) = std::str::from_utf8(buf.as_slice()) { + // ext_proc writes ":443"; keep only the IP so the + // port stays this filter's configuration, not a parsed value. + if let Some((ip, _port)) = addr.rsplit_once(':') { + if ip.parse::().is_ok() { + cache().insert( + key, + Binding { + worker_ip: ip.to_string(), + expires_at: Instant::now() + + Duration::from_secs(self.config.cache_ttl_seconds), + }, + ); + } + } + } + } + } + abi::envoy_dynamic_module_type_on_http_filter_response_headers_status::Continue + } +}