diff --git a/api/v1alpha1/envoygateway_helpers.go b/api/v1alpha1/envoygateway_helpers.go index 661a7961330..3202f216de7 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, EnvoyProxyPatch: true, } diff --git a/api/v1alpha1/envoygateway_types.go b/api/v1alpha1/envoygateway_types.go index 47236688f6a..e088b1d7091 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;EnvoyProxyPatch;ApplyClientTrafficPolicyHTTP1SettingsToClientsOnly +// +kubebuilder:validation:Enum=XDSNameSchemeV2;EndpointSliceIndex;PerResourceSystemCASecret;EndpointFastPath;EnvoyProxyPatch;ApplyClientTrafficPolicyHTTP1SettingsToClientsOnly type RuntimeFlag string const ( @@ -192,6 +192,14 @@ const ( // 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" + // EnvoyProxyPatch enables applying the Kubernetes resource `patch` fields configured on // EnvoyProxy's Kubernetes provider settings. It is enabled by default to preserve // pre-existing behavior. Because EnvoyProxy is commonly namespace-scoped and 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 3bf59ac7819..b6dc7b7f15c 100644 --- a/internal/gatewayapi/route.go +++ b/internal/gatewayapi/route.go @@ -30,6 +30,7 @@ import ( "github.com/envoyproxy/gateway/internal/gatewayapi/resource" "github.com/envoyproxy/gateway/internal/gatewayapi/status" "github.com/envoyproxy/gateway/internal/ir" + endpointsutil "github.com/envoyproxy/gateway/internal/utils/endpoints" labelsutil "github.com/envoyproxy/gateway/internal/utils/labels" "github.com/envoyproxy/gateway/internal/utils/regex" ) @@ -3051,9 +3052,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)) @@ -3086,6 +3088,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 { @@ -3101,6 +3112,7 @@ func (t *Translator) processServiceImportDestinationSetting( Endpoints: endpoints, AddressType: addrType, Metadata: buildResourceMetadata(serviceImport, new(gwapiv1.SectionName(strconv.Itoa(int(*backendRef.Port))))), + EndpointSource: endpointSource, }, nil } @@ -3113,8 +3125,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 @@ -3158,6 +3171,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) @@ -3172,6 +3194,7 @@ func (t *Translator) processServiceDestinationSetting( AddressType: addrType, PreferLocal: processPreferLocalZone(service), Metadata: buildResourceMetadata(service, new(gwapiv1.SectionName(strconv.Itoa(int(*backendRef.Port))))), + EndpointSource: endpointSource, }, nil } @@ -3365,66 +3388,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 082d7e17f1e..42dc9624158 100644 --- a/internal/gatewayapi/runner/runner.go +++ b/internal/gatewayapi/runner/runner.go @@ -331,6 +331,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 5c91f13053d..ca15c84af29 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 d687432dbb4..012a51521ab 100644 --- a/internal/ir/xds.go +++ b/internal/ir/xds.go @@ -2502,6 +2502,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 87bcc6ba7e8..858a87bdfa2 100644 --- a/internal/ir/zz_generated.deepcopy.go +++ b/internal/ir/zz_generated.deepcopy.go @@ -1673,6 +1673,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. @@ -1752,6 +1757,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 8516c963c7e..61f962226c8 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 { @@ -295,6 +296,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 @@ -336,6 +409,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 { @@ -370,7 +450,7 @@ func (r *gatewayAPIReconciler) Reconcile(ctx context.Context, _ reconcile.Reques // - Referenced resources: Services, ServiceImports, EndpointSlices, Secrets, ConfigMaps ... gwcResources := make(resource.ControllerResources, 0, len(managedGCs)) for _, managedGC := range managedGCs { - gwcResource, err := r.reconcileGatewayClass(ctx, managedGC, gcStatusToDelete, logger) + gwcResource, err := r.reconcileGatewayClass(ctx, managedGC, gcStatusToDelete, referencedBackends, logger) if err != nil { // A transient error here aborts the entire Reconcile call: managedGCs after this // one are left unprocessed this pass, and GatewayAPIResources is never Store()'d @@ -380,6 +460,23 @@ func (r *gatewayAPIReconciler) Reconcile(ctx context.Context, _ reconcile.Reques gwcResources = append(gwcResources, gwcResource) } + // 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: // 1. ensure identical resources are not retranslated // and updates are avoided by the watchable layer @@ -414,7 +511,7 @@ func (r *gatewayAPIReconciler) Reconcile(ctx context.Context, _ reconcile.Reques // A transient error aborts processing for this GatewayClass and is returned to the caller. // The caller (Reconcile) in turn aborts the entire Reconcile call on any such error, so the // workqueue retries the whole batch - this mirrors the original inline behavior exactly. -func (r *gatewayAPIReconciler) reconcileGatewayClass(ctx context.Context, managedGC *gwapiv1.GatewayClass, gcStatusToDelete sets.Set[types.NamespacedName], logger logging.Logger) (*resource.Resources, error) { +func (r *gatewayAPIReconciler) reconcileGatewayClass(ctx context.Context, managedGC *gwapiv1.GatewayClass, gcStatusToDelete sets.Set[types.NamespacedName], referencedBackends sets.Set[string], logger logging.Logger) (*resource.Resources, error) { // Initialize resource types. gwcResource := resource.NewResources() gwcResource.GatewayClass = managedGC @@ -665,6 +762,19 @@ func (r *gatewayAPIReconciler) reconcileGatewayClass(ctx context.Context, manage } } + 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))) + } + } + return gwcResource, nil } @@ -2703,6 +2813,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..60fe9212441 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" @@ -52,10 +55,15 @@ type SnapshotCacheWithCallbacks interface { cachev3.SnapshotCache serverv3.Callbacks GenerateNewSnapshot(string, types.XdsResources, context.Context) 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 @@ -124,6 +132,56 @@ func (s *snapshotCache) GenerateNewSnapshot(irKey string, resources types.XdsRes return 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 + } + xdsSnapshotUpdateTotal.WithSuccess(nodeIDLabel.Value(node)).Increment() + } + + return nil +} + // newSnapshotVersion increments the current snapshotVersion // and returns as a string. func (s *snapshotCache) newSnapshotVersion() string { diff --git a/internal/xds/cache/snapshotcache_test.go b/internal/xds/cache/snapshotcache_test.go index 5b460d7e83d..f863bfd1f66 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,83 @@ 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"), + }, + } + err := sc.GenerateNewSnapshot(irKey, resources, context.Background()) + require.NoError(t, err) + + 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..2563cd9965a --- /dev/null +++ b/internal/xds/runner/endpoint_fast_path.go @@ -0,0 +1,339 @@ +// 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 ( + "errors" + "reflect" + "slices" + "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 + + // 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. + lastEndpoints map[string]*message.EndpointUpdate +} + +func newEndpointFastPath(c cache.SnapshotCacheWithCallbacks, extMgr extension.Manager, logger logging.Logger) *endpointFastPath { + return &endpointFastPath{ + logger: logger, + cache: c, + extensionModifiesEndpoints: extensionModifiesEndpoints(extMgr), + edsContextsByIRKey: make(map[string]map[string][]*xdstypes.ClusterLoadAssignmentContext), + 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) +} + +// 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 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) + 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) + } + } + + f.edsContextsByIRKey[irKey] = ctxs +} + +// 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 ClusterLoadAssignment contexts for a deleted IR key. +func (f *endpointFastPath) OnDelete(irKey string) { + f.mu.Lock() + defer f.mu.Unlock() + delete(f.edsContextsByIRKey, irKey) +} + +// 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 { + 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. +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.edsContextsByIRKey { + if _, ok := ctxs[backendKey]; !ok { + continue + } + found = true + if f.extensionModifiesEndpoints { + 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 map[string][]*xdstypes.ClusterLoadAssignmentContext, 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.ClusterLoadAssignmentContext + seen = make(map[*xdstypes.ClusterLoadAssignmentContext]struct{}) + ) + + for _, ec := range ctxs[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..4e435d636fa --- /dev/null +++ b/internal/xds/runner/endpoint_fast_path_test.go @@ -0,0 +1,390 @@ +// 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" + "slices" + "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 + snapshotErr error + 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) error { + f.snapshots = append(f.snapshots, irKey) + f.snapshotResources = append(f.snapshotResources, resources) + return f.snapshotErr +} + +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.ClusterLoadAssignmentContext { + const clusterName = "cluster-a" + return &xdstypes.ClusterLoadAssignmentContext{ + 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.AddClusterLoadAssignmentContext(testFastPathContext()) + fp.updateEDSForFullSnapshot("gw/eg", &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.edsContextsByIRKey["gw/eg"][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 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 ClusterLoadAssignment context captured by the build (as a real translation would). +func publishFullBuildTable(ec *xdstypes.ClusterLoadAssignmentContext) *xdstypes.ResourceVersionTable { + table := &xdstypes.ResourceVersionTable{} + table.AddClusterLoadAssignmentContext(ec) + table.XdsResources = xdstypes.XdsResources{ + resourcev3.EndpointType: {translator.BuildClusterLoadAssignment(ec)}, + } + return table +} + +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. + 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, r.generateSnapshot("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.edsContextsByIRKey["gw/eg"][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) + 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 + // 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, 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])) +} + +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. + 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, 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 + // update is patched into it rather than being lost. + ec := testFastPathContext() + ec.Settings[0].EndpointSource.Name = "pending-svc" + table := publishFullBuildTable(ec) + 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) + 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) +} + +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.edsContextsByIRKey, "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.edsContextsByIRKey, "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 f549f733cad..f42ebfeabf8 100644 --- a/internal/xds/runner/runner.go +++ b/internal/xds/runner/runner.go @@ -39,6 +39,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" @@ -93,6 +94,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 // done tracks goroutines started by Start so that Close can block until // they have all exited, ensuring shared state they write to is not @@ -161,6 +165,7 @@ func (r *Runner) Close() error { // 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. @@ -235,12 +240,28 @@ func (r *Runner) Start(ctx context.Context) error { 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) r.done.Go(func() { r.translateFromSubscription(sub) }) + + if r.endpointFastPath != nil { + epSub := r.ProviderResources.EndpointUpdates.Subscribe(ctx) + r.done.Go(func() { + r.endpointFastPath.subscribe(epSub, r.Name()) + }) + } r.Logger.Info("started") return err } @@ -310,6 +331,9 @@ func (r *Runner) translateFromSubscription(sub <-chan watchable.Snapshot[string, ) if update.Delete { + 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 @@ -371,12 +395,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 { - if r.cache == nil { + switch r.cache { + case nil: r.Logger.Error(err, "failed to init snapshot cache") errChan <- err - } else { - // Update snapshot cache - if err := r.cache.GenerateNewSnapshot(key, result.XdsResources, traceCtx); err != nil { + default: + if err := r.generateSnapshot(key, val.XdsIR, result, traceCtx); err != nil { r.Logger.Error(err, "failed to generate a snapshot") errChan <- err } @@ -421,6 +445,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.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.edsContextsByIRKey, irKey) + } + return err +} + func (r *Runner) loadTLSConfig() (*tls.Config, error) { var certPath, keyPath, caPath string diff --git a/internal/xds/translator/endpoint_fast_path.go b/internal/xds/translator/endpoint_fast_path.go new file mode 100644 index 00000000000..82de8128ebc --- /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 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.ClusterLoadAssignmentContext) *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 + }) +} + +// 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 buildClusterLoadAssignmentContext(args *xdsClusterArgs, lb *ir.LoadBalancer) *types.ClusterLoadAssignmentContext { + 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.ClusterLoadAssignmentContext{ + 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 54766940ddc..00afe210801 100644 --- a/internal/xds/translator/translator.go +++ b/internal/xds/translator/translator.go @@ -1430,6 +1430,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 := buildClusterLoadAssignmentContext(args, &lb); ec != nil { + tCtx.AddClusterLoadAssignmentContext(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..da11727d4a9 --- /dev/null +++ b/internal/xds/types/endpointcontext.go @@ -0,0 +1,48 @@ +// 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" + "github.com/envoyproxy/gateway/internal/message" +) + +// 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 ClusterLoadAssignmentContext 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 +} + +// 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) + } +} diff --git a/internal/xds/types/resourceversiontable.go b/internal/xds/types/resourceversiontable.go index ad0124914be..4d140ddde22 100644 --- a/internal/xds/types/resourceversiontable.go +++ b/internal/xds/types/resourceversiontable.go @@ -32,6 +32,11 @@ type ResourceVersionTable struct { XdsResources EnvoyPatchPolicyStatuses GlobalResourceStatus + + // 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/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 1b6da23c65f..94c73215ee6 100644 --- a/site/content/en/latest/api/extension_types.md +++ b/site/content/en/latest/api/extension_types.md @@ -6447,6 +6447,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.
| | `EnvoyProxyPatch` | EnvoyProxyPatch enables applying the Kubernetes resource `patch` fields configured on
EnvoyProxy's Kubernetes provider settings. It is enabled by default to preserve
pre-existing behavior. Because EnvoyProxy is commonly namespace-scoped and
tenant-authored, a patch may grant arbitrary access to resources applied by Envoy
Gateway's more privileged ServiceAccount; disable this flag in multi-tenant clusters
where tenants can author their own EnvoyProxy resources.
| | `ApplyClientTrafficPolicyHTTP1SettingsToClientsOnly` | ApplyClientTrafficPolicyHTTP1SettingsToClientsOnly restricts the ClientTrafficPolicy http1 fields
(enableTrailers, preserveHeaderCase, http10) to client connections only.
When enabled, these fields no longer affect backend HTTP/1 protocol settings.
Use BackendTrafficPolicy.http1 to configure backend HTTP/1 protocol settings.
Disabled by default to preserve existing behavior. Users are encouraged to migrate
to BackendTrafficPolicy.http1 for backend settings and enable this flag to opt in
to the new behavior. This flag will be enabled by default in a future release.
| diff --git a/site/package.json b/site/package.json index ec886944104..34007ffb5ed 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", diff --git a/test/go.mod b/test/go.mod index 9fecd7e05c2..68ef914c9b0 100644 --- a/test/go.mod +++ b/test/go.mod @@ -251,6 +251,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 69d071f5c6f..8948f8b16a7 100644 --- a/test/go.sum +++ b/test/go.sum @@ -155,6 +155,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= @@ -589,6 +591,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=