From edb2e453dcf54b60d611d3d0bd88ad977830411a Mon Sep 17 00:00:00 2001 From: Renuka Fernando Date: Fri, 17 Jul 2026 21:53:33 +0530 Subject: [PATCH 1/2] feat(policy-engine): reuse unchanged policy chains on xDS update xDS is State-of-the-World, so every artifact change pushes the full snapshot and HandlePolicyChainUpdate rebuilt every route's chain, re-running each policy's GetPolicy factory. Redeploying one API thus re-instantiated every other API's policies, churning pools/state. Reconcile each snapshot against the last-applied set: - Hash each route's behavioral config (canonical JSON, order-sensitive, excluding volatile Metadata fields) and reuse the existing chain pointer when unchanged; rebuild only changed/new routes; drop removed. - Fail-safe to rebuild on signature errors; add per-route decision debug logs and per-snapshot reused/rebuilt/new/removed counts. - Add signature + reconciliation tests incl. a completeness guard. --- .../internal/xdsclient/handler.go | 97 ++++- .../internal/xdsclient/reconcile_test.go | 339 ++++++++++++++++++ .../internal/xdsclient/signature.go | 99 +++++ 3 files changed, 530 insertions(+), 5 deletions(-) create mode 100644 gateway/gateway-runtime/policy-engine/internal/xdsclient/reconcile_test.go create mode 100644 gateway/gateway-runtime/policy-engine/internal/xdsclient/signature.go diff --git a/gateway/gateway-runtime/policy-engine/internal/xdsclient/handler.go b/gateway/gateway-runtime/policy-engine/internal/xdsclient/handler.go index f38cb0167..3c7fcb3d2 100644 --- a/gateway/gateway-runtime/policy-engine/internal/xdsclient/handler.go +++ b/gateway/gateway-runtime/policy-engine/internal/xdsclient/handler.go @@ -53,6 +53,13 @@ type StoredPolicyConfig struct { TransportMetadata *TransportMetadata `json:"transport_metadata,omitempty"` } +// appliedRoute records the signature and built chain last applied for a route, +// so a subsequent snapshot can reuse the chain when the config is unchanged. +type appliedRoute struct { + signature string + chain *registry.PolicyChain +} + // ResourceHandler handles xDS resource updates type ResourceHandler struct { kernel *kernel.Kernel @@ -62,6 +69,13 @@ type ResourceHandler struct { lazyResourceHandler *LazyResourceHandler subscriptionStore *policyenginev1.SubscriptionStore subscriptionHandler *SubscriptionStateHandler + + // lastApplied maps routeKey -> the signature and chain currently applied. + // Used by HandlePolicyChainUpdate to reuse unchanged chains instead of + // re-invoking each policy's GetPolicy factory on every SotW snapshot. + // Safe without a lock: HandlePolicyChainUpdate runs serially on the single + // ADS recv goroutine. + lastApplied map[string]appliedRoute } // NewResourceHandler creates a new ResourceHandler @@ -77,6 +91,7 @@ func NewResourceHandler(k *kernel.Kernel, reg *registry.PolicyRegistry) *Resourc lazyResourceHandler: NewLazyResourceHandler(lazyResourceStore, slog.Default()), subscriptionStore: subStore, subscriptionHandler: NewSubscriptionStateHandler(subStore, slog.Default()), + lastApplied: make(map[string]appliedRoute), } } @@ -163,17 +178,74 @@ func (h *ResourceHandler) HandlePolicyChainUpdate(ctx context.Context, resources } } - // Build all policy chains (can fail if policy not found or validation fails) + // Reconcile against the last-applied set: reuse the existing chain when a + // route's behavioral config is unchanged (skipping every GetPolicy factory + // call for that route), rebuild only changed/new routes, and drop routes + // absent from this snapshot. This keeps a redeploy of one API from + // re-instantiating every other API's policies (xDS is State-of-the-World, so + // every snapshot carries all routes). chains := make(map[string]*registry.PolicyChain) + nextApplied := make(map[string]appliedRoute, len(configsWithMetadata)) + var reused, rebuilt, added int + for _, cwm := range configsWithMetadata { - chain, err := h.buildPolicyChain(cwm.config.RouteKey, cwm.config, cwm.metadata) + routeKey := cwm.config.RouteKey + + // Fail-safe: if the signature can't be computed, treat the route as + // changed and rebuild — never risk serving a stale chain. + sig, sigErr := routeSignature(cwm.config, cwm.metadata) + + if sigErr == nil { + if prev, ok := h.lastApplied[routeKey]; ok && prev.signature == sig { + chains[routeKey] = prev.chain + nextApplied[routeKey] = prev + reused++ + slog.DebugContext(ctx, "Policy chain reconcile decision", + "route", routeKey, "decision", "REUSED", + "policyCount", len(prev.chain.Policies), "sig", shortSig(sig)) + continue + } + } else { + slog.WarnContext(ctx, "Failed to compute route signature, forcing rebuild", + "route", routeKey, "error", sigErr) + } + + _, existed := h.lastApplied[routeKey] + + chain, err := h.buildPolicyChain(routeKey, cwm.config, cwm.metadata) if err != nil { slog.ErrorContext(ctx, "Failed to build policy chain for route, skipping", - "route", cwm.config.RouteKey, + "route", routeKey, "error", err) continue // Skip this route but process others } - chains[cwm.config.RouteKey] = chain + chains[routeKey] = chain + // Only remember the signature when we could compute it; a failed + // signature stays absent so the route rebuilds again next snapshot. + if sigErr == nil { + nextApplied[routeKey] = appliedRoute{signature: sig, chain: chain} + } + + decision := "NEW" + if existed { + decision = "REBUILT" + rebuilt++ + } else { + added++ + } + slog.DebugContext(ctx, "Policy chain reconcile decision", + "route", routeKey, "decision", decision, + "policyCount", len(chain.Policies), "sig", shortSig(sig)) + } + + // Report routes that dropped out of the snapshot (removed APIs/operations). + removed := 0 + for routeKey := range h.lastApplied { + if _, present := chains[routeKey]; !present { + removed++ + slog.DebugContext(ctx, "Policy chain reconcile decision", + "route", routeKey, "decision", "REMOVED") + } } // Log the full set of routes being applied so we can detect missing routes @@ -189,6 +261,10 @@ func (h *ResourceHandler) HandlePolicyChainUpdate(ctx context.Context, resources // observe new policy chains with stale sensitive values (which would bypass redaction). h.kernel.ApplyWholeRoutesAndSensitiveValues(chains, allSensitiveValues) + // Commit the reconciliation state so the next snapshot diffs against exactly + // what was just applied. + h.lastApplied = nextApplied + // Record metrics for policy chains loaded metrics.PolicyChainsLoaded.WithLabelValues("ads").Set(float64(len(chains))) @@ -208,7 +284,11 @@ func (h *ResourceHandler) HandlePolicyChainUpdate(ctx context.Context, resources slog.InfoContext(ctx, "Policy chain update completed successfully", "version", version, - "total_routes", len(chains)) + "total_routes", len(chains), + "reused", reused, + "rebuilt", rebuilt, + "new", added, + "removed", removed) return nil } @@ -384,6 +464,13 @@ func (h *ResourceHandler) buildPolicyChain(routeKey string, config *policyengine hasResponseBodyPolicy := false for _, policyConfig := range config.Policies { + // WARNING: routeSignature (signature.go) must capture every input this + // function feeds into the policy instance. New keys read from + // policyConfig.Parameters and new PolicyInstance fields are covered + // automatically (Parameters and PolicyInstance are hashed wholesale), but + // a NEW apiMetadata field added below is NOT — routeSignatureView projects + // Metadata explicitly. If you consume another apiMetadata.* field here, add + // it to routeSignatureView or reconciliation will reuse stale chains. metadata := policy.PolicyMetadata{ RouteName: routeKey, APIId: apiMetadata.APIId, diff --git a/gateway/gateway-runtime/policy-engine/internal/xdsclient/reconcile_test.go b/gateway/gateway-runtime/policy-engine/internal/xdsclient/reconcile_test.go new file mode 100644 index 000000000..a823888c4 --- /dev/null +++ b/gateway/gateway-runtime/policy-engine/internal/xdsclient/reconcile_test.go @@ -0,0 +1,339 @@ +/* + * Copyright (c) 2026, WSO2 LLC. (https://www.wso2.com). + * + * WSO2 LLC. licenses this file to you 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 xdsclient + +import ( + "context" + "reflect" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/anypb" + "google.golang.org/protobuf/types/known/structpb" + + "github.com/wso2/api-platform/gateway/gateway-runtime/policy-engine/internal/kernel" + "github.com/wso2/api-platform/gateway/gateway-runtime/policy-engine/internal/metrics" + "github.com/wso2/api-platform/gateway/gateway-runtime/policy-engine/internal/registry" + policy "github.com/wso2/api-platform/sdk/core/policy/v1alpha2" + policyenginev1 "github.com/wso2/api-platform/sdk/core/policyengine" +) + +// ============================================================================= +// Test helpers +// ============================================================================= + +// skipPolicy is a minimal policy.Policy that declares no processing (SKIP modes) +// so chain-build produces no warnings. +type skipPolicy struct{} + +func (skipPolicy) Mode() policy.ProcessingMode { + return policy.ProcessingMode{ + RequestHeaderMode: policy.HeaderModeSkip, + ResponseHeaderMode: policy.HeaderModeSkip, + RequestBodyMode: policy.BodyModeSkip, + ResponseBodyMode: policy.BodyModeSkip, + } +} + +// regWithCounters builds a registry whose factories count how many times each +// policy's GetPolicy factory is invoked. names are "name:version" keys. +func regWithCounters(t *testing.T, names ...string) (*registry.PolicyRegistry, map[string]*int) { + t.Helper() + reg := ®istry.PolicyRegistry{Policies: make(map[string]*registry.PolicyEntry)} + require.NoError(t, reg.SetConfig(map[string]interface{}{})) + counts := make(map[string]*int) + for _, nv := range names { + c := new(int) + counts[nv] = c + parts := strings.SplitN(nv, ":", 2) + require.Len(t, parts, 2, "policy key must be name:version") + reg.Policies[nv] = ®istry.PolicyEntry{ + Definition: &policy.PolicyDefinition{Name: parts[0], Version: parts[1]}, + Factory: func(_ policy.PolicyMetadata, _ map[string]interface{}) (policy.Policy, error) { + *c++ + return skipPolicy{}, nil + }, + } + } + return reg, counts +} + +// mustResource wraps a StoredPolicyConfig map as the double-wrapped *anypb.Any +// the ADS stream delivers. +func mustResource(t *testing.T, stored map[string]interface{}) *anypb.Any { + t.Helper() + ps, err := structpb.NewStruct(stored) + require.NoError(t, err) + sb, err := proto.Marshal(ps) + require.NoError(t, err) + inner := &anypb.Any{TypeUrl: "type.googleapis.com/google.protobuf.Struct", Value: sb} + ib, err := proto.Marshal(inner) + require.NoError(t, err) + return &anypb.Any{TypeUrl: PolicyChainTypeURL, Value: ib} +} + +// storedConfig builds a StoredPolicyConfig map with the given routes. updatedAt +// feeds updated_at/resource_version — volatile fields that must NOT affect the +// signature. +func storedConfig(id, apiID, apiName, apiVersion string, updatedAt int, routes ...interface{}) map[string]interface{} { + return map[string]interface{}{ + "id": id, + "version": 1, + "configuration": map[string]interface{}{ + "metadata": map[string]interface{}{ + "api_id": apiID, + "api_name": apiName, + "version": apiVersion, + "created_at": 0, + "updated_at": updatedAt, + "resource_version": updatedAt, + "context": "/" + apiName, + }, + "routes": routes, + }, + } +} + +func route(routeKey string, policies ...interface{}) interface{} { + return map[string]interface{}{"route_key": routeKey, "policies": policies} +} + +func pol(name, version string, params map[string]interface{}) interface{} { + if params == nil { + params = map[string]interface{}{} + } + return map[string]interface{}{ + "name": name, + "version": version, + "enabled": true, + "parameters": params, + } +} + +// ============================================================================= +// Reconciliation behaviour (handler level) +// ============================================================================= + +// The original bug: redeploying one API re-ran every other API's GetPolicy. +func TestHandlePolicyChainUpdate_ReusesUnchangedRoutesOnUnrelatedRedeploy(t *testing.T) { + metrics.Init() + reg, counts := regWithCounters(t, "polA:v1", "polB:v1") + k := kernel.NewKernel() + h := NewResourceHandler(k, reg) + ctx := context.Background() + + rA := route("rA", pol("polA", "v1", map[string]interface{}{"x": "1"})) + rB := route("rB", pol("polB", "v1", map[string]interface{}{"y": "1"})) + + require.NoError(t, h.HandlePolicyChainUpdate(ctx, []*anypb.Any{ + mustResource(t, storedConfig("A", "apiA", "A", "v1", 1, rA)), + mustResource(t, storedConfig("B", "apiB", "B", "v1", 1, rB)), + }, "1")) + require.Equal(t, 1, *counts["polA:v1"]) + require.Equal(t, 1, *counts["polB:v1"]) + chainA1 := k.GetPolicyChain("rA") + require.NotNil(t, chainA1) + + // Redeploy B (param changed); A is byte-identical, B's metadata bumped. + rBChanged := route("rB", pol("polB", "v1", map[string]interface{}{"y": "2"})) + require.NoError(t, h.HandlePolicyChainUpdate(ctx, []*anypb.Any{ + mustResource(t, storedConfig("A", "apiA", "A", "v1", 1, rA)), + mustResource(t, storedConfig("B", "apiB", "B", "v1", 2, rBChanged)), + }, "2")) + + assert.Equal(t, 1, *counts["polA:v1"], "A reused: its GetPolicy must NOT re-run on B's redeploy") + assert.Equal(t, 2, *counts["polB:v1"], "B changed: must rebuild") + assert.Same(t, chainA1, k.GetPolicyChain("rA"), "reused chain pointer must be identical") + assert.NotSame(t, chainA1, k.GetPolicyChain("rB")) +} + +// Bumping only volatile metadata (updated_at/resource_version) must not rebuild. +func TestHandlePolicyChainUpdate_ReuseAcrossVolatileMetadataBump(t *testing.T) { + metrics.Init() + reg, counts := regWithCounters(t, "polA:v1") + k := kernel.NewKernel() + h := NewResourceHandler(k, reg) + ctx := context.Background() + + rA := route("rA", pol("polA", "v1", map[string]interface{}{"x": "1"})) + require.NoError(t, h.HandlePolicyChainUpdate(ctx, + []*anypb.Any{mustResource(t, storedConfig("A", "apiA", "A", "v1", 10, rA))}, "1")) + require.Equal(t, 1, *counts["polA:v1"]) + + // Identical behavioural config, only volatile metadata differs. + require.NoError(t, h.HandlePolicyChainUpdate(ctx, + []*anypb.Any{mustResource(t, storedConfig("A", "apiA", "A", "v1", 999, rA))}, "2")) + assert.Equal(t, 1, *counts["polA:v1"], "volatile metadata bump must not trigger a rebuild") +} + +// Reordering the policies within a chain must rebuild (execution order matters). +func TestHandlePolicyChainUpdate_ReorderRebuilds(t *testing.T) { + metrics.Init() + reg, counts := regWithCounters(t, "polA:v1", "polB:v1") + k := kernel.NewKernel() + h := NewResourceHandler(k, reg) + ctx := context.Background() + + pA := pol("polA", "v1", map[string]interface{}{"x": "1"}) + pB := pol("polB", "v1", map[string]interface{}{"y": "1"}) + + require.NoError(t, h.HandlePolicyChainUpdate(ctx, + []*anypb.Any{mustResource(t, storedConfig("A", "apiA", "A", "v1", 1, route("rA", pA, pB)))}, "1")) + require.Equal(t, 1, *counts["polA:v1"]) + require.Equal(t, 1, *counts["polB:v1"]) + + // Same policies, swapped order. + require.NoError(t, h.HandlePolicyChainUpdate(ctx, + []*anypb.Any{mustResource(t, storedConfig("A", "apiA", "A", "v1", 1, route("rA", pB, pA)))}, "2")) + assert.Equal(t, 2, *counts["polA:v1"], "reorder must rebuild the chain") + assert.Equal(t, 2, *counts["polB:v1"]) +} + +// Routes absent from a later snapshot must be dropped from the kernel. +func TestHandlePolicyChainUpdate_RemovesAbsentRoutes(t *testing.T) { + metrics.Init() + reg, _ := regWithCounters(t, "polA:v1") + k := kernel.NewKernel() + h := NewResourceHandler(k, reg) + ctx := context.Background() + + rA := route("rA", pol("polA", "v1", nil)) + require.NoError(t, h.HandlePolicyChainUpdate(ctx, + []*anypb.Any{mustResource(t, storedConfig("A", "apiA", "A", "v1", 1, rA))}, "1")) + require.NotNil(t, k.GetPolicyChain("rA")) + + // Empty snapshot: A is gone. + require.NoError(t, h.HandlePolicyChainUpdate(ctx, []*anypb.Any{}, "2")) + assert.Nil(t, k.GetPolicyChain("rA"), "absent route must be dropped from the kernel") +} + +// ============================================================================= +// routeSignature — field sensitivity +// ============================================================================= + +func baseChainAndMeta() (*policyenginev1.PolicyChain, policyenginev1.Metadata) { + cfg := &policyenginev1.PolicyChain{ + RouteKey: "rA", + Policies: []policyenginev1.PolicyInstance{ + {Name: "polA", Version: "v1", Enabled: true, Parameters: map[string]interface{}{"x": "1"}}, + {Name: "polB", Version: "v1", Enabled: true, Parameters: map[string]interface{}{"y": "1"}}, + }, + } + md := policyenginev1.Metadata{APIId: "apiA", APIName: "A", Version: "v1"} + return cfg, md +} + +func sig(t *testing.T, cfg *policyenginev1.PolicyChain, md policyenginev1.Metadata) string { + t.Helper() + s, err := routeSignature(cfg, md) + require.NoError(t, err) + return s +} + +func TestRouteSignature_StableAndSensitive(t *testing.T) { + cfg, md := baseChainAndMeta() + base := sig(t, cfg, md) + + // Identical inputs -> identical signature. + cfg2, md2 := baseChainAndMeta() + assert.Equal(t, base, sig(t, cfg2, md2), "identical config must yield identical signature") + + strPtr := func(s string) *string { return &s } + + // Each of these must CHANGE the signature. + changed := []struct { + name string + apply func(*policyenginev1.PolicyChain, *policyenginev1.Metadata) + }{ + {"param value", func(c *policyenginev1.PolicyChain, _ *policyenginev1.Metadata) { c.Policies[0].Parameters["x"] = "2" }}, + {"param added", func(c *policyenginev1.PolicyChain, _ *policyenginev1.Metadata) { c.Policies[0].Parameters["z"] = "1" }}, + {"param removed", func(c *policyenginev1.PolicyChain, _ *policyenginev1.Metadata) { delete(c.Policies[0].Parameters, "x") }}, + {"reorder", func(c *policyenginev1.PolicyChain, _ *policyenginev1.Metadata) { + c.Policies[0], c.Policies[1] = c.Policies[1], c.Policies[0] + }}, + {"name", func(c *policyenginev1.PolicyChain, _ *policyenginev1.Metadata) { c.Policies[0].Name = "polC" }}, + {"version", func(c *policyenginev1.PolicyChain, _ *policyenginev1.Metadata) { c.Policies[0].Version = "v2" }}, + {"enabled", func(c *policyenginev1.PolicyChain, _ *policyenginev1.Metadata) { c.Policies[0].Enabled = false }}, + {"executionCondition", func(c *policyenginev1.PolicyChain, _ *policyenginev1.Metadata) { + c.Policies[0].ExecutionCondition = strPtr("x == 1") + }}, + {"routeKey", func(c *policyenginev1.PolicyChain, _ *policyenginev1.Metadata) { c.RouteKey = "rB" }}, + {"apiId", func(_ *policyenginev1.PolicyChain, m *policyenginev1.Metadata) { m.APIId = "apiB" }}, + {"apiName", func(_ *policyenginev1.PolicyChain, m *policyenginev1.Metadata) { m.APIName = "B" }}, + {"apiVersion", func(_ *policyenginev1.PolicyChain, m *policyenginev1.Metadata) { m.Version = "v2" }}, + } + for _, tc := range changed { + c, m := baseChainAndMeta() + tc.apply(c, &m) + assert.NotEqualf(t, base, sig(t, c, m), "%s must change the signature", tc.name) + } + + // Volatile / non-behavioural metadata must NOT change the signature. + unchanged := []struct { + name string + apply func(*policyenginev1.Metadata) + }{ + {"created_at", func(m *policyenginev1.Metadata) { m.CreatedAt = 123 }}, + {"updated_at", func(m *policyenginev1.Metadata) { m.UpdatedAt = 456 }}, + {"resource_version", func(m *policyenginev1.Metadata) { m.ResourceVersion = 789 }}, + {"context", func(m *policyenginev1.Metadata) { m.Context = "/changed" }}, + } + for _, tc := range unchanged { + c, m := baseChainAndMeta() + tc.apply(&m) + assert.Equalf(t, base, sig(t, c, m), "%s must NOT change the signature", tc.name) + } +} + +// Completeness guard: every field of routeSignatureView must influence the +// signature, and every field must have a mutator here. If someone adds a field +// to the view without a mutator, this fails — forcing them to keep coverage. +func TestRouteSignatureView_Completeness(t *testing.T) { + base := routeSignatureView{ + RouteKey: "r", + Policies: []policyenginev1.PolicyInstance{{Name: "p", Version: "v1", Enabled: true, Parameters: map[string]interface{}{"a": "b"}}}, + APIId: "i", + APIName: "n", + APIVersion: "v", + } + + mutators := map[string]func(*routeSignatureView){ + "RouteKey": func(v *routeSignatureView) { v.RouteKey = "r2" }, + "Policies": func(v *routeSignatureView) { v.Policies = []policyenginev1.PolicyInstance{{Name: "other"}} }, + "APIId": func(v *routeSignatureView) { v.APIId = "i2" }, + "APIName": func(v *routeSignatureView) { v.APIName = "n2" }, + "APIVersion": func(v *routeSignatureView) { v.APIVersion = "v2" }, + } + + require.Equal(t, reflect.TypeOf(base).NumField(), len(mutators), + "add a mutator for every routeSignatureView field so completeness stays guaranteed") + + baseSig, err := signatureOf(base) + require.NoError(t, err) + for name, mutate := range mutators { + v := base + mutate(&v) + got, err := signatureOf(v) + require.NoError(t, err) + assert.NotEqualf(t, baseSig, got, "mutating field %s must change the signature", name) + } +} diff --git a/gateway/gateway-runtime/policy-engine/internal/xdsclient/signature.go b/gateway/gateway-runtime/policy-engine/internal/xdsclient/signature.go new file mode 100644 index 000000000..170ba2cd5 --- /dev/null +++ b/gateway/gateway-runtime/policy-engine/internal/xdsclient/signature.go @@ -0,0 +1,99 @@ +/* + * Copyright (c) 2025, WSO2 LLC. (https://www.wso2.com). + * + * WSO2 LLC. licenses this file to you 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 xdsclient + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + + policyenginev1 "github.com/wso2/api-platform/sdk/core/policyengine" +) + +// routeSignatureView is the exact set of inputs buildPolicyChain consumes to +// construct a route's policy chain. Its purpose is change detection for +// per-route reconciliation: two snapshots that produce the SAME view must yield +// byte-identical JSON, and any change that would alter the built chain must +// alter the view. +// +// Deliberately EXCLUDED: the volatile Metadata fields (CreatedAt, UpdatedAt, +// ResourceVersion) which the controller bumps on every push, and Context which +// buildPolicyChain does not read — including any of them would make every route +// look changed on every snapshot and silently defeat the optimization. +// +// KEEP IN SYNC WITH buildPolicyChain (handler.go). New keys inside Parameters and +// new PolicyInstance fields are covered automatically (both are hashed wholesale). +// But Metadata is PROJECTED here, not hashed wholesale — so if buildPolicyChain +// starts consuming another apiMetadata.* field, it MUST be added to this view too, +// or reconciliation will reuse a stale chain. The completeness test does NOT catch +// that omission; it only proves the fields already listed here affect the hash. +// +// If a NEW field is added here, extend TestRouteSignatureView_Completeness with a +// mutator so a change to it is proven to alter the signature. +type routeSignatureView struct { + RouteKey string `json:"route_key"` + // Whole PolicyInstance list, in order. PolicyInstance carries only behavioral + // fields (Name, Version, Enabled, ExecutionCondition, Parameters), so it is + // marshaled wholesale — new behavioral fields are covered automatically. + // Order is significant (execution order) and preserved by JSON arrays. + Policies []policyenginev1.PolicyInstance `json:"policies"` + APIId string `json:"api_id"` + APIName string `json:"api_name"` + APIVersion string `json:"api_version"` +} + +// routeSignature returns a stable content hash of the behavioral configuration +// that produces a route's policy chain. Identical config always yields the same +// signature; any change to a build input yields a different one. +// +// Canonicalization: encoding/json emits struct fields in declaration order and +// sorts map keys, so the output is deterministic for a given value. Parameters +// arrive as JSON-native types (float64/string/…) from the original unmarshal, so +// re-marshaling here round-trips stably. ${config} references are not resolved +// here on purpose: resolution is process-constant (from baked-in SystemParameters +// against PE-local config) and changes only on a restart, which rebuilds every +// chain from scratch anyway. +func routeSignature(config *policyenginev1.PolicyChain, md policyenginev1.Metadata) (string, error) { + return signatureOf(routeSignatureView{ + RouteKey: config.RouteKey, + Policies: config.Policies, + APIId: md.APIId, + APIName: md.APIName, + APIVersion: md.Version, + }) +} + +// signatureOf hashes a projection view. Kept separate so tests can exercise +// field sensitivity directly. +func signatureOf(view routeSignatureView) (string, error) { + b, err := json.Marshal(view) + if err != nil { + return "", err + } + sum := sha256.Sum256(b) + return hex.EncodeToString(sum[:]), nil +} + +// shortSig returns the first 8 hex chars of a signature for compact logging. +func shortSig(sig string) string { + if len(sig) <= 8 { + return sig + } + return sig[:8] +} From 066695c7a2490adacbe77d31189bf775eec5f8f8 Mon Sep 17 00:00:00 2001 From: Renuka Fernando Date: Fri, 17 Jul 2026 22:13:50 +0530 Subject: [PATCH 2/2] fix(policy-engine): don't report failed routes as REMOVED A route present in the snapshot but dropped due to a validation or rebuild failure was logged and counted as REMOVED, because the check tested membership in the successfully-built chains. Track the route keys seen in the snapshot separately and base removal detection on that set, so present-but-rejected routes (already reported by their own warning/error log) are not double-reported. Add a failed-rebuild test asserting the dropped route is not logged or counted as REMOVED. Addresses CodeRabbit review on #2741. --- .../internal/xdsclient/handler.go | 12 ++- .../internal/xdsclient/reconcile_test.go | 100 ++++++++++++++++++ 2 files changed, 111 insertions(+), 1 deletion(-) diff --git a/gateway/gateway-runtime/policy-engine/internal/xdsclient/handler.go b/gateway/gateway-runtime/policy-engine/internal/xdsclient/handler.go index 3c7fcb3d2..478e183b2 100644 --- a/gateway/gateway-runtime/policy-engine/internal/xdsclient/handler.go +++ b/gateway/gateway-runtime/policy-engine/internal/xdsclient/handler.go @@ -112,6 +112,13 @@ func (h *ResourceHandler) HandlePolicyChainUpdate(ctx context.Context, resources configsWithMetadata := make([]policyChainWithMetadata, 0) var allSensitiveValues []string + // snapshotRouteKeys records every route key that appeared in this snapshot, + // including routes that fail validation or rebuild below. Removal detection + // keys off this set (not the successfully-built chains) so a route that is + // present-but-rejected is not misreported as REMOVED — its own warning/error + // log already explains why it was dropped. + snapshotRouteKeys := make(map[string]bool) + for i, resource := range resources { if resource.TypeUrl != PolicyChainTypeURL { slog.WarnContext(ctx, "Skipping resource with unexpected type", @@ -165,6 +172,7 @@ func (h *ResourceHandler) HandlePolicyChainUpdate(ctx context.Context, resources // Validate each route configuration and pair with metadata // Note: We log errors but continue to avoid NACK loops when policies are missing for _, config := range routeConfigs { + snapshotRouteKeys[config.RouteKey] = true if err := h.validatePolicyChainConfig(config); err != nil { slog.WarnContext(ctx, "Skipping invalid route configuration", "route", config.RouteKey, @@ -239,9 +247,11 @@ func (h *ResourceHandler) HandlePolicyChainUpdate(ctx context.Context, resources } // Report routes that dropped out of the snapshot (removed APIs/operations). + // A route absent from chains but still present in the snapshot failed + // validation/rebuild — it is logged elsewhere, not counted as REMOVED. removed := 0 for routeKey := range h.lastApplied { - if _, present := chains[routeKey]; !present { + if !snapshotRouteKeys[routeKey] { removed++ slog.DebugContext(ctx, "Policy chain reconcile decision", "route", routeKey, "decision", "REMOVED") diff --git a/gateway/gateway-runtime/policy-engine/internal/xdsclient/reconcile_test.go b/gateway/gateway-runtime/policy-engine/internal/xdsclient/reconcile_test.go index a823888c4..0e925de77 100644 --- a/gateway/gateway-runtime/policy-engine/internal/xdsclient/reconcile_test.go +++ b/gateway/gateway-runtime/policy-engine/internal/xdsclient/reconcile_test.go @@ -20,8 +20,11 @@ package xdsclient import ( "context" + "fmt" + "log/slog" "reflect" "strings" + "sync" "testing" "github.com/stretchr/testify/assert" @@ -337,3 +340,100 @@ func TestRouteSignatureView_Completeness(t *testing.T) { assert.NotEqualf(t, baseSig, got, "mutating field %s must change the signature", name) } } + +// ============================================================================= +// Failed rebuild must not be reported as REMOVED +// ============================================================================= + +// captureHandler collects slog records so tests can assert on decision logs. +type captureHandler struct { + mu sync.Mutex + records []slog.Record +} + +func (h *captureHandler) Enabled(context.Context, slog.Level) bool { return true } +func (h *captureHandler) Handle(_ context.Context, r slog.Record) error { + h.mu.Lock() + defer h.mu.Unlock() + h.records = append(h.records, r.Clone()) + return nil +} +func (h *captureHandler) WithAttrs([]slog.Attr) slog.Handler { return h } +func (h *captureHandler) WithGroup(string) slog.Handler { return h } + +func (h *captureHandler) attr(r slog.Record, key string) (any, bool) { + var val any + var found bool + r.Attrs(func(a slog.Attr) bool { + if a.Key == key { + val, found = a.Value.Any(), true + return false + } + return true + }) + return val, found +} + +// A route present in the snapshot whose rebuild fails must be dropped but must +// NOT be logged or counted as REMOVED (its build error is reported separately). +func TestHandlePolicyChainUpdate_FailedRebuildNotReportedRemoved(t *testing.T) { + metrics.Init() + reg := ®istry.PolicyRegistry{Policies: make(map[string]*registry.PolicyEntry)} + require.NoError(t, reg.SetConfig(map[string]interface{}{})) + reg.Policies["polOK:v1"] = ®istry.PolicyEntry{ + Definition: &policy.PolicyDefinition{Name: "polOK", Version: "v1"}, + Factory: func(policy.PolicyMetadata, map[string]interface{}) (policy.Policy, error) { + return skipPolicy{}, nil + }, + } + // polBad passes validation (it is registered) but its factory errors, so + // buildPolicyChain fails — exercising the rebuild-failure path. + reg.Policies["polBad:v1"] = ®istry.PolicyEntry{ + Definition: &policy.PolicyDefinition{Name: "polBad", Version: "v1"}, + Factory: func(policy.PolicyMetadata, map[string]interface{}) (policy.Policy, error) { + return nil, fmt.Errorf("boom") + }, + } + + k := kernel.NewKernel() + h := NewResourceHandler(k, reg) + ctx := context.Background() + + // Snapshot 1: rB applied successfully. + require.NoError(t, h.HandlePolicyChainUpdate(ctx, + []*anypb.Any{mustResource(t, storedConfig("B", "apiB", "B", "v1", 1, + route("rB", pol("polOK", "v1", map[string]interface{}{"x": "1"}))))}, "1")) + require.NotNil(t, k.GetPolicyChain("rB")) + + // Capture logs only for snapshot 2. + capH := &captureHandler{} + old := slog.Default() + slog.SetDefault(slog.New(capH)) + defer slog.SetDefault(old) + + // Snapshot 2: rB still present but now references a policy whose factory errors. + require.NoError(t, h.HandlePolicyChainUpdate(ctx, + []*anypb.Any{mustResource(t, storedConfig("B", "apiB", "B", "v1", 2, + route("rB", pol("polBad", "v1", map[string]interface{}{"x": "1"}))))}, "2")) + + // rB is dropped from the kernel (build failed)... + assert.Nil(t, k.GetPolicyChain("rB")) + + // ...but must NOT be logged as REMOVED, and the summary count must be 0. + sawSummary := false + for _, r := range capH.records { + switch r.Message { + case "Policy chain reconcile decision": + if dec, ok := capH.attr(r, "decision"); ok && dec == "REMOVED" { + rk, _ := capH.attr(r, "route") + t.Fatalf("route %v wrongly logged as REMOVED after a failed rebuild", rk) + } + case "Policy chain update completed successfully": + sawSummary = true + removedVal, ok := capH.attr(r, "removed") + require.True(t, ok, "summary must include a removed count") + assert.EqualValues(t, 0, removedVal, "failed rebuild must not count as removed") + } + } + require.True(t, sawSummary, "expected a completion summary log") +}