diff --git a/pkg/kgateway/proxy_syncer/effective_endpoints.go b/pkg/kgateway/proxy_syncer/effective_endpoints.go index 2270fc327cc..14aeb76dea4 100644 --- a/pkg/kgateway/proxy_syncer/effective_endpoints.go +++ b/pkg/kgateway/proxy_syncer/effective_endpoints.go @@ -31,11 +31,10 @@ func newFinalBackendEndpoints( final.AttachedPolicies = backend.AttachedPolicies final.ClusterName = backend.ClusterName() final.UpstreamResourceName = backend.ResourceName() - for locality, endpoints := range raw.LbEps { - for _, endpoint := range endpoints { - final.Add(locality, endpoint) - } - } + // Reuse the endpoint protos AND their precomputed equality hash instead + // of re-Adding every endpoint: Add re-marshals each LbEndpoint proto + // (HashProtoWithHasher), which is a major allocation source at scale. + final.ReuseEndpointsFrom(raw) // A same-named EDS cluster can still re-warm when policy changes CDS. // Bump only the endpoint version so Envoy receives a fresh CLA response. if policyHash := backendEndpointVersionHash(backend); policyHash != 0 { diff --git a/pkg/kgateway/proxy_syncer/gateway_backend_variants.go b/pkg/kgateway/proxy_syncer/gateway_backend_variants.go index aaf7e264af5..c3ba6753f4e 100644 --- a/pkg/kgateway/proxy_syncer/gateway_backend_variants.go +++ b/pkg/kgateway/proxy_syncer/gateway_backend_variants.go @@ -93,11 +93,10 @@ func newGatewayBackendVariantEndpoints( } clone := ir.NewEndpointsForBackend(*variant.backend) - for locality, endpoints := range base.LbEps { - for _, endpoint := range endpoints { - clone.Add(locality, endpoint) - } - } + // The endpoint protos are shared with the base backend; reuse the + // precomputed equality hash as well to avoid re-marshaling every + // LbEndpoint proto (HashProtoWithHasher) on each recompute. + clone.ReuseEndpointsFrom(base) return clone }, krtopts.ToOptions("GatewayBackendClientCertificateVariantEndpoints")...) diff --git a/pkg/pluginsdk/ir/model.go b/pkg/pluginsdk/ir/model.go index 0a85b0a1d05..5ac9fb41a27 100644 --- a/pkg/pluginsdk/ir/model.go +++ b/pkg/pluginsdk/ir/model.go @@ -6,6 +6,7 @@ import ( "fmt" "hash/fnv" "maps" + "slices" "strings" envoyendpointv3 "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3" @@ -269,6 +270,32 @@ func (e *EndpointsForBackend) Add(l PodLocality, emd EndpointWithMd) { e.LbEps[l] = append(e.LbEps[l], emd) } +// ReuseEndpointsFrom copies the endpoint entries and their precomputed +// equality hash from base, avoiding a per-endpoint proto re-hash (which +// marshals every LbEndpoint). The equality hashes are identical to what a full +// re-Add of every endpoint would produce: epsEqualityHash only depends on the +// endpoint set (shared with base), and the final hash mixes in this copy's own +// upstreamHash for backend identity. +// The per-locality slices are cloned to avoid append aliasing with base. +func (e *EndpointsForBackend) ReuseEndpointsFrom(base *EndpointsForBackend) { + e.epsEqualityHash = base.epsEqualityHash + e.LbEpsEqualityHash = e.upstreamHash + // Mirror what Add() computes. Distinguish the empty set by endpoint count, + // not by hash value: a non-empty set can xor to a zero epsEqualityHash + // (e.g. duplicate endpoints across localities), and Add() would still + // produce hash(0, upstreamHash) in that case. + nonEmpty := false + for l, eps := range base.LbEps { + if len(eps) > 0 { + nonEmpty = true + } + e.LbEps[l] = slices.Clone(eps) + } + if nonEmpty { + e.LbEpsEqualityHash = hash(e.epsEqualityHash, e.upstreamHash) + } +} + func (c EndpointsForBackend) ResourceName() string { return c.UpstreamResourceName } diff --git a/pkg/pluginsdk/ir/model_endpoints_test.go b/pkg/pluginsdk/ir/model_endpoints_test.go new file mode 100644 index 00000000000..1dae61cf120 --- /dev/null +++ b/pkg/pluginsdk/ir/model_endpoints_test.go @@ -0,0 +1,89 @@ +package ir + +import ( + "fmt" + "testing" +) + +func epsBackend(name string) BackendObjectIR { + return NewBackendObjectIR(ObjectSource{ + Group: "", + Kind: "Service", + Namespace: "default", + Name: name, + }, 80, "", "") +} + +func TestReuseEndpointsFromMatchesReAdd(t *testing.T) { + base := NewEndpointsForBackend(epsBackend("svc")) + for i := 0; i < 50; i++ { + loc := PodLocality{Region: "us-east-1", Zone: fmt.Sprintf("zone-%d", i%3)} + base.Add(loc, EndpointWithMd{ + EndpointMd: EndpointMetadata{Labels: map[string]string{"i": fmt.Sprint(i)}}, + }) + } + + // reference: rebuild by re-Adding every endpoint + readded := NewEndpointsForBackend(epsBackend("svc-variant")) + for loc, eps := range base.LbEps { + for _, ep := range eps { + readded.Add(loc, ep) + } + } + + clone := NewEndpointsForBackend(epsBackend("svc-variant")) + clone.ReuseEndpointsFrom(base) + + if clone.LbEpsEqualityHash != readded.LbEpsEqualityHash { + t.Fatalf("LbEpsEqualityHash mismatch: reuse=%d readd=%d", clone.LbEpsEqualityHash, readded.LbEpsEqualityHash) + } + if clone.LbEpsEqualityHash == base.LbEpsEqualityHash { + t.Fatal("variant hash should differ from base due to different upstream identity") + } + if len(clone.LbEps) != len(base.LbEps) { + t.Fatalf("locality count mismatch: got %d want %d", len(clone.LbEps), len(base.LbEps)) + } + for loc, eps := range base.LbEps { + got := clone.LbEps[loc] + if len(got) != len(eps) { + t.Fatalf("endpoint count mismatch in %v: got %d want %d", loc, len(got), len(eps)) + } + // verify slice was cloned (no aliasing of backing arrays) + if len(got) > 0 && &got[0] == &eps[0] { + t.Fatalf("endpoint slice for %v aliases base backing array", loc) + } + } +} + +func TestReuseEndpointsFromNonEmptyZeroHash(t *testing.T) { + // two identical (locality, endpoint) entries xor to a zero epsEqualityHash; + // a re-Add still produces hash(0, upstreamHash), so reuse must too. + base := NewEndpointsForBackend(epsBackend("svc")) + loc := PodLocality{Region: "us-east-1", Zone: "zone-a"} + emd := EndpointWithMd{EndpointMd: EndpointMetadata{Labels: map[string]string{"i": "1"}}} + base.Add(loc, emd) + base.Add(loc, emd) + if base.epsEqualityHash != 0 { + t.Fatalf("expected xor of identical endpoints to be 0, got %d", base.epsEqualityHash) + } + + readded := NewEndpointsForBackend(epsBackend("svc-variant")) + readded.Add(loc, emd) + readded.Add(loc, emd) + + clone := NewEndpointsForBackend(epsBackend("svc-variant")) + clone.ReuseEndpointsFrom(base) + if clone.LbEpsEqualityHash != readded.LbEpsEqualityHash { + t.Fatalf("zero-hash non-empty reuse mismatch: reuse=%d readd=%d", clone.LbEpsEqualityHash, readded.LbEpsEqualityHash) + } +} + +func TestReuseEndpointsFromEmpty(t *testing.T) { + base := NewEndpointsForBackend(epsBackend("svc")) + clone := NewEndpointsForBackend(epsBackend("svc-variant")) + clone.ReuseEndpointsFrom(base) + readded := NewEndpointsForBackend(epsBackend("svc-variant")) + if clone.LbEpsEqualityHash != readded.LbEpsEqualityHash { + t.Fatalf("empty reuse hash mismatch: reuse=%d readd=%d", clone.LbEpsEqualityHash, readded.LbEpsEqualityHash) + } +}