Increase Envoy dataplane efficiency by adding egress policy cache

.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

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

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,

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)",

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 <value> 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

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)
+		}
+	})
+}

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)
+		}
 	}
 }
 

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.

cmd/dataplane/envoy/Dockerfile

@@ -12,32 +12,33 @@
 # 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 \
     clang \
     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

cmd/dataplane/envoy/dynamic-modules/.gitignore

@@ -0,0 +1,3 @@
+target/
+egress-policy/target/
+egress-policy-cache/target/