From 578e49ec6dd12129ad1334c64e29f95d1d3340c0 Mon Sep 17 00:00:00 2001 From: Renuka Fernando Date: Mon, 13 Jul 2026 09:04:07 +0530 Subject: [PATCH] perf(policy-engine): drop per-request deep copy of policy params MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit deepCopyParams did a full JSON marshal/unmarshal round-trip on spec.Parameters.Raw at all six execution phases on every request. Params are an immutable snapshot published once at chain-build time and shared read-only across concurrent requests, so the copy was pure overhead — it accounted for ~30% of the ext_proc CPU tree. - Pass spec.Parameters.Raw directly to each policy On* method - Remove deepCopyParams and the now-unused encoding/json import - Drop the three orphaned TestDeepCopyParams_* tests Verified with go test -race on the executor and kernel packages (kernel covers the production chain-build/publish path). An A/B benchmark with a realistic nested param map showed ~30-40x faster execution and ~14x fewer allocations per policy. --- .../policy-engine/internal/executor/chain.go | 81 +++++-------------- .../internal/executor/chain_test.go | 56 ------------- 2 files changed, 18 insertions(+), 119 deletions(-) diff --git a/gateway/gateway-runtime/policy-engine/internal/executor/chain.go b/gateway/gateway-runtime/policy-engine/internal/executor/chain.go index 29f3c8a3d9..587d2f2190 100644 --- a/gateway/gateway-runtime/policy-engine/internal/executor/chain.go +++ b/gateway/gateway-runtime/policy-engine/internal/executor/chain.go @@ -20,7 +20,6 @@ package executor import ( "context" - "encoding/json" "fmt" "log/slog" "strings" @@ -141,13 +140,9 @@ func (c *ChainExecutor) ExecuteRequestHeaderPolicies( } } - params, err := deepCopyParams(spec.Parameters.Raw) - if err != nil { - span.End() - return nil, fmt.Errorf("failed to clone parameters for policy %s:%s: %w", spec.Name, spec.Version, err) - } - - action := headerPol.OnRequestHeaders(ctx, reqCtx, params) + // spec.Parameters.Raw is an immutable snapshot published at chain-build time and + // shared read-only across concurrent requests; policies must not mutate it. + action := headerPol.OnRequestHeaders(ctx, reqCtx, spec.Parameters.Raw) executionTime := time.Since(policyStartTime) // Apply header mutations to reqCtx so subsequent policies and CEL conditions see the mutated state @@ -308,16 +303,10 @@ func (c *ChainExecutor) ExecuteRequestPolicies(ctx context.Context, policyList [ } } - // Deep-copy params to prevent a policy from mutating the shared spec map - // across concurrent requests (nested maps/slices require a full deep copy). - params, err := deepCopyParams(spec.Parameters.Raw) - if err != nil { - span.End() - return nil, fmt.Errorf("failed to clone parameters for policy %s:%s: %w", spec.Name, spec.Version, err) - } - + // spec.Parameters.Raw is an immutable snapshot published at chain-build time and + // shared read-only across concurrent requests; policies must not mutate it. slog.Debug("[body] calling OnRequestBody", "policy", spec.Name, "version", spec.Version, "route", route) - action := rp.OnRequestBody(ctx, reqCtx, params) + action := rp.OnRequestBody(ctx, reqCtx, spec.Parameters.Raw) executionTime := time.Since(policyStartTime) // Record policy execution metrics @@ -467,13 +456,9 @@ func (c *ChainExecutor) ExecuteResponseHeaderPolicies( } } - params, err := deepCopyParams(spec.Parameters.Raw) - if err != nil { - span.End() - return nil, fmt.Errorf("failed to clone parameters for policy %s:%s: %w", spec.Name, spec.Version, err) - } - - action := headerPol.OnResponseHeaders(ctx, respCtx, params) + // spec.Parameters.Raw is an immutable snapshot published at chain-build time and + // shared read-only across concurrent requests; policies must not mutate it. + action := headerPol.OnResponseHeaders(ctx, respCtx, spec.Parameters.Raw) executionTime := time.Since(policyStartTime) // Apply header mutations to respCtx so subsequent policies and CEL conditions see the mutated state @@ -629,16 +614,10 @@ func (c *ChainExecutor) ExecuteResponsePolicies(ctx context.Context, policyList } } - // Deep-copy params to prevent a policy from mutating the shared spec map - // across concurrent requests (nested maps/slices require a full deep copy). - params, err := deepCopyParams(spec.Parameters.Raw) - if err != nil { - span.End() - return nil, fmt.Errorf("failed to clone parameters for policy %s:%s: %w", spec.Name, spec.Version, err) - } - + // spec.Parameters.Raw is an immutable snapshot published at chain-build time and + // shared read-only across concurrent requests; policies must not mutate it. slog.Debug("[body] calling OnResponseBody", "policy", spec.Name, "version", spec.Version, "route", route) - action := rp.OnResponseBody(ctx, respCtx, params) + action := rp.OnResponseBody(ctx, respCtx, spec.Parameters.Raw) executionTime := time.Since(policyStartTime) // Record policy execution metrics @@ -788,14 +767,10 @@ func (c *ChainExecutor) ExecuteStreamingRequestPolicies( } } - params, err := deepCopyParams(spec.Parameters.Raw) - if err != nil { - span.End() - return nil, fmt.Errorf("failed to clone parameters for policy %s:%s: %w", spec.Name, spec.Version, err) - } - + // spec.Parameters.Raw is an immutable snapshot published at chain-build time and + // shared read-only across concurrent requests; policies must not mutate it. slog.Debug("[streaming] calling OnRequestBodyChunk", "policy", spec.Name, "version", spec.Version, "route", route, "end_of_stream", currentChunk.EndOfStream) - action := streamingPol.OnRequestBodyChunk(ctx, reqCtx, currentChunk, params) + action := streamingPol.OnRequestBodyChunk(ctx, reqCtx, currentChunk, spec.Parameters.Raw) executionTime := time.Since(policyStartTime) metrics.PolicyExecutionsTotal.WithLabelValues(spec.Name, spec.Version, api, route, "executed").Inc() @@ -936,14 +911,10 @@ func (c *ChainExecutor) ExecuteStreamingResponsePolicies( } } - params, err := deepCopyParams(spec.Parameters.Raw) - if err != nil { - span.End() - return nil, fmt.Errorf("failed to clone parameters for policy %s:%s: %w", spec.Name, spec.Version, err) - } - + // spec.Parameters.Raw is an immutable snapshot published at chain-build time and + // shared read-only across concurrent requests; policies must not mutate it. slog.Debug("[streaming] calling OnResponseBodyChunk", "policy", spec.Name, "version", spec.Version, "route", route, "end_of_stream", currentChunk.EndOfStream) - action := streamingPol.OnResponseBodyChunk(ctx, respCtx, currentChunk, params) + action := streamingPol.OnResponseBodyChunk(ctx, respCtx, currentChunk, spec.Parameters.Raw) executionTime := time.Since(policyStartTime) metrics.PolicyExecutionsTotal.WithLabelValues(spec.Name, spec.Version, api, route, "executed").Inc() @@ -1066,22 +1037,6 @@ func applyResponseModifications(ctx *policy.ResponseContext, mods *policy.Downst } } -// deepCopyParams returns a deep copy of a map[string]interface{} via a JSON round-trip. -func deepCopyParams(src map[string]interface{}) (map[string]interface{}, error) { - if len(src) == 0 { - return make(map[string]interface{}), nil - } - b, err := json.Marshal(src) - if err != nil { - return nil, err - } - var dst map[string]interface{} - if err := json.Unmarshal(b, &dst); err != nil { - return nil, err - } - return dst, nil -} - // ─── ChainExecutor ──────────────────────────────────────────────────────────── // ChainExecutor represents the policy chain execution engine diff --git a/gateway/gateway-runtime/policy-engine/internal/executor/chain_test.go b/gateway/gateway-runtime/policy-engine/internal/executor/chain_test.go index 59a9fcb2dc..a6408a78c9 100644 --- a/gateway/gateway-runtime/policy-engine/internal/executor/chain_test.go +++ b/gateway/gateway-runtime/policy-engine/internal/executor/chain_test.go @@ -744,62 +744,6 @@ func TestResponseExecutionResult(t *testing.T) { assert.Nil(t, result.FinalAction) } -// ============================================================================= -// Tests for deepCopyParams -// ============================================================================= - -// TestDeepCopyParams_MutationIsolation verifies that a policy mutating params -// during execution does not affect the original spec map or a concurrent request. -func TestDeepCopyParams_MutationIsolation(t *testing.T) { - original := map[string]interface{}{ - "level1": map[string]interface{}{ - "nested": "original", - }, - } - - copy1, err := deepCopyParams(original) - require.NoError(t, err) - - // Simulate a policy mutating a nested value - copy1["level1"].(map[string]interface{})["nested"] = "mutated" - - // A second copy (next request) should still see the original value - copy2, err := deepCopyParams(original) - require.NoError(t, err) - - assert.Equal(t, "original", copy2["level1"].(map[string]interface{})["nested"], - "mutation of copy1 must not affect the source map used by copy2") - assert.Equal(t, "original", original["level1"].(map[string]interface{})["nested"], - "mutation of copy1 must not affect the original source map") -} - -// TestDeepCopyParams_EmptyMap verifies that an empty map returns an empty copy. -func TestDeepCopyParams_EmptyMap(t *testing.T) { - result, err := deepCopyParams(map[string]interface{}{}) - require.NoError(t, err) - assert.NotNil(t, result) - assert.Empty(t, result) -} - -// TestDeepCopyParams_SliceIsolation verifies that slice values are also deep-copied. -func TestDeepCopyParams_SliceIsolation(t *testing.T) { - original := map[string]interface{}{ - "items": []interface{}{"a", "b", "c"}, - } - - copy1, err := deepCopyParams(original) - require.NoError(t, err) - - // Mutate the slice in copy1 - copy1["items"] = []interface{}{"x", "y"} - - copy2, err := deepCopyParams(original) - require.NoError(t, err) - - assert.Equal(t, []interface{}{"a", "b", "c"}, copy2["items"], - "slice mutation in copy1 must not affect subsequent copies") -} - // TestExecuteResponsePolicies_ImmediateResponseShortCircuit verifies that a // policy returning ImmediateResponse from OnResponse short-circuits the chain. func TestExecuteResponsePolicies_ImmediateResponseShortCircuit(t *testing.T) {