diff --git a/.github/workflows/pr-workflow.yaml b/.github/workflows/pr-workflow.yaml index 199c874c60..f8bff9bfdd 100644 --- a/.github/workflows/pr-workflow.yaml +++ b/.github/workflows/pr-workflow.yaml @@ -90,8 +90,8 @@ jobs: ~/.cargo/registry/index ~/.cargo/registry/cache ~/.cargo/git/db - cmd/dataplane/envoy/dynamic-modules/*/target - key: cargo-${{ runner.os }}-${{ hashFiles('cmd/dataplane/envoy/dynamic-modules/*/Cargo.lock') }} + cmd/dataplane/envoy/dynamic-modules/target + key: cargo-${{ runner.os }}-${{ hashFiles('cmd/dataplane/envoy/dynamic-modules/Cargo.lock') }} restore-keys: cargo-${{ runner.os }}- - name: dynamic module tests run: | @@ -154,6 +154,7 @@ jobs: ) }} env: E2E_ATENET_DATAPLANE: ${{ matrix.dataplane }} + E2E_ENVOY_CONCURRENCY: "1" # The installer requires a credential provider selection, and deploys the # bundled Kubernetes Secrets provider with the control plane. Job-wide # because every ate-setup invocation reads it, including the control @@ -229,6 +230,8 @@ jobs: - name: Install Agent Substrate (${{ matrix.dataplane }}) # The dataplane selection applies to both the ingress router and egress # gateway. + env: + E2E_ENVOY_CONCURRENCY: "1" run: hack/install-ate-kind.sh --deploy-ate-system --atenet-dataplane=${{ matrix.dataplane }} - name: Enable NFS # Load NFS kernel modules so in-cluster NFS server and CSI driver can run. @@ -243,6 +246,7 @@ jobs: # redeploys the control plane, so retain the selected dataplane. env: ATE_ATENET_DATAPLANE: ${{ matrix.dataplane }} + E2E_ENVOY_CONCURRENCY: "1" run: hack/run-microvm-demo-kind.sh - name: Deploy gVisor counter demo run: hack/install-ate-kind.sh --deploy-demo-counter diff --git a/cmd/ate-setup/internal/config/config.go b/cmd/ate-setup/internal/config/config.go index 7e7042398a..58f08e92db 100644 --- a/cmd/ate-setup/internal/config/config.go +++ b/cmd/ate-setup/internal/config/config.go @@ -200,6 +200,9 @@ type Config struct { // BenchmarkActorMemory is the memory limit for benchmark actors // (BENCHMARK_ACTOR_MEMORY). Empty leaves the workload default in place. BenchmarkActorMemory string + // EnvoyConcurrency overrides the Envoy egress gateway worker thread count + // (E2E_ENVOY_CONCURRENCY). Empty leaves Envoy's default concurrency in place. + EnvoyConcurrency string // kubeconfigEnv is what ScriptEnv exports as $KUBECONFIG. Unlike Kubeconfig // it may be a PATH-style list of files, which kubectl understands and diff --git a/cmd/ate-setup/internal/config/load.go b/cmd/ate-setup/internal/config/load.go index 0d6bec4461..3adabed3a6 100644 --- a/cmd/ate-setup/internal/config/load.go +++ b/cmd/ate-setup/internal/config/load.go @@ -249,6 +249,7 @@ func buildConfig(root string, env map[string]string, r *Resolved) (*Config, erro AnthropicAPIKey: r.String("demo.anthropicAPIKey"), OtlpEndpoint: r.String("otlpEndpoint"), BenchmarkActorMemory: r.String("benchmark.actorMemory"), + EnvoyConcurrency: r.String("e2e.envoyConcurrency"), kubeconfigEnv: kubeconfigEnv, shellEnv: env, resolved: r, diff --git a/cmd/ate-setup/internal/config/setting.go b/cmd/ate-setup/internal/config/setting.go index e52d783a4e..a14f20a6bb 100644 --- a/cmd/ate-setup/internal/config/setting.go +++ b/cmd/ate-setup/internal/config/setting.go @@ -265,6 +265,11 @@ var Registry = []Setting{ Kind: KindString, Usage: "Memory limit for benchmark actors", }, + { + Key: "e2e.envoyConcurrency", Env: "E2E_ENVOY_CONCURRENCY", Flag: "e2e-envoy-concurrency", + Kind: KindString, + Usage: "Number of worker threads for Envoy dataplane", + }, { Key: "record.dir", Env: "ATE_RECORD_DIR", Flag: "record-dir", Kind: KindString, Usage: "Directory install records are written to (default: the user cache directory)", diff --git a/cmd/ate-setup/internal/steps/overlay.go b/cmd/ate-setup/internal/steps/overlay.go index 390dd8a6ef..6b7ed7819a 100644 --- a/cmd/ate-setup/internal/steps/overlay.go +++ b/cmd/ate-setup/internal/steps/overlay.go @@ -179,6 +179,7 @@ func (e *Env) renderAtenetEgressManifest(ctx context.Context, provider config.Cr return nil, err } raw = e.patchEnvoyDataplaneImage(raw, imageReference) + raw = e.patchEnvoyConcurrency(raw, e.Cfg.EnvoyConcurrency) rendered, err := e.renderBytes(raw) if err != nil { return nil, err @@ -192,6 +193,38 @@ func (e *Env) patchEnvoyDataplaneImage(raw []byte, imageRef string) []byte { return bytes.ReplaceAll(raw, []byte("${ENVOY_DATAPLANE_IMAGE}"), []byte(imageRef)) } +// patchEnvoyConcurrency replaces the - ${E2E_ENVOY_CONCURRENCY} placeholder in +// the manifest with --concurrency when concurrency is set, or removes +// the placeholder line when empty. It also replaces the +// stats_flush_on_admin: ${E2E_ENVOY_STATS_FLUSH_ON_ADMIN} placeholder in the +// Envoy bootstrap config with stats_flush_on_admin: true when concurrency is +// "1", or removes the placeholder line otherwise. +func (e *Env) patchEnvoyConcurrency(raw []byte, concurrency string) []byte { + const ( + concurrencyPlaceholder = "- ${E2E_ENVOY_CONCURRENCY}" + statsFlushPlaceholder = "stats_flush_on_admin: ${E2E_ENVOY_STATS_FLUSH_ON_ADMIN}" + ) + var out []string + for _, line := range strings.Split(string(raw), "\n") { + switch strings.TrimSpace(line) { + case concurrencyPlaceholder: + if concurrency != "" { + indent := line[:len(line)-len(strings.TrimLeft(line, " \t"))] + out = append(out, indent+"- --concurrency", fmt.Sprintf("%s- %q", indent, concurrency)) + } + continue + case statsFlushPlaceholder: + if concurrency == "1" { + indent := line[:len(line)-len(strings.TrimLeft(line, " \t"))] + out = append(out, indent+"stats_flush_on_admin: true") + } + continue + } + out = append(out, line) + } + return []byte(strings.Join(out, "\n")) +} + // patchAtenetEgressInject replaces the #ATE_EGRESS_INJECT_FLAGS marker in the // egress sidecar's args with the credential-provider flags, or removes it when // injection is off, which leaves the gateway with no provider. It takes the diff --git a/cmd/ate-setup/internal/steps/overlay_test.go b/cmd/ate-setup/internal/steps/overlay_test.go index 440febd241..32f64d8aa6 100644 --- a/cmd/ate-setup/internal/steps/overlay_test.go +++ b/cmd/ate-setup/internal/steps/overlay_test.go @@ -817,7 +817,7 @@ func TestRenderAtenetEgressManifestPrebuilt(t *testing.T) { if !strings.Contains(string(out), "image: "+want) { t.Errorf("rendered manifest does not install %s:\n%s", want, out) } - for _, leftover := range []string{"${ENVOY_DATAPLANE_IMAGE}", "ko://"} { + for _, leftover := range []string{"${ENVOY_DATAPLANE_IMAGE}", "${E2E_ENVOY_CONCURRENCY}", "${E2E_ENVOY_STATS_FLUSH_ON_ADMIN}", "ko://"} { if strings.Contains(string(out), leftover) { t.Errorf("rendered manifest still contains %q", leftover) } @@ -862,3 +862,29 @@ func TestPatchEnvoyDataplaneImage(t *testing.T) { t.Errorf("patchEnvoyDataplaneImage() = %q, want %q", got, want) } } + +func TestPatchEnvoyConcurrency(t *testing.T) { + e := &Env{} + raw := []byte("data:\n envoy.yaml: |\n stats_flush_on_admin: ${E2E_ENVOY_STATS_FLUSH_ON_ADMIN}\ncontainers:\n- name: envoy\n args:\n - -c\n - /etc/envoy/envoy.yaml\n - ${E2E_ENVOY_CONCURRENCY}\n") + + t.Run("set to 1", func(t *testing.T) { + want := "data:\n envoy.yaml: |\n stats_flush_on_admin: true\ncontainers:\n- name: envoy\n args:\n - -c\n - /etc/envoy/envoy.yaml\n - --concurrency\n - \"1\"\n" + if got := string(e.patchEnvoyConcurrency(raw, "1")); got != want { + t.Errorf("patchEnvoyConcurrency(1) = %q, want %q", got, want) + } + }) + + t.Run("set to other value", func(t *testing.T) { + want := "data:\n envoy.yaml: |\ncontainers:\n- name: envoy\n args:\n - -c\n - /etc/envoy/envoy.yaml\n - --concurrency\n - \"2\"\n" + if got := string(e.patchEnvoyConcurrency(raw, "2")); got != want { + t.Errorf("patchEnvoyConcurrency(2) = %q, want %q", got, want) + } + }) + + t.Run("unset", func(t *testing.T) { + want := "data:\n envoy.yaml: |\ncontainers:\n- name: envoy\n args:\n - -c\n - /etc/envoy/envoy.yaml\n" + if got := string(e.patchEnvoyConcurrency(raw, "")); got != want { + t.Errorf("patchEnvoyConcurrency(\"\") = %q, want %q", got, want) + } + }) +} diff --git a/cmd/atenet/internal/router/egresspolicymanifest_test.go b/cmd/atenet/internal/router/egresspolicymanifest_test.go index 84e4c9f511..7e7fb01ab4 100644 --- a/cmd/atenet/internal/router/egresspolicymanifest_test.go +++ b/cmd/atenet/internal/router/egresspolicymanifest_test.go @@ -172,8 +172,8 @@ func TestEgressManifestsNameEveryExtProcChain(t *testing.T) { t.Run(path, func(t *testing.T) { var got []string for _, lc := range allChains(bootstrapTree(t, path)) { - h := hcm(lc.chain) - if h == nil || filterIndex(list(h, "http_filters"), extProcFilter) < 0 { + cfg, _, _ := extProcOf(lc.chain) + if cfg == nil { continue } name := str(lc.chain, "name") @@ -196,11 +196,19 @@ func TestEgressManifestsNameEveryExtProcChain(t *testing.T) { // among the http_filters, or nil and -1. func extProcOf(chain node) (node, int, []node) { filters := list(hcm(chain), "http_filters") - i := filterIndex(filters, extProcFilter) - if i < 0 { - return nil, -1, filters + if i := filterIndex(filters, extProcFilter); i >= 0 { + return child(filters[i], "typed_config"), i, filters + } + if i := filterIndex(filters, "envoy.filters.http.composite"); i >= 0 { + matchers := list(child(child(child(filters[i], "typed_config"), "matcher"), "matcher_list"), "matchers") + for _, m := range matchers { + action := child(child(child(m, "on_match"), "action"), "typed_config") + if inner := child(action, "typed_config"); str(inner, "name") == extProcFilter { + return child(inner, "typed_config"), i, filters + } + } } - return child(filters[i], "typed_config"), i, filters + return nil, -1, filters } // Every ext_proc in the egress gateway fails closed, talks to the co-located @@ -654,10 +662,9 @@ func TestEgressManifestsConnectLegHandsTheSNIRulesToTheInnerListener(t *testing. } writers := filterStateWriters(tree, extproc.EgressPolicyMetadataNamespace) - if len(writers) != 1 { - t.Fatalf("%s is set by %d filters, want exactly the CONNECT leg's copy of its answer", extproc.EgressPolicyMetadataNamespace, len(writers)) + if len(writers) != 2 { + t.Fatalf("%s is set by %d entries, want 2 (ext_proc dynamic metadata and cached filter state)", extproc.EgressPolicyMetadataNamespace, len(writers)) } - entry := writers[0] at := -1 for i, f := range filters { if str(f, "name") != setFilterStateFilter { @@ -675,18 +682,21 @@ func TestEgressManifestsConnectLegHandsTheSNIRulesToTheInnerListener(t *testing. if at < extProcAt { t.Errorf("%s is set at http_filters[%d], before ext_proc at [%d]; the metadata it reads does not exist yet", extproc.EgressPolicyMetadataNamespace, at, extProcAt) } - format := child(entry, "format_string") - if got := str(child(format, "text_format_source"), "inline_string"); got != extproc.EgressPolicyMetadataFormat { - t.Errorf("%s is set from %q, want %q", extproc.EgressPolicyMetadataNamespace, got, extproc.EgressPolicyMetadataFormat) - } - if got := str(entry, "factory_key"); got != "envoy.string" { - t.Errorf("%s uses factory %q, want envoy.string, the string accessor the module reads", extproc.EgressPolicyMetadataNamespace, got) - } - if skip, _ := entry["skip_if_empty"].(bool); !skip { - t.Errorf("%s is not skip_if_empty; an absent answer would be written as unparseable JSON", extproc.EgressPolicyMetadataNamespace) - } - if str(entry, "shared_with_upstream") == "" { - t.Errorf("%s is not shared with upstream; the inner listener would never see it", extproc.EgressPolicyMetadataNamespace) + wantFormats := []string{extproc.EgressPolicyMetadataFormat, extproc.EgressPolicyCachedFormat} + for i, entry := range writers { + format := child(entry, "format_string") + if got := str(child(format, "text_format_source"), "inline_string"); got != wantFormats[i] { + t.Errorf("%s writer[%d] is set from %q, want %q", extproc.EgressPolicyMetadataNamespace, i, got, wantFormats[i]) + } + if got := str(entry, "factory_key"); got != "envoy.string" { + t.Errorf("%s writer[%d] uses factory %q, want envoy.string, the string accessor the module reads", extproc.EgressPolicyMetadataNamespace, i, got) + } + if skip, _ := entry["skip_if_empty"].(bool); !skip { + t.Errorf("%s writer[%d] is not skip_if_empty; an absent answer would be written as unparseable JSON", extproc.EgressPolicyMetadataNamespace, i) + } + if str(entry, "shared_with_upstream") == "" { + t.Errorf("%s writer[%d] is not shared with upstream; the inner listener would never see it", extproc.EgressPolicyMetadataNamespace, i) + } } } diff --git a/cmd/atenet/internal/router/extproc/attributes.go b/cmd/atenet/internal/router/extproc/attributes.go index f347e58d43..15f483be5a 100644 --- a/cmd/atenet/internal/router/extproc/attributes.go +++ b/cmd/atenet/internal/router/extproc/attributes.go @@ -142,6 +142,13 @@ const UpstreamDynamicHostFilterStateKey = "envoy.upstream.dynamic_host" // EgressPolicyMetadataFormat renders EgressPolicyMetadataNamespace as JSON. const EgressPolicyMetadataFormat = "%DYNAMIC_METADATA(" + EgressPolicyMetadataNamespace + ")%" +// EgressPolicyCachedFilterStateKey holds the cached SNI rules written by the +// egress-policy-cache module before set_filter_state shares them upstream. +const EgressPolicyCachedFilterStateKey = EgressPolicyMetadataNamespace + ".cached" + +// EgressPolicyCachedFormat reads EgressPolicyCachedFilterStateKey as a plain string. +const EgressPolicyCachedFormat = "%FILTER_STATE(" + EgressPolicyCachedFilterStateKey + ":PLAIN)%" + // OriginalDstFilterStateKey is Envoy's filter-state key for the address an // ORIGINAL_DST cluster dials. The gateway never writes it and dials only by // name; the request legs read it as an attribute. diff --git a/cmd/dataplane/envoy/Dockerfile b/cmd/dataplane/envoy/Dockerfile index 9de411563c..23be3766fb 100644 --- a/cmd/dataplane/envoy/Dockerfile +++ b/cmd/dataplane/envoy/Dockerfile @@ -12,10 +12,10 @@ # See the License for the specific language governing permissions and # limitations under the License. -# Stage 1: Build the release version of the Rust dynamic module +# Stage 1: Build the release version of the Rust dynamic modules FROM rust:bookworm@sha256:93ce27a88655056a51dbdd8f5f2d7ddc071c7b0070fb288a37b5a285fc83971e AS builder -WORKDIR /usr/src/dynamic-modules/egress-policy +WORKDIR /usr/src/dynamic-modules # Install build dependencies required by bindgen for generating Envoy ABI bindings RUN apt-get update && apt-get install -y --no-install-recommends \ @@ -23,21 +23,22 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ libclang-dev \ && rm -rf /var/lib/apt/lists/* -# Copy rust module source code -COPY dynamic-modules/egress-policy/ . +# Copy rust module workspace source code +COPY dynamic-modules/ . -# Build release version of the dynamic module shared library +# Build release version of the dynamic module shared libraries RUN cargo build --release --locked -# Stage 2: Create the runtime image containing Envoy and the built dynamic module shared library +# Stage 2: Create the runtime image containing Envoy and the built dynamic module shared libraries # When changing Envoy version, the version of SDK in the Cargo.toml files SHOULD match. # Pin the multi-platform image index, not one platform's manifest, so that the # --platform the installer passes (the host architecture on kind, linux/amd64 # otherwise) selects a matching Envoy binary. FROM envoyproxy/envoy:v1.39.1@sha256:57e14a549d7bd43c8d3f6d03e8cfa653e037d4b38e133acd9b54f38c524401b4 -# Copy the built dynamic module shared library from the builder stage -COPY --from=builder /usr/src/dynamic-modules/egress-policy/target/release/libenvoy_substrate_egress_policy.so /usr/local/lib/libenvoy_substrate_egress_policy.so +# Copy the built dynamic module shared libraries from the builder stage +COPY --from=builder /usr/src/dynamic-modules/target/release/libenvoy_substrate_egress_policy.so /usr/local/lib/libenvoy_substrate_egress_policy.so +COPY --from=builder /usr/src/dynamic-modules/target/release/libenvoy_substrate_egress_policy_cache.so /usr/local/lib/libenvoy_substrate_egress_policy_cache.so # Set default search path for Envoy dynamic modules ENV ENVOY_DYNAMIC_MODULES_SEARCH_PATH=/usr/local/lib diff --git a/cmd/dataplane/envoy/dynamic-modules/.gitignore b/cmd/dataplane/envoy/dynamic-modules/.gitignore new file mode 100644 index 0000000000..7233fc6f32 --- /dev/null +++ b/cmd/dataplane/envoy/dynamic-modules/.gitignore @@ -0,0 +1,3 @@ +target/ +egress-policy/target/ +egress-policy-cache/target/ diff --git a/cmd/dataplane/envoy/dynamic-modules/egress-policy/Cargo.lock b/cmd/dataplane/envoy/dynamic-modules/Cargo.lock similarity index 83% rename from cmd/dataplane/envoy/dynamic-modules/egress-policy/Cargo.lock rename to cmd/dataplane/envoy/dynamic-modules/Cargo.lock index d5aa3de254..61f3b30498 100644 --- a/cmd/dataplane/envoy/dynamic-modules/egress-policy/Cargo.lock +++ b/cmd/dataplane/envoy/dynamic-modules/Cargo.lock @@ -11,6 +11,12 @@ dependencies = [ "memchr", ] +[[package]] +name = "allocator-api2" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" + [[package]] name = "anstyle" version = "1.0.14" @@ -39,9 +45,9 @@ dependencies = [ [[package]] name = "bitflags" -version = "2.13.1" +version = "2.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da" +checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" [[package]] name = "cexpr" @@ -54,9 +60,9 @@ dependencies = [ [[package]] name = "cfg-if" -version = "1.0.4" +version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600" [[package]] name = "clang-sys" @@ -90,6 +96,18 @@ dependencies = [ "mockall", ] +[[package]] +name = "equivalent" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" + +[[package]] +name = "foldhash" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" + [[package]] name = "fragile" version = "2.1.0" @@ -111,6 +129,17 @@ version = "0.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e4eba85ea1d0a966a983acd07deee566e67395d2d96b6fb39e62b5a833f1eb0b" +[[package]] +name = "hashbrown" +version = "0.16.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" +dependencies = [ + "allocator-api2", + "equivalent", + "foldhash", +] + [[package]] name = "itertools" version = "0.13.0" @@ -128,9 +157,9 @@ checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" [[package]] name = "libc" -version = "0.2.189" +version = "0.2.190" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" +checksum = "ce5d3ddc6d3fa000eb1536d85e147bfe31aacaba692ed6a876f95cb7c855be78" [[package]] name = "libloading" @@ -148,6 +177,15 @@ version = "0.4.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" +[[package]] +name = "lru" +version = "0.16.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f66e8d5d03f609abc3a39e6f08e4164ebf1447a732906d39eb9b99b7919ef39" +dependencies = [ + "hashbrown", +] + [[package]] name = "memchr" version = "2.8.3" @@ -343,6 +381,17 @@ dependencies = [ "serde_json", ] +[[package]] +name = "substrate-envoy-egress-policy-cache" +version = "0.1.0" +dependencies = [ + "envoy-proxy-dynamic-modules-rust-sdk", + "lru", + "serde", + "serde_json", + "thread_local", +] + [[package]] name = "syn" version = "2.0.119" @@ -371,11 +420,20 @@ version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f50febec83f5ee1df3015341d8bd429f2d1cc62bcba7ea2076759d315084683" +[[package]] +name = "thread_local" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ad99c4c6d32803332c548b1af0540b357b3f5fc0be8f6c6bfe8b2e6ae784070" +dependencies = [ + "cfg-if", +] + [[package]] name = "unicode-ident" -version = "1.0.24" +version = "1.0.26" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954" [[package]] name = "windows-link" diff --git a/cmd/dataplane/envoy/dynamic-modules/Cargo.toml b/cmd/dataplane/envoy/dynamic-modules/Cargo.toml new file mode 100644 index 0000000000..2a5e3f8475 --- /dev/null +++ b/cmd/dataplane/envoy/dynamic-modules/Cargo.toml @@ -0,0 +1,20 @@ +# 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. + +[workspace] +resolver = "3" +members = [ + "egress-policy", + "egress-policy-cache", +] diff --git a/cmd/dataplane/envoy/dynamic-modules/egress-policy-cache/Cargo.toml b/cmd/dataplane/envoy/dynamic-modules/egress-policy-cache/Cargo.toml new file mode 100644 index 0000000000..0e78356930 --- /dev/null +++ b/cmd/dataplane/envoy/dynamic-modules/egress-policy-cache/Cargo.toml @@ -0,0 +1,33 @@ +# 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. + +[package] +name = "substrate-envoy-egress-policy-cache" +version = "0.1.0" +edition = "2024" + +[lib] +name = "envoy_substrate_egress_policy_cache" +crate-type = ["cdylib"] + +[dependencies] +# Pin to v1.39.1 commit SHA +# When changing the SDK version, the version of Envoy in the cmd/dataplane/envoy/Dockerfile SHOULD match. +# SDK version SHOULD NOT be newer than Envoy's in the dataplane image, as extensions will not load if they use +# APIs not supported by the Envoy binary in the image. +envoy-proxy-dynamic-modules-rust-sdk = { git = "https://github.com/envoyproxy/envoy", rev = "b579d07d3ad7ee11d32b105e91a5a39ad24718d7" } +lru = "0.16" +serde = { version = "1.0", features = ["derive"] } +serde_json = "1.0" +thread_local = "1.1" diff --git a/cmd/dataplane/envoy/dynamic-modules/egress-policy-cache/README.md b/cmd/dataplane/envoy/dynamic-modules/egress-policy-cache/README.md new file mode 100644 index 0000000000..13cb110bad --- /dev/null +++ b/cmd/dataplane/envoy/dynamic-modules/egress-policy-cache/README.md @@ -0,0 +1,138 @@ +# Envoy Substrate Egress Policy Cache - Rust Dynamic Module + +An Envoy dynamic module, written in Rust, that runs as an HTTP filter on the +egress gateway's outer `CONNECT` listener and caches egress policy SNI rules +per actor certificate and destination port in a thread-local LRU cache so +repeat `CONNECT` requests from the same actor to the same destination can +use locally cached policy and avoid external callout to get the policy. + +It is safe because external callout evaluates if there are policies that allow +given actor to egress to specific destination port and then provides SNI and host +matching rules to the dataplane. Since local cache is populated only when CONNECT +was allowed and keyed by the actor ID and destination port it safe to use cached +SNI and host matching rules. The freshness configuration `cache_ttl` determines +responsiveness of the dataplane to policy changes. + +## How it works + +Each `EgressPolicyCacheFilterConfig` owns a `ThreadLocal` store of +`LruCache` instances (`max_cache_items` capacity per +Envoy worker thread). Each `CachedPolicy` entry stores the serialized egress +policy JSON string alongside the `Instant` (`stored_at`) when it was cached, +keyed by `;`, where `` is +the downstream peer certificate's SHA-256 digest +(`connection.sha256_peer_certificate_digest`) and `` is the +destination port extracted from the CONNECT `:authority` request header. + +### Request path (`on_request_headers`) + +When an actor opens a `CONNECT` tunnel on the outer listener: + +1. The filter builds the cache key `;` from + `connection.sha256_peer_certificate_digest` and the port from the CONNECT `:authority` + header, and looks up cached policy in worker thread local LRU cache: + - **If unexpired entry was found.**: + - Writes the cached policy JSON to filter state under key + `dev.ate.policy.egress.cached`. + - Sets filter state key `dev.ate.policy.egress.skip_callout` to `"true"` to skip the ext_proc + callout. + - Sets `has_cached_policy = true` on the per-stream + `EgressPolicyCacheFilter`. + - Sets `dev.ate.egress:dialed_port` in the dynamic metadata, so that inner + request is sent to the port in the CONNECT authority. + - Increments the `ate_egress.connect_cache_hit` counter. + - **If expired entry (`stored_at.elapsed() > cache_ttl`) was found**: + - Removes the expired entry from the LRU cache without setting filter + state. + - Increments the `ate_egress.connect_cache_miss` counter. + - **Cache miss (or missing certificate digest / destination port)**: + - Increments the `ate_egress.connect_cache_miss` counter. +2. Subsequent filters in the outer HTTP filter chain act on the result: + - `envoy.filters.http.composite` checks whether + `dev.ate.policy.egress.skip_callout` equals `"true"` in filter state and + invokes `envoy.filters.http.ext_proc` only when it does not. + - `envoy.filters.http.set_filter_state` copies either + `%DYNAMIC_METADATA(dev.ate.policy.egress)%` (on a cache miss, populated by + `ext_proc`) or `%FILTER_STATE(dev.ate.policy.egress.cached:PLAIN)%` (on a + cache hit) into the `dev.ate.policy.egress` filter state with + `shared_with_upstream: ONCE`, making it available to the inner listener's + `egress-policy` listener filter. + +### Response path (`on_response_headers`) + +When the `CONNECT` response headers arrive: + +1. If `cache_enabled` is `false`, the filter returns `Continue` immediately. +2. If `:status` is `200` and `self.has_cached_policy` is `false`: + - Builds cache value from the `dev.ate.policy.egress` filer state. + - Inserts `CachedPolicy { policy, stored_at: Instant::now() }` into the + current worker thread's LRU cache at the the `;` key. +3. If `self.has_cached_policy` is `true` (the policy for this stream came from + the cache), caching is skipped so the entry's original `stored_at` timestamp + is preserved until `cache_ttl` expires. + +## Filter configuration + +The filter accepts an optional JSON configuration object in `filter_config` +(an empty config defaults to `{}`): + +| Field | Type | Default | Description | +|---|---|---|---| +| `cache_ttl` | `u64` (seconds) | `5` | Time-to-live in seconds for cached policy entries before they expire | +| `cache_enabled` | `bool` | `true` | Enables or disables reading from and writing to the policy cache | +| `max_cache_items` (or `max-cache-items`) | `usize` | `1000` | Maximum number of entries in each worker thread's LRU cache | + +Example JSON configuration: + +```json +{ + "cache_ttl": 5, + "cache_enabled": true, + "max_cache_items": 1000 +} +``` + +## Metrics + +The filter defines two Envoy counters on config initialization: + +| Counter name | Envoy stat name | Description | +|---|---|---| +| `ate_egress.connect_cache_hit` | `dynamic_modules.custom.egress_policy_cache.ate_egress.connect_cache_hit` | Incremented when a non-expired policy is found in the cache on `CONNECT` | +| `ate_egress.connect_cache_miss` | `dynamic_modules.custom.egress_policy_cache.ate_egress.connect_cache_miss` | Incremented when no entry or an expired entry is found in the cache on `CONNECT` | + +## Building + +Prerequisites: Rust toolchain (Cargo, rustc 1.75+), plus `clang` and +`libclang-dev` for the SDK's bindgen step. + +```bash +cargo build --release +``` + +The compiled shared object will be located at +`target/release/libenvoy_substrate_egress_policy_cache.so`. +`cmd/dataplane/envoy/Dockerfile` builds it and packages it into the Envoy +dataplane image. + +## Testing + +```bash +cargo test +``` + +Or from the repository root: + +```bash +hack/test-dynamic-modules.sh +``` + +> **Note:** Because the LRU cache is thread-local per Envoy worker thread, set +> `export E2E_ENVOY_CONCURRENCY=1` before deploying `ate-system` when running +> the networking E2E suite (`./internal/e2e/suites/networking`) to test caching +> behavior (`TestActorEgressPolicyCache` and +> `TestActorEgressPolicyCacheExpiration`). This configures Envoy with a single +> worker thread (`--concurrency 1`) so consecutive `CONNECT` requests are +> routed to the same thread-local cache instance. + +See `manifests/ate-install/atenet-egress.yaml` for the complete configuration. diff --git a/cmd/dataplane/envoy/dynamic-modules/egress-policy-cache/src/lib.rs b/cmd/dataplane/envoy/dynamic-modules/egress-policy-cache/src/lib.rs new file mode 100644 index 0000000000..8a902916cc --- /dev/null +++ b/cmd/dataplane/envoy/dynamic-modules/egress-policy-cache/src/lib.rs @@ -0,0 +1,923 @@ +// 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. + +//! The egress-policy-cache HTTP filter dynamic module. + +use envoy_proxy_dynamic_modules_rust_sdk::{ + abi::{ + envoy_dynamic_module_type_attribute_id, + envoy_dynamic_module_type_on_http_filter_request_headers_status, + envoy_dynamic_module_type_on_http_filter_response_headers_status, + }, + declare_init_functions, envoy_log_error, EnvoyCounterId, EnvoyHttpFilter, + EnvoyHttpFilterConfig, HttpFilter, HttpFilterConfig, +}; +use lru::LruCache; +use serde::Deserialize; +use std::cell::RefCell; +use std::num::NonZeroUsize; +use std::sync::Arc; +use std::time::{Duration, Instant}; +use thread_local::ThreadLocal; + +/// Key holding the egress policy SNI rules JSON. +pub const ATE_POLICY_EGRESS: &str = "dev.ate.policy.egress"; + +/// Key holding the cached egress policy SNI rules JSON before it is copied to +/// upstream-shared filter state. +// TODO(yanavlasov): this is a temporary workaround for the Rust dynamic module API +// limitation that does not allow storing filter state shared with upstream. +// The dev.ate.policy.egress.cached filter state is later copied into +// dev.ate.policy.egress filter state by the set_filter_state HTTP filter. +// Once this limitation is addressed this filter can set "dev.ate.policy.egress" filter +// state directly and share it with upstream. +pub const ATE_POLICY_EGRESS_CACHED: &str = "dev.ate.policy.egress.cached"; + +/// Filter state key set to `"true"` on a cache hit so the composite filter can +/// skip the `ext_proc` callout. +pub const ATE_POLICY_EGRESS_SKIP_CALLOUT: &str = "dev.ate.policy.egress.skip_callout"; + +/// Filter state value for [`ATE_POLICY_EGRESS_SKIP_CALLOUT`] when the `ext_proc` +/// callout should be skipped. +pub const SKIP_CALLOUT_TRUE: &str = "true"; + +/// Dynamic metadata namespace for egress attributes set on CONNECT. +pub const ATE_EGRESS_METADATA_NAMESPACE: &str = "dev.ate.egress"; + +/// Dynamic metadata key holding the destination port dialed by the actor. +pub const ATE_EGRESS_DIALED_PORT_KEY: &str = "dialed_port"; + +/// Counter name for egress policy cache hits on CONNECT. +pub const CONNECT_CACHE_HIT_COUNTER: &str = "ate_egress.connect_cache_hit"; + +/// Counter name for egress policy cache misses on CONNECT. +pub const CONNECT_CACHE_MISS_COUNTER: &str = "ate_egress.connect_cache_miss"; + +/// Cached egress policy value and the timestamp when it was stored. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CachedPolicy { + pub policy: String, + pub stored_at: Instant, +} + +impl CachedPolicy { + pub fn new(policy: String, stored_at: Instant) -> Self { + Self { policy, stored_at } + } +} + +/// Thread-local store of LRU caches with `String` keys and [`CachedPolicy`] values. +pub type ThreadLocalCache = ThreadLocal>>; + +declare_init_functions!(init, new_http_filter_config_fn); + +/// Called when the dynamic module is loaded into Envoy. +fn init() -> bool { + true +} + +/// Called when a new HTTP filter configuration is created. +fn new_http_filter_config_fn( + envoy_filter_config: &mut EC, + _filter_name: &str, + filter_config: &[u8], +) -> Option>> { + let config = parse_config(filter_config)?; + let cache_hit_counter = envoy_filter_config + .define_counter(CONNECT_CACHE_HIT_COUNTER) + .ok()?; + let cache_miss_counter = envoy_filter_config + .define_counter(CONNECT_CACHE_MISS_COUNTER) + .ok()?; + Some(Box::new(EgressPolicyCacheFilterConfig::with_counters( + config, + cache_hit_counter, + cache_miss_counter, + ))) +} + +fn parse_config(filter_config: &[u8]) -> Option { + let raw = std::str::from_utf8(filter_config).ok()?; + let json_str = if raw.trim().is_empty() { "{}" } else { raw }; + serde_json::from_str(json_str) + .map_err(|e| { + envoy_log_error!("egress_policy_cache: invalid filter config: {}", e); + }) + .ok() +} + +/// Configuration for the egress-policy-cache HTTP filter. +#[derive(Debug, Clone, Deserialize, PartialEq)] +pub struct Config { + #[serde(default = "default_cache_ttl", deserialize_with = "deserialize_secs")] + pub cache_ttl: Duration, + #[serde(default = "default_cache_enabled")] + pub cache_enabled: bool, + #[serde(default = "default_max_cache_items", alias = "max-cache-items")] + pub max_cache_items: usize, +} + +fn default_cache_ttl() -> Duration { + Duration::from_secs(5) +} + +fn default_cache_enabled() -> bool { + true +} + +fn default_max_cache_items() -> usize { + 1000 +} + +fn deserialize_secs<'de, D>(deserializer: D) -> Result +where + D: serde::Deserializer<'de>, +{ + let secs = u64::deserialize(deserializer)?; + Ok(Duration::from_secs(secs)) +} + +impl Default for Config { + fn default() -> Self { + Self { + cache_ttl: default_cache_ttl(), + cache_enabled: default_cache_enabled(), + max_cache_items: default_max_cache_items(), + } + } +} + +fn new_lru_cache(max_cache_items: usize) -> RefCell> { + let cap = NonZeroUsize::new(max_cache_items).unwrap_or(NonZeroUsize::MIN); + RefCell::new(LruCache::new(cap)) +} + +/// Per-filter-chain configuration for the egress policy cache filter. +pub struct EgressPolicyCacheFilterConfig { + pub config: Config, + pub cache: Arc, + pub cache_hit_counter: EnvoyCounterId, + pub cache_miss_counter: EnvoyCounterId, +} + +impl EgressPolicyCacheFilterConfig { + pub fn new(config: Config) -> Self { + Self::with_counters(config, EnvoyCounterId(0), EnvoyCounterId(1)) + } + + pub fn with_counters( + config: Config, + cache_hit_counter: EnvoyCounterId, + cache_miss_counter: EnvoyCounterId, + ) -> Self { + Self { + config, + cache: Arc::new(ThreadLocal::new()), + cache_hit_counter, + cache_miss_counter, + } + } + + /// Returns a reference to the current thread's LRU cache, initializing it + /// with capacity `config.max_cache_items` on first access. + pub fn local_cache(&self) -> &RefCell> { + let max_items = self.config.max_cache_items; + self.cache.get_or(|| new_lru_cache(max_items)) + } + + /// Creates a concrete [`EgressPolicyCacheFilter`] bound to this configuration. + pub fn create_filter(&self) -> EgressPolicyCacheFilter { + EgressPolicyCacheFilter { + config: self.config.clone(), + cache: Arc::clone(&self.cache), + cache_hit_counter: self.cache_hit_counter, + cache_miss_counter: self.cache_miss_counter, + has_cached_policy: false, + } + } +} + +impl HttpFilterConfig for EgressPolicyCacheFilterConfig { + fn new_http_filter(&self, _envoy: &mut EHF) -> Box> { + Box::new(self.create_filter()) + } +} + +/// Per-stream HTTP filter instance for caching egress policy decisions. +pub struct EgressPolicyCacheFilter { + pub config: Config, + pub cache: Arc, + pub cache_hit_counter: EnvoyCounterId, + pub cache_miss_counter: EnvoyCounterId, + pub has_cached_policy: bool, +} + +impl EgressPolicyCacheFilter { + /// Returns a reference to the current thread's LRU cache, initializing it + /// with capacity `config.max_cache_items` on first access. + pub fn local_cache(&self) -> &RefCell> { + let max_items = self.config.max_cache_items; + self.cache.get_or(|| new_lru_cache(max_items)) + } +} + +fn is_http_200(envoy_filter: &EHF) -> bool { + if let Some(status) = envoy_filter.get_response_header_value(":status") { + return status.as_slice() == b"200"; + } + false +} + +fn read_egress_policy(envoy_filter: &EHF) -> Option { + if let Some(buf) = envoy_filter.get_filter_state_bytes(ATE_POLICY_EGRESS.as_bytes()) + && let Ok(s) = std::str::from_utf8(buf.as_slice()) + && !s.is_empty() { + return Some(s.to_owned()); + } + None +} + +fn read_peer_cert_digest(envoy_filter: &EHF) -> Option { + let buf = envoy_filter.get_attribute_string( + envoy_dynamic_module_type_attribute_id::ConnectionSha256PeerCertificateDigest, + )?; + let s = std::str::from_utf8(buf.as_slice()).ok()?; + if s.is_empty() { + return None; + } + Some(s.to_owned()) +} + +fn extract_destination_port(authority: &str) -> Option<&str> { + let (host, port) = authority.rsplit_once(':')?; + if host.is_empty() || (host.contains(':') && !host.ends_with(']')) { + return None; + } + let port_num: u16 = port.parse().ok()?; + if port_num == 0 { + return None; + } + Some(port) +} + +fn read_destination_port(envoy_filter: &EHF) -> Option { + let buf = envoy_filter.get_request_header_value(":authority")?; + let authority = std::str::from_utf8(buf.as_slice()).ok()?; + extract_destination_port(authority).map(str::to_owned) +} + +fn build_cache_key(envoy_filter: &EHF) -> Option<(String, String)> { + let cert_digest = read_peer_cert_digest(envoy_filter)?; + let port = read_destination_port(envoy_filter)?; + let key = format!("{cert_digest};{port}"); + Some((key, port)) +} + +impl HttpFilter for EgressPolicyCacheFilter { + fn on_request_headers( + &mut self, + envoy_filter: &mut EHF, + _end_of_stream: bool, + ) -> envoy_dynamic_module_type_on_http_filter_request_headers_status { + let cached_policy = if let Some((cache_key, port)) = build_cache_key(envoy_filter) { + let mut cache = self.local_cache().borrow_mut(); + if let Some(entry) = cache.get(&cache_key) { + if entry.stored_at.elapsed() > self.config.cache_ttl { + cache.pop(&cache_key); + None + } else { + Some((entry.policy.clone(), port)) + } + } else { + None + } + } else { + None + }; + + if let Some((policy, port)) = cached_policy { + self.has_cached_policy = true; + envoy_filter.set_filter_state_bytes(ATE_POLICY_EGRESS_CACHED.as_bytes(), policy.as_bytes()); + envoy_filter.set_filter_state_bytes( + ATE_POLICY_EGRESS_SKIP_CALLOUT.as_bytes(), + SKIP_CALLOUT_TRUE.as_bytes(), + ); + envoy_filter.set_dynamic_metadata_string( + ATE_EGRESS_METADATA_NAMESPACE, + ATE_EGRESS_DIALED_PORT_KEY, + &port, + ); + let _ = envoy_filter.increment_counter(self.cache_hit_counter, 1); + } else { + let _ = envoy_filter.increment_counter(self.cache_miss_counter, 1); + } + 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, + ) -> envoy_dynamic_module_type_on_http_filter_response_headers_status { + // Store policy in cache in case it was authorized (200 response) on a + // cache miss. Actor identity comes from cert's subject. Use cert digest + // concatenated with the destination port as key, since SNI rules depend on + // both the actor identity and the dialed port. + if self.config.cache_enabled + && !self.has_cached_policy + && is_http_200(envoy_filter) + && let Some(policy) = read_egress_policy(envoy_filter) + && let Some((cache_key, _)) = build_cache_key(envoy_filter) + { + self.local_cache().borrow_mut().put( + cache_key, + CachedPolicy::new(policy, Instant::now()), + ); + } + envoy_dynamic_module_type_on_http_filter_response_headers_status::Continue + } +} + +#[cfg(test)] +mod tests { + use super::*; + use envoy_proxy_dynamic_modules_rust_sdk::{ + abi::{ + envoy_dynamic_module_type_http_callout_init_result, + envoy_dynamic_module_type_metrics_result, + }, + EnvoyBuffer, EnvoyCounterVecId, EnvoyGaugeId, EnvoyGaugeVecId, + EnvoyHistogramId, EnvoyHistogramVecId, EnvoyHttpFilterConfigScheduler, + MockEnvoyHttpFilter, + }; + + #[derive(Default)] + struct TestEnvoyHttpFilterConfig { + defined_counters: Vec, + } + + impl EnvoyHttpFilterConfig for TestEnvoyHttpFilterConfig { + fn define_counter( + &mut self, + name: &str, + ) -> Result { + self.defined_counters.push(name.to_string()); + Ok(EnvoyCounterId(self.defined_counters.len())) + } + fn define_counter_vec( + &mut self, + _: &str, + _: &[&str], + ) -> Result { + todo!() + } + fn define_gauge( + &mut self, + _: &str, + ) -> Result { + todo!() + } + fn define_gauge_vec( + &mut self, + _: &str, + _: &[&str], + ) -> Result { + todo!() + } + fn define_histogram( + &mut self, + _: &str, + ) -> Result { + todo!() + } + fn define_histogram_vec( + &mut self, + _: &str, + _: &[&str], + ) -> Result { + todo!() + } + fn increment_counter( + &self, + _: EnvoyCounterId, + _: u64, + ) -> Result<(), envoy_dynamic_module_type_metrics_result> { + todo!() + } + fn increment_counter_vec( + &self, + _: EnvoyCounterVecId, + _: &[&str], + _: u64, + ) -> Result<(), envoy_dynamic_module_type_metrics_result> { + todo!() + } + fn increase_gauge( + &self, + _: EnvoyGaugeId, + _: u64, + ) -> Result<(), envoy_dynamic_module_type_metrics_result> { + todo!() + } + fn increase_gauge_vec( + &self, + _: EnvoyGaugeVecId, + _: &[&str], + _: u64, + ) -> Result<(), envoy_dynamic_module_type_metrics_result> { + todo!() + } + fn decrease_gauge( + &self, + _: EnvoyGaugeId, + _: u64, + ) -> Result<(), envoy_dynamic_module_type_metrics_result> { + todo!() + } + fn decrease_gauge_vec( + &self, + _: EnvoyGaugeVecId, + _: &[&str], + _: u64, + ) -> Result<(), envoy_dynamic_module_type_metrics_result> { + todo!() + } + fn set_gauge( + &self, + _: EnvoyGaugeId, + _: u64, + ) -> Result<(), envoy_dynamic_module_type_metrics_result> { + todo!() + } + fn set_gauge_vec( + &self, + _: EnvoyGaugeVecId, + _: &[&str], + _: u64, + ) -> Result<(), envoy_dynamic_module_type_metrics_result> { + todo!() + } + fn record_histogram_value( + &self, + _: EnvoyHistogramId, + _: u64, + ) -> Result<(), envoy_dynamic_module_type_metrics_result> { + todo!() + } + fn record_histogram_value_vec( + &self, + _: EnvoyHistogramVecId, + _: &[&str], + _: u64, + ) -> Result<(), envoy_dynamic_module_type_metrics_result> { + todo!() + } + fn new_scheduler(&self) -> Box { + todo!() + } + fn send_http_callout<'a>( + &mut self, + _: &'a str, + _: &'a [(&'a str, &'a [u8])], + _: Option<&'a [u8]>, + _: u64, + ) -> (envoy_dynamic_module_type_http_callout_init_result, u64) { + todo!() + } + fn start_http_stream<'a>( + &mut self, + _: &'a str, + _: &'a [(&'a str, &'a [u8])], + _: Option<&'a [u8]>, + _: bool, + _: u64, + ) -> (envoy_dynamic_module_type_http_callout_init_result, u64) { + todo!() + } + unsafe fn send_http_stream_data(&mut self, _: u64, _: &[u8], _: bool) -> bool { + todo!() + } + unsafe fn send_http_stream_trailers<'a>( + &mut self, + _: u64, + _: &'a [(&'a str, &'a [u8])], + ) -> bool { + todo!() + } + unsafe fn reset_http_stream(&mut self, _: u64) { + todo!() + } + } + + #[test] + fn test_config_defaults_and_overrides() { + assert_eq!( + parse_config(b""), + Some(Config { + cache_ttl: Duration::from_secs(5), + cache_enabled: true, + max_cache_items: 1000, + }) + ); + assert_eq!( + parse_config(b"{}"), + Some(Config { + cache_ttl: Duration::from_secs(5), + cache_enabled: true, + max_cache_items: 1000, + }) + ); + assert_eq!( + parse_config( + br#"{"cache_ttl": 10, "cache_enabled": false, "max_cache_items": 500}"# + ), + Some(Config { + cache_ttl: Duration::from_secs(10), + cache_enabled: false, + max_cache_items: 500, + }) + ); + assert_eq!( + parse_config(br#"{"max-cache-items": 250}"#), + Some(Config { + cache_ttl: Duration::from_secs(5), + cache_enabled: true, + max_cache_items: 250, + }) + ); + assert_eq!(parse_config(b"not-json"), None); + } + + #[test] + fn test_new_http_filter_config_defines_counters() { + let mut test_config = TestEnvoyHttpFilterConfig::default(); + let filter_config = new_http_filter_config_fn::< + TestEnvoyHttpFilterConfig, + MockEnvoyHttpFilter, + >(&mut test_config, "egress_policy_cache", b"{}"); + assert!(filter_config.is_some()); + assert_eq!( + test_config.defined_counters, + vec![ + CONNECT_CACHE_HIT_COUNTER.to_string(), + CONNECT_CACHE_MISS_COUNTER.to_string(), + ] + ); + } + + #[test] + fn test_extract_destination_port() { + assert_eq!(extract_destination_port("10.0.0.1:443"), Some("443")); + assert_eq!(extract_destination_port("[2001:db8::1]:8443"), Some("8443")); + assert_eq!(extract_destination_port("example.com:80"), Some("80")); + assert_eq!(extract_destination_port("example.com"), None); + assert_eq!(extract_destination_port(":443"), None); + assert_eq!(extract_destination_port("10.0.0.1:0"), None); + assert_eq!(extract_destination_port("10.0.0.1:70000"), None); + assert_eq!(extract_destination_port("2001:db8::1"), None); + assert_eq!(extract_destination_port("[2001:db8::1]"), None); + } + + #[test] + fn test_headers_continue() { + let hit_id = EnvoyCounterId(1); + let miss_id = EnvoyCounterId(2); + let config = + EgressPolicyCacheFilterConfig::with_counters(Config::default(), hit_id, miss_id); + let mut mock_filter = MockEnvoyHttpFilter::new(); + mock_filter + .expect_get_attribute_string() + .withf(|id| { + *id == envoy_dynamic_module_type_attribute_id::ConnectionSha256PeerCertificateDigest + }) + .returning(|_| Some(EnvoyBuffer::new(b"unknown_digest"))); + mock_filter + .expect_get_request_header_value() + .withf(|key| key == ":authority") + .returning(|_| Some(EnvoyBuffer::new(b"10.0.0.1:443"))); + mock_filter + .expect_increment_counter() + .withf(move |id, val| *id == miss_id && *val == 1) + .return_const(Ok(())) + .once(); + mock_filter + .expect_get_response_header_value() + .withf(|key| key == ":status") + .returning(|_| None); + let mut filter = config.new_http_filter(&mut mock_filter); + + assert_eq!( + filter.on_request_headers(&mut mock_filter, false), + envoy_dynamic_module_type_on_http_filter_request_headers_status::Continue + ); + assert_eq!( + filter.on_response_headers(&mut mock_filter, false), + envoy_dynamic_module_type_on_http_filter_response_headers_status::Continue + ); + } + + #[test] + fn test_on_request_headers_cache_hit_writes_filter_state_and_increments_hit_counter() { + let hit_id = EnvoyCounterId(1); + let miss_id = EnvoyCounterId(2); + let config = + EgressPolicyCacheFilterConfig::with_counters(Config::default(), hit_id, miss_id); + let expected_policy = r#"{"rules":[{"pattern":"*.example.com","mode":"mitm"}]}"#; + config.local_cache().borrow_mut().put( + "abc123digest;443".to_string(), + CachedPolicy::new(expected_policy.to_string(), Instant::now()), + ); + + let mut mock_filter = MockEnvoyHttpFilter::new(); + mock_filter + .expect_get_attribute_string() + .withf(|id| { + *id == envoy_dynamic_module_type_attribute_id::ConnectionSha256PeerCertificateDigest + }) + .returning(|_| Some(EnvoyBuffer::new(b"abc123digest"))); + mock_filter + .expect_get_request_header_value() + .withf(|key| key == ":authority") + .returning(|_| Some(EnvoyBuffer::new(b"10.0.0.1:443"))); + mock_filter + .expect_set_filter_state_bytes() + .withf(move |key, val| { + key == ATE_POLICY_EGRESS_CACHED.as_bytes() && val == expected_policy.as_bytes() + }) + .return_const(true) + .once(); + mock_filter + .expect_set_filter_state_bytes() + .withf(|key, val| { + key == ATE_POLICY_EGRESS_SKIP_CALLOUT.as_bytes() && val == SKIP_CALLOUT_TRUE.as_bytes() + }) + .return_const(true) + .once(); + mock_filter + .expect_set_dynamic_metadata_string() + .withf(|namespace, key, val| { + namespace == ATE_EGRESS_METADATA_NAMESPACE + && key == ATE_EGRESS_DIALED_PORT_KEY + && val == "443" + }) + .return_const(()) + .once(); + mock_filter + .expect_increment_counter() + .withf(move |id, val| *id == hit_id && *val == 1) + .return_const(Result::<(), envoy_dynamic_module_type_metrics_result>::Ok(())) + .once(); + + let mut filter = config.create_filter(); + assert!(!filter.has_cached_policy); + assert_eq!( + filter.on_request_headers(&mut mock_filter, false), + envoy_dynamic_module_type_on_http_filter_request_headers_status::Continue + ); + assert!(filter.has_cached_policy); + } + + #[test] + fn test_on_request_headers_expired_cache_entry_is_removed_and_increments_miss_counter() { + let hit_id = EnvoyCounterId(1); + let miss_id = EnvoyCounterId(2); + let config = EgressPolicyCacheFilterConfig::with_counters( + Config { + cache_ttl: Duration::from_secs(5), + cache_enabled: true, + max_cache_items: 1000, + }, + hit_id, + miss_id, + ); + config.local_cache().borrow_mut().put( + "abc123digest;443".to_string(), + CachedPolicy::new( + r#"{"rules":[{"pattern":"*.example.com","mode":"mitm"}]}"#.to_string(), + Instant::now() - Duration::from_secs(10), + ), + ); + + let mut mock_filter = MockEnvoyHttpFilter::new(); + mock_filter + .expect_get_attribute_string() + .withf(|id| { + *id == envoy_dynamic_module_type_attribute_id::ConnectionSha256PeerCertificateDigest + }) + .returning(|_| Some(EnvoyBuffer::new(b"abc123digest"))); + mock_filter + .expect_get_request_header_value() + .withf(|key| key == ":authority") + .returning(|_| Some(EnvoyBuffer::new(b"10.0.0.1:443"))); + mock_filter.expect_set_filter_state_bytes().never(); + mock_filter.expect_set_dynamic_metadata_string().never(); + mock_filter + .expect_increment_counter() + .withf(move |id, val| *id == miss_id && *val == 1) + .return_const(Result::<(), envoy_dynamic_module_type_metrics_result>::Ok(())) + .once(); + + let mut filter = config.create_filter(); + assert_eq!( + filter.on_request_headers(&mut mock_filter, false), + envoy_dynamic_module_type_on_http_filter_request_headers_status::Continue + ); + assert!(!filter.has_cached_policy); + assert!(config.local_cache().borrow_mut().get("abc123digest;443").is_none()); + } + + #[test] + fn test_on_response_headers_caches_on_200_with_policy_and_cert_digest() { + let config = EgressPolicyCacheFilterConfig::new(Config::default()); + let mut mock_filter = MockEnvoyHttpFilter::new(); + mock_filter + .expect_get_response_header_value() + .withf(|key| key == ":status") + .returning(|_| Some(EnvoyBuffer::new(b"200"))); + mock_filter + .expect_get_filter_state_bytes() + .withf(|key| key == ATE_POLICY_EGRESS.as_bytes()) + .returning(|_| { + Some(EnvoyBuffer::new( + br#"{"rules":[{"pattern":"*.example.com","mode":"mitm"}]}"#, + )) + }); + mock_filter + .expect_get_attribute_string() + .withf(|id| { + *id == envoy_dynamic_module_type_attribute_id::ConnectionSha256PeerCertificateDigest + }) + .returning(|_| Some(EnvoyBuffer::new(b"abc123digest"))); + mock_filter + .expect_get_request_header_value() + .withf(|key| key == ":authority") + .returning(|_| Some(EnvoyBuffer::new(b"10.0.0.1:443"))); + + let mut filter = config.new_http_filter(&mut mock_filter); + assert_eq!( + filter.on_response_headers(&mut mock_filter, false), + envoy_dynamic_module_type_on_http_filter_response_headers_status::Continue + ); + + let cached = config + .local_cache() + .borrow_mut() + .get("abc123digest;443") + .cloned() + .expect("expected entry in cache"); + assert_eq!( + cached.policy, + r#"{"rules":[{"pattern":"*.example.com","mode":"mitm"}]}"# + ); + assert!(cached.stored_at.elapsed() < Duration::from_secs(5)); + } + + #[test] + fn test_on_response_headers_skips_cache_when_already_cached() { + let config = EgressPolicyCacheFilterConfig::new(Config::default()); + let initial_stored_at = Instant::now() - Duration::from_secs(3); + config.local_cache().borrow_mut().put( + "abc123digest;443".to_string(), + CachedPolicy::new( + r#"{"rules":[{"pattern":"*.example.com","mode":"mitm"}]}"#.to_string(), + initial_stored_at, + ), + ); + + let mut mock_filter = MockEnvoyHttpFilter::new(); + mock_filter + .expect_get_response_header_value() + .withf(|key| key == ":status") + .returning(|_| Some(EnvoyBuffer::new(b"200"))); + mock_filter.expect_get_filter_state_bytes().never(); + + let mut filter = config.create_filter(); + filter.has_cached_policy = true; + assert_eq!( + filter.on_response_headers(&mut mock_filter, false), + envoy_dynamic_module_type_on_http_filter_response_headers_status::Continue + ); + + let cached = config + .local_cache() + .borrow_mut() + .get("abc123digest;443") + .cloned() + .expect("expected entry in cache"); + assert_eq!(cached.stored_at, initial_stored_at); + } + + #[test] + fn test_on_response_headers_skips_cache_when_disabled() { + let config = EgressPolicyCacheFilterConfig::new(Config { + cache_ttl: Duration::from_secs(5), + cache_enabled: false, + max_cache_items: 1000, + }); + let mut mock_filter = MockEnvoyHttpFilter::new(); + mock_filter.expect_get_response_header_value().never(); + + let mut filter = config.new_http_filter(&mut mock_filter); + assert_eq!( + filter.on_response_headers(&mut mock_filter, false), + envoy_dynamic_module_type_on_http_filter_response_headers_status::Continue + ); + assert_eq!(config.local_cache().borrow().len(), 0); + } + + #[test] + fn test_on_response_headers_skips_cache_on_non_200() { + let config = EgressPolicyCacheFilterConfig::new(Config::default()); + let mut mock_filter = MockEnvoyHttpFilter::new(); + mock_filter + .expect_get_response_header_value() + .withf(|key| key == ":status") + .returning(|_| Some(EnvoyBuffer::new(b"403"))); + + let mut filter = config.new_http_filter(&mut mock_filter); + assert_eq!( + filter.on_response_headers(&mut mock_filter, false), + envoy_dynamic_module_type_on_http_filter_response_headers_status::Continue + ); + assert_eq!(config.local_cache().borrow().len(), 0); + } + + #[test] + fn test_thread_local_lru_cache_shared_across_filters_and_isolated_across_threads() { + let filter_config = Arc::new(EgressPolicyCacheFilterConfig::new(Config { + cache_ttl: Duration::from_secs(5), + cache_enabled: true, + max_cache_items: 2, + })); + + let filter1 = filter_config.create_filter(); + let filter2 = filter_config.create_filter(); + + assert_eq!(filter1.local_cache().borrow().cap().get(), 2); + + let now = Instant::now(); + filter1 + .local_cache() + .borrow_mut() + .put("k1".to_string(), CachedPolicy::new("v1".to_string(), now)); + filter1 + .local_cache() + .borrow_mut() + .put("k2".to_string(), CachedPolicy::new("v2".to_string(), now)); + + // Second filter created from the same config on the same thread sees the entries. + assert_eq!( + filter2 + .local_cache() + .borrow_mut() + .get("k1") + .map(|e| e.policy.as_str()), + Some("v1") + ); + + // Inserting a 3rd item evicts the least-recently-used item ("k2"). + filter2 + .local_cache() + .borrow_mut() + .put("k3".to_string(), CachedPolicy::new("v3".to_string(), now)); + assert_eq!(filter1.local_cache().borrow_mut().get("k2"), None); + assert_eq!( + filter1 + .local_cache() + .borrow_mut() + .get("k1") + .map(|e| e.policy.as_str()), + Some("v1") + ); + assert_eq!( + filter1 + .local_cache() + .borrow_mut() + .get("k3") + .map(|e| e.policy.as_str()), + Some("v3") + ); + + // A filter on another worker thread gets its own empty thread-local cache. + let filter_config_clone = Arc::clone(&filter_config); + let other_thread_len = std::thread::spawn(move || { + let other_filter = filter_config_clone.create_filter(); + assert_eq!(other_filter.local_cache().borrow().cap().get(), 2); + other_filter.local_cache().borrow().len() + }) + .join() + .unwrap(); + assert_eq!(other_thread_len, 0); + } +} diff --git a/cmd/dataplane/envoy/dynamic-modules/egress-policy/.gitignore b/cmd/dataplane/envoy/dynamic-modules/egress-policy/.gitignore deleted file mode 100644 index 2f7896d1d1..0000000000 --- a/cmd/dataplane/envoy/dynamic-modules/egress-policy/.gitignore +++ /dev/null @@ -1 +0,0 @@ -target/ diff --git a/demos/egress/main.go b/demos/egress/main.go index 66607e054c..123127ed45 100644 --- a/demos/egress/main.go +++ b/demos/egress/main.go @@ -48,6 +48,15 @@ const ( type fetchRequest struct { URL string `json:"url"` + // Host, when set, overrides the outbound HTTP request's Host header + // without changing the dial destination or TLS SNI derived from URL. + Host string `json:"host,omitempty"` + // This flag allows tests to send each request over a new connection, + // resulting in a new CONNECT tunnel from the atunnel, rather than re-using + // existing connections in the http dialer connection pool and corresponding CONNECT + // tunnels in the atunnel. + // This is useful when testing egress policy caching behavior in the egress gateway. + DisableKeepAlive bool `json:"disableKeepAlive,omitempty"` } type fetchResponse struct { @@ -133,6 +142,13 @@ func newHandler(client *http.Client) http.Handler { writeJSON(w, http.StatusBadRequest, fetchResponse{Error: fmt.Sprintf("invalid URL: %v", err)}) return } + if input.Host != "" { + outbound.Host = input.Host + } + if input.DisableKeepAlive { + outbound.Close = true + client.CloseIdleConnections() + } if traceparent := r.Header.Get("traceparent"); traceparent != "" { outbound.Header.Set("traceparent", traceparent) } @@ -144,6 +160,7 @@ func newHandler(client *http.Client) http.Handler { defer response.Body.Close() body, err := io.ReadAll(io.LimitReader(response.Body, maxResponseBody)) + _ = response.Body.Close() if err != nil { writeJSON(w, http.StatusBadGateway, fetchResponse{Error: fmt.Sprintf("reading response: %v", err)}) return diff --git a/demos/egress/main_test.go b/demos/egress/main_test.go index c164d3371f..bdb4ffe8b1 100644 --- a/demos/egress/main_test.go +++ b/demos/egress/main_test.go @@ -69,6 +69,59 @@ func TestFetch(t *testing.T) { } } +func TestFetchDisableKeepAlive(t *testing.T) { + client := &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) { + if !r.Close { + t.Errorf("upstream request Close = false, want true when disableKeepAlive is set") + } + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader("ok")), + Header: make(http.Header), + }, nil + })} + + payload, err := json.Marshal(fetchRequest{URL: "https://allowed.example/", DisableKeepAlive: true}) + if err != nil { + t.Fatal(err) + } + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(string(payload))) + newHandler(client).ServeHTTP(recorder, request) + + if recorder.Code != http.StatusOK { + t.Fatalf("status = %d, want %d", recorder.Code, http.StatusOK) + } +} + +func TestFetchHostOverride(t *testing.T) { + client := &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) { + if r.URL.Host != "allowed.example" { + t.Errorf("upstream URL.Host = %q, want %q", r.URL.Host, "allowed.example") + } + if r.Host != "allowed.example:1" { + t.Errorf("upstream Host = %q, want %q", r.Host, "allowed.example:1") + } + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader("ok")), + Header: make(http.Header), + }, nil + })} + + payload, err := json.Marshal(fetchRequest{URL: "https://allowed.example/", Host: "allowed.example:1"}) + if err != nil { + t.Fatal(err) + } + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(string(payload))) + newHandler(client).ServeHTTP(recorder, request) + + if recorder.Code != http.StatusOK { + t.Fatalf("status = %d, want %d", recorder.Code, http.StatusOK) + } +} + func TestFetchHTTPAndHTTPS(t *testing.T) { httpSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { _, _ = io.WriteString(w, "from http") diff --git a/docs/operator-install.md b/docs/operator-install.md index 902382a6d1..b1f6ec8a0c 100644 --- a/docs/operator-install.md +++ b/docs/operator-install.md @@ -286,6 +286,7 @@ command. A setting with no default is unset unless something supplies it. | `demo.counter.storageClass` | `STORAGE_CLASS` | `--storage-class` | `standard` | `deploy demo counter` | | `demo.counter.withExternalVolume` | `ATE_DEMO_COUNTER_WITH_EXTERNAL_VOLUME` | `--with-external-volume` | `false` | `deploy demo counter` | | `docker.buildFlags` | `DOCKER_BUILD_FLAGS` | `--docker-build-flags` | — | (global) | +| `e2e.envoyConcurrency` | `E2E_ENVOY_CONCURRENCY` | `--e2e-envoy-concurrency` | — | (global) | | `gcp.clusterLocation` | `CLUSTER_LOCATION` | `--gcp-cluster-location` | — | (global) | | `gcp.clusterName` | `CLUSTER_NAME` | `--gcp-cluster-name` | — | (global) | | `gcp.projectID` | `PROJECT_ID` | `--gcp-project-id` | — | (global) | diff --git a/internal/e2e/atenet_dataplane.go b/internal/e2e/atenet_dataplane.go index 25dc80ce3e..7ade51647b 100644 --- a/internal/e2e/atenet_dataplane.go +++ b/internal/e2e/atenet_dataplane.go @@ -37,6 +37,7 @@ type AtenetDataplane interface { IsEgressPolicyDenied(status int, body string) bool PlatformMetricPrefixes([]string) []string RouteDurationSeen(context.Context, string) (bool, error) + IsDataplaneConcurrencyDisabled() bool } // CurrentAtenetDataplane returns the implementation selected for this test @@ -77,6 +78,10 @@ func (envoyAtenetDataplane) RouteDurationSeen(_ context.Context, collectorScrape return len(MissingPlatformMetrics(collectorScrape, []string{"atenet_router_route_duration"})) == 0, nil } +func (envoyAtenetDataplane) IsDataplaneConcurrencyDisabled() bool { + return os.Getenv("E2E_ENVOY_CONCURRENCY") == "1" +} + type agentGatewayAtenetDataplane struct{} func (agentGatewayAtenetDataplane) NewParkingObserver(context.Context) (ParkingObserver, error) { @@ -111,6 +116,8 @@ func (agentGatewayAtenetDataplane) RouteDurationSeen(ctx context.Context, _ stri return len(MissingPlatformMetrics(scrape, []string{"agentgateway_atenet_router_route_duration_seconds"})) == 0, nil } +func (agentGatewayAtenetDataplane) IsDataplaneConcurrencyDisabled() bool { return false } + type agentGatewayParkingObserver struct{} func (agentGatewayParkingObserver) WaitForCount(ctx context.Context, cond func(int) bool) (int, error) { diff --git a/internal/e2e/collector_metrics.go b/internal/e2e/collector_metrics.go index 293a63ddae..7044227580 100644 --- a/internal/e2e/collector_metrics.go +++ b/internal/e2e/collector_metrics.go @@ -25,7 +25,9 @@ import ( "github.com/agent-substrate/substrate/internal/ateclient" "github.com/agent-substrate/substrate/internal/portforward" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" ) const ( @@ -35,6 +37,25 @@ const ( // AgentGateway exposes native Prometheus metrics; it does not export these // instruments through the OTLP collector. agentGatewayRouterStatsPort = 15020 + // The egress Envoy exposes its Prometheus stats on the admin listener. + egressEnvoyAdminPort = 15000 + + // EgressExtProcIdentityStatPrefix is the Envoy ext_proc filter stat_prefix on + // the outer CONNECT leg in atenet-egress. + EgressExtProcIdentityStatPrefix = "egress_identity" + // EgressExtProcPolicyMITMStatPrefix is the Envoy ext_proc filter stat_prefix + // on the inner MITM HTTP leg in atenet-egress. + EgressExtProcPolicyMITMStatPrefix = "egress_policy_mitm" + // EgressExtProcPolicyClearTextStatPrefix is the Envoy ext_proc filter stat_prefix + // on the inner clear text HTTP leg in atenet-egress. + EgressExtProcPolicyClearTextStatPrefix = "egress_policy_cleartext" + + // EgressConnectCacheHitCounter is the Envoy dynamic-module counter name for + // CONNECT policy cache hits in atenet-egress. + EgressConnectCacheHitCounter = "ate_egress_connect_cache_hit" + // EgressConnectCacheMissCounter is the Envoy dynamic-module counter name for + // CONNECT policy cache misses in atenet-egress. + EgressConnectCacheMissCounter = "ate_egress_connect_cache_miss" ) // PlatformMetricPrefixes are the Prometheus metric-name prefixes (OTLP dots @@ -129,6 +150,131 @@ func ScrapeCollectorMetrics(ctx context.Context) (string, error) { return string(body), nil } +// ScrapeEgressEnvoyMetrics port-forwards each ready atenet-egress pod's Envoy +// admin port and reads its /stats/prometheus endpoint, returning the +// concatenated exposition text. +func ScrapeEgressEnvoyMetrics(ctx context.Context) (string, error) { + config, err := ateclient.LoadKubeConfig(KubeConfig, KubeContext) + if err != nil { + return "", fmt.Errorf("loading kubeconfig: %w", err) + } + clientset, err := kubernetes.NewForConfig(config) + if err != nil { + return "", fmt.Errorf("creating k8s client: %w", err) + } + + ns := SystemNamespace() + pods, err := clientset.CoreV1().Pods(ns).List(ctx, metav1.ListOptions{ + LabelSelector: "app=atenet-egress", + }) + if err != nil { + return "", fmt.Errorf("listing atenet-egress pods: %w", err) + } + + var b strings.Builder + ready := 0 + for i := range pods.Items { + pod := &pods.Items[i] + if !portforward.IsPodReady(pod) { + continue + } + ready++ + body, err := scrapeEgressEnvoyPod(ctx, config, clientset, ns, pod.Name) + if err != nil { + return "", err + } + b.WriteString(body) + } + if ready == 0 { + return "", fmt.Errorf("no ready atenet-egress pods in %s", ns) + } + return b.String(), nil +} + +func scrapeEgressEnvoyPod(ctx context.Context, config *rest.Config, clientset kubernetes.Interface, namespace, podName string) (string, error) { + localPort, stop, err := portforward.PodPortForward(ctx, config, clientset, namespace, podName, egressEnvoyAdminPort) + if err != nil { + return "", err + } + defer stop() + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf("http://127.0.0.1:%d/stats/prometheus", localPort), nil) + if err != nil { + return "", err + } + resp, err := (&http.Client{Timeout: 30 * time.Second}).Do(req) + if err != nil { + return "", fmt.Errorf("scraping egress Envoy metrics on %s: %w", podName, err) + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + return "", fmt.Errorf("reading egress Envoy metrics on %s: %w", podName, err) + } + if resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("egress Envoy metrics on %s returned %d: %s", podName, resp.StatusCode, body) + } + return string(body), nil +} + +// EgressExtProcStreamCounts returns the envoy_http_ext_proc__streams_started +// count for each egress ext_proc stat_prefix in an Envoy /stats/prometheus +// scrape, summed across instances. +func EgressExtProcStreamCounts(scrape string) map[string]int { + counts := map[string]int{} + for _, line := range strings.Split(scrape, "\n") { + line = strings.TrimSpace(line) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + name := metricNameFromLine(line) + rest, ok := strings.CutPrefix(name, "envoy_http_ext_proc_") + if !ok { + continue + } + statPrefix, ok := strings.CutSuffix(rest, "_streams_started") + if !ok { + statPrefix, ok = strings.CutSuffix(rest, "_streams_started_total") + } + if !ok || statPrefix == "" { + continue + } + fields := strings.Fields(line) + v, err := strconv.ParseFloat(fields[len(fields)-1], 64) + if err != nil { + continue + } + counts[statPrefix] += int(v) + } + return counts +} + +// EgressPolicyCacheCounts returns the CONNECT policy cache hit and miss counts +// from an Envoy /stats/prometheus scrape, summed across instances. +func EgressPolicyCacheCounts(scrape string) (hits, misses int) { + for _, line := range strings.Split(scrape, "\n") { + line = strings.TrimSpace(line) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + name := strings.TrimSuffix(metricNameFromLine(line), "_total") + fields := strings.Fields(line) + v, err := strconv.ParseFloat(fields[len(fields)-1], 64) + if err != nil { + continue + } + switch name { + case "envoy_dynamicmodulescustom_" + EgressConnectCacheHitCounter, + "envoy_" + EgressConnectCacheHitCounter: + hits += int(v) + case "envoy_dynamicmodulescustom_" + EgressConnectCacheMissCounter, + "envoy_" + EgressConnectCacheMissCounter: + misses += int(v) + } + } + return hits, misses +} + // MissingPlatformMetrics returns the prefixes with no matching series in the // Prometheus exposition text. A metric matches when its name equals a prefix or // begins with prefix+"_"; the underscore boundary stops "ate_actor_restore" from diff --git a/internal/e2e/collector_metrics_test.go b/internal/e2e/collector_metrics_test.go index b0efb72843..8ca74923e3 100644 --- a/internal/e2e/collector_metrics_test.go +++ b/internal/e2e/collector_metrics_test.go @@ -257,3 +257,98 @@ func TestStatesNotAdvanced(t *testing.T) { }) } } + +const sampleEgressEnvoyScrape = `# TYPE envoy_http_ext_proc_egress_identity_streams_started counter +envoy_http_ext_proc_egress_identity_streams_started{envoy_http_conn_manager_prefix="egress_connect"} 6 +# TYPE envoy_http_ext_proc_egress_policy_mitm_streams_started counter +envoy_http_ext_proc_egress_policy_mitm_streams_started{envoy_http_conn_manager_prefix="mitm_http"} 1 +# TYPE envoy_http_ext_proc_egress_policy_cleartext_streams_started counter +envoy_http_ext_proc_egress_policy_cleartext_streams_started{envoy_http_conn_manager_prefix="mitm_cleartext"} 0 +# TYPE envoy_http_ext_proc_egress_identity_streams_closed counter +envoy_http_ext_proc_egress_identity_streams_closed{envoy_http_conn_manager_prefix="egress_connect"} 6 +# TYPE envoy_dynamicmodulescustom_ate_egress_connect_cache_hit counter +envoy_dynamicmodulescustom_ate_egress_connect_cache_hit{} 4 +# TYPE envoy_dynamicmodulescustom_ate_egress_connect_cache_miss counter +envoy_dynamicmodulescustom_ate_egress_connect_cache_miss{} 1 +` + +func TestEgressExtProcStreamCounts(t *testing.T) { + tests := []struct { + name string + scrape string + want map[string]int + }{ + { + name: "per-leg ext_proc counters", + scrape: sampleEgressEnvoyScrape, + want: map[string]int{ + EgressExtProcIdentityStatPrefix: 6, + EgressExtProcPolicyMITMStatPrefix: 1, + "egress_policy_cleartext": 0, + }, + }, + { + name: "summed across replicas", + scrape: sampleEgressEnvoyScrape + sampleEgressEnvoyScrape, + want: map[string]int{ + EgressExtProcIdentityStatPrefix: 12, + EgressExtProcPolicyMITMStatPrefix: 2, + "egress_policy_cleartext": 0, + }, + }, + { + name: "empty scrape", + scrape: "", + want: map[string]int{}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := EgressExtProcStreamCounts(tt.scrape) + if len(got) != len(tt.want) { + t.Fatalf("EgressExtProcStreamCounts() = %v, want %v", got, tt.want) + } + for prefix, want := range tt.want { + if got[prefix] != want { + t.Errorf("EgressExtProcStreamCounts()[%q] = %d, want %d", prefix, got[prefix], want) + } + } + }) + } +} + +func TestEgressPolicyCacheCounts(t *testing.T) { + tests := []struct { + name string + scrape string + wantHits int + wantMisses int + }{ + { + name: "single replica", + scrape: sampleEgressEnvoyScrape, + wantHits: 4, + wantMisses: 1, + }, + { + name: "summed across replicas", + scrape: sampleEgressEnvoyScrape + sampleEgressEnvoyScrape, + wantHits: 8, + wantMisses: 2, + }, + { + name: "empty scrape", + scrape: "", + wantHits: 0, + wantMisses: 0, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gotHits, gotMisses := EgressPolicyCacheCounts(tt.scrape) + if gotHits != tt.wantHits || gotMisses != tt.wantMisses { + t.Errorf("EgressPolicyCacheCounts() = (%d, %d), want (%d, %d)", gotHits, gotMisses, tt.wantHits, tt.wantMisses) + } + }) + } +} diff --git a/internal/e2e/suites/networking/egresspolicy_test.go b/internal/e2e/suites/networking/egresspolicy_test.go index feb7e4268d..a214084953 100644 --- a/internal/e2e/suites/networking/egresspolicy_test.go +++ b/internal/e2e/suites/networking/egresspolicy_test.go @@ -99,7 +99,10 @@ func TestActorEgressRequiresPolicy(t *testing.T) { // example.com and nothing else, and waits until it is routable. func hostnamePolicyActor(t *testing.T, ctx context.Context) (*e2e.RouterClient, resources.ActorRef) { t.Helper() - _, actorName, _ := createAndResumeActorWithEgress(t, ctx, "egress-sni", e2e.EgressFixture(), e2e.EgressAllowHTTPS("example.com"), e2e.EgressAllowPassthrough("example.edu")) + _, actorName, _ := createAndResumeActorWithEgress(t, ctx, "egress-sni", e2e.EgressFixture(), + e2e.EgressAllowHTTPS("example.com"), + e2e.EgressAllowPassthrough("example.edu"), + e2e.EgressAllowHTTP("example.net")) router := mustRouterClient(t, ctx) t.Cleanup(func() { router.Close() }) actorRef := resources.ActorRef{Atespace: networkingAtespace, Name: actorName} @@ -154,6 +157,198 @@ func TestActorEgressHTTPSByHostnamePassthrough(t *testing.T) { } } +// TestActorEgressPolicyCache sends 5 requests to example.com and checks that +// the egress policy cache counters record 4 hits and 1 miss, and that the +// outer CONNECT ext_proc filter is invoked only once. +func TestActorEgressPolicyCache(t *testing.T) { + ctx := context.Background() + router, actorRef := hostnamePolicyActor(t, ctx) + + beforeScrape, err := e2e.ScrapeEgressEnvoyMetrics(ctx) + if err != nil { + t.Fatalf("ScrapeEgressEnvoyMetrics before fetches: %v", err) + } + beforeHits, beforeMisses := e2e.EgressPolicyCacheCounts(beforeScrape) + beforeExtProc := e2e.EgressExtProcStreamCounts(beforeScrape) + + const numRequests = 5 + payload := []byte(`{"url":"https://example.com/","disableKeepAlive":true}`) + for i := range numRequests { + status, body := postThroughEgressActorUntil(t, ctx, router, actorRef, "/", payload, reached) + if status != http.StatusOK { + t.Fatalf("request %d to example.com returned HTTP %d, want 200; body: %s", i+1, status, body) + } + } + + if !e2e.CurrentAtenetDataplane().IsDataplaneConcurrencyDisabled() { + t.Log("skipping counter verification: dataplane concurrency is not disabled") + return + } + + afterScrape, err := e2e.ScrapeEgressEnvoyMetrics(ctx) + if err != nil { + t.Fatalf("ScrapeEgressEnvoyMetrics after fetches: %v", err) + } + afterHits, afterMisses := e2e.EgressPolicyCacheCounts(afterScrape) + afterExtProc := e2e.EgressExtProcStreamCounts(afterScrape) + + const ( + wantHits = 4 + wantMisses = 1 + wantExtProcCalls = 1 + ) + if got := afterHits - beforeHits; got != wantHits { + t.Errorf("egress policy cache hits delta = %d, want %d", got, wantHits) + } + if got := afterMisses - beforeMisses; got != wantMisses { + t.Errorf("egress policy cache misses delta = %d, want %d", got, wantMisses) + } + if got := afterExtProc[e2e.EgressExtProcIdentityStatPrefix] - beforeExtProc[e2e.EgressExtProcIdentityStatPrefix]; got != wantExtProcCalls { + t.Errorf("egress_identity ext_proc streams_started delta = %d, want %d", got, wantExtProcCalls) + } +} + +// TestActorEgressPolicyCacheExpiration verifies that cached egress policy +// entries expire after the configured cache_ttl. +func TestActorEgressPolicyCacheExpiration(t *testing.T) { + ctx := context.Background() + router, actorRef := hostnamePolicyActor(t, ctx) + + beforeScrape, err := e2e.ScrapeEgressEnvoyMetrics(ctx) + if err != nil { + t.Fatalf("ScrapeEgressEnvoyMetrics before fetches: %v", err) + } + beforeHits, beforeMisses := e2e.EgressPolicyCacheCounts(beforeScrape) + beforeExtProc := e2e.EgressExtProcStreamCounts(beforeScrape) + + payload := []byte(`{"url":"https://example.com/","disableKeepAlive":true}`) + for i := range 2 { + status, body := postThroughEgressActorUntil(t, ctx, router, actorRef, "/", payload, reached) + if status != http.StatusOK { + t.Fatalf("pre-expiration request %d to example.com returned HTTP %d, want 200; body: %s", i+1, status, body) + } + } + + // Wait longer than the 5s default cache_ttl so the cached policy expires. + time.Sleep(6 * time.Second) + + for i := range 2 { + status, body := postThroughEgressActorUntil(t, ctx, router, actorRef, "/", payload, reached) + if status != http.StatusOK { + t.Fatalf("post-expiration request %d to example.com returned HTTP %d, want 200; body: %s", i+1, status, body) + } + } + + if !e2e.CurrentAtenetDataplane().IsDataplaneConcurrencyDisabled() { + t.Log("skipping counter verification: dataplane concurrency is not disabled") + return + } + + afterScrape, err := e2e.ScrapeEgressEnvoyMetrics(ctx) + if err != nil { + t.Fatalf("ScrapeEgressEnvoyMetrics after fetches: %v", err) + } + afterHits, afterMisses := e2e.EgressPolicyCacheCounts(afterScrape) + afterExtProc := e2e.EgressExtProcStreamCounts(afterScrape) + + const ( + wantHits = 2 + wantMisses = 2 + wantExtProcCalls = 2 + ) + if got := afterHits - beforeHits; got != wantHits { + t.Errorf("egress policy cache hits delta = %d, want %d", got, wantHits) + } + if got := afterMisses - beforeMisses; got != wantMisses { + t.Errorf("egress policy cache misses delta = %d, want %d", got, wantMisses) + } + if got := afterExtProc[e2e.EgressExtProcIdentityStatPrefix] - beforeExtProc[e2e.EgressExtProcIdentityStatPrefix]; got != wantExtProcCalls { + t.Errorf("egress_identity ext_proc streams_started delta = %d, want %d", got, wantExtProcCalls) + } +} + +// TestActorEgressPolicyCachePerPort sends 5 requests to http://example.net and +// then 5 requests to https://example.com, verifying that per-port policy +// caching records 2 cache misses, 8 cache hits, and 2 outer CONNECT ext_proc +// callouts. +func TestActorEgressPolicyCachePerPort(t *testing.T) { + ctx := context.Background() + router, actorRef := hostnamePolicyActor(t, ctx) + + beforeScrape, err := e2e.ScrapeEgressEnvoyMetrics(ctx) + if err != nil { + t.Fatalf("ScrapeEgressEnvoyMetrics before fetches: %v", err) + } + beforeHits, beforeMisses := e2e.EgressPolicyCacheCounts(beforeScrape) + beforeExtProc := e2e.EgressExtProcStreamCounts(beforeScrape) + + const numRequests = 5 + for _, targetURL := range []string{"http://example.net/", "https://example.com/"} { + payload := []byte(fmt.Sprintf(`{"url":%q,"disableKeepAlive":true}`, targetURL)) + for i := range numRequests { + status, body := postThroughEgressActorUntil(t, ctx, router, actorRef, "/", payload, reached) + if status != http.StatusOK { + t.Fatalf("request %d to %s returned HTTP %d, want 200; body: %s", i+1, targetURL, status, body) + } + } + } + + if !e2e.CurrentAtenetDataplane().IsDataplaneConcurrencyDisabled() { + t.Log("skipping counter verification: dataplane concurrency is not disabled") + return + } + + afterScrape, err := e2e.ScrapeEgressEnvoyMetrics(ctx) + if err != nil { + t.Fatalf("ScrapeEgressEnvoyMetrics after fetches: %v", err) + } + afterHits, afterMisses := e2e.EgressPolicyCacheCounts(afterScrape) + afterExtProc := e2e.EgressExtProcStreamCounts(afterScrape) + + const ( + wantHits = 8 + wantMisses = 2 + wantExtProcCalls = 2 + ) + if got := afterHits - beforeHits; got != wantHits { + t.Errorf("egress policy cache hits delta = %d, want %d", got, wantHits) + } + if got := afterMisses - beforeMisses; got != wantMisses { + t.Errorf("egress policy cache misses delta = %d, want %d", got, wantMisses) + } + if got := afterExtProc[e2e.EgressExtProcIdentityStatPrefix] - beforeExtProc[e2e.EgressExtProcIdentityStatPrefix]; got != wantExtProcCalls { + t.Errorf("egress_identity ext_proc streams_started delta = %d, want %d", got, wantExtProcCalls) + } +} + +// TestActorEgressDialedPortOverridesHostHeader sends requests to +// https://example.com/ (default TCP port 443) with "example.com:1" in the Host +// header and to http://example.net/ (default TCP port 80) with "example.net:2" +// in the Host header, verifying that the egress gateway dials the TCP port the +// actor connected to rather than the port from the Host header. +// Sending 2 requests per target verifies that cached policy correctly sets the +// destination port filter state. +func TestActorEgressDialedPortOverridesHostHeader(t *testing.T) { + ctx := context.Background() + router, actorRef := hostnamePolicyActor(t, ctx) + + for _, tc := range []struct { + url string + host string + }{ + {url: "https://example.com/", host: "example.com:1"}, + {url: "http://example.net/", host: "example.net:2"}, + } { + payload := []byte(fmt.Sprintf(`{"url":%q,"host":%q,"disableKeepAlive":true}`, tc.url, tc.host)) + for i := range 2 { + status, body := postThroughEgressActorUntil(t, ctx, router, actorRef, "/", payload, reached) + if status != http.StatusOK { + t.Fatalf("request %d to %s with Host %s returned HTTP %d, want 200; body: %s", i+1, tc.url, tc.host, status, body) + } + } + } +} + // fetchThroughEgressActorUntil is fetchThroughEgressActor with the caller // deciding which answer is final. func fetchThroughEgressActorUntil(t *testing.T, ctx context.Context, router *e2e.RouterClient, actorRef resources.ActorRef, url string, done func(status int, body []byte) bool) (int, []byte) { diff --git a/internal/portforward/portforward.go b/internal/portforward/portforward.go index 7fa750b251..4b79b327a8 100644 --- a/internal/portforward/portforward.go +++ b/internal/portforward/portforward.go @@ -48,7 +48,7 @@ func ServicePortForward(ctx context.Context, config *rest.Config, clientset kube if err != nil { return 0, nil, err } - return podPortForward(ctx, config, clientset, namespace, pod.Name, targetPort) + return PodPortForward(ctx, config, clientset, namespace, pod.Name, targetPort) } func firstReadyPodForService(ctx context.Context, clientset kubernetes.Interface, namespace, service string) (*corev1.Pod, *corev1.Service, error) { @@ -102,7 +102,10 @@ func resolveTargetPort(svc *corev1.Service, pod *corev1.Pod, servicePort int32) return 0, fmt.Errorf("service %s has no port %d", svc.Name, servicePort) } -func podPortForward(ctx context.Context, config *rest.Config, clientset kubernetes.Interface, namespace, podName string, targetPort int32) (int, func(), error) { +// PodPortForward forwards a random local port to targetPort of the named pod. +// It returns the chosen local port and a stop func the caller must invoke to +// tear the tunnel down. +func PodPortForward(ctx context.Context, config *rest.Config, clientset kubernetes.Interface, namespace, podName string, targetPort int32) (int, func(), error) { req := clientset.CoreV1().RESTClient().Post(). Resource("pods"). Namespace(namespace). diff --git a/manifests/ate-install/atenet-egress.yaml b/manifests/ate-install/atenet-egress.yaml index c8bf828581..e999e2268a 100644 --- a/manifests/ate-install/atenet-egress.yaml +++ b/manifests/ate-install/atenet-egress.yaml @@ -56,6 +56,9 @@ data: # dialling the pod IP, which is IPv4 on an IPv4 cluster. socket_address: { address: "::", ipv4_compat: true, port_value: 15000 } + # E2E_ENVOY_STATS_FLUSH_ON_ADMIN is substituted in renderAtenetEgressManifest + stats_flush_on_admin: ${E2E_ENVOY_STATS_FLUSH_ON_ADMIN} + # Required by the mitm_listener below. An internal listener has no socket, # so it is reachable only through a cluster with an envoy_internal_address; # without this extension registered, that listener silently fails to load. @@ -177,51 +180,81 @@ data: text_format_source: inline_string: "%REQ(:AUTHORITY)%" omit_empty_values: true + - name: envoy.filters.http.dynamic_modules + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.dynamic_modules.v3.DynamicModuleFilter + dynamic_module_config: + name: envoy_substrate_egress_policy_cache + filter_name: envoy_substrate_egress_policy_cache # The CONNECT leg. The sidecar authenticates the actor from its # certificate (see cmd/atenet/internal/router/egress) and returns # the SNI rules in dev.ate.policy.egress and the dialed port in # dev.ate.egress. Fails closed if the router is down. - - name: envoy.filters.http.ext_proc + # Skipped when envoy_substrate_egress_policy_cache already set + # dev.ate.policy.egress.skip_callout to "true" in filter state on a cache hit. + - name: envoy.filters.http.composite typed_config: - "@type": type.googleapis.com/envoy.extensions.filters.http.ext_proc.v3.ExternalProcessor - stat_prefix: egress_identity - grpc_service: - envoy_grpc: - cluster_name: ext_proc_server - timeout: 2s - failure_mode_allow: false - # Set ext_proc timeout to 5s (from default 200ms) to handle slower ext_proc transactions. - message_timeout: 5s - # How the ext_proc server tells egress from ingress. It applies - # opposite trust models to the two directions, and dispatches on - # this Envoy-asserted filter chain name so that no client can - # select the egress path by crafting a request. The value must - # match EgressFilterChainName in - # cmd/atenet/internal/router/extproc/dispatch.go; renaming the - # filter chain above without updating it fails closed (the - # request is classified as ingress, which this --mode=egress - # instance does not serve, and is refused with a 404). - request_attributes: - - xds.filter_chain_name - processing_mode: - request_header_mode: SEND - response_header_mode: SKIP - request_body_mode: NONE - response_body_mode: NONE - request_trailer_mode: SKIP - response_trailer_mode: SKIP - # This leg authenticates and never rewrites; the sidecar may - # not touch routing or Envoy's own headers here. - mutation_rules: - disallow_system: true - disallow_is_error: true - # Envoy silently drops metadata from a namespace not listed - # here, and the filter state below would never be set. - metadata_options: - receiving_namespaces: - untyped: - - dev.ate.egress - - dev.ate.policy.egress + "@type": type.googleapis.com/envoy.extensions.filters.http.composite.v3.Composite + matcher: + matcher_list: + matchers: + - predicate: + not_matcher: + single_predicate: + input: + name: filter_state + typed_config: + "@type": type.googleapis.com/envoy.extensions.matching.common_inputs.network.v3.FilterStateInput + key: dev.ate.policy.egress.skip_callout + value_match: + exact: "true" + on_match: + action: + name: composite-action + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.composite.v3.ExecuteFilterAction + typed_config: + name: envoy.filters.http.ext_proc + typed_config: + "@type": type.googleapis.com/envoy.extensions.filters.http.ext_proc.v3.ExternalProcessor + stat_prefix: egress_identity + grpc_service: + envoy_grpc: + cluster_name: ext_proc_server + timeout: 2s + failure_mode_allow: false + # Set ext_proc timeout to 5s (from default 200ms) to handle slower ext_proc transactions. + message_timeout: 5s + # How the ext_proc server tells egress from ingress. It applies + # opposite trust models to the two directions, and dispatches on + # this Envoy-asserted filter chain name so that no client can + # select the egress path by crafting a request. The value must + # match EgressFilterChainName in + # cmd/atenet/internal/router/extproc/dispatch.go; renaming the + # filter chain above without updating it fails closed (the + # request is classified as ingress, which this --mode=egress + # instance does not serve, and is refused with a 404). + request_attributes: + - xds.filter_chain_name + processing_mode: + request_header_mode: SEND + response_header_mode: SKIP + request_body_mode: NONE + response_body_mode: NONE + request_trailer_mode: SKIP + response_trailer_mode: SKIP + # This leg authenticates and never rewrites; the sidecar may + # not touch routing or Envoy's own headers here. + mutation_rules: + disallow_system: true + disallow_is_error: true + # Envoy silently drops metadata from a namespace not listed + # here, and the filter state below would never be set. + metadata_options: + receiving_namespaces: + untyped: + - dev.ate.egress + - dev.ate.policy.egress # The CONNECT leg's answers, copied into filter state shared with # the inner listener. Must stay after ext_proc. - name: envoy.filters.http.set_filter_state @@ -249,6 +282,14 @@ data: text_format_source: inline_string: "%DYNAMIC_METADATA(dev.ate.policy.egress)%" omit_empty_values: true + - object_key: dev.ate.policy.egress + factory_key: envoy.string + skip_if_empty: true + shared_with_upstream: ONCE + format_string: + text_format_source: + inline_string: "%FILTER_STATE(dev.ate.policy.egress.cached:PLAIN)%" + omit_empty_values: true - name: envoy.filters.http.router typed_config: "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router @@ -1262,6 +1303,8 @@ spec: - atenet-egress - --service-cluster - atenet-egress + # E2E_ENVOY_CONCURRENCY is substituted in renderAtenetEgressManifest + - ${E2E_ENVOY_CONCURRENCY} # Prevents Envoy from fast-exiting on SIGTERM before the ext-proc # sidecar finishes its drain sequence. Polls for the drain-complete # marker the sidecar writes on the shared emptyDir, terminating Envoy