From b6e2dcd4081cc0d825d168b3e2f0c23553106681 Mon Sep 17 00:00:00 2001 From: "Huabing (Robin) Zhao" Date: Mon, 14 Sep 2026 00:08:17 -0700 Subject: [PATCH 1/4] perf(xds): add opt-in endpoint fast path for EDS-only pushes Endpoint updates currently reach Envoy only by riding along in a full rebuild, so their freshness is bound to whatever a full translation costs. This adds an opt-in fast path behind the EndpointFastPath runtime flag: the provider publishes EndpointSlice changes on a dedicated channel, and the xDS runner rebuilds only the affected ClusterLoadAssignments from per-cluster context captured during the last successful translation, patching them into the snapshot with only the EDS version bumped. Full translations still run for every endpoint event and remain authoritative; endpoint state the fast path has published is folded into a racing build before it is published, so a build cannot regress it. Changes that are not load-assignment-only, such as an endpoint count crossing zero or a gateway using EnvoyPatchPolicy or endpoint-modifying extension hooks, are left to the full path. Signed-off-by: Huabing (Robin) Zhao Signed-off-by: Huabing (Robin) Zhao --- api/v1alpha1/envoygateway_helpers.go | 1 + api/v1alpha1/envoygateway_types.go | 10 +- internal/gatewayapi/contexts.go | 11 +- internal/gatewayapi/resource/resource.go | 26 +- internal/gatewayapi/route.go | 94 ++--- internal/gatewayapi/runner/runner.go | 1 + internal/gatewayapi/translator.go | 5 + internal/ir/xds.go | 22 ++ internal/ir/zz_generated.deepcopy.go | 20 + internal/message/types.go | 73 ++++ internal/provider/kubernetes/controller.go | 111 ++++++ internal/utils/endpoints/endpoints.go | 101 +++++ internal/xds/cache/snapshotcache.go | 77 +++- internal/xds/cache/snapshotcache_test.go | 85 +++++ internal/xds/runner/endpoint_fast_path.go | 351 ++++++++++++++++++ .../xds/runner/endpoint_fast_path_test.go | 304 +++++++++++++++ internal/xds/runner/runner.go | 38 +- internal/xds/translator/endpoint_fast_path.go | 59 +++ internal/xds/translator/translator.go | 6 + internal/xds/types/endpointcontext.go | 36 ++ internal/xds/types/resourceversiontable.go | 5 + .../10022-endpoint-fast-path.md | 6 + site/content/en/latest/api/extension_types.md | 1 + site/package.json | 8 +- 24 files changed, 1354 insertions(+), 97 deletions(-) create mode 100644 internal/utils/endpoints/endpoints.go create mode 100644 internal/xds/runner/endpoint_fast_path.go create mode 100644 internal/xds/runner/endpoint_fast_path_test.go create mode 100644 internal/xds/translator/endpoint_fast_path.go create mode 100644 internal/xds/types/endpointcontext.go create mode 100644 release-notes/current/performance_improvements/10022-endpoint-fast-path.md diff --git a/api/v1alpha1/envoygateway_helpers.go b/api/v1alpha1/envoygateway_helpers.go index 5a6d8f2a72a..a16cda16134 100644 --- a/api/v1alpha1/envoygateway_helpers.go +++ b/api/v1alpha1/envoygateway_helpers.go @@ -173,6 +173,7 @@ func (e *EnvoyGateway) GetEnvoyProxyDefaultSpec() *EnvoyProxySpec { var defaultRuntimeFlags = map[RuntimeFlag]bool{ XDSNameSchemeV2: false, EndpointSliceIndex: true, + EndpointFastPath: false, } // IsEnabled checks if an experimental Gateway API is enabled in the EnvoyGateway configuration. diff --git a/api/v1alpha1/envoygateway_types.go b/api/v1alpha1/envoygateway_types.go index c24fd3615e0..602ffc15fab 100644 --- a/api/v1alpha1/envoygateway_types.go +++ b/api/v1alpha1/envoygateway_types.go @@ -170,7 +170,7 @@ type GatewayAPISettings struct { // RuntimeFlag defines a runtime flag used to guard breaking changes or risky experimental features in new Envoy Gateway releases. // A runtime flag may be enabled or disabled by default and can be toggled through the EnvoyGateway resource. // +enum -// +kubebuilder:validation:Enum=XDSNameSchemeV2;EndpointSliceIndex;PerResourceSystemCASecret +// +kubebuilder:validation:Enum=XDSNameSchemeV2;EndpointSliceIndex;PerResourceSystemCASecret;EndpointFastPath type RuntimeFlag string const ( @@ -191,6 +191,14 @@ const ( // upgrades — Envoy must warm the new system_ca_certificates secret before clusters can use // it, which may cause a brief disruption to new connections on first enable. PerResourceSystemCASecret RuntimeFlag = "PerResourceSystemCASecret" //nolint:gosec // not a credential + + // EndpointFastPath enables propagating EndpointSlice updates to Envoy as EDS-only + // pushes without waiting for a full translation: endpoint changes are patched into + // the current xDS snapshot using cached per-cluster context from the last successful + // full translation. Endpoint changes that affect configuration (e.g. a backend's + // endpoint count crossing zero) still take the full path, and full translations + // remain authoritative. Disabled by default. + EndpointFastPath RuntimeFlag = "EndpointFastPath" ) // RuntimeFlags provide a mechanism to guard breaking changes or risky experimental features in new Envoy Gateway releases. diff --git a/internal/gatewayapi/contexts.go b/internal/gatewayapi/contexts.go index 0e9330ab6f0..98c882de7d9 100644 --- a/internal/gatewayapi/contexts.go +++ b/internal/gatewayapi/contexts.go @@ -21,6 +21,7 @@ import ( "github.com/envoyproxy/gateway/internal/gatewayapi/status" "github.com/envoyproxy/gateway/internal/ir" "github.com/envoyproxy/gateway/internal/utils" + endpointsutil "github.com/envoyproxy/gateway/internal/utils/endpoints" ) // GatewayContext wraps a Gateway and provides helper methods for @@ -1015,15 +1016,9 @@ func (t *TranslatorContext) GetEndpointSlicesForBackend(svcNamespace, svcName, b func (t *TranslatorContext) SetEndpointSlicesForBackend(slices []*discoveryv1.EndpointSlice) { t.EndpointSliceMap = make(map[backendServiceKey][]*discoveryv1.EndpointSlice) - var kind, svcName string for _, slice := range slices { - if name, ok := slice.Labels[discoveryv1.LabelServiceName]; ok { - kind = resource.KindService - svcName = name - } else if name, ok := slice.Labels[mcsapiv1a1.LabelServiceName]; ok { - kind = resource.KindServiceImport - svcName = name - } else { + kind, svcName, ok := endpointsutil.BackendForSlice(slice) + if !ok { continue } diff --git a/internal/gatewayapi/resource/resource.go b/internal/gatewayapi/resource/resource.go index cbb80a5097d..1f79155ada1 100644 --- a/internal/gatewayapi/resource/resource.go +++ b/internal/gatewayapi/resource/resource.go @@ -183,6 +183,22 @@ func (c *ControllerResourcesContext) StoredAtTime() time.Time { return c.StoredAt } +// SortEndpointSlices sorts EndpointSlices by creation timestamp, then +// namespace/name. It is shared with the provider's endpoint fast path +// publisher, which must derive endpoints in the same order as a full +// translation. +func SortEndpointSlices(endpointSlices []*discoveryv1.EndpointSlice) { + sort.Slice(endpointSlices, func(i, j int) bool { + if endpointSlices[i].CreationTimestamp.Equal(&endpointSlices[j].CreationTimestamp) { + if endpointSlices[i].Namespace != endpointSlices[j].Namespace { + return endpointSlices[i].Namespace < endpointSlices[j].Namespace + } + return endpointSlices[i].Name < endpointSlices[j].Name + } + return endpointSlices[i].CreationTimestamp.Before(&endpointSlices[j].CreationTimestamp) + }) +} + // DeepCopy creates a new ControllerResourcesContext. // The Context field is preserved (not deep copied) since contexts are meant to be passed around. func (c *ControllerResourcesContext) DeepCopy() *ControllerResourcesContext { @@ -376,15 +392,7 @@ func (r *Resources) Sort() { }) // Sort EndpointSlices by creation timestamp, then namespace/name - sort.Slice(r.EndpointSlices, func(i, j int) bool { - if r.EndpointSlices[i].CreationTimestamp.Equal(&r.EndpointSlices[j].CreationTimestamp) { - if r.EndpointSlices[i].Namespace != r.EndpointSlices[j].Namespace { - return r.EndpointSlices[i].Namespace < r.EndpointSlices[j].Namespace - } - return r.EndpointSlices[i].Name < r.EndpointSlices[j].Name - } - return r.EndpointSlices[i].CreationTimestamp.Before(&r.EndpointSlices[j].CreationTimestamp) - }) + SortEndpointSlices(r.EndpointSlices) // Sort Secrets by creation timestamp, then namespace/name sort.Slice(r.Secrets, func(i, j int) bool { diff --git a/internal/gatewayapi/route.go b/internal/gatewayapi/route.go index 855949547d0..a1720a5880f 100644 --- a/internal/gatewayapi/route.go +++ b/internal/gatewayapi/route.go @@ -29,6 +29,7 @@ import ( "github.com/envoyproxy/gateway/internal/gatewayapi/status" "github.com/envoyproxy/gateway/internal/ir" "github.com/envoyproxy/gateway/internal/utils" + endpointsutil "github.com/envoyproxy/gateway/internal/utils/endpoints" labelsutil "github.com/envoyproxy/gateway/internal/utils/labels" "github.com/envoyproxy/gateway/internal/utils/regex" ) @@ -3016,9 +3017,10 @@ func (t *Translator) processServiceImportDestinationSetting( btpRoutingType *egv1a1.RoutingType, ) (*ir.DestinationSetting, status.Error) { var ( - endpoints []*ir.DestinationEndpoint - addrType *ir.DestinationAddressType - protocol = defaultProtocol + endpoints []*ir.DestinationEndpoint + addrType *ir.DestinationAddressType + endpointSource *ir.EndpointSource + protocol = defaultProtocol ) serviceImport := t.GetServiceImport(backendNamespace, string(backendRef.Name)) @@ -3051,6 +3053,15 @@ func (t *Translator) processServiceImportDestinationSetting( status.RouteReasonEndpointsNotFound, ) } + if t.EndpointFastPathEnabled { + endpointSource = &ir.EndpointSource{ + Kind: resource.KindServiceImport, + Namespace: backendNamespace, + Name: string(backendRef.Name), + PortName: servicePort.Name, + Protocol: getServicePortProtocol(servicePort.Protocol), + } + } } else { // Use ServiceImport IPs for routing for _, ip := range backendIps { @@ -3066,6 +3077,7 @@ func (t *Translator) processServiceImportDestinationSetting( Endpoints: endpoints, AddressType: addrType, Metadata: buildResourceMetadata(serviceImport, new(gwapiv1.SectionName(strconv.Itoa(int(*backendRef.Port))))), + EndpointSource: endpointSource, }, nil } @@ -3078,8 +3090,9 @@ func (t *Translator) processServiceDestinationSetting( btpRoutingType *egv1a1.RoutingType, ) (*ir.DestinationSetting, status.Error) { var ( - endpoints []*ir.DestinationEndpoint - addrType *ir.DestinationAddressType + endpoints []*ir.DestinationEndpoint + addrType *ir.DestinationAddressType + endpointSource *ir.EndpointSource ) protocol := defaultProtocol @@ -3123,6 +3136,15 @@ func (t *Translator) processServiceDestinationSetting( status.RouteReasonEndpointsNotFound, ) } + if t.EndpointFastPathEnabled { + endpointSource = &ir.EndpointSource{ + Kind: KindDerefOr(backendRef.Kind, resource.KindService), + Namespace: backendNamespace, + Name: string(backendRef.Name), + PortName: servicePort.Name, + Protocol: getServicePortProtocol(servicePort.Protocol), + } + } } else { // Use Service ClusterIP routing ep := ir.NewDestEndpoint(nil, service.Spec.ClusterIP, uint32(*backendRef.Port), false, nil) @@ -3137,6 +3159,7 @@ func (t *Translator) processServiceDestinationSetting( AddressType: addrType, PreferLocal: processPreferLocalZone(service), Metadata: buildResourceMetadata(service, new(gwapiv1.SectionName(strconv.Itoa(int(*backendRef.Port))))), + EndpointSource: endpointSource, }, nil } @@ -3330,66 +3353,7 @@ func (t *Translator) processAllowedListenersForParentRefs( } func getIREndpointsFromEndpointSlices(endpointSlices []*discoveryv1.EndpointSlice, portName string, portProtocol corev1.Protocol) ([]*ir.DestinationEndpoint, *ir.DestinationAddressType) { - var ( - dstEndpoints []*ir.DestinationEndpoint - dstAddrType *ir.DestinationAddressType - ) - - addrTypeMap := make(map[ir.DestinationAddressType]int) - for _, endpointSlice := range endpointSlices { - if endpointSlice.AddressType == discoveryv1.AddressTypeFQDN { - addrTypeMap[ir.FQDN]++ - } else { - addrTypeMap[ir.IP]++ - } - endpoints := getIREndpointsFromEndpointSlice(endpointSlice, portName, portProtocol) - dstEndpoints = append(dstEndpoints, endpoints...) - } - - for addrTypeState, addrTypeCounts := range addrTypeMap { - if addrTypeCounts == len(endpointSlices) { - dstAddrType = new(addrTypeState) - break - } - } - - if len(addrTypeMap) > 0 && dstAddrType == nil { - dstAddrType = new(ir.MIXED) - } - - return dstEndpoints, dstAddrType -} - -func getIREndpointsFromEndpointSlice(endpointSlice *discoveryv1.EndpointSlice, portName string, portProtocol corev1.Protocol) []*ir.DestinationEndpoint { - var endpoints []*ir.DestinationEndpoint - for _, endpoint := range endpointSlice.Endpoints { - for _, endpointPort := range endpointSlice.Ports { - // Check if the endpoint port matches the service port - if *endpointPort.Name != portName || *endpointPort.Protocol != portProtocol { - continue - } - conditions := endpoint.Conditions - - // Unknown Serving/Terminating (nil) should fall-back to Ready, see https://pkg.go.dev/k8s.io/api/discovery/v1#EndpointConditions - // So drain the endpoint if: - // 1. Both `Terminating` and `Serving` are != null, and either `Terminating=true` or `Serving=false` - // 2. Or `Ready=false` - var draining bool - if conditions.Serving != nil && conditions.Terminating != nil { - draining = *conditions.Terminating || !*conditions.Serving - } else { - draining = conditions.Ready != nil && !*conditions.Ready - } - - for _, address := range endpoint.Addresses { - ep := ir.NewDestEndpoint(nil, address, uint32(*endpointPort.Port), draining, endpoint.Zone) - endpoints = append(endpoints, ep) - } - - } - } - - return endpoints + return endpointsutil.EndpointsFromSlices(endpointSlices, portName, portProtocol) } // isCustomBackendResource checks if the given group and kind match any of the configured custom backend resources diff --git a/internal/gatewayapi/runner/runner.go b/internal/gatewayapi/runner/runner.go index 68e25000e98..28daddb95bb 100644 --- a/internal/gatewayapi/runner/runner.go +++ b/internal/gatewayapi/runner/runner.go @@ -329,6 +329,7 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re MergeGateways: gatewayapi.IsMergeGatewaysEnabled(resources), MergeBackends: gatewayapi.ResolveMergeBackendsConfig(resources), PerResourceSystemCASecret: r.EnvoyGateway.RuntimeFlags.IsEnabled(egv1a1.PerResourceSystemCASecret), + EndpointFastPathEnabled: r.EnvoyGateway.RuntimeFlags.IsEnabled(egv1a1.EndpointFastPath), WasmCache: r.wasmCache, RunningOnHost: r.EnvoyGateway.Provider != nil && r.EnvoyGateway.Provider.IsRunningOnHost(), InfraRemotelyManaged: r.EnvoyGateway.Provider != nil && r.EnvoyGateway.Provider.IsInfraManagedRemotely(), diff --git a/internal/gatewayapi/translator.go b/internal/gatewayapi/translator.go index 99e9c368e2a..03ebb92702e 100644 --- a/internal/gatewayapi/translator.go +++ b/internal/gatewayapi/translator.go @@ -140,6 +140,11 @@ type Translator struct { // sharing a single system_ca_certificates secret. Disabled by default (shared secret used). PerResourceSystemCASecret bool + // EndpointFastPathEnabled records the endpoint source on EndpointSlice-backed + // destination settings so the xDS runner can re-resolve endpoint updates + // without a full translation. + EndpointFastPathEnabled bool + // GatewayNamespaceMode is true if controller uses gateway namespace mode for infra deployments. GatewayNamespaceMode bool diff --git a/internal/ir/xds.go b/internal/ir/xds.go index b7ed7d211e5..f10cfbcf317 100644 --- a/internal/ir/xds.go +++ b/internal/ir/xds.go @@ -2333,6 +2333,28 @@ type DestinationSetting struct { // * invalid 500 // * without endpoints 503 Invalid bool `json:"invalid,omitempty" yaml:"invalid,omitempty"` + // EndpointSource identifies the Kubernetes backend this setting's endpoints are + // derived from, so endpoint updates can be re-resolved without a full translation. + // Only set for EndpointSlice-backed (EDS) settings when the EndpointFastPath + // runtime flag is enabled. + EndpointSource *EndpointSource `json:"endpointSource,omitempty" yaml:"endpointSource,omitempty"` +} + +// EndpointSource identifies the Kubernetes backend an EDS-managed DestinationSetting's +// endpoints are derived from, along with the parameters needed to filter that backend's +// EndpointSlices the same way the full translation does. +// +kubebuilder:object:generate=true +type EndpointSource struct { + // Kind is the backend kind: Service or ServiceImport. + Kind string `json:"kind" yaml:"kind"` + // Namespace of the backend. + Namespace string `json:"namespace" yaml:"namespace"` + // Name of the backend. + Name string `json:"name" yaml:"name"` + // PortName is the service port name used to filter EndpointSlice ports. + PortName string `json:"portName,omitempty" yaml:"portName,omitempty"` + // Protocol is the service port protocol used to filter EndpointSlice ports. + Protocol corev1.Protocol `json:"protocol,omitempty" yaml:"protocol,omitempty"` } // Validate the fields within the DestinationSetting structure diff --git a/internal/ir/zz_generated.deepcopy.go b/internal/ir/zz_generated.deepcopy.go index 1e53d0c1b3d..1e3307ea45a 100644 --- a/internal/ir/zz_generated.deepcopy.go +++ b/internal/ir/zz_generated.deepcopy.go @@ -1608,6 +1608,11 @@ func (in *DestinationSetting) DeepCopyInto(out *DestinationSetting) { *out = new(ResourceMetadata) (*in).DeepCopyInto(*out) } + if in.EndpointSource != nil { + in, out := &in.EndpointSource, &out.EndpointSource + *out = new(EndpointSource) + **out = **in + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DestinationSetting. @@ -1687,6 +1692,21 @@ func (in *EndpointOverrideExtractFrom) DeepCopy() *EndpointOverrideExtractFrom { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *EndpointSource) DeepCopyInto(out *EndpointSource) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new EndpointSource. +func (in *EndpointSource) DeepCopy() *EndpointSource { + if in == nil { + return nil + } + out := new(EndpointSource) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *EnvoyExtensionFeatures) DeepCopyInto(out *EnvoyExtensionFeatures) { *out = *in diff --git a/internal/message/types.go b/internal/message/types.go index 3ad56ad445b..30468596828 100644 --- a/internal/message/types.go +++ b/internal/message/types.go @@ -12,6 +12,7 @@ import ( "github.com/telepresenceio/watchable" "go.opentelemetry.io/otel/trace" + discoveryv1 "k8s.io/api/discovery/v1" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" gwapiv1 "sigs.k8s.io/gateway-api/apis/v1" @@ -27,6 +28,12 @@ type ProviderResources struct { // a group of gateway API and other related resources with trace context. GatewayAPIResources watchable.Map[string, *resource.ControllerResourcesContext] + // EndpointUpdates is a map from a backend key (EndpointUpdate.Key()) to the + // backend's current EndpointSlices, published by the provider for the + // endpoint fast path. Only written when the EndpointFastPath runtime flag + // is enabled. + EndpointUpdates watchable.Map[string, *EndpointUpdate] + // GatewayAPIStatuses is a group of gateway api // resource statuses maps. GatewayAPIStatuses @@ -74,6 +81,10 @@ func (p *ProviderResources) GetResourcesKey() string { func (p *ProviderResources) Close() { p.GatewayAPIResources.Close() + // EndpointUpdates is deliberately not closed: it is written from informer + // event handlers that are not synchronized with this Close, and a Store on + // a closed watchable.Map panics. Its subscribers terminate through their + // own subscription contexts instead. p.GatewayAPIStatuses.Close() p.PolicyStatuses.Close() p.ExtensionStatuses.Close() @@ -275,6 +286,65 @@ func (x *InfraIRWithContext) Equal(other *InfraIRWithContext) bool { return reflect.DeepEqual(x.Infra, other.Infra) } +// EndpointUpdate carries the current set of EndpointSlices for one backend +// (Service or ServiceImport). It is published by the provider whenever the +// backend's endpoints change and consumed by the xDS runner's endpoint fast +// path. +type EndpointUpdate struct { + // Kind is the backend kind: Service or ServiceImport. + Kind string + // Namespace of the backend. + Namespace string + // Name of the backend. + Name string + // EndpointSlices is the full current set of EndpointSlices for the + // backend, so each update is self-contained under coalescing. + EndpointSlices []*discoveryv1.EndpointSlice +} + +// BackendKey returns the EndpointUpdates map key identifying a backend. It is +// the single definition of that format: the provider keys published updates by +// it, the xDS runner indexes endpoint contexts by it, and the reconcile prunes +// retained updates by it — they must agree. +func BackendKey(kind, namespace, name string) string { + return kind + "/" + namespace + "/" + name +} + +// Key returns the watchable map key for this update's backend. +func (e *EndpointUpdate) Key() string { + return BackendKey(e.Kind, e.Namespace, e.Name) +} + +// DeepCopy creates a new EndpointUpdate. +func (e *EndpointUpdate) DeepCopy() *EndpointUpdate { + if e == nil { + return nil + } + out := &EndpointUpdate{ + Kind: e.Kind, + Namespace: e.Namespace, + Name: e.Name, + } + if e.EndpointSlices != nil { + out.EndpointSlices = make([]*discoveryv1.EndpointSlice, len(e.EndpointSlices)) + for i, s := range e.EndpointSlices { + out.EndpointSlices[i] = s.DeepCopy() + } + } + return out +} + +// Equal compares two EndpointUpdates. +func (e *EndpointUpdate) Equal(other *EndpointUpdate) bool { + if e == nil || other == nil { + return e == other + } + return e.Kind == other.Kind && + e.Namespace == other.Namespace && + e.Name == other.Name && + reflect.DeepEqual(e.EndpointSlices, other.EndpointSlices) +} + // InfraIR message type InfraIR struct { watchable.Map[string, *InfraIRWithContext] @@ -285,6 +355,9 @@ type MessageName string const ( // XDSIRMessageName is a message containing xds-ir translated from provider-resources XDSIRMessageName MessageName = "xds-ir" + // EndpointSlicesMessageName is a message containing a backend's EndpointSlices, + // published by the provider for the endpoint fast path + EndpointSlicesMessageName MessageName = "endpointslices" // InfraIRMessageName is a message containing infra-ir translated from provider-resources InfraIRMessageName MessageName = "infra-ir" // ProviderResourcesMessageName is a message containing gw-api and envoy gateway resources from the provider diff --git a/internal/provider/kubernetes/controller.go b/internal/provider/kubernetes/controller.go index e122598885b..6e2f5e5c947 100644 --- a/internal/provider/kubernetes/controller.go +++ b/internal/provider/kubernetes/controller.go @@ -53,6 +53,7 @@ import ( "github.com/envoyproxy/gateway/internal/message" workqueuemetrics "github.com/envoyproxy/gateway/internal/metrics/workqueue" "github.com/envoyproxy/gateway/internal/utils" + endpointsutil "github.com/envoyproxy/gateway/internal/utils/endpoints" ) var skipNameValidation = func() *bool { @@ -301,6 +302,78 @@ func (r *gatewayAPIReconciler) endpointSliceIndexEnabled() bool { return flags.IsEnabled(egv1a1.EndpointSliceIndex) } +// endpointFastPathEnabled returns true when endpoint updates should be published +// on the EndpointUpdates channel for the xDS runner's endpoint fast path. +func (r *gatewayAPIReconciler) endpointFastPathEnabled() bool { + var flags *egv1a1.RuntimeFlags + if r != nil && r.envoyGateway != nil { + flags = r.envoyGateway.RuntimeFlags + } + return flags.IsEnabled(egv1a1.EndpointFastPath) +} + +// publishEndpointUpdate publishes the full current set of EndpointSlices for the +// backend that eps belongs to, so the xDS runner can propagate the change +// without waiting for a full translation. +func (r *gatewayAPIReconciler) publishEndpointUpdate(ctx context.Context, eps *discoveryv1.EndpointSlice) { + if !r.endpointFastPathEnabled() { + return + } + // Informer events can still arrive while the manager is shutting down; + // there is no point publishing an update nothing will consume. + if ctx.Err() != nil { + return + } + + kind, name, ok := endpointsutil.BackendForSlice(eps) + if !ok { + return + } + namespace := eps.Namespace + endpointSliceIndex, endpointSliceLabelKey := serviceEndpointSliceIndex, discoveryv1.LabelServiceName + if kind == resource.KindServiceImport { + endpointSliceIndex, endpointSliceLabelKey = serviceImportEndpointSliceIndex, mcsapiv1a1.LabelServiceName + } + + opts := []client.ListOption{client.InNamespace(namespace)} + if r.endpointSliceIndexEnabled() { + opts = append(opts, client.MatchingFields{endpointSliceIndex: name}) + } else { + opts = append(opts, client.MatchingLabels{endpointSliceLabelKey: name}) + } + + endpointSliceList := new(discoveryv1.EndpointSliceList) + if err := r.client.List(ctx, endpointSliceList, opts...); err != nil { + r.log.Error(err, "failed to list EndpointSlices for endpoint update", + "kind", kind, "namespace", namespace, "name", name) + return + } + + update := &message.EndpointUpdate{ + Kind: kind, + Namespace: namespace, + Name: name, + } + for i := range endpointSliceList.Items { + update.EndpointSlices = append(update.EndpointSlices, &endpointSliceList.Items[i]) + } + // Sort with the same comparator the resource tree uses, so the fast path + // derives endpoints in the same order as a full translation. + resource.SortEndpointSlices(update.EndpointSlices) + if len(update.EndpointSlices) == 0 { + // The backend's last EndpointSlice is gone; drop the retained entry so + // the channel doesn't accumulate state for deleted backends. Consumers + // treat the deletion as a skip and leave it to the full path. + r.resources.EndpointUpdates.Delete(update.Key()) + return + } + r.resources.EndpointUpdates.Store(update.Key(), update) + message.PublishMetric(message.Metadata{ + Runner: string(egv1a1.LogComponentProviderRunner), + Message: message.EndpointSlicesMessageName, + }, 1) +} + func byNamespaceSelectorEnabled(eg *egv1a1.EnvoyGateway) bool { if eg.Provider == nil { return false @@ -343,6 +416,13 @@ func (r *gatewayAPIReconciler) Reconcile(ctx context.Context, _ reconcile.Reques ) logger.Info("reconciling gateways") + // Collect the backends referenced by any route this reconcile, to prune + // retained endpoint updates for backends that are no longer referenced. + var referencedBackends sets.Set[string] + if r.endpointFastPathEnabled() { + referencedBackends = sets.New[string]() + } + // Get the GatewayClasses managed by the Envoy Gateway Controller. managedGCs, err = r.managedGatewayClasses(ctx) if err != nil { @@ -624,6 +704,36 @@ func (r *gatewayAPIReconciler) Reconcile(ctx context.Context, _ reconcile.Reques gcLogger.Error(err, "failed adding finalizer to gatewayClass") } } + + if referencedBackends != nil { + // Collect from the backendRefs themselves, not from the resources + // they resolved to: a reference to a Service that does not exist is + // still a reference, and must not look prunable. + for _, ref := range gwcResourceMapping.allAssociatedBackendRefs { + kind := gatewayapi.KindDerefOr(ref.Kind, resource.KindService) + if kind != resource.KindService && kind != resource.KindServiceImport { + continue + } + referencedBackends.Insert(message.BackendKey(kind, string(*ref.Namespace), string(ref.Name))) + } + } + } + + // Drop retained endpoint updates for backends no longer referenced by any + // route, so the EndpointUpdates channel does not accumulate stale state. + // The filter collects the stale keys and admits nothing, so no retained + // EndpointSlice set is deep copied just to enumerate the keys. + if referencedBackends != nil { + var staleBackends []string + r.resources.EndpointUpdates.LoadAllMatching(func(key string, _ *message.EndpointUpdate) bool { + if !referencedBackends.Has(key) { + staleBackends = append(staleBackends, key) + } + return false + }) + for _, key := range staleBackends { + r.resources.EndpointUpdates.Delete(key) + } } // Sort before storing to: @@ -2659,6 +2769,7 @@ func (r *gatewayAPIReconciler) watchResources(ctx context.Context, mgr manager.M if err := c.Watch( source.Kind(mgr.GetCache(), &discoveryv1.EndpointSlice{}, handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, si *discoveryv1.EndpointSlice) []reconcile.Request { + r.publishEndpointUpdate(ctx, si) return r.enqueueClass(ctx, si) }), esPredicates...)); err != nil { diff --git a/internal/utils/endpoints/endpoints.go b/internal/utils/endpoints/endpoints.go new file mode 100644 index 00000000000..98867919e61 --- /dev/null +++ b/internal/utils/endpoints/endpoints.go @@ -0,0 +1,101 @@ +// Copyright Envoy Gateway Authors +// SPDX-License-Identifier: Apache-2.0 +// The full text of the Apache license is available in the LICENSE file at +// the root of the repo. + +// Package endpoints converts Kubernetes EndpointSlices into IR destination +// endpoints. It is shared by the Gateway API translator (full translation) +// and the xDS runner's endpoint fast path, so both compute endpoints +// identically. +package endpoints + +import ( + corev1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" + mcsapiv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1" + + "github.com/envoyproxy/gateway/internal/gatewayapi/resource" + "github.com/envoyproxy/gateway/internal/ir" +) + +// BackendForSlice returns the backend an EndpointSlice belongs to, derived from +// its service-name labels. A slice carrying both labels belongs to the Service: +// that precedence decides which backend's endpoints a slice contributes to, so +// the full translation and the endpoint fast path must apply it identically. +// ok is false when the slice carries neither label. +func BackendForSlice(slice *discoveryv1.EndpointSlice) (kind, name string, ok bool) { + if name, ok := slice.Labels[discoveryv1.LabelServiceName]; ok { + return resource.KindService, name, true + } + if name, ok := slice.Labels[mcsapiv1a1.LabelServiceName]; ok { + return resource.KindServiceImport, name, true + } + return "", "", false +} + +// EndpointsFromSlices converts the given EndpointSlices into IR destination +// endpoints, keeping only ports matching portName and portProtocol, and +// returns them together with the resulting address type (IP, FQDN, or MIXED +// when the slices disagree). +func EndpointsFromSlices(endpointSlices []*discoveryv1.EndpointSlice, portName string, portProtocol corev1.Protocol) ([]*ir.DestinationEndpoint, *ir.DestinationAddressType) { + var ( + dstEndpoints []*ir.DestinationEndpoint + dstAddrType *ir.DestinationAddressType + ) + + addrTypeMap := make(map[ir.DestinationAddressType]int) + for _, endpointSlice := range endpointSlices { + if endpointSlice.AddressType == discoveryv1.AddressTypeFQDN { + addrTypeMap[ir.FQDN]++ + } else { + addrTypeMap[ir.IP]++ + } + endpoints := endpointsFromSlice(endpointSlice, portName, portProtocol) + dstEndpoints = append(dstEndpoints, endpoints...) + } + + for addrTypeState, addrTypeCounts := range addrTypeMap { + if addrTypeCounts == len(endpointSlices) { + dstAddrType = new(addrTypeState) + break + } + } + + if len(addrTypeMap) > 0 && dstAddrType == nil { + dstAddrType = new(ir.MIXED) + } + + return dstEndpoints, dstAddrType +} + +func endpointsFromSlice(endpointSlice *discoveryv1.EndpointSlice, portName string, portProtocol corev1.Protocol) []*ir.DestinationEndpoint { + var endpoints []*ir.DestinationEndpoint + for _, endpoint := range endpointSlice.Endpoints { + for _, endpointPort := range endpointSlice.Ports { + // Check if the endpoint port matches the service port + if *endpointPort.Name != portName || *endpointPort.Protocol != portProtocol { + continue + } + conditions := endpoint.Conditions + + // Unknown Serving/Terminating (nil) should fall-back to Ready, see https://pkg.go.dev/k8s.io/api/discovery/v1#EndpointConditions + // So drain the endpoint if: + // 1. Both `Terminating` and `Serving` are != null, and either `Terminating=true` or `Serving=false` + // 2. Or `Ready=false` + var draining bool + if conditions.Serving != nil && conditions.Terminating != nil { + draining = *conditions.Terminating || !*conditions.Serving + } else { + draining = conditions.Ready != nil && !*conditions.Ready + } + + for _, address := range endpoint.Addresses { + ep := ir.NewDestEndpoint(nil, address, uint32(*endpointPort.Port), draining, endpoint.Zone) + endpoints = append(endpoints, ep) + } + + } + } + + return endpoints +} diff --git a/internal/xds/cache/snapshotcache.go b/internal/xds/cache/snapshotcache.go index 31d3a31de30..80f3925ec16 100644 --- a/internal/xds/cache/snapshotcache.go +++ b/internal/xds/cache/snapshotcache.go @@ -15,6 +15,7 @@ package cache import ( "context" + "errors" "fmt" "math" "strconv" @@ -23,7 +24,9 @@ import ( corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" discoveryv3 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3" + envoytypes "github.com/envoyproxy/go-control-plane/pkg/cache/types" cachev3 "github.com/envoyproxy/go-control-plane/pkg/cache/v3" + resourcev3 "github.com/envoyproxy/go-control-plane/pkg/resource/v3" serverv3 "github.com/envoyproxy/go-control-plane/pkg/server/v3" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/trace" @@ -51,11 +54,16 @@ var ( type SnapshotCacheWithCallbacks interface { cachev3.SnapshotCache serverv3.Callbacks - GenerateNewSnapshot(string, types.XdsResources, context.Context) error + GenerateNewSnapshot(string, types.XdsResources, context.Context) (bool, error) + UpdateEndpointResources(string, []envoytypes.Resource) error SnapshotHasIrKey(string) bool GetIrKeys() []string } +// ErrNoSnapshot is returned by UpdateEndpointResources when the IR key has no +// snapshot to patch, so callers can tell "nothing to do" from a real failure. +var ErrNoSnapshot = errors.New("no snapshot for the IR key") + type snapshotMap map[string]*cachev3.Snapshot type nodeInfoMap map[int64]*corev3.Node @@ -78,7 +86,12 @@ type snapshotCache struct { // GenerateNewSnapshot takes a table of resources (the output from the IR->xDS // translator) and updates the snapshot version. -func (s *snapshotCache) GenerateNewSnapshot(irKey string, resources types.XdsResources, ctx context.Context) error { +// +// The returned bool reports whether the snapshot was committed. It is false only +// when building it failed, leaving the previous snapshot untouched; it is true +// even when publishing to a node then failed, so callers can keep state in sync +// with the committed snapshot while still surfacing the error. +func (s *snapshotCache) GenerateNewSnapshot(irKey string, resources types.XdsResources, ctx context.Context) (bool, error) { s.mu.Lock() defer s.mu.Unlock() @@ -98,10 +111,11 @@ func (s *snapshotCache) GenerateNewSnapshot(irKey string, resources types.XdsRes ) if err != nil { xdsSnapshotCreateTotal.WithFailure(metrics.ReasonError).Increment() - return err + return false, err } xdsSnapshotCreateTotal.WithSuccess().Increment() + // Commit point. // Delete snapshot from cache if resources are nil if resources == nil { delete(s.lastSnapshot, irKey) @@ -110,15 +124,68 @@ func (s *snapshotCache) GenerateNewSnapshot(irKey string, resources types.XdsRes s.lastSnapshot[irKey] = snapshot } + // The snapshot is committed above, so a per-node publish failure is + // reported with committed=true: the error still propagates to the caller + // (and its error channel), while a node whose publish failed is served the + // committed snapshot when it reconnects. for _, node := range s.getNodeIDs(irKey) { s.log.Debugf("Generating a snapshot with Node %s", node) if err = s.SetSnapshot(context.TODO(), node, snapshot); err != nil { + xdsSnapshotUpdateTotal.WithFailure(metrics.ReasonError, nodeIDLabel.Value(node)).Increment() + return true, err + } + xdsSnapshotUpdateTotal.WithSuccess(nodeIDLabel.Value(node)).Increment() + } + + return true, nil +} + +// UpdateEndpointResources patches only the EDS resources of irKey's snapshot: the +// given ClusterLoadAssignments replace the same-named ones, the EDS version is +// bumped, and every other type keeps its version, so non-EDS resources are not +// resent. Returns ErrNoSnapshot when irKey has no snapshot to patch. +func (s *snapshotCache) UpdateEndpointResources(irKey string, assignments []envoytypes.Resource) error { + s.mu.Lock() + defer s.mu.Unlock() + + oldSnapshot, ok := s.lastSnapshot[irKey] + if !ok || oldSnapshot == nil { + return ErrNoSnapshot + } + + // Merge the updated CLAs over the existing EDS resources. + edsIndex := cachev3.GetResponseType(resourcev3.EndpointType) + oldEDS := oldSnapshot.Resources[edsIndex] + merged := make([]envoytypes.Resource, 0, len(oldEDS.Items)+len(assignments)) + replaced := make(map[string]struct{}, len(assignments)) + for _, cla := range assignments { + merged = append(merged, cla) + replaced[cachev3.GetResourceName(cla)] = struct{}{} + } + for name, res := range oldEDS.Items { + if _, ok := replaced[name]; !ok { + merged = append(merged, res.Resource) + } + } + + // Clone the snapshot with the merged EDS resources: the Resources array is + // copied by value, so only the EDS entry gets a new value. VersionMap is left + // nil; go-control-plane rebuilds it lazily, deriving each resource's version by + // hashing its marshaled bytes, so only the CLAs that changed are resent. + newSnapshot := &cachev3.Snapshot{Resources: oldSnapshot.Resources} + newSnapshot.Resources[edsIndex] = cachev3.NewResources(s.newSnapshotVersion(), merged) + + s.lastSnapshot[irKey] = newSnapshot + + for _, node := range s.getNodeIDs(irKey) { + s.log.Debugf("Updating endpoint resources in snapshot with Node %s", node) + + if err := s.SetSnapshot(context.TODO(), node, newSnapshot); err != nil { xdsSnapshotUpdateTotal.WithFailure(metrics.ReasonError, nodeIDLabel.Value(node)).Increment() return err - } else { - xdsSnapshotUpdateTotal.WithSuccess(nodeIDLabel.Value(node)).Increment() } + xdsSnapshotUpdateTotal.WithSuccess(nodeIDLabel.Value(node)).Increment() } return nil diff --git a/internal/xds/cache/snapshotcache_test.go b/internal/xds/cache/snapshotcache_test.go index 5b460d7e83d..565690e3c59 100644 --- a/internal/xds/cache/snapshotcache_test.go +++ b/internal/xds/cache/snapshotcache_test.go @@ -11,8 +11,11 @@ import ( "sync" "testing" + clusterv3 "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3" corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" + endpointv3 "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3" discoveryv3 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3" + envoytypes "github.com/envoyproxy/go-control-plane/pkg/cache/types" cachev3 "github.com/envoyproxy/go-control-plane/pkg/cache/v3" resourcev3 "github.com/envoyproxy/go-control-plane/pkg/resource/v3" "github.com/stretchr/testify/require" @@ -20,6 +23,7 @@ import ( egv1a1 "github.com/envoyproxy/gateway/api/v1alpha1" "github.com/envoyproxy/gateway/internal/logging" + "github.com/envoyproxy/gateway/internal/xds/types" ) func newTestSnapshotCache(t *testing.T) *snapshotCache { @@ -124,3 +128,84 @@ func TestOnStreamDeltaResponseConcurrentAccess(t *testing.T) { } wg.Wait() } + +// TestUpdateEndpointResources verifies that the endpoint fast path patch bumps +// only the EDS resource version, replaces only the given CLAs, keeps everything +// else version-stable, and never mutates the previous snapshot. +func TestUpdateEndpointResources(t *testing.T) { + const irKey = "test-cluster" + + sc := newTestSnapshotCache(t) + + cla := func(name, address string) *endpointv3.ClusterLoadAssignment { + return &endpointv3.ClusterLoadAssignment{ + ClusterName: name, + Endpoints: []*endpointv3.LocalityLbEndpoints{{ + LbEndpoints: []*endpointv3.LbEndpoint{{ + HostIdentifier: &endpointv3.LbEndpoint_Endpoint{ + Endpoint: &endpointv3.Endpoint{ + Address: &corev3.Address{ + Address: &corev3.Address_SocketAddress{ + SocketAddress: &corev3.SocketAddress{ + Address: address, + PortSpecifier: &corev3.SocketAddress_PortValue{PortValue: 8080}, + }, + }, + }, + }, + }, + }}, + }}, + } + } + + resources := types.XdsResources{ + resourcev3.ClusterType: {&clusterv3.Cluster{Name: "cluster-a"}, &clusterv3.Cluster{Name: "cluster-b"}}, + resourcev3.EndpointType: { + cla("cluster-a", "1.1.1.1"), + cla("cluster-b", "2.2.2.2"), + }, + } + committed, err := sc.GenerateNewSnapshot(irKey, resources, context.Background()) + require.NoError(t, err) + require.True(t, committed) + + oldSnapshot := sc.lastSnapshot[irKey] + oldEDSVersion := oldSnapshot.GetVersion(resourcev3.EndpointType) + oldCDSVersion := oldSnapshot.GetVersion(resourcev3.ClusterType) + + // Patching an unknown IR key is a no-op. + err = sc.UpdateEndpointResources("unknown", []envoytypes.Resource{cla("cluster-a", "3.3.3.3")}) + require.ErrorIs(t, err, ErrNoSnapshot) + + require.NoError(t, sc.UpdateEndpointResources(irKey, []envoytypes.Resource{cla("cluster-a", "3.3.3.3")})) + + newSnapshot := sc.lastSnapshot[irKey] + require.NotSame(t, oldSnapshot, newSnapshot) + + // Only the EDS version is bumped. + require.NotEqual(t, oldEDSVersion, newSnapshot.GetVersion(resourcev3.EndpointType)) + require.Equal(t, oldCDSVersion, newSnapshot.GetVersion(resourcev3.ClusterType)) + + // The patched CLA is replaced, the other one is kept. + eds := newSnapshot.GetResources(resourcev3.EndpointType) + require.Len(t, eds, 2) + gotA := eds["cluster-a"].(*endpointv3.ClusterLoadAssignment) + require.Equal(t, "3.3.3.3", gotA.GetEndpoints()[0].GetLbEndpoints()[0].GetEndpoint().GetAddress().GetSocketAddress().GetAddress()) + gotB := eds["cluster-b"].(*endpointv3.ClusterLoadAssignment) + require.Equal(t, "2.2.2.2", gotB.GetEndpoints()[0].GetLbEndpoints()[0].GetEndpoint().GetAddress().GetSocketAddress().GetAddress()) + + // The previous snapshot is untouched. + oldA := oldSnapshot.GetResources(resourcev3.EndpointType)["cluster-a"].(*endpointv3.ClusterLoadAssignment) + require.Equal(t, "1.1.1.1", oldA.GetEndpoints()[0].GetLbEndpoints()[0].GetEndpoint().GetAddress().GetSocketAddress().GetAddress()) + + // The patched snapshot carries no delta version map, so go-control-plane + // rebuilds it on demand and it reflects the patched endpoints. + require.Nil(t, newSnapshot.VersionMap) + require.NoError(t, newSnapshot.ConstructVersionMap()) + require.NoError(t, oldSnapshot.ConstructVersionMap()) + oldEDSVM := oldSnapshot.GetVersionMap(resourcev3.EndpointType) + newEDSVM := newSnapshot.GetVersionMap(resourcev3.EndpointType) + require.NotEqual(t, oldEDSVM["cluster-a"], newEDSVM["cluster-a"]) + require.Equal(t, oldEDSVM["cluster-b"], newEDSVM["cluster-b"]) +} diff --git a/internal/xds/runner/endpoint_fast_path.go b/internal/xds/runner/endpoint_fast_path.go new file mode 100644 index 00000000000..ae325321c91 --- /dev/null +++ b/internal/xds/runner/endpoint_fast_path.go @@ -0,0 +1,351 @@ +// Copyright Envoy Gateway Authors +// SPDX-License-Identifier: Apache-2.0 +// The full text of the Apache license is available in the LICENSE file at +// the root of the repo. + +package runner + +import ( + "context" + "errors" + "reflect" + "sync" + + envoytypes "github.com/envoyproxy/go-control-plane/pkg/cache/types" + cachev3 "github.com/envoyproxy/go-control-plane/pkg/cache/v3" + resourcev3 "github.com/envoyproxy/go-control-plane/pkg/resource/v3" + "github.com/telepresenceio/watchable" + + egv1a1 "github.com/envoyproxy/gateway/api/v1alpha1" + extension "github.com/envoyproxy/gateway/internal/extension/types" + "github.com/envoyproxy/gateway/internal/ir" + "github.com/envoyproxy/gateway/internal/logging" + "github.com/envoyproxy/gateway/internal/message" + "github.com/envoyproxy/gateway/internal/metrics" + endpointsutil "github.com/envoyproxy/gateway/internal/utils/endpoints" + "github.com/envoyproxy/gateway/internal/xds/cache" + "github.com/envoyproxy/gateway/internal/xds/translator" + xdstypes "github.com/envoyproxy/gateway/internal/xds/types" +) + +var ( + endpointFastPathPatchTotal = metrics.NewCounter( + "endpoint_fastpath_patch_total", + "Total number of EDS-only snapshot patches applied by the endpoint fast path.", + ) + + endpointFastPathSkippedTotal = metrics.NewCounter( + "endpoint_fastpath_skipped_total", + "Total number of endpoint updates skipped by the endpoint fast path, by reason.", + ) + + skipReasonLabel = metrics.NewLabel("reason") +) + +const ( + skipReasonNoContext = "no_context" + skipReasonDisabledForKey = "disabled_for_key" + skipReasonZeroTransition = "zero_transition" + skipReasonNoSnapshot = "no_snapshot" + skipReasonAddressTypeChange = "address_type_change" +) + +// endpointFastPath propagates endpoint updates to Envoy as EDS-only snapshot patches, +// rebuilding the affected clusters' ClusterLoadAssignments from the contexts +// captured at the last successful full translation. Semantics are best-effort: +// fresh endpoints against last-known-good config, with full translations +// remaining authoritative. +type endpointFastPath struct { + logger logging.Logger + cache cache.SnapshotCacheWithCallbacks + + // extensionModifiesEndpoints is true when a registered extension server + // subscribes to a post-xDS hook that can modify clusters or endpoints, + // in which case a regenerated CLA could lose the extension's changes and + // the fast path stands down entirely. + extensionModifiesEndpoints bool + + // mu serializes full-snapshot publishes and fast-path patches, so a patch + // can never interleave with a snapshot rebuild or its context swap. + mu sync.Mutex + // contexts holds, per IR key, the endpoint contexts captured at the last + // successful full translation. + contexts map[string]*irKeyContexts + // lastEndpoints holds the most recent endpoint update per backend key, so + // it can be re-applied on top of a full snapshot that raced with it. + // Entries are dropped only when the provider deletes the backend. Pruning + // against the committed contexts here would race an in-flight build that + // is about to introduce a backend it has no context for yet. + lastEndpoints map[string]*message.EndpointUpdate +} + +// irKeyContexts is the endpoint context index for one IR key. +type irKeyContexts struct { + // enabled is false when this IR key must not take the fast path — its IR + // contains EnvoyPatchPolicies, whose patches (and statuses) are only + // handled by the full translation. + enabled bool + // byBackend maps a backend key (message.EndpointUpdate.Key()) to the + // endpoint contexts of the clusters fed by that backend. + byBackend map[string][]*xdstypes.EndpointContext +} + +func newEndpointFastPath(c cache.SnapshotCacheWithCallbacks, extMgr extension.Manager, logger logging.Logger) *endpointFastPath { + return &endpointFastPath{ + logger: logger, + cache: c, + extensionModifiesEndpoints: extensionModifiesEndpoints(extMgr), + contexts: make(map[string]*irKeyContexts), + lastEndpoints: make(map[string]*message.EndpointUpdate), + } +} + +// extensionModifiesEndpoints reports whether a registered extension subscribes +// to a post-xDS hook that can modify clusters or endpoints. Extension mutations +// are applied in place and are invisible afterwards, so the decision must be +// made from hook registration, not from diffing results. +func extensionModifiesEndpoints(extMgr extension.Manager) bool { + if extMgr == nil { + return false + } + for _, hook := range []egv1a1.XDSTranslatorHook{egv1a1.XDSEndpoints, egv1a1.XDSCluster, egv1a1.XDSTranslation} { + client, err := extMgr.GetPostXDSHookClient(hook) + if err != nil || client != nil { + return true + } + } + return false +} + +func endpointSourceKey(es *ir.EndpointSource) string { + return message.BackendKey(es.Kind, es.Namespace, es.Name) +} + +// OnFullBuild hands a completed translation to the snapshot cache. The endpoint +// state this path already published is folded into the build's own EDS resources +// first, so a build that raced an endpoint change does not regress it. The +// captured contexts replace the previous ones only if the snapshot committed. +func (f *endpointFastPath) OnFullBuild(irKey string, xdsIR *ir.Xds, table *xdstypes.ResourceVersionTable, ctx context.Context) error { + f.mu.Lock() + defer f.mu.Unlock() + + ctxs := f.buildContexts(xdsIR, table) + if ctxs.enabled { + for backendKey := range ctxs.byBackend { + update, ok := f.lastEndpoints[backendKey] + if !ok { + continue + } + // Mutations target the private contexts built above; skips (zero + // transition, address-type change) leave the build's own endpoint + // state in place. + assignments, skipReason := f.patchContexts(ctxs, update) + if skipReason != "" { + // Counted here too: this build publishes endpoint state the + // fast path knows is already superseded. + endpointFastPathSkippedTotal.With(skipReasonLabel.Value(skipReason)).Increment() + continue + } + replaceEndpointResources(table, assignments) + } + } + + committed, err := f.cache.GenerateNewSnapshot(irKey, table.XdsResources, ctx) + if committed { + // The snapshot is live, so the contexts must describe it even when + // publishing to some node failed — otherwise later patches would be + // built from the previous translation's settings. + f.contexts[irKey] = ctxs + } + return err +} + +// replaceEndpointResources swaps the given ClusterLoadAssignments into the +// translation table's EDS resources by cluster name. +func replaceEndpointResources(table *xdstypes.ResourceVersionTable, assignments []envoytypes.Resource) { + if len(assignments) == 0 { + return + } + byName := make(map[string]envoytypes.Resource, len(assignments)) + for _, assignment := range assignments { + byName[cachev3.GetResourceName(assignment)] = assignment + } + eds := table.XdsResources[resourcev3.EndpointType] + for i, res := range eds { + if replacement, ok := byName[cachev3.GetResourceName(res)]; ok { + eds[i] = replacement + } + } +} + +// OnDelete drops the endpoint contexts for a deleted IR key. +func (f *endpointFastPath) OnDelete(irKey string) { + f.mu.Lock() + defer f.mu.Unlock() + delete(f.contexts, irKey) +} + +// buildContexts builds the endpoint context index from a full translation's +// captured contexts, without committing it. +func (f *endpointFastPath) buildContexts(xdsIR *ir.Xds, table *xdstypes.ResourceVersionTable) *irKeyContexts { + ctxs := &irKeyContexts{ + enabled: len(xdsIR.EnvoyPatchPolicies) == 0 && !f.extensionModifiesEndpoints, + byBackend: make(map[string][]*xdstypes.EndpointContext), + } + for _, ec := range table.EndpointContexts { + for _, ds := range ec.Settings { + if ds.EndpointSource != nil { + key := endpointSourceKey(ds.EndpointSource) + ctxs.byBackend[key] = append(ctxs.byBackend[key], ec) + } + } + } + return ctxs +} + +// subscribe consumes endpoint updates until the subscription context is +// cancelled. The EndpointUpdates map itself is never closed, because the +// provider writes it from informer handlers that Close cannot synchronize with. +func (f *endpointFastPath) subscribe(sub <-chan watchable.Snapshot[string, *message.EndpointUpdate], runnerName string) { + message.HandleSubscription( + f.logger, + message.Metadata{Runner: runnerName, Message: message.EndpointSlicesMessageName}, sub, + func(update message.Update[string, *message.EndpointUpdate], _ chan error) { + message.PublishRunnerEventMetric(runnerName, update.Delete) + if update.Delete || update.Value == nil { + // The provider deletes an entry when the backend's last + // EndpointSlice is gone; drop the retained state and leave the + // transition to the full path. + f.mu.Lock() + delete(f.lastEndpoints, update.Key) + f.mu.Unlock() + return + } + f.handleUpdate(update.Value) + }, + ) + f.logger.Info("endpoint fast path subscriber shutting down") +} + +// handleUpdate applies one endpoint update: for every IR key whose contexts +// reference the updated backend, the affected clusters' CLAs are rebuilt and +// patched into the current snapshot with only the EDS version bumped. +func (f *endpointFastPath) handleUpdate(update *message.EndpointUpdate) { + logger := f.logger + + f.mu.Lock() + defer f.mu.Unlock() + + backendKey := update.Key() + f.lastEndpoints[backendKey] = update + + found := false + for irKey, ctxs := range f.contexts { + if _, ok := ctxs.byBackend[backendKey]; !ok { + continue + } + found = true + if !ctxs.enabled { + endpointFastPathSkippedTotal.With(skipReasonLabel.Value(skipReasonDisabledForKey)).Increment() + continue + } + assignments, skipReason := f.patchContexts(ctxs, update) + if skipReason != "" { + endpointFastPathSkippedTotal.With(skipReasonLabel.Value(skipReason)).Increment() + continue + } + if len(assignments) == 0 { + // The endpoint state is already reflected in the snapshot (e.g. the + // full path got there first); nothing to patch. + continue + } + // The full reconcile enqueued by the same endpoint event is the backstop + // for both failure paths, so the contexts are left describing the new + // endpoints either way. + switch err := f.cache.UpdateEndpointResources(irKey, assignments); { + case errors.Is(err, cache.ErrNoSnapshot): + endpointFastPathSkippedTotal.With(skipReasonLabel.Value(skipReasonNoSnapshot)).Increment() + case err != nil: + logger.Error(err, "endpoint fast path: failed to patch endpoint resources", "key", irKey, "backend", backendKey) + default: + endpointFastPathPatchTotal.Increment() + logger.Info("endpoint fast path: patched endpoints", "key", irKey, "backend", backendKey, "clusters", len(assignments)) + } + } + if !found { + endpointFastPathSkippedTotal.With(skipReasonLabel.Value(skipReasonNoContext)).Increment() + } +} + +// patchContexts recomputes the endpoints fed by the updated backend and returns +// the rebuilt CLAs of the clusters that changed. A non-empty skip reason means +// the contexts were left untouched: the change is not CLA-only and belongs to +// the full path. Must be called with f.mu held. +func (f *endpointFastPath) patchContexts(ctxs *irKeyContexts, update *message.EndpointUpdate) ([]envoytypes.Resource, string) { + backendKey := update.Key() + + type pendingChange struct { + ds *ir.DestinationSetting + endpoints []*ir.DestinationEndpoint + addrType *ir.DestinationAddressType + } + var ( + pending []pendingChange + changedClusters []*xdstypes.EndpointContext + seen = make(map[*xdstypes.EndpointContext]struct{}) + ) + + for _, ec := range ctxs.byBackend[backendKey] { + if _, ok := seen[ec]; ok { + continue + } + seen[ec] = struct{}{} + changed := false + for _, ds := range ec.Settings { + if ds.EndpointSource == nil || endpointSourceKey(ds.EndpointSource) != backendKey { + continue + } + endpoints, addrType := endpointsutil.EndpointsFromSlices(update.EndpointSlices, ds.EndpointSource.PortName, ds.EndpointSource.Protocol) + // A setting captured in a context always had endpoints at full + // translation time; dropping to zero restructures clusters, routes + // and statuses, so it must take the full path. + if len(endpoints) == 0 { + return nil, skipReasonZeroTransition + } + if !equalAddressType(addrType, ds.AddressType) { + return nil, skipReasonAddressTypeChange + } + if reflect.DeepEqual(endpoints, ds.Endpoints) { + continue + } + pending = append(pending, pendingChange{ + ds: ds, + endpoints: endpoints, + addrType: addrType, + }) + changed = true + } + if changed { + changedClusters = append(changedClusters, ec) + } + } + + // All settings validated; apply the changes to the cached contexts so they + // track the last-applied endpoint state, then rebuild the affected CLAs. + for _, p := range pending { + p.ds.Endpoints = p.endpoints + p.ds.AddressType = p.addrType + } + assignments := make([]envoytypes.Resource, 0, len(changedClusters)) + for _, ec := range changedClusters { + assignments = append(assignments, translator.BuildClusterLoadAssignment(ec)) + } + return assignments, "" +} + +func equalAddressType(a, b *ir.DestinationAddressType) bool { + if a == nil || b == nil { + return a == b + } + return *a == *b +} diff --git a/internal/xds/runner/endpoint_fast_path_test.go b/internal/xds/runner/endpoint_fast_path_test.go new file mode 100644 index 00000000000..ce7ddfbfce8 --- /dev/null +++ b/internal/xds/runner/endpoint_fast_path_test.go @@ -0,0 +1,304 @@ +// Copyright Envoy Gateway Authors +// SPDX-License-Identifier: Apache-2.0 +// The full text of the Apache license is available in the LICENSE file at +// the root of the repo. + +package runner + +import ( + "context" + "os" + "testing" + + endpointv3 "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3" + envoytypes "github.com/envoyproxy/go-control-plane/pkg/cache/types" + resourcev3 "github.com/envoyproxy/go-control-plane/pkg/resource/v3" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + egv1a1 "github.com/envoyproxy/gateway/api/v1alpha1" + "github.com/envoyproxy/gateway/internal/ir" + "github.com/envoyproxy/gateway/internal/logging" + "github.com/envoyproxy/gateway/internal/message" + "github.com/envoyproxy/gateway/internal/xds/cache" + "github.com/envoyproxy/gateway/internal/xds/translator" + xdstypes "github.com/envoyproxy/gateway/internal/xds/types" +) + +type patchCall struct { + irKey string + assignments []envoytypes.Resource +} + +// fakeCache records snapshot and patch calls; the embedded interface is nil and +// only the overridden methods may be called. +type fakeCache struct { + cache.SnapshotCacheWithCallbacks + snapshots []string + snapshotResources []xdstypes.XdsResources + patches []patchCall + patchErr error + // patchCommitted is what UpdateEndpointResources reports alongside + // patchErr: true models a snapshot that went live but failed to reach + // some node. + patchCommitted bool +} + +func (f *fakeCache) GenerateNewSnapshot(irKey string, resources xdstypes.XdsResources, _ context.Context) (bool, error) { + f.snapshots = append(f.snapshots, irKey) + f.snapshotResources = append(f.snapshotResources, resources) + return true, nil +} + +func (f *fakeCache) UpdateEndpointResources(irKey string, assignments []envoytypes.Resource) error { + if f.patchErr != nil { + if f.patchCommitted { + f.patches = append(f.patches, patchCall{irKey: irKey, assignments: assignments}) + } + return f.patchErr + } + f.patches = append(f.patches, patchCall{irKey: irKey, assignments: assignments}) + return nil +} + +func testEndpointSlice(svcName string, addresses ...string) *discoveryv1.EndpointSlice { + return &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: svcName + "-abc", Labels: map[string]string{discoveryv1.LabelServiceName: svcName}}, + AddressType: discoveryv1.AddressTypeIPv4, + Endpoints: []discoveryv1.Endpoint{{ + Addresses: addresses, + Conditions: discoveryv1.EndpointConditions{Ready: new(true)}, + }}, + Ports: []discoveryv1.EndpointPort{{ + Name: new("http"), + Protocol: new(corev1.ProtocolTCP), + Port: new(int32(8080)), + }}, + } +} + +func testFastPathContext() *xdstypes.EndpointContext { + const clusterName = "cluster-a" + return &xdstypes.EndpointContext{ + ClusterName: clusterName, + Settings: []*ir.DestinationSetting{{ + Name: clusterName + "/backend/0", + Weight: new(uint32(1)), + Protocol: ir.HTTP, + Endpoints: []*ir.DestinationEndpoint{ir.NewDestEndpoint(nil, "1.1.1.1", 8080, false, nil)}, + AddressType: new(ir.IP), + EndpointSource: &ir.EndpointSource{ + Kind: "Service", + Namespace: "default", + Name: "svc", + PortName: "http", + Protocol: corev1.ProtocolTCP, + }, + }}, + } +} + +func testFastPath(fc *fakeCache) *endpointFastPath { + fp := newEndpointFastPath(fc, nil, logging.DefaultLogger(os.Stderr, egv1a1.LogLevelInfo)) + table := &xdstypes.ResourceVersionTable{} + table.AddEndpointContext(testFastPathContext()) + fp.contexts["gw/eg"] = fp.buildContexts(&ir.Xds{}, table) + return fp +} + +func testUpdate(addresses ...string) *message.EndpointUpdate { + return &message.EndpointUpdate{ + Kind: "Service", + Namespace: "default", + Name: "svc", + EndpointSlices: []*discoveryv1.EndpointSlice{testEndpointSlice("svc", addresses...)}, + } +} + +func claAddresses(t *testing.T, res envoytypes.Resource) []string { + t.Helper() + cla, ok := res.(*endpointv3.ClusterLoadAssignment) + require.True(t, ok) + var addrs []string + for _, locality := range cla.GetEndpoints() { + for _, ep := range locality.GetLbEndpoints() { + addrs = append(addrs, ep.GetEndpoint().GetAddress().GetSocketAddress().GetAddress()) + } + } + return addrs +} + +func TestFastPathPatchesChangedEndpoints(t *testing.T) { + fc := &fakeCache{} + fp := testFastPath(fc) + + fp.handleUpdate(testUpdate("2.2.2.2", "3.3.3.3")) + + require.Len(t, fc.patches, 1) + require.Equal(t, "gw/eg", fc.patches[0].irKey) + require.Len(t, fc.patches[0].assignments, 1) + require.ElementsMatch(t, []string{"2.2.2.2", "3.3.3.3"}, claAddresses(t, fc.patches[0].assignments[0])) + + // The cached context tracks the applied endpoint state, so replaying the + // same update is a no-op. + fp.handleUpdate(testUpdate("2.2.2.2", "3.3.3.3")) + require.Len(t, fc.patches, 1) +} + +func TestFastPathSkipsZeroEndpointTransition(t *testing.T) { + fc := &fakeCache{} + fp := testFastPath(fc) + + update := testUpdate() + update.EndpointSlices = nil + fp.handleUpdate(update) + + require.Empty(t, fc.patches) + // The cached context still holds the last-applied endpoints. + require.Len(t, fp.contexts["gw/eg"].byBackend[update.Key()][0].Settings[0].Endpoints, 1) +} + +func TestFastPathSkipsAddressTypeChange(t *testing.T) { + fc := &fakeCache{} + fp := testFastPath(fc) + + update := testUpdate("foo.example.com") + update.EndpointSlices[0].AddressType = discoveryv1.AddressTypeFQDN + fp.handleUpdate(update) + + require.Empty(t, fc.patches) +} + +func TestFastPathSkipsUnknownBackend(t *testing.T) { + fc := &fakeCache{} + fp := testFastPath(fc) + + update := testUpdate("2.2.2.2") + update.Name = "other-svc" + fp.handleUpdate(update) + + require.Empty(t, fc.patches) +} + +func TestFastPathDisabledForKeyWithEnvoyPatchPolicies(t *testing.T) { + fc := &fakeCache{} + fp := testFastPath(fc) + + // A full build whose IR carries EnvoyPatchPolicies disables the fast path + // for the key: patches could target ClusterLoadAssignments. + table := &xdstypes.ResourceVersionTable{} + table.AddEndpointContext(testFastPathContext()) + fp.contexts["gw/eg"] = fp.buildContexts(&ir.Xds{EnvoyPatchPolicies: []*ir.EnvoyPatchPolicy{{}}}, table) + + fp.handleUpdate(testUpdate("2.2.2.2")) + require.Empty(t, fc.patches) +} + +// publishFullBuildTable builds a translation table whose EDS resources match +// the endpoint context captured by the build (as a real translation would). +func publishFullBuildTable(ec *xdstypes.EndpointContext) *xdstypes.ResourceVersionTable { + table := &xdstypes.ResourceVersionTable{} + table.AddEndpointContext(ec) + table.XdsResources = xdstypes.XdsResources{ + resourcev3.EndpointType: {translator.BuildClusterLoadAssignment(ec)}, + } + return table +} + +func TestFastPathReappliesLatestEndpointsOnFullBuild(t *testing.T) { + fc := &fakeCache{} + fp := testFastPath(fc) + + // The fast path has seen endpoint state newer than what the (racing) full + // build translated. + fp.handleUpdate(testUpdate("9.9.9.9")) + require.Len(t, fc.patches, 1) + + // The full build still carries the old endpoints; publishing it must patch + // the state the fast path already published into the build's own EDS + // resources, so the snapshot goes out fresh in a single push. + table := publishFullBuildTable(testFastPathContext()) + require.NoError(t, fp.OnFullBuild("gw/eg", &ir.Xds{}, table, context.Background())) + + require.Equal(t, []string{"gw/eg"}, fc.snapshots) + eds := fc.snapshotResources[0][resourcev3.EndpointType] + require.Len(t, eds, 1) + require.ElementsMatch(t, []string{"9.9.9.9"}, claAddresses(t, eds[0])) + // No second push happened. + require.Len(t, fc.patches, 1) +} + +func TestFastPathKeepsContextWhenPatchCommittedWithError(t *testing.T) { + fc := &fakeCache{} + fp := testFastPath(fc) + + // The patched snapshot went live but publishing to a node failed. The + // cached context must keep describing the live snapshot, so a later rebuild + // of the same cluster (e.g. triggered by another backend it serves) does not + // regress these endpoints. + fc.patchErr = context.DeadlineExceeded + fc.patchCommitted = true + fp.handleUpdate(testUpdate("2.2.2.2")) + require.Len(t, fc.patches, 1) + + applied := fp.contexts["gw/eg"].byBackend[testUpdate().Key()][0].Settings[0].Endpoints + require.Len(t, applied, 1) + require.Equal(t, "2.2.2.2", applied[0].Host) +} + +func TestFastPathLeavesUncommittedPatchToTheFullPath(t *testing.T) { + fc := &fakeCache{} + fp := testFastPath(fc) + + // Nothing was committed. The contexts still track the new endpoints, so the + // fast path does not retry; the full build triggered by the same event is + // the backstop and re-applies the retained endpoints. + fc.patchErr = context.DeadlineExceeded + fp.handleUpdate(testUpdate("2.2.2.2")) + require.Empty(t, fc.patches) + + table := publishFullBuildTable(testFastPathContext()) + require.NoError(t, fp.OnFullBuild("gw/eg", &ir.Xds{}, table, context.Background())) + require.ElementsMatch(t, []string{"2.2.2.2"}, + claAddresses(t, fc.snapshotResources[0][resourcev3.EndpointType][0])) +} + +func TestFastPathRetainsEndpointsForBackendsNotYetInAnyContext(t *testing.T) { + fc := &fakeCache{} + fp := testFastPath(fc) + + // An endpoint update arrives for a backend no committed context references + // yet, because the build that will introduce it is still in flight. + pending := testUpdate("7.7.7.7") + pending.Name = "pending-svc" + fp.handleUpdate(pending) + require.Empty(t, fc.patches) + + // An unrelated IR key's build commits. It must not drop the retained + // update: the in-flight build still needs it. + other := publishFullBuildTable(testFastPathContext()) + require.NoError(t, fp.OnFullBuild("gw/other", &ir.Xds{}, other, context.Background())) + require.Contains(t, fp.lastEndpoints, pending.Key()) + + // When the in-flight build lands carrying older endpoints, the retained + // update is patched into it rather than being lost. + ec := testFastPathContext() + ec.Settings[0].EndpointSource.Name = "pending-svc" + table := publishFullBuildTable(ec) + require.NoError(t, fp.OnFullBuild("gw/pending", &ir.Xds{}, table, context.Background())) + + eds := fc.snapshotResources[len(fc.snapshotResources)-1][resourcev3.EndpointType] + require.Len(t, eds, 1) + require.ElementsMatch(t, []string{"7.7.7.7"}, claAddresses(t, eds[0])) +} + +func TestFastPathOnDelete(t *testing.T) { + fc := &fakeCache{} + fp := testFastPath(fc) + + fp.OnDelete("gw/eg") + fp.handleUpdate(testUpdate("2.2.2.2")) + require.Empty(t, fc.patches) +} diff --git a/internal/xds/runner/runner.go b/internal/xds/runner/runner.go index 20d52cd51df..299c725db8c 100644 --- a/internal/xds/runner/runner.go +++ b/internal/xds/runner/runner.go @@ -92,6 +92,9 @@ type Config struct { type Runner struct { Config + // endpointFastPath is non-nil when the EndpointFastPath runtime flag is enabled; it + // patches endpoint updates into the snapshot without a full translation. + endpointFastPath *endpointFastPath } func New(cfg *Config) *Runner { @@ -152,6 +155,7 @@ func (r *Runner) Close() error { return nil } // Start starts the xds-server runner func (r *Runner) Start(ctx context.Context) error { r.Logger = r.Logger.WithName(r.Name()).WithValues("runner", r.Name()) + endpointFastPath := r.EnvoyGateway.RuntimeFlags.IsEnabled(egv1a1.EndpointFastPath) r.cache = cache.NewSnapshotCache(true, r.Logger) // Set up the gRPC server and register the xDS handler. @@ -224,10 +228,24 @@ func (r *Runner) Start(ctx context.Context) error { // Start and listen xDS gRPC Server. go r.serveXdsServer(ctx) + // Set up the endpoint fast path: endpoint updates published by the provider + // are patched into the snapshot as EDS-only updates, without waiting for a + // full translation. r.endpointFastPath must be assigned before the translate + // goroutine below starts reading it. + if endpointFastPath && r.ProviderResources != nil { + r.endpointFastPath = newEndpointFastPath(r.cache, r.ExtensionManager, r.Logger) + r.Logger.Info("endpoint fast path enabled") + } + // Do not call .Subscribe() inside Goroutine since it is supposed to be called from the same // Goroutine where Close() is called. sub := r.XdsIR.Subscribe(ctx) go r.translateFromSubscription(sub) + + if r.endpointFastPath != nil { + epSub := r.ProviderResources.EndpointUpdates.Subscribe(ctx) + go r.endpointFastPath.subscribe(epSub, r.Name()) + } r.Logger.Info("started") return err } @@ -297,7 +315,10 @@ func (r *Runner) translateFromSubscription(sub <-chan watchable.Snapshot[string, ) if update.Delete { - if err := r.cache.GenerateNewSnapshot(key, nil, traceCtx); err != nil { + if r.endpointFastPath != nil { + r.endpointFastPath.OnDelete(key) + } + if _, err := r.cache.GenerateNewSnapshot(key, nil, traceCtx); err != nil { traceLogger.Error(err, "failed to delete the snapshot") errChan <- err } @@ -358,12 +379,21 @@ func (r *Runner) translateFromSubscription(sub <-chan watchable.Snapshot[string, // Note: invalid EnvoyPatchPolicies are considered user-level errors and will not prevent the snapshot from being updated. if err == nil { if result.XdsResources != nil { - if r.cache == nil { + switch { + case r.cache == nil: r.Logger.Error(err, "failed to init snapshot cache") errChan <- err - } else { + case r.endpointFastPath != nil: + // Publish through the fast path so the snapshot, the + // endpoint contexts, and the re-applied endpoint state + // are swapped atomically with respect to endpoint patches. + if err := r.endpointFastPath.OnFullBuild(key, val.XdsIR, result, traceCtx); err != nil { + r.Logger.Error(err, "failed to generate a snapshot") + errChan <- err + } + default: // Update snapshot cache - if err := r.cache.GenerateNewSnapshot(key, result.XdsResources, traceCtx); err != nil { + if _, err := r.cache.GenerateNewSnapshot(key, result.XdsResources, traceCtx); err != nil { r.Logger.Error(err, "failed to generate a snapshot") errChan <- err } diff --git a/internal/xds/translator/endpoint_fast_path.go b/internal/xds/translator/endpoint_fast_path.go new file mode 100644 index 00000000000..ad890ede098 --- /dev/null +++ b/internal/xds/translator/endpoint_fast_path.go @@ -0,0 +1,59 @@ +// Copyright Envoy Gateway Authors +// SPDX-License-Identifier: Apache-2.0 +// The full text of the Apache license is available in the LICENSE file at +// the root of the repo. + +package translator + +import ( + "slices" + + endpointv3 "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3" + + "github.com/envoyproxy/gateway/internal/ir" + "github.com/envoyproxy/gateway/internal/xds/types" +) + +// BuildClusterLoadAssignment rebuilds a cluster's ClusterLoadAssignment from an +// endpoint context captured during full translation. It is the endpoint fast +// path's entry point into the same CLA construction used by full translation, +// so both paths produce identical output for the same inputs. +func BuildClusterLoadAssignment(ec *types.EndpointContext) *endpointv3.ClusterLoadAssignment { + return buildXdsClusterLoadAssignment(ec.ClusterName, ec.Settings, ec.HealthCheck, ec.PreferLocal, ec.WeightedZones) +} + +// endpointFastPathEnabled reports whether the endpoint fast path can serve this +// cluster. The gatewayapi translation only populates EndpointSource when the +// EndpointFastPath runtime flag is on and the backend is EndpointSlice-backed. +func endpointFastPathEnabled(settings []*ir.DestinationSetting) bool { + return slices.ContainsFunc(settings, func(ds *ir.DestinationSetting) bool { + return ds != nil && ds.EndpointSource != nil + }) +} + +// buildEndpointContext captures everything the endpoint fast path needs to rebuild +// this cluster's CLA without a full translation, or nil when the fast path is off. +// The context owns deep copies: the IR it is built from stays live in the watchable +// layer, which compares it for equality on the next publish. +func buildEndpointContext(args *xdsClusterArgs, lb *ir.LoadBalancer) *types.EndpointContext { + if !endpointFastPathEnabled(args.settings) { + return nil + } + settings := make([]*ir.DestinationSetting, len(args.settings)) + for i, s := range args.settings { + settings[i] = s.DeepCopy() + } + ec := &types.EndpointContext{ + ClusterName: args.name, + Settings: settings, + HealthCheck: args.healthCheck.DeepCopy(), + PreferLocal: lb.PreferLocal.DeepCopy(), + } + if lb.WeightedZones != nil { + ec.WeightedZones = make([]ir.WeightedZoneConfig, len(lb.WeightedZones)) + for i, wz := range lb.WeightedZones { + wz.DeepCopyInto(&ec.WeightedZones[i]) + } + } + return ec +} diff --git a/internal/xds/translator/translator.go b/internal/xds/translator/translator.go index aaf8c708c60..b49913dc1ac 100644 --- a/internal/xds/translator/translator.go +++ b/internal/xds/translator/translator.go @@ -1350,6 +1350,12 @@ func addXdsCluster(tCtx *types.ResourceVersionTable, args *xdsClusterArgs) error if err := tCtx.AddXdsResource(resourcev3.EndpointType, xdsEndpoints); err != nil { return err } + // Remember how this CLA was built so the endpoint fast path can rebuild it + // on an EndpointSlice change. Nil unless the EndpointFastPath flag is on and + // the cluster is EndpointSlice-backed. + if ec := buildEndpointContext(args, &lb); ec != nil { + tCtx.AddEndpointContext(ec) + } case EndpointTypeDNS: xdsCluster.LoadAssignment = xdsEndpoints case EndpointTypeDynamicResolver: diff --git a/internal/xds/types/endpointcontext.go b/internal/xds/types/endpointcontext.go new file mode 100644 index 00000000000..802fa8078fd --- /dev/null +++ b/internal/xds/types/endpointcontext.go @@ -0,0 +1,36 @@ +// Copyright Envoy Gateway Authors +// SPDX-License-Identifier: Apache-2.0 +// The full text of the Apache license is available in the LICENSE file at +// the root of the repo. + +package types + +import ( + "github.com/envoyproxy/gateway/internal/ir" +) + +// EndpointContext captures, for one EDS cluster, the inputs needed to rebuild its +// ClusterLoadAssignment outside a full translation. It is recorded during full +// translation and consumed by the xDS runner's endpoint fast path. +type EndpointContext struct { + // ClusterName is the xDS cluster (and CLA) name. + ClusterName string + // Settings are the cluster's destination settings from the translated IR. + // Their order matters: the per-endpoint TLS transport socket match metadata + // embeds the setting index. + Settings []*ir.DestinationSetting + // HealthCheck carries the active health check overrides applied to endpoints. + HealthCheck *ir.HealthCheck + // PreferLocal is the cluster-level zone-aware routing config. + PreferLocal *ir.PreferLocalZone + // WeightedZones is the explicit per-zone weighting config. + WeightedZones []ir.WeightedZoneConfig +} + +// AddEndpointContext records the endpoint context for an EDS cluster. +func (t *ResourceVersionTable) AddEndpointContext(ec *EndpointContext) { + if t.EndpointContexts == nil { + t.EndpointContexts = make(map[string]*EndpointContext) + } + t.EndpointContexts[ec.ClusterName] = ec +} diff --git a/internal/xds/types/resourceversiontable.go b/internal/xds/types/resourceversiontable.go index ad0124914be..4f20b8b5e00 100644 --- a/internal/xds/types/resourceversiontable.go +++ b/internal/xds/types/resourceversiontable.go @@ -32,6 +32,11 @@ type ResourceVersionTable struct { XdsResources EnvoyPatchPolicyStatuses GlobalResourceStatus + + // EndpointContexts maps an EDS cluster name to the context needed to rebuild + // its ClusterLoadAssignment on the endpoint fast path. Only populated when + // the EndpointFastPath runtime flag is enabled. + EndpointContexts map[string]*EndpointContext } // GetXdsResources retrieves the translated xds resources saved in the translator context. diff --git a/release-notes/current/performance_improvements/10022-endpoint-fast-path.md b/release-notes/current/performance_improvements/10022-endpoint-fast-path.md new file mode 100644 index 00000000000..91c3b567644 --- /dev/null +++ b/release-notes/current/performance_improvements/10022-endpoint-fast-path.md @@ -0,0 +1,6 @@ +Added an opt-in endpoint fast path (`EndpointFastPath` runtime flag): EndpointSlice +updates are propagated to Envoy as EDS-only pushes using cached per-cluster context +from the last successful translation, so endpoint freshness no longer waits for a +full translation. Endpoint changes that affect configuration (e.g. a backend's +endpoint count crossing zero, or clusters affected by EnvoyPatchPolicy or extension +server hooks) still take the full path, which remains authoritative. diff --git a/site/content/en/latest/api/extension_types.md b/site/content/en/latest/api/extension_types.md index 125f4a1f204..87c441ab245 100644 --- a/site/content/en/latest/api/extension_types.md +++ b/site/content/en/latest/api/extension_types.md @@ -6132,6 +6132,7 @@ _Appears in:_ | `XDSNameSchemeV2` | XDSNameSchemeV2 indicates that the xds name scheme v2 is used.
* The listener name will be generated using the protocol and port of the listener.
| | `EndpointSliceIndex` | EndpointSliceIndex indicates that field indexes are used to look up EndpointSlices by backend.
It is enabled by default to reduce CPU usage for EndpointSlice lookups in large clusters.
If the additional controller memory usage for the indexes becomes a concern,
consider disabling this flag.
| | `PerResourceSystemCASecret` | PerResourceSystemCASecret restores the pre-1.x behavior of emitting one SDS secret per
BackendTLSPolicy or Backend resource that uses WellKnownCACertificates: System, instead
of sharing a single system_ca_certificates secret across all of them.
Disabled by default (i.e. the shared secret is used). Enable this flag to opt out during
upgrades — Envoy must warm the new system_ca_certificates secret before clusters can use
it, which may cause a brief disruption to new connections on first enable.
| +| `EndpointFastPath` | EndpointFastPath enables propagating EndpointSlice updates to Envoy as EDS-only
pushes without waiting for a full translation: endpoint changes are patched into
the current xDS snapshot using cached per-cluster context from the last successful
full translation. Endpoint changes that affect configuration (e.g. a backend's
endpoint count crossing zero) still take the full path, and full translations
remain authoritative. Disabled by default.
| #### RuntimeFlags diff --git a/site/package.json b/site/package.json index df796d7df9f..efd79a61621 100644 --- a/site/package.json +++ b/site/package.json @@ -8,10 +8,10 @@ "author": "Docsy Authors", "license": "Apache-2.0", "bugs": "https://github.com/google/docsy-example/issues", - "spelling": "cSpell:ignore docsy hugo htmltest precheck postbuild rtlcss -", + "spelling": "cSpell:ignore docsy hugo linkinator precheck rtlcss -", "scripts": { "_build": "npm run _hugo-dev --", - "_check:links": "echo IMPLEMENTATION PENDING for check-links; echo", + "_check:links": "make -C .. docs-check-links", "_hugo": "hugo --cleanDestinationDir", "_hugo-dev": "npm run _hugo -- -e dev -DFE", "_local": "npx cross-env HUGO_MODULE_WORKSPACE=docsy.work", @@ -19,15 +19,13 @@ "build:preview": "npm run _hugo-dev -- --minify --baseURL \"${DEPLOY_PRIME_URL:-/}\"", "build:production": "npm run _hugo -- --minify", "build": "npm run _build -- ", - "check:links:all": "HTMLTEST_ARGS= npm run _check:links", + "check:links:all": "npm run _check:links", "check:links": "npm run _check:links", "clean": "rm -Rf public/* resources", "local": "npm run _local -- npm run", "make:public": "git init -b main public", "precheck:links:all": "npm run build", "precheck:links": "npm run build", - "postbuild:preview": "npm run _check:links", - "postbuild:production": "npm run _check:links", "serve": "npm run _serve", "test": "npm run check:links", "update:dep": "npm install --save-dev autoprefixer@latest postcss-cli@latest", From b8c26f530fad3d2caaca93fd9b2f092d485b994c Mon Sep 17 00:00:00 2001 From: "Huabing (Robin) Zhao" Date: Wed, 16 Sep 2026 23:22:43 -0700 Subject: [PATCH 2/4] refactor: keep full snapshot publication in xDS runner Signed-off-by: Huabing (Robin) Zhao --- internal/xds/cache/snapshotcache.go | 23 ++++------- internal/xds/cache/snapshotcache_test.go | 3 +- internal/xds/runner/endpoint_fast_path.go | 22 +++------- .../xds/runner/endpoint_fast_path_test.go | 41 ++++++++++++++++--- internal/xds/runner/runner.go | 36 ++++++++++------ 5 files changed, 71 insertions(+), 54 deletions(-) diff --git a/internal/xds/cache/snapshotcache.go b/internal/xds/cache/snapshotcache.go index 80f3925ec16..60fe9212441 100644 --- a/internal/xds/cache/snapshotcache.go +++ b/internal/xds/cache/snapshotcache.go @@ -54,7 +54,7 @@ var ( type SnapshotCacheWithCallbacks interface { cachev3.SnapshotCache serverv3.Callbacks - GenerateNewSnapshot(string, types.XdsResources, context.Context) (bool, error) + GenerateNewSnapshot(string, types.XdsResources, context.Context) error UpdateEndpointResources(string, []envoytypes.Resource) error SnapshotHasIrKey(string) bool GetIrKeys() []string @@ -86,12 +86,7 @@ type snapshotCache struct { // GenerateNewSnapshot takes a table of resources (the output from the IR->xDS // translator) and updates the snapshot version. -// -// The returned bool reports whether the snapshot was committed. It is false only -// when building it failed, leaving the previous snapshot untouched; it is true -// even when publishing to a node then failed, so callers can keep state in sync -// with the committed snapshot while still surfacing the error. -func (s *snapshotCache) GenerateNewSnapshot(irKey string, resources types.XdsResources, ctx context.Context) (bool, error) { +func (s *snapshotCache) GenerateNewSnapshot(irKey string, resources types.XdsResources, ctx context.Context) error { s.mu.Lock() defer s.mu.Unlock() @@ -111,11 +106,10 @@ func (s *snapshotCache) GenerateNewSnapshot(irKey string, resources types.XdsRes ) if err != nil { xdsSnapshotCreateTotal.WithFailure(metrics.ReasonError).Increment() - return false, err + return err } xdsSnapshotCreateTotal.WithSuccess().Increment() - // Commit point. // Delete snapshot from cache if resources are nil if resources == nil { delete(s.lastSnapshot, irKey) @@ -124,21 +118,18 @@ func (s *snapshotCache) GenerateNewSnapshot(irKey string, resources types.XdsRes s.lastSnapshot[irKey] = snapshot } - // The snapshot is committed above, so a per-node publish failure is - // reported with committed=true: the error still propagates to the caller - // (and its error channel), while a node whose publish failed is served the - // committed snapshot when it reconnects. for _, node := range s.getNodeIDs(irKey) { s.log.Debugf("Generating a snapshot with Node %s", node) if err = s.SetSnapshot(context.TODO(), node, snapshot); err != nil { xdsSnapshotUpdateTotal.WithFailure(metrics.ReasonError, nodeIDLabel.Value(node)).Increment() - return true, err + return err + } else { + xdsSnapshotUpdateTotal.WithSuccess(nodeIDLabel.Value(node)).Increment() } - xdsSnapshotUpdateTotal.WithSuccess(nodeIDLabel.Value(node)).Increment() } - return true, nil + return nil } // UpdateEndpointResources patches only the EDS resources of irKey's snapshot: the diff --git a/internal/xds/cache/snapshotcache_test.go b/internal/xds/cache/snapshotcache_test.go index 565690e3c59..f863bfd1f66 100644 --- a/internal/xds/cache/snapshotcache_test.go +++ b/internal/xds/cache/snapshotcache_test.go @@ -166,9 +166,8 @@ func TestUpdateEndpointResources(t *testing.T) { cla("cluster-b", "2.2.2.2"), }, } - committed, err := sc.GenerateNewSnapshot(irKey, resources, context.Background()) + err := sc.GenerateNewSnapshot(irKey, resources, context.Background()) require.NoError(t, err) - require.True(t, committed) oldSnapshot := sc.lastSnapshot[irKey] oldEDSVersion := oldSnapshot.GetVersion(resourcev3.EndpointType) diff --git a/internal/xds/runner/endpoint_fast_path.go b/internal/xds/runner/endpoint_fast_path.go index ae325321c91..22b81104b13 100644 --- a/internal/xds/runner/endpoint_fast_path.go +++ b/internal/xds/runner/endpoint_fast_path.go @@ -6,7 +6,6 @@ package runner import ( - "context" "errors" "reflect" "sync" @@ -121,14 +120,10 @@ func endpointSourceKey(es *ir.EndpointSource) string { return message.BackendKey(es.Kind, es.Namespace, es.Name) } -// OnFullBuild hands a completed translation to the snapshot cache. The endpoint -// state this path already published is folded into the build's own EDS resources -// first, so a build that raced an endpoint change does not regress it. The -// captured contexts replace the previous ones only if the snapshot committed. -func (f *endpointFastPath) OnFullBuild(irKey string, xdsIR *ir.Xds, table *xdstypes.ResourceVersionTable, ctx context.Context) error { - f.mu.Lock() - defer f.mu.Unlock() - +// prepareFullSnapshot prepares endpoint contexts for a completed translation and folds +// newer endpoint state into its EDS resources. The caller must hold f.mu through +// snapshot publication and clear the contexts if publication fails. +func (f *endpointFastPath) prepareFullSnapshot(irKey string, xdsIR *ir.Xds, table *xdstypes.ResourceVersionTable) { ctxs := f.buildContexts(xdsIR, table) if ctxs.enabled { for backendKey := range ctxs.byBackend { @@ -150,14 +145,7 @@ func (f *endpointFastPath) OnFullBuild(irKey string, xdsIR *ir.Xds, table *xdsty } } - committed, err := f.cache.GenerateNewSnapshot(irKey, table.XdsResources, ctx) - if committed { - // The snapshot is live, so the contexts must describe it even when - // publishing to some node failed — otherwise later patches would be - // built from the previous translation's settings. - f.contexts[irKey] = ctxs - } - return err + f.contexts[irKey] = ctxs } // replaceEndpointResources swaps the given ClusterLoadAssignments into the diff --git a/internal/xds/runner/endpoint_fast_path_test.go b/internal/xds/runner/endpoint_fast_path_test.go index ce7ddfbfce8..dfd96d6c806 100644 --- a/internal/xds/runner/endpoint_fast_path_test.go +++ b/internal/xds/runner/endpoint_fast_path_test.go @@ -36,6 +36,7 @@ type patchCall struct { // only the overridden methods may be called. type fakeCache struct { cache.SnapshotCacheWithCallbacks + snapshotErr error snapshots []string snapshotResources []xdstypes.XdsResources patches []patchCall @@ -46,10 +47,10 @@ type fakeCache struct { patchCommitted bool } -func (f *fakeCache) GenerateNewSnapshot(irKey string, resources xdstypes.XdsResources, _ context.Context) (bool, error) { +func (f *fakeCache) GenerateNewSnapshot(irKey string, resources xdstypes.XdsResources, _ context.Context) error { f.snapshots = append(f.snapshots, irKey) f.snapshotResources = append(f.snapshotResources, resources) - return true, nil + return f.snapshotErr } func (f *fakeCache) UpdateEndpointResources(irKey string, assignments []envoytypes.Resource) error { @@ -210,6 +211,7 @@ func publishFullBuildTable(ec *xdstypes.EndpointContext) *xdstypes.ResourceVersi func TestFastPathReappliesLatestEndpointsOnFullBuild(t *testing.T) { fc := &fakeCache{} fp := testFastPath(fc) + r := &Runner{cache: fc, endpointFastPath: fp} // The fast path has seen endpoint state newer than what the (racing) full // build translated. @@ -220,7 +222,7 @@ func TestFastPathReappliesLatestEndpointsOnFullBuild(t *testing.T) { // the state the fast path already published into the build's own EDS // resources, so the snapshot goes out fresh in a single push. table := publishFullBuildTable(testFastPathContext()) - require.NoError(t, fp.OnFullBuild("gw/eg", &ir.Xds{}, table, context.Background())) + require.NoError(t, r.generateSnapshot("gw/eg", &ir.Xds{}, table, context.Background())) require.Equal(t, []string{"gw/eg"}, fc.snapshots) eds := fc.snapshotResources[0][resourcev3.EndpointType] @@ -251,6 +253,7 @@ func TestFastPathKeepsContextWhenPatchCommittedWithError(t *testing.T) { func TestFastPathLeavesUncommittedPatchToTheFullPath(t *testing.T) { fc := &fakeCache{} fp := testFastPath(fc) + r := &Runner{cache: fc, endpointFastPath: fp} // Nothing was committed. The contexts still track the new endpoints, so the // fast path does not retry; the full build triggered by the same event is @@ -260,7 +263,7 @@ func TestFastPathLeavesUncommittedPatchToTheFullPath(t *testing.T) { require.Empty(t, fc.patches) table := publishFullBuildTable(testFastPathContext()) - require.NoError(t, fp.OnFullBuild("gw/eg", &ir.Xds{}, table, context.Background())) + require.NoError(t, r.generateSnapshot("gw/eg", &ir.Xds{}, table, context.Background())) require.ElementsMatch(t, []string{"2.2.2.2"}, claAddresses(t, fc.snapshotResources[0][resourcev3.EndpointType][0])) } @@ -268,6 +271,7 @@ func TestFastPathLeavesUncommittedPatchToTheFullPath(t *testing.T) { func TestFastPathRetainsEndpointsForBackendsNotYetInAnyContext(t *testing.T) { fc := &fakeCache{} fp := testFastPath(fc) + r := &Runner{cache: fc, endpointFastPath: fp} // An endpoint update arrives for a backend no committed context references // yet, because the build that will introduce it is still in flight. @@ -279,7 +283,7 @@ func TestFastPathRetainsEndpointsForBackendsNotYetInAnyContext(t *testing.T) { // An unrelated IR key's build commits. It must not drop the retained // update: the in-flight build still needs it. other := publishFullBuildTable(testFastPathContext()) - require.NoError(t, fp.OnFullBuild("gw/other", &ir.Xds{}, other, context.Background())) + require.NoError(t, r.generateSnapshot("gw/other", &ir.Xds{}, other, context.Background())) require.Contains(t, fp.lastEndpoints, pending.Key()) // When the in-flight build lands carrying older endpoints, the retained @@ -287,7 +291,7 @@ func TestFastPathRetainsEndpointsForBackendsNotYetInAnyContext(t *testing.T) { ec := testFastPathContext() ec.Settings[0].EndpointSource.Name = "pending-svc" table := publishFullBuildTable(ec) - require.NoError(t, fp.OnFullBuild("gw/pending", &ir.Xds{}, table, context.Background())) + require.NoError(t, r.generateSnapshot("gw/pending", &ir.Xds{}, table, context.Background())) eds := fc.snapshotResources[len(fc.snapshotResources)-1][resourcev3.EndpointType] require.Len(t, eds, 1) @@ -302,3 +306,28 @@ func TestFastPathOnDelete(t *testing.T) { fp.handleUpdate(testUpdate("2.2.2.2")) require.Empty(t, fc.patches) } + +func TestFastPathFullBuildFailureDisablesPatchesUntilRecovery(t *testing.T) { + fc := &fakeCache{} + fp := testFastPath(fc) + r := &Runner{cache: fc, endpointFastPath: fp} + + fc.snapshotErr = context.DeadlineExceeded + table := publishFullBuildTable(testFastPathContext()) + require.ErrorIs(t, r.generateSnapshot("gw/eg", &ir.Xds{}, table, context.Background()), fc.snapshotErr) + require.NotContains(t, fp.contexts, "gw/eg") + + fp.handleUpdate(testUpdate("2.2.2.2")) + require.Empty(t, fc.patches) + + fc.snapshotErr = nil + table = publishFullBuildTable(testFastPathContext()) + require.NoError(t, r.generateSnapshot("gw/eg", &ir.Xds{}, table, context.Background())) + require.Contains(t, fp.contexts, "gw/eg") + require.ElementsMatch(t, []string{"2.2.2.2"}, + claAddresses(t, fc.snapshotResources[len(fc.snapshotResources)-1][resourcev3.EndpointType][0])) + + fp.handleUpdate(testUpdate("3.3.3.3")) + require.Len(t, fc.patches, 1) + require.ElementsMatch(t, []string{"3.3.3.3"}, claAddresses(t, fc.patches[0].assignments[0])) +} diff --git a/internal/xds/runner/runner.go b/internal/xds/runner/runner.go index 299c725db8c..7a224a63f1f 100644 --- a/internal/xds/runner/runner.go +++ b/internal/xds/runner/runner.go @@ -38,6 +38,7 @@ import ( extension "github.com/envoyproxy/gateway/internal/extension/types" "github.com/envoyproxy/gateway/internal/infrastructure/host" "github.com/envoyproxy/gateway/internal/infrastructure/kubernetes/ratelimit" + "github.com/envoyproxy/gateway/internal/ir" "github.com/envoyproxy/gateway/internal/message" "github.com/envoyproxy/gateway/internal/xds/bootstrap" "github.com/envoyproxy/gateway/internal/xds/cache" @@ -318,7 +319,7 @@ func (r *Runner) translateFromSubscription(sub <-chan watchable.Snapshot[string, if r.endpointFastPath != nil { r.endpointFastPath.OnDelete(key) } - if _, err := r.cache.GenerateNewSnapshot(key, nil, traceCtx); err != nil { + if err := r.cache.GenerateNewSnapshot(key, nil, traceCtx); err != nil { traceLogger.Error(err, "failed to delete the snapshot") errChan <- err } @@ -379,21 +380,12 @@ func (r *Runner) translateFromSubscription(sub <-chan watchable.Snapshot[string, // Note: invalid EnvoyPatchPolicies are considered user-level errors and will not prevent the snapshot from being updated. if err == nil { if result.XdsResources != nil { - switch { - case r.cache == nil: + switch r.cache { + case nil: r.Logger.Error(err, "failed to init snapshot cache") errChan <- err - case r.endpointFastPath != nil: - // Publish through the fast path so the snapshot, the - // endpoint contexts, and the re-applied endpoint state - // are swapped atomically with respect to endpoint patches. - if err := r.endpointFastPath.OnFullBuild(key, val.XdsIR, result, traceCtx); err != nil { - r.Logger.Error(err, "failed to generate a snapshot") - errChan <- err - } default: - // Update snapshot cache - if _, err := r.cache.GenerateNewSnapshot(key, result.XdsResources, traceCtx); err != nil { + if err := r.generateSnapshot(key, val.XdsIR, result, traceCtx); err != nil { r.Logger.Error(err, "failed to generate a snapshot") errChan <- err } @@ -438,6 +430,24 @@ func (r *Runner) translateFromSubscription(sub <-chan watchable.Snapshot[string, r.Logger.Info("subscriber shutting down") } +// generateSnapshot serializes full-snapshot publication with endpoint patches, +// keeping the snapshot and the fast-path contexts aligned. +func (r *Runner) generateSnapshot(irKey string, xdsIR *ir.Xds, table *xtypes.ResourceVersionTable, ctx context.Context) error { + if r.endpointFastPath != nil { + r.endpointFastPath.mu.Lock() + defer r.endpointFastPath.mu.Unlock() + r.endpointFastPath.prepareFullSnapshot(irKey, xdsIR, table) + } + + err := r.cache.GenerateNewSnapshot(irKey, table.XdsResources, ctx) + if err != nil && r.endpointFastPath != nil { + // Creation and publication failures can leave different snapshots live. + // Fall back to the full pipeline until a successful build restores contexts. + delete(r.endpointFastPath.contexts, irKey) + } + return err +} + func (r *Runner) loadTLSConfig() (*tls.Config, error) { var certPath, keyPath, caPath string From ab3fd75ae9944edf0ef7e1636d6b642818ee9c7e Mon Sep 17 00:00:00 2001 From: "Huabing (Robin) Zhao" Date: Thu, 17 Sep 2026 00:09:43 -0700 Subject: [PATCH 3/4] fix: scope endpoint fast path around endpoint patches Signed-off-by: Huabing (Robin) Zhao --- internal/xds/runner/endpoint_fast_path.go | 27 +++++++++++++++---- .../xds/runner/endpoint_fast_path_test.go | 4 ++- 2 files changed, 25 insertions(+), 6 deletions(-) diff --git a/internal/xds/runner/endpoint_fast_path.go b/internal/xds/runner/endpoint_fast_path.go index 22b81104b13..1b91fda6278 100644 --- a/internal/xds/runner/endpoint_fast_path.go +++ b/internal/xds/runner/endpoint_fast_path.go @@ -80,10 +80,11 @@ type endpointFastPath struct { // irKeyContexts is the endpoint context index for one IR key. type irKeyContexts struct { - // enabled is false when this IR key must not take the fast path — its IR - // contains EnvoyPatchPolicies, whose patches (and statuses) are only - // handled by the full translation. + // enabled is false when an extension can modify generated xDS resources. enabled bool + // patchedClusters contains EDS clusters modified by an applicable + // EnvoyPatchPolicy. Those clusters must use the full translation path. + patchedClusters map[string]struct{} // byBackend maps a backend key (message.EndpointUpdate.Key()) to the // endpoint contexts of the clusters fed by that backend. byBackend map[string][]*xdstypes.EndpointContext @@ -177,8 +178,9 @@ func (f *endpointFastPath) OnDelete(irKey string) { // captured contexts, without committing it. func (f *endpointFastPath) buildContexts(xdsIR *ir.Xds, table *xdstypes.ResourceVersionTable) *irKeyContexts { ctxs := &irKeyContexts{ - enabled: len(xdsIR.EnvoyPatchPolicies) == 0 && !f.extensionModifiesEndpoints, - byBackend: make(map[string][]*xdstypes.EndpointContext), + enabled: !f.extensionModifiesEndpoints, + patchedClusters: patchedEndpointClusters(xdsIR), + byBackend: make(map[string][]*xdstypes.EndpointContext), } for _, ec := range table.EndpointContexts { for _, ds := range ec.Settings { @@ -191,6 +193,18 @@ func (f *endpointFastPath) buildContexts(xdsIR *ir.Xds, table *xdstypes.Resource return ctxs } +func patchedEndpointClusters(xdsIR *ir.Xds) map[string]struct{} { + clusters := make(map[string]struct{}) + for _, policy := range xdsIR.EnvoyPatchPolicies { + for _, patch := range policy.JSONPatches { + if patch.Type == resourcev3.EndpointType && patch.Name != "" { + clusters[patch.Name] = struct{}{} + } + } + } + return clusters +} + // subscribe consumes endpoint updates until the subscription context is // cancelled. The EndpointUpdates map itself is never closed, because the // provider writes it from informer handlers that Close cannot synchronize with. @@ -287,6 +301,9 @@ func (f *endpointFastPath) patchContexts(ctxs *irKeyContexts, update *message.En if _, ok := seen[ec]; ok { continue } + if _, patched := ctxs.patchedClusters[ec.ClusterName]; patched { + continue + } seen[ec] = struct{}{} changed := false for _, ds := range ec.Settings { diff --git a/internal/xds/runner/endpoint_fast_path_test.go b/internal/xds/runner/endpoint_fast_path_test.go index dfd96d6c806..c2776a06f55 100644 --- a/internal/xds/runner/endpoint_fast_path_test.go +++ b/internal/xds/runner/endpoint_fast_path_test.go @@ -191,7 +191,9 @@ func TestFastPathDisabledForKeyWithEnvoyPatchPolicies(t *testing.T) { // for the key: patches could target ClusterLoadAssignments. table := &xdstypes.ResourceVersionTable{} table.AddEndpointContext(testFastPathContext()) - fp.contexts["gw/eg"] = fp.buildContexts(&ir.Xds{EnvoyPatchPolicies: []*ir.EnvoyPatchPolicy{{}}}, table) + fp.contexts["gw/eg"] = fp.buildContexts(&ir.Xds{EnvoyPatchPolicies: []*ir.EnvoyPatchPolicy{{ + JSONPatches: []*ir.JSONPatchConfig{{Type: resourcev3.EndpointType, Name: "cluster-a"}}, + }}}, table) fp.handleUpdate(testUpdate("2.2.2.2")) require.Empty(t, fc.patches) From 79972c1b404948b8384883f4533f867e4fb925be Mon Sep 17 00:00:00 2001 From: "Huabing (Robin) Zhao" Date: Thu, 17 Sep 2026 02:46:46 -0700 Subject: [PATCH 4/4] refactor: simplify EDS fast-path contexts Signed-off-by: Huabing (Robin) Zhao --- internal/xds/runner/endpoint_fast_path.go | 103 +++++++---------- .../xds/runner/endpoint_fast_path_test.go | 105 +++++++++++++----- internal/xds/runner/runner.go | 4 +- internal/xds/translator/endpoint_fast_path.go | 12 +- internal/xds/translator/translator.go | 4 +- internal/xds/types/endpointcontext.go | 26 +++-- internal/xds/types/resourceversiontable.go | 8 +- test/go.mod | 1 + test/go.sum | 6 + 9 files changed, 163 insertions(+), 106 deletions(-) diff --git a/internal/xds/runner/endpoint_fast_path.go b/internal/xds/runner/endpoint_fast_path.go index 1b91fda6278..2563cd9965a 100644 --- a/internal/xds/runner/endpoint_fast_path.go +++ b/internal/xds/runner/endpoint_fast_path.go @@ -8,6 +8,7 @@ package runner import ( "errors" "reflect" + "slices" "sync" envoytypes "github.com/envoyproxy/go-control-plane/pkg/cache/types" @@ -67,35 +68,22 @@ type endpointFastPath struct { // mu serializes full-snapshot publishes and fast-path patches, so a patch // can never interleave with a snapshot rebuild or its context swap. mu sync.Mutex - // contexts holds, per IR key, the endpoint contexts captured at the last - // successful full translation. - contexts map[string]*irKeyContexts + + // edsContextsByIRKey maps each IR key to its backend index of ClusterLoadAssignment + // contexts from the last successful full translation. + edsContextsByIRKey map[string]map[string][]*xdstypes.ClusterLoadAssignmentContext + // lastEndpoints holds the most recent endpoint update per backend key, so // it can be re-applied on top of a full snapshot that raced with it. - // Entries are dropped only when the provider deletes the backend. Pruning - // against the committed contexts here would race an in-flight build that - // is about to introduce a backend it has no context for yet. lastEndpoints map[string]*message.EndpointUpdate } -// irKeyContexts is the endpoint context index for one IR key. -type irKeyContexts struct { - // enabled is false when an extension can modify generated xDS resources. - enabled bool - // patchedClusters contains EDS clusters modified by an applicable - // EnvoyPatchPolicy. Those clusters must use the full translation path. - patchedClusters map[string]struct{} - // byBackend maps a backend key (message.EndpointUpdate.Key()) to the - // endpoint contexts of the clusters fed by that backend. - byBackend map[string][]*xdstypes.EndpointContext -} - func newEndpointFastPath(c cache.SnapshotCacheWithCallbacks, extMgr extension.Manager, logger logging.Logger) *endpointFastPath { return &endpointFastPath{ logger: logger, cache: c, extensionModifiesEndpoints: extensionModifiesEndpoints(extMgr), - contexts: make(map[string]*irKeyContexts), + edsContextsByIRKey: make(map[string]map[string][]*xdstypes.ClusterLoadAssignmentContext), lastEndpoints: make(map[string]*message.EndpointUpdate), } } @@ -121,18 +109,33 @@ func endpointSourceKey(es *ir.EndpointSource) string { return message.BackendKey(es.Kind, es.Namespace, es.Name) } -// prepareFullSnapshot prepares endpoint contexts for a completed translation and folds -// newer endpoint state into its EDS resources. The caller must hold f.mu through -// snapshot publication and clear the contexts if publication fails. -func (f *endpointFastPath) prepareFullSnapshot(irKey string, xdsIR *ir.Xds, table *xdstypes.ResourceVersionTable) { - ctxs := f.buildContexts(xdsIR, table) - if ctxs.enabled { - for backendKey := range ctxs.byBackend { +// updateEDSForFullSnapshot applies retained endpoint updates to a completed +// translation's EDS resources and replaces the stored EDS contexts. The caller +// must hold f.mu through snapshot publication and clear the contexts if publication fails. +func (f *endpointFastPath) updateEDSForFullSnapshot(irKey string, xdsIR *ir.Xds, table *xdstypes.ResourceVersionTable) { + ctxs := table.EDSContexts + // Exclude clusters targeted by endpoint patches before replaying retained + // updates so neither replay nor later fast-path updates overwrite policy changes. + if clustersWithPatches := clustersWithEndpointPatches(xdsIR); len(clustersWithPatches) > 0 { + for backendKey, contexts := range ctxs { + contexts = slices.DeleteFunc(contexts, func(ec *xdstypes.ClusterLoadAssignmentContext) bool { + _, hasEndpointPatch := clustersWithPatches[ec.ClusterName] + return hasEndpointPatch + }) + if len(contexts) == 0 { + delete(ctxs, backendKey) + } else { + ctxs[backendKey] = contexts + } + } + } + if !f.extensionModifiesEndpoints { + for backendKey := range ctxs { update, ok := f.lastEndpoints[backendKey] if !ok { continue } - // Mutations target the private contexts built above; skips (zero + // Mutations target the private contexts captured by the build; skips (zero // transition, address-type change) leave the build's own endpoint // state in place. assignments, skipReason := f.patchContexts(ctxs, update) @@ -146,7 +149,7 @@ func (f *endpointFastPath) prepareFullSnapshot(irKey string, xdsIR *ir.Xds, tabl } } - f.contexts[irKey] = ctxs + f.edsContextsByIRKey[irKey] = ctxs } // replaceEndpointResources swaps the given ClusterLoadAssignments into the @@ -167,33 +170,16 @@ func replaceEndpointResources(table *xdstypes.ResourceVersionTable, assignments } } -// OnDelete drops the endpoint contexts for a deleted IR key. +// OnDelete drops the ClusterLoadAssignment contexts for a deleted IR key. func (f *endpointFastPath) OnDelete(irKey string) { f.mu.Lock() defer f.mu.Unlock() - delete(f.contexts, irKey) + delete(f.edsContextsByIRKey, irKey) } -// buildContexts builds the endpoint context index from a full translation's -// captured contexts, without committing it. -func (f *endpointFastPath) buildContexts(xdsIR *ir.Xds, table *xdstypes.ResourceVersionTable) *irKeyContexts { - ctxs := &irKeyContexts{ - enabled: !f.extensionModifiesEndpoints, - patchedClusters: patchedEndpointClusters(xdsIR), - byBackend: make(map[string][]*xdstypes.EndpointContext), - } - for _, ec := range table.EndpointContexts { - for _, ds := range ec.Settings { - if ds.EndpointSource != nil { - key := endpointSourceKey(ds.EndpointSource) - ctxs.byBackend[key] = append(ctxs.byBackend[key], ec) - } - } - } - return ctxs -} - -func patchedEndpointClusters(xdsIR *ir.Xds) map[string]struct{} { +// clustersWithEndpointPatches returns the cluster names targeted by +// ClusterLoadAssignment patches in the IR's EnvoyPatchPolicies. +func clustersWithEndpointPatches(xdsIR *ir.Xds) map[string]struct{} { clusters := make(map[string]struct{}) for _, policy := range xdsIR.EnvoyPatchPolicies { for _, patch := range policy.JSONPatches { @@ -242,12 +228,12 @@ func (f *endpointFastPath) handleUpdate(update *message.EndpointUpdate) { f.lastEndpoints[backendKey] = update found := false - for irKey, ctxs := range f.contexts { - if _, ok := ctxs.byBackend[backendKey]; !ok { + for irKey, ctxs := range f.edsContextsByIRKey { + if _, ok := ctxs[backendKey]; !ok { continue } found = true - if !ctxs.enabled { + if f.extensionModifiesEndpoints { endpointFastPathSkippedTotal.With(skipReasonLabel.Value(skipReasonDisabledForKey)).Increment() continue } @@ -283,7 +269,7 @@ func (f *endpointFastPath) handleUpdate(update *message.EndpointUpdate) { // the rebuilt CLAs of the clusters that changed. A non-empty skip reason means // the contexts were left untouched: the change is not CLA-only and belongs to // the full path. Must be called with f.mu held. -func (f *endpointFastPath) patchContexts(ctxs *irKeyContexts, update *message.EndpointUpdate) ([]envoytypes.Resource, string) { +func (f *endpointFastPath) patchContexts(ctxs map[string][]*xdstypes.ClusterLoadAssignmentContext, update *message.EndpointUpdate) ([]envoytypes.Resource, string) { backendKey := update.Key() type pendingChange struct { @@ -293,17 +279,14 @@ func (f *endpointFastPath) patchContexts(ctxs *irKeyContexts, update *message.En } var ( pending []pendingChange - changedClusters []*xdstypes.EndpointContext - seen = make(map[*xdstypes.EndpointContext]struct{}) + changedClusters []*xdstypes.ClusterLoadAssignmentContext + seen = make(map[*xdstypes.ClusterLoadAssignmentContext]struct{}) ) - for _, ec := range ctxs.byBackend[backendKey] { + for _, ec := range ctxs[backendKey] { if _, ok := seen[ec]; ok { continue } - if _, patched := ctxs.patchedClusters[ec.ClusterName]; patched { - continue - } seen[ec] = struct{}{} changed := false for _, ds := range ec.Settings { diff --git a/internal/xds/runner/endpoint_fast_path_test.go b/internal/xds/runner/endpoint_fast_path_test.go index c2776a06f55..4e435d636fa 100644 --- a/internal/xds/runner/endpoint_fast_path_test.go +++ b/internal/xds/runner/endpoint_fast_path_test.go @@ -8,6 +8,7 @@ package runner import ( "context" "os" + "slices" "testing" endpointv3 "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3" @@ -80,9 +81,9 @@ func testEndpointSlice(svcName string, addresses ...string) *discoveryv1.Endpoin } } -func testFastPathContext() *xdstypes.EndpointContext { +func testFastPathContext() *xdstypes.ClusterLoadAssignmentContext { const clusterName = "cluster-a" - return &xdstypes.EndpointContext{ + return &xdstypes.ClusterLoadAssignmentContext{ ClusterName: clusterName, Settings: []*ir.DestinationSetting{{ Name: clusterName + "/backend/0", @@ -104,8 +105,8 @@ func testFastPathContext() *xdstypes.EndpointContext { func testFastPath(fc *fakeCache) *endpointFastPath { fp := newEndpointFastPath(fc, nil, logging.DefaultLogger(os.Stderr, egv1a1.LogLevelInfo)) table := &xdstypes.ResourceVersionTable{} - table.AddEndpointContext(testFastPathContext()) - fp.contexts["gw/eg"] = fp.buildContexts(&ir.Xds{}, table) + table.AddClusterLoadAssignmentContext(testFastPathContext()) + fp.updateEDSForFullSnapshot("gw/eg", &ir.Xds{}, table) return fp } @@ -158,7 +159,7 @@ func TestFastPathSkipsZeroEndpointTransition(t *testing.T) { require.Empty(t, fc.patches) // The cached context still holds the last-applied endpoints. - require.Len(t, fp.contexts["gw/eg"].byBackend[update.Key()][0].Settings[0].Endpoints, 1) + require.Len(t, fp.edsContextsByIRKey["gw/eg"][update.Key()][0].Settings[0].Endpoints, 1) } func TestFastPathSkipsAddressTypeChange(t *testing.T) { @@ -183,27 +184,81 @@ func TestFastPathSkipsUnknownBackend(t *testing.T) { require.Empty(t, fc.patches) } -func TestFastPathDisabledForKeyWithEnvoyPatchPolicies(t *testing.T) { - fc := &fakeCache{} - fp := testFastPath(fc) - - // A full build whose IR carries EnvoyPatchPolicies disables the fast path - // for the key: patches could target ClusterLoadAssignments. - table := &xdstypes.ResourceVersionTable{} - table.AddEndpointContext(testFastPathContext()) - fp.contexts["gw/eg"] = fp.buildContexts(&ir.Xds{EnvoyPatchPolicies: []*ir.EnvoyPatchPolicy{{ - JSONPatches: []*ir.JSONPatchConfig{{Type: resourcev3.EndpointType, Name: "cluster-a"}}, - }}}, table) - - fp.handleUpdate(testUpdate("2.2.2.2")) - require.Empty(t, fc.patches) +func TestFastPathSkipsClustersWithEndpointPatches(t *testing.T) { + for _, tc := range []struct { + name string + patches []*ir.JSONPatchConfig + skipped []string + }{ + { + name: "matching endpoint patch", + patches: []*ir.JSONPatchConfig{{Type: resourcev3.EndpointType, Name: "cluster-a"}}, + skipped: []string{"cluster-a"}, + }, + { + name: "other resource type", + patches: []*ir.JSONPatchConfig{{Type: resourcev3.ClusterType, Name: "cluster-a"}}, + }, + { + name: "other cluster", + patches: []*ir.JSONPatchConfig{{Type: resourcev3.EndpointType, Name: "cluster-other"}}, + }, + { + name: "all clusters patched", + patches: []*ir.JSONPatchConfig{ + {Type: resourcev3.EndpointType, Name: "cluster-a"}, + {Type: resourcev3.EndpointType, Name: "cluster-b"}, + }, + skipped: []string{"cluster-a", "cluster-b"}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + fc := &fakeCache{} + fp := testFastPath(fc) + r := &Runner{cache: fc, endpointFastPath: fp} + retained := testUpdate("2.2.2.2") + fp.lastEndpoints[retained.Key()] = retained + + // Both clusters share a backend; only the targeted clusters must + // retain their full translation's endpoints during replay. + table := publishFullBuildTable(testFastPathContext()) + other := testFastPathContext() + other.ClusterName = "cluster-b" + table.AddClusterLoadAssignmentContext(other) + table.XdsResources[resourcev3.EndpointType] = append(table.XdsResources[resourcev3.EndpointType], translator.BuildClusterLoadAssignment(other)) + xdsIR := &ir.Xds{EnvoyPatchPolicies: []*ir.EnvoyPatchPolicy{{JSONPatches: tc.patches}}} + require.NoError(t, r.generateSnapshot("gw/eg", xdsIR, table, context.Background())) + for _, res := range fc.snapshotResources[0][resourcev3.EndpointType] { + cla := res.(*endpointv3.ClusterLoadAssignment) + if slices.Contains(tc.skipped, cla.ClusterName) { + require.Equal(t, []string{"1.1.1.1"}, claAddresses(t, res)) + } else { + require.Equal(t, []string{"2.2.2.2"}, claAddresses(t, res)) + } + } + + // Later updates must apply the same exclusions. + fp.handleUpdate(testUpdate("3.3.3.3")) + if len(tc.skipped) == 2 { + require.Empty(t, fc.patches) + require.NotContains(t, fp.edsContextsByIRKey["gw/eg"], retained.Key()) + return + } + require.Len(t, fc.patches, 1) + require.Len(t, fc.patches[0].assignments, 2-len(tc.skipped)) + for _, res := range fc.patches[0].assignments { + require.NotContains(t, tc.skipped, res.(*endpointv3.ClusterLoadAssignment).ClusterName) + require.Equal(t, []string{"3.3.3.3"}, claAddresses(t, res)) + } + }) + } } // publishFullBuildTable builds a translation table whose EDS resources match -// the endpoint context captured by the build (as a real translation would). -func publishFullBuildTable(ec *xdstypes.EndpointContext) *xdstypes.ResourceVersionTable { +// the ClusterLoadAssignment context captured by the build (as a real translation would). +func publishFullBuildTable(ec *xdstypes.ClusterLoadAssignmentContext) *xdstypes.ResourceVersionTable { table := &xdstypes.ResourceVersionTable{} - table.AddEndpointContext(ec) + table.AddClusterLoadAssignmentContext(ec) table.XdsResources = xdstypes.XdsResources{ resourcev3.EndpointType: {translator.BuildClusterLoadAssignment(ec)}, } @@ -247,7 +302,7 @@ func TestFastPathKeepsContextWhenPatchCommittedWithError(t *testing.T) { fp.handleUpdate(testUpdate("2.2.2.2")) require.Len(t, fc.patches, 1) - applied := fp.contexts["gw/eg"].byBackend[testUpdate().Key()][0].Settings[0].Endpoints + applied := fp.edsContextsByIRKey["gw/eg"][testUpdate().Key()][0].Settings[0].Endpoints require.Len(t, applied, 1) require.Equal(t, "2.2.2.2", applied[0].Host) } @@ -317,7 +372,7 @@ func TestFastPathFullBuildFailureDisablesPatchesUntilRecovery(t *testing.T) { fc.snapshotErr = context.DeadlineExceeded table := publishFullBuildTable(testFastPathContext()) require.ErrorIs(t, r.generateSnapshot("gw/eg", &ir.Xds{}, table, context.Background()), fc.snapshotErr) - require.NotContains(t, fp.contexts, "gw/eg") + require.NotContains(t, fp.edsContextsByIRKey, "gw/eg") fp.handleUpdate(testUpdate("2.2.2.2")) require.Empty(t, fc.patches) @@ -325,7 +380,7 @@ func TestFastPathFullBuildFailureDisablesPatchesUntilRecovery(t *testing.T) { fc.snapshotErr = nil table = publishFullBuildTable(testFastPathContext()) require.NoError(t, r.generateSnapshot("gw/eg", &ir.Xds{}, table, context.Background())) - require.Contains(t, fp.contexts, "gw/eg") + require.Contains(t, fp.edsContextsByIRKey, "gw/eg") require.ElementsMatch(t, []string{"2.2.2.2"}, claAddresses(t, fc.snapshotResources[len(fc.snapshotResources)-1][resourcev3.EndpointType][0])) diff --git a/internal/xds/runner/runner.go b/internal/xds/runner/runner.go index 7a224a63f1f..bd188e06560 100644 --- a/internal/xds/runner/runner.go +++ b/internal/xds/runner/runner.go @@ -436,14 +436,14 @@ func (r *Runner) generateSnapshot(irKey string, xdsIR *ir.Xds, table *xtypes.Res if r.endpointFastPath != nil { r.endpointFastPath.mu.Lock() defer r.endpointFastPath.mu.Unlock() - r.endpointFastPath.prepareFullSnapshot(irKey, xdsIR, table) + r.endpointFastPath.updateEDSForFullSnapshot(irKey, xdsIR, table) } err := r.cache.GenerateNewSnapshot(irKey, table.XdsResources, ctx) if err != nil && r.endpointFastPath != nil { // Creation and publication failures can leave different snapshots live. // Fall back to the full pipeline until a successful build restores contexts. - delete(r.endpointFastPath.contexts, irKey) + delete(r.endpointFastPath.edsContextsByIRKey, irKey) } return err } diff --git a/internal/xds/translator/endpoint_fast_path.go b/internal/xds/translator/endpoint_fast_path.go index ad890ede098..82de8128ebc 100644 --- a/internal/xds/translator/endpoint_fast_path.go +++ b/internal/xds/translator/endpoint_fast_path.go @@ -14,11 +14,11 @@ import ( "github.com/envoyproxy/gateway/internal/xds/types" ) -// BuildClusterLoadAssignment rebuilds a cluster's ClusterLoadAssignment from an -// endpoint context captured during full translation. It is the endpoint fast +// BuildClusterLoadAssignment rebuilds a cluster's ClusterLoadAssignment from a +// ClusterLoadAssignment context captured during full translation. It is the endpoint fast // path's entry point into the same CLA construction used by full translation, // so both paths produce identical output for the same inputs. -func BuildClusterLoadAssignment(ec *types.EndpointContext) *endpointv3.ClusterLoadAssignment { +func BuildClusterLoadAssignment(ec *types.ClusterLoadAssignmentContext) *endpointv3.ClusterLoadAssignment { return buildXdsClusterLoadAssignment(ec.ClusterName, ec.Settings, ec.HealthCheck, ec.PreferLocal, ec.WeightedZones) } @@ -31,11 +31,11 @@ func endpointFastPathEnabled(settings []*ir.DestinationSetting) bool { }) } -// buildEndpointContext captures everything the endpoint fast path needs to rebuild +// buildClusterLoadAssignmentContext captures everything the endpoint fast path needs to rebuild // this cluster's CLA without a full translation, or nil when the fast path is off. // The context owns deep copies: the IR it is built from stays live in the watchable // layer, which compares it for equality on the next publish. -func buildEndpointContext(args *xdsClusterArgs, lb *ir.LoadBalancer) *types.EndpointContext { +func buildClusterLoadAssignmentContext(args *xdsClusterArgs, lb *ir.LoadBalancer) *types.ClusterLoadAssignmentContext { if !endpointFastPathEnabled(args.settings) { return nil } @@ -43,7 +43,7 @@ func buildEndpointContext(args *xdsClusterArgs, lb *ir.LoadBalancer) *types.Endp for i, s := range args.settings { settings[i] = s.DeepCopy() } - ec := &types.EndpointContext{ + ec := &types.ClusterLoadAssignmentContext{ ClusterName: args.name, Settings: settings, HealthCheck: args.healthCheck.DeepCopy(), diff --git a/internal/xds/translator/translator.go b/internal/xds/translator/translator.go index b49913dc1ac..cfa77bfd4c8 100644 --- a/internal/xds/translator/translator.go +++ b/internal/xds/translator/translator.go @@ -1353,8 +1353,8 @@ func addXdsCluster(tCtx *types.ResourceVersionTable, args *xdsClusterArgs) error // Remember how this CLA was built so the endpoint fast path can rebuild it // on an EndpointSlice change. Nil unless the EndpointFastPath flag is on and // the cluster is EndpointSlice-backed. - if ec := buildEndpointContext(args, &lb); ec != nil { - tCtx.AddEndpointContext(ec) + if ec := buildClusterLoadAssignmentContext(args, &lb); ec != nil { + tCtx.AddClusterLoadAssignmentContext(ec) } case EndpointTypeDNS: xdsCluster.LoadAssignment = xdsEndpoints diff --git a/internal/xds/types/endpointcontext.go b/internal/xds/types/endpointcontext.go index 802fa8078fd..da11727d4a9 100644 --- a/internal/xds/types/endpointcontext.go +++ b/internal/xds/types/endpointcontext.go @@ -7,12 +7,13 @@ package types import ( "github.com/envoyproxy/gateway/internal/ir" + "github.com/envoyproxy/gateway/internal/message" ) -// EndpointContext captures, for one EDS cluster, the inputs needed to rebuild its +// ClusterLoadAssignmentContext captures, for one EDS cluster, the inputs needed to rebuild its // ClusterLoadAssignment outside a full translation. It is recorded during full // translation and consumed by the xDS runner's endpoint fast path. -type EndpointContext struct { +type ClusterLoadAssignmentContext struct { // ClusterName is the xDS cluster (and CLA) name. ClusterName string // Settings are the cluster's destination settings from the translated IR. @@ -27,10 +28,21 @@ type EndpointContext struct { WeightedZones []ir.WeightedZoneConfig } -// AddEndpointContext records the endpoint context for an EDS cluster. -func (t *ResourceVersionTable) AddEndpointContext(ec *EndpointContext) { - if t.EndpointContexts == nil { - t.EndpointContexts = make(map[string]*EndpointContext) +// AddClusterLoadAssignmentContext records the context for an EDS cluster. +func (t *ResourceVersionTable) AddClusterLoadAssignmentContext(ec *ClusterLoadAssignmentContext) { + if t.EDSContexts == nil { + t.EDSContexts = make(map[string][]*ClusterLoadAssignmentContext) + } + seen := make(map[string]struct{}) + for _, setting := range ec.Settings { + if setting.EndpointSource == nil { + continue + } + key := message.BackendKey(setting.EndpointSource.Kind, setting.EndpointSource.Namespace, setting.EndpointSource.Name) + if _, ok := seen[key]; ok { + continue + } + seen[key] = struct{}{} + t.EDSContexts[key] = append(t.EDSContexts[key], ec) } - t.EndpointContexts[ec.ClusterName] = ec } diff --git a/internal/xds/types/resourceversiontable.go b/internal/xds/types/resourceversiontable.go index 4f20b8b5e00..4d140ddde22 100644 --- a/internal/xds/types/resourceversiontable.go +++ b/internal/xds/types/resourceversiontable.go @@ -33,10 +33,10 @@ type ResourceVersionTable struct { EnvoyPatchPolicyStatuses GlobalResourceStatus - // EndpointContexts maps an EDS cluster name to the context needed to rebuild - // its ClusterLoadAssignment on the endpoint fast path. Only populated when - // the EndpointFastPath runtime flag is enabled. - EndpointContexts map[string]*EndpointContext + // EDSContexts maps a backend key to the contexts needed to rebuild its + // ClusterLoadAssignments on the endpoint fast path. Only populated when the + // EndpointFastPath runtime flag is enabled. + EDSContexts map[string][]*ClusterLoadAssignmentContext } // GetXdsResources retrieves the translated xds resources saved in the translator context. diff --git a/test/go.mod b/test/go.mod index 080cb1d9770..499950a90bb 100644 --- a/test/go.mod +++ b/test/go.mod @@ -250,6 +250,7 @@ require ( github.com/spf13/pflag v1.0.10 // indirect github.com/spf13/viper v1.21.0 // indirect github.com/subosito/gotenv v1.6.0 // indirect + github.com/telepresenceio/watchable v0.0.0-20220726211108-9bb86f92afa7 // indirect github.com/tklauser/go-sysconf v0.4.0 // indirect github.com/tklauser/numcpus v0.12.0 // indirect github.com/vishvananda/netlink v1.3.1 // indirect diff --git a/test/go.sum b/test/go.sum index 2f4e81bb72e..39de69fc3cb 100644 --- a/test/go.sum +++ b/test/go.sum @@ -157,6 +157,8 @@ github.com/creack/pty v1.1.24 h1:bJrF4RRfyJnbTJqzRLHzcGaZK1NeM5kTC9jGgovnR1s= github.com/creack/pty v1.1.24/go.mod h1:08sCNb52WyoAwi2QDyzUCTgcvVFhUzewun7wtTfvcwE= github.com/cyphar/filepath-securejoin v0.7.0 h1:s0Y3ITPy6sQn5xt54DuYvTF8hu134ooYLUb58DX/HjE= github.com/cyphar/filepath-securejoin v0.7.0/go.mod h1:ymLGms/u3BYaviIiuKFnUx8EkQEZeK6cInNoAPJA3o4= +github.com/datawire/dlib v1.3.0 h1:KkmyXU1kwm3oPBk1ypR70YbcOlEXWzEbx5RE0iRXTGk= +github.com/datawire/dlib v1.3.0/go.mod h1:NiGDmetmbkBvtznpWSx6C0vA0s0LK9aHna3LJDqjruk= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= @@ -593,6 +595,10 @@ github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWD github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8= github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU= +github.com/telepresenceio/telepresence/rpc/v2 v2.6.8 h1:q5V85LBT9bA/c4YPa/kMvJGyKZDgBPJTftlAMqJx7j4= +github.com/telepresenceio/telepresence/rpc/v2 v2.6.8/go.mod h1:VlgfRoXaW6Tl8IZbHmMWhITne8HY09/wOFtABHGj3ic= +github.com/telepresenceio/watchable v0.0.0-20220726211108-9bb86f92afa7 h1:GMw3nEaOVyi+tNiGko5kAeRtoiEIpXNHmISyZ7fpw14= +github.com/telepresenceio/watchable v0.0.0-20220726211108-9bb86f92afa7/go.mod h1:ihJ97e2gsd8GuzFF/I3B1qcik3XZLpXjumQifXi8Slg= github.com/tklauser/go-sysconf v0.4.0 h1:7H0uAN+7RkwWRaxhYXDLqa5V3LPrJeV8wmD9dRUgPQU= github.com/tklauser/go-sysconf v0.4.0/go.mod h1:8mTNWyog7H+MpKijp4VmKJAd2bbYQ2zuUwkYRbUArPI= github.com/tklauser/numcpus v0.12.0 h1:NR85qdvHA9pFse3x3weVZ0r0ST8R6l5RHbZrlRaqob4=