upstream-grpc-go/internal/xds/resolver/xds_resolver.go

/*
 * Copyright 2019 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

// Package resolver implements the xds resolver, that does LDS and RDS to find
// the cluster to use.
package resolver

import (
	"context"
	"fmt"
	rand "math/rand/v2"
	"slices"
	"strings"
	"sync/atomic"

	"google.golang.org/grpc"
	estats "google.golang.org/grpc/experimental/stats"
	"google.golang.org/grpc/internal"
	"google.golang.org/grpc/internal/grpclog"
	"google.golang.org/grpc/internal/grpcsync"
	iresolver "google.golang.org/grpc/internal/resolver"
	"google.golang.org/grpc/internal/wrr"
	"google.golang.org/grpc/internal/xds/bootstrap"
	"google.golang.org/grpc/internal/xds/httpfilter"
	rinternal "google.golang.org/grpc/internal/xds/resolver/internal"
	"google.golang.org/grpc/internal/xds/xdsclient"
	"google.golang.org/grpc/internal/xds/xdsclient/xdsresource"
	"google.golang.org/grpc/internal/xds/xdsdepmgr"
	"google.golang.org/grpc/resolver"
)

// Scheme is the xDS resolver's scheme.
//
// TODO(easwars): Rename this package as xdsresolver so that this is accessed as
// xdsresolver.Scheme
const Scheme = "xds"

// newBuilderWithConfigForTesting creates a new xds resolver builder using a
// specific xds bootstrap config, so tests can use multiple xDS clients in
// different ClientConns at the same time. The builder creates a new pool with
// the provided config and a new xDS client in that pool.
func newBuilderWithConfigForTesting(config []byte) (resolver.Builder, error) {
	return &xdsResolverBuilder{
		newXDSClient: func(name string, mr estats.MetricsRecorder) (xdsclient.XDSClient, func(), error) {
			config, err := bootstrap.NewConfigFromContents(config)
			if err != nil {
				return nil, nil, err
			}
			pool := xdsclient.NewPool(config)
			return pool.NewClientForTesting(xdsclient.OptionsForTesting{
				Name:            name,
				MetricsRecorder: mr,
			})
		},
	}, nil
}

// newBuilderWithPoolForTesting creates a new xds resolver builder using the
// specific xds client pool, so that tests have complete control over the exact
// specific xds client pool being used.
func newBuilderWithPoolForTesting(pool *xdsclient.Pool) (resolver.Builder, error) {
	return &xdsResolverBuilder{
		newXDSClient: func(name string, mr estats.MetricsRecorder) (xdsclient.XDSClient, func(), error) {
			return pool.NewClientForTesting(xdsclient.OptionsForTesting{
				Name:            name,
				MetricsRecorder: mr,
			})
		},
	}, nil
}

// newBuilderWithClientForTesting creates a new xds resolver builder using the
// specific xDS client, so that tests have complete control over the exact
// specific xDS client being used.
func newBuilderWithClientForTesting(client xdsclient.XDSClient) (resolver.Builder, error) {
	return &xdsResolverBuilder{
		newXDSClient: func(string, estats.MetricsRecorder) (xdsclient.XDSClient, func(), error) {
			// Returning an empty close func here means that the responsibility
			// of closing the client lies with the caller.
			return client, func() {}, nil
		},
	}, nil
}

func init() {
	resolver.Register(&xdsResolverBuilder{})
	internal.NewXDSResolverWithConfigForTesting = newBuilderWithConfigForTesting
	internal.NewXDSResolverWithPoolForTesting = newBuilderWithPoolForTesting
	internal.NewXDSResolverWithClientForTesting = newBuilderWithClientForTesting

	rinternal.NewWRR = wrr.NewRandom
	rinternal.NewXDSClient = xdsclient.DefaultPool.NewClient
}

type xdsResolverBuilder struct {
	newXDSClient func(string, estats.MetricsRecorder) (xdsclient.XDSClient, func(), error)
}

// Build helps implement the resolver.Builder interface.
//
// The xds bootstrap process is performed (and a new xDS client is built) every
// time an xds resolver is built.
func (b *xdsResolverBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (_ resolver.Resolver, retErr error) {
	// Initialize the xDS client.
	newXDSClient := rinternal.NewXDSClient.(func(string, estats.MetricsRecorder) (xdsclient.XDSClient, func(), error))
	if b.newXDSClient != nil {
		newXDSClient = b.newXDSClient
	}
	client, xdsClientClose, err := newXDSClient(target.String(), opts.MetricsRecorder)
	if err != nil {
		return nil, fmt.Errorf("xds: failed to create xds-client: %v", err)
	}

	template, err := sanityChecksOnBootstrapConfig(target, client)
	if err != nil {
		xdsClientClose()
		return nil, err
	}
	ldsResourceName := bootstrap.PopulateResourceTemplate(template, target.Endpoint())

	ctx, cancel := context.WithCancel(context.Background())
	r := &xdsResolver{
		cc:              cc,
		xdsClient:       client,
		xdsClientClose:  xdsClientClose,
		activeClusters:  make(map[string]*clusterInfo),
		activePlugins:   make(map[string]*clusterInfo),
		httpFilters:     make(map[clientFilterKey]httpfilter.ClientFilter),
		channelID:       rand.Uint64(),
		ldsResourceName: ldsResourceName,
		target:          target.String(),
		metricsRecorder: opts.MetricsRecorder,

		// serializer used to synchronize the following:
		// - updates from the dependency manager. This could lead to generation
		// of new service config if resolution is complete.
		// - completion of an RPC to a removed cluster causing the associated
		// ref count to become zero, resulting in generation of new service
		// config.
		// - stopping of a config selector that results in generation of new
		// service config.
		serializer:       grpcsync.NewCallbackSerializer(ctx),
		serializerCancel: cancel,
	}
	r.logger = prefixLogger(r)
	r.logger.Infof("Creating resolver for target: %+v", target)

	dmSet := make(chan struct{})
	// Schedule a callback that blocks until r.dm is set i.e xdsdepmgr.New()
	// returns. This acts as a gatekeeper: even if dependency manager sends the
	// updates before the xdsdepmgr.New() has a chance to return, they will be
	// queued behind this blocker and processed only after initialization is
	// complete.
	r.serializer.TrySchedule(func(ctx context.Context) {
		select {
		case <-dmSet:
		case <-ctx.Done():
		}
	})
	r.dm = xdsdepmgr.New(r.ldsResourceName, opts.Authority, r.xdsClient, r)
	close(dmSet)
	return r, nil
}

// Performs the following sanity checks:
//   - Verifies that the bootstrap configuration is not empty.
//   - Verifies that if xDS credentials are specified by the user, the
//     bootstrap configuration contains certificate providers.
//   - Verifies that if the provided dial target contains an authority, the
//     bootstrap configuration contains server config for that authority.
//
// Returns the listener resource name template to use. If any of the above
// validations fail, a non-nil error is returned.
func sanityChecksOnBootstrapConfig(target resolver.Target, client xdsclient.XDSClient) (string, error) {
	bootstrapConfig := client.BootstrapConfig()
	if bootstrapConfig == nil {
		// This is never expected to happen after a successful xDS client
		// creation. Defensive programming.
		return "", fmt.Errorf("xds: bootstrap configuration is empty")
	}

	// Find the client listener template to use from the bootstrap config:
	// - If authority is not set in the target, use the top level template
	// - If authority is set, use the template from the authority map.
	template := bootstrapConfig.ClientDefaultListenerResourceNameTemplate()
	if authority := target.URL.Host; authority != "" {
		authorities := bootstrapConfig.Authorities()
		if authorities == nil {
			return "", fmt.Errorf("xds: authority %q specified in dial target %q is not found in the bootstrap file", authority, target)
		}
		a := authorities[authority]
		if a == nil {
			return "", fmt.Errorf("xds: authority %q specified in dial target %q is not found in the bootstrap file", authority, target)
		}
		if a.ClientListenerResourceNameTemplate != "" {
			// This check will never be false, because
			// ClientListenerResourceNameTemplate is required to start with
			// xdstp://, and has a default value (not an empty string) if unset.
			template = a.ClientListenerResourceNameTemplate
		}
	}
	return template, nil
}

// Name helps implement the resolver.Builder interface.
func (*xdsResolverBuilder) Scheme() string {
	return Scheme
}

// xdsResolver implements the resolver.Resolver interface.
//
// It manages the dependency manager which in turn manages all xDS resource
// watches. It receives the xDS resource config and passes them to ClientConn.
type xdsResolver struct {
	// The following fields are initialized at creation time and are read-only
	// after that.
	cc              resolver.ClientConn
	logger          *grpclog.PrefixLogger
	ldsResourceName string
	dm              *xdsdepmgr.DependencyManager
	xdsClient       xdsclient.XDSClient
	xdsClientClose  func()
	channelID       uint64 // Unique random ID for the channel owning this resolver.
	target          string
	metricsRecorder estats.MetricsRecorder

	// All methods on the xdsResolver type except for the ones invoked by gRPC,
	// i.e ResolveNow() and Close(), are guaranteed to execute in the context of
	// this serializer's callback. We use the serializer because these shared
	// states are accessed by each RPC when it is committed, and so
	// serializer is preffered over a mutex.
	serializer       *grpcsync.CallbackSerializer
	serializerCancel context.CancelFunc

	// The following fields are accessed only from within the serializer
	// callbacks.
	xdsConfig *xdsresource.XDSConfig
	// activeClusters is a map from cluster name to information about the
	// weighted cluster that includes a reference count and load balancing
	// configuration. These counts are used only by the resolver. The current
	// configSelector holds one reference, and each ongoing RPC holds an
	// additional reference. When the count hits zero, the resolver removes the
	// cluster from this map and calls unsubscribe. This signals the dependency
	// manager to stop the xDS watch once its own reference count reaches zero.
	activeClusters map[string]*clusterInfo
	// activePlugins is a map from cluster specifier plugin name to information
	// about the cluster specifier plugin that includes a ref count and load
	// balancing configuration. These counts are used only by the resolver. The
	// current configSelector holds one reference, and each ongoing RPC holds an
	// additional reference. When the count hits zero, the resolver removes the
	// plugin name from this map.
	activePlugins     map[string]*clusterInfo
	curConfigSelector stoppableConfigSelector
	// httpFilters is a map from client filter key to client filter instance. It
	// lives here so that the resolver can reuse filter instances across config
	// updates when the same filter is specified, and to be able to clean up
	// filter instances that are no longer used.
	httpFilters map[clientFilterKey]httpfilter.ClientFilter
}

// ResolveNow calls RequestDNSReresolution on the dependency manager.
func (r *xdsResolver) ResolveNow(opts resolver.ResolveNowOptions) {
	if r.dm != nil {
		r.dm.RequestDNSReresolution(opts)
	}
}

func (r *xdsResolver) Close() {
	// Cancel the context passed to the serializer and wait for any scheduled
	// callbacks to complete. Canceling the context ensures that no new
	// callbacks will be scheduled.
	r.serializerCancel()
	<-r.serializer.Done()

	if r.dm != nil {
		r.dm.Close()
	}
	if r.curConfigSelector != nil {
		r.curConfigSelector.stop()
	}
	for _, cf := range r.httpFilters {
		cf.Close()
	}
	// Release the xDS client only after the filters are torn down: filters
	// may hold side channels whose credentials are owned by the client's
	// bootstrap config and are released when the last reference to the
	// client is dropped.
	if r.xdsClientClose != nil {
		r.xdsClientClose()
	}
	// Drop field references so the resolver isn't retained across ClientConn
	// close via any cross-object retention chain (e.g. ClientConn holding the
	// ServiceConfig that captures the configSelector, shared xDS-client
	// watcher metadata, channelz trace entries). Each nil individually looks
	// harmless; the combination breaks the multi-path retention observed when
	// callers repeatedly Close and redial DirectPath channels.
	r.cc = nil
	r.curConfigSelector = nil
	r.dm = nil
	r.xdsClient = nil
	r.xdsClientClose = nil
	r.httpFilters = nil
	r.activeClusters = nil
	r.activePlugins = nil
	r.xdsConfig = nil
	r.logger.Infof("Shutdown")
}

// Update is called when there is a new xDS config available from the dependency
// manager and does the following:
//   - creates a new config selector (this involves incrementing references to
//     clusters owned by this config selector).
//   - stops the old config selector (this involves decrementing references to
//     clusters owned by this config selector).
//   - prunes active clusters and pushes a new service config to the channel.
//   - updates the current config selector used by the resolver.
func (r *xdsResolver) Update(config *xdsresource.XDSConfig) {
	r.serializer.TrySchedule(func(context.Context) {
		r.xdsConfig = config
		cs, err := r.newConfigSelector()
		if err != nil {
			r.onResourceError(err)
			return
		}
		r.sendNewServiceConfig(cs)

		if r.curConfigSelector != nil {
			r.curConfigSelector.stop()
		}
		r.curConfigSelector = cs
	})
}

func (r *xdsResolver) Error(err error) {
	r.serializer.TrySchedule(func(context.Context) {
		r.onResourceError(err)
	})
}

// sendNewServiceConfig prunes active clusters, generates a new service config
// based on the current set of active clusters, and sends an update to the
// channel with that service config and the provided config selector. It is safe
// to ignore the error from `cc.UpdateState` because the clientconn uses the
// given configSelector even if it returns an error.
//
// Only executed in the context of a serializer callback.
func (r *xdsResolver) sendNewServiceConfig(cs stoppableConfigSelector) {
	// Delete entries from r.activeClusters with zero references;
	// otherwise serviceConfigJSON will generate a config including
	// them.
	r.pruneActiveClustersAndPlugins()

	if errCS, ok := cs.(*erroringConfigSelector); ok {
		// Send an empty config, which picks pick-first, with no address, and
		// puts the ClientConn into transient failure.
		//
		// This call to UpdateState is expected to return ErrBadResolverState
		// since pick_first doesn't like an update with no addresses.
		r.cc.UpdateState(resolver.State{ServiceConfig: r.cc.ParseServiceConfig("{}")})

		// Send a resolver error to pick_first so that RPCs will fail with a
		// more meaningful error, as opposed to one that says that pick_first
		// received no addresses.
		r.cc.ReportError(errCS.err)
		return
	}

	sc := serviceConfigJSON(r.activeClusters, r.activePlugins)
	if r.logger.V(2) {
		r.logger.Infof("For Listener resource %q and RouteConfiguration resource %q, generated service config: %s", r.ldsResourceName, r.xdsConfig.Listener.APIListener.RouteConfigName, string(sc))
	}

	// Send the update to the ClientConn.
	state := iresolver.SetConfigSelector(resolver.State{
		ServiceConfig: r.cc.ParseServiceConfig(string(sc)),
	}, cs)
	state = xdsresource.SetXDSConfig(state, r.xdsConfig)
	state = xdsdepmgr.SetXDSClusterSubscriber(state, r.dm)
	r.cc.UpdateState(xdsclient.SetClient(state, r.xdsClient))
}

// newConfigSelector creates a new config selector using the most recently
// received listener and route config updates. May add entries to
// r.activeClusters for previously-unseen clusters.
//
// Only executed in the context of a serializer callback.
func (r *xdsResolver) newConfigSelector() (_ *configSelector, err error) {
	cs := &configSelector{
		channelID: r.channelID,
		xdsNodeID: r.xdsClient.BootstrapConfig().Node().GetId(),
		sendNewServiceConfig: func() {
			r.serializer.TrySchedule(func(context.Context) {
				r.sendNewServiceConfig(r.curConfigSelector)
			})
		},
		virtualHost: virtualHost{
			retryConfig: r.xdsConfig.VirtualHost.RetryConfig,
		},
		routes:           make([]route, len(r.xdsConfig.VirtualHost.Routes)),
		clusters:         make(map[string]*clusterInfo),
		plugins:          make(map[string]*clusterInfo),
		httpFilterConfig: r.xdsConfig.Listener.APIListener.HTTPFilters,
		xdsConfig:        r.xdsConfig,
	}

	defer func() {
		if err != nil {
			// Stop the config selector if an error occurs during construction
			// to ensure that interceptors that were created successfully before
			// the error are cleaned up.
			cs.stop()
		}
	}()

	for i, rt := range r.xdsConfig.VirtualHost.Routes {
		clusters := rinternal.NewWRR.(func() wrr.WRR)()
		interceptors := []httpfilter.ClientInterceptor{}
		// TODO: Carve out the common logic between the ClusterSpecifierPlugin
		// and WeightedClusters.
		if rt.ClusterSpecifierPlugin != "" {
			clusterName := clusterSpecifierPluginPrefix + rt.ClusterSpecifierPlugin
			interceptor, err := r.newInterceptor(r.xdsConfig.Listener.APIListener.HTTPFilters, nil, rt.HTTPFilterConfigOverride, r.xdsConfig.VirtualHost.HTTPFilterConfigOverride)
			if err != nil {
				// Clean up any interceptors that were successfully built
				// for the current route before this error occurred. Note
				// that this is not handled by the call to cs.stop() in the
				// deferred function.
				for _, i := range interceptors {
					i.Close()
				}
				return nil, err
			}
			routeCluster := &routeCluster{
				name:        clusterName,
				interceptor: interceptor,
			}
			rc := grpcsync.NewRefCounted(routeCluster, func() { interceptor.Close() })
			cs.routes[i].routeClusters = append(cs.routes[i].routeClusters, rc)
			clusters.Add(rc, 1)
			ci := r.addOrGetActiveClusterInfo(clusterName, "")
			ci.cfg = xdsChildConfig{ChildPolicy: balancerConfig(r.xdsConfig.RouteConfig.ClusterSpecifierPlugins[rt.ClusterSpecifierPlugin])}
			cs.plugins[clusterName] = ci
		} else {
			for _, wc := range rt.WeightedClusters {
				clusterName := clusterPrefix + wc.Name
				interceptor, err := r.newInterceptor(r.xdsConfig.Listener.APIListener.HTTPFilters, wc.HTTPFilterConfigOverride, rt.HTTPFilterConfigOverride, r.xdsConfig.VirtualHost.HTTPFilterConfigOverride)
				if err != nil {
					// Clean up any interceptors that were successfully built
					// for the current route before this error occurred. Note
					// that this is not handled by the call to cs.stop() in the
					// deferred function.
					for _, i := range interceptors {
						i.Close()
					}
					return nil, err
				}
				routeCluster := &routeCluster{
					name:        clusterName,
					interceptor: interceptor,
				}
				rc := grpcsync.NewRefCounted(routeCluster, func() { interceptor.Close() })
				cs.routes[i].routeClusters = append(cs.routes[i].routeClusters, rc)
				clusters.Add(rc, int64(wc.Weight))
				ci := r.addOrGetActiveClusterInfo(clusterName, wc.Name)
				ci.cfg = xdsChildConfig{ChildPolicy: newBalancerConfig(cdsName, cdsBalancerConfig{Cluster: wc.Name})}
				cs.clusters[clusterName] = ci
			}
		}
		cs.routes[i].clusters = clusters
		cs.routes[i].m = xdsresource.RouteToMatcher(rt)
		cs.routes[i].actionType = rt.ActionType
		if rt.MaxStreamDuration == nil {
			cs.routes[i].maxStreamDuration = r.xdsConfig.Listener.APIListener.MaxStreamDuration
		} else {
			cs.routes[i].maxStreamDuration = *rt.MaxStreamDuration
		}

		cs.routes[i].retryConfig = rt.RetryConfig
		cs.routes[i].hashPolicies = rt.HashPolicies
		cs.routes[i].autoHostRewrite = rt.AutoHostRewrite
	}

	// Account for this config selector's clusters. Do this after no further
	// errors may occur. Note: cs.clusters are pointers to entries in
	// activeClusters.
	for _, ci := range cs.clusters {
		ci.refCount.Add(1)
	}
	for _, ci := range cs.plugins {
		ci.refCount.Add(1)
	}

	// Cleanup filter instances that are no longer specified in the current
	// listener resource.
	filtersInNewConfig := make(map[clientFilterKey]bool)
	for _, filter := range r.xdsConfig.Listener.APIListener.HTTPFilters {
		filtersInNewConfig[newClientFilterKey(&filter)] = true
	}
	for key, cf := range r.httpFilters {
		if _, ok := filtersInNewConfig[key]; ok {
			continue
		}
		cf.Close()
		delete(r.httpFilters, key)
	}

	return cs, nil
}

// pruneActiveClustersAndPlugins removes entries from activeClusters and
// activePlugins that have a reference count of zero. For clusters, it also
// invokes the unsubscribe function to signal the dependency manager to stop the
// xDS watch. Because cluster specifier plugins do not have their own watches,
// they are simply removed from the map without an unsubscribe call.
//
// Only executed in the context of a serializer callback.
func (r *xdsResolver) pruneActiveClustersAndPlugins() {
	for cluster, ci := range r.activeClusters {
		if ci.refCount.Load() == 0 {
			ci.unsubscribe()
			delete(r.activeClusters, cluster)
		}
	}
	for cluster, ci := range r.activePlugins {
		if ci.refCount.Load() == 0 {
			delete(r.activePlugins, cluster)
		}
	}
}

// addOrGetActiveClusterInfo returns the clusterInfo for the provided key,
// creating it if it does not exist. It accepts the following parameters:
//   - key: Formatted as "cluster:<name>" or "cluster_specifier_plugin:<name>",
//     this is the lookup key for the activeClusters or activePlugins maps.
//   - name: The actual xDS resource name used to initiate a CDS watch.
//     If empty (e.g., for plugins), no resource watch is triggered.
//
// This function manages entry creation and xDS subscriptions but does not
// increment the reference count of the returned clusterInfo.
func (r *xdsResolver) addOrGetActiveClusterInfo(key string, name string) *clusterInfo {
	if name == "" {
		ci, ok := r.activePlugins[key]
		if !ok {
			ci = &clusterInfo{}
			r.activePlugins[key] = ci
		}
		return ci
	}
	ci, ok := r.activeClusters[key]
	if !ok {
		ci = &clusterInfo{unsubscribe: r.dm.SubscribeToCluster(name)}
		r.activeClusters[key] = ci
	}
	return ci
}

type clusterInfo struct {
	// refCount is the number of references to this cluster.
	refCount atomic.Int32
	// cfg is the child configuration for this cluster, containing either the
	// csp config or the cds cluster config.
	cfg xdsChildConfig
	// unsubscribe is the function to call to unsubscribe from this cluster's
	// CDS resource. It is populated only for clusters in activeClusters and not
	// for cluster specifier plugins. When invoked, it decrements the reference
	// count in the dependency manager; once that count reaches zero, the
	// underlying CDS watch is terminated. Plugins do not have associated
	// watches and therefore do not require an unsubscribe function.
	unsubscribe func()
}

// Contains common functionality to be executed when resources of either type
// are removed.
//
// Only executed in the context of a serializer callback.
func (r *xdsResolver) onResourceError(err error) {
	// We cannot remove clusters from the service config that have ongoing RPCs.
	// Instead, what we can do is to send an erroring config selector
	// along with normal service config. This will ensure that new RPCs will
	// fail, and once the active RPCs complete, the reference counts on the
	// clusters will come down to zero. At that point, we will send an empty
	// service config with no addresses. This results in the pick-first
	// LB policy being configured on the channel, and since there are no
	// address, pick-first will put the channel in TRANSIENT_FAILURE.
	cs := newErroringConfigSelector(err, r.xdsClient.BootstrapConfig().Node().GetId())
	r.sendNewServiceConfig(cs)

	// Stop and dereference the active config selector, if one exists.
	if r.curConfigSelector != nil {
		r.curConfigSelector.stop()
	}
	r.curConfigSelector = cs
}

// newInterceptor builds a chain of client interceptors for the given filters
// and override configuration. The cluster override has the highest priority,
// followed by the route override, and finally the virtual host override.
//
// Only executed in the context of a serializer callback.
func (r *xdsResolver) newInterceptor(filters []xdsresource.HTTPFilter, clusterOverride, routeOverride, virtualHostOverride map[string]httpfilter.FilterConfig) (_ httpfilter.ClientInterceptor, err error) {
	interceptors := make([]httpfilter.ClientInterceptor, 0, len(filters))
	defer func() {
		// Clean up any interceptors that were successfully built before the
		// error occurred, to avoid leaking resources.
		if err != nil {
			for _, i := range interceptors {
				i.Close()
			}
		}
	}()
	for _, filter := range filters {
		override := clusterOverride[filter.Name]
		if override == nil {
			override = routeOverride[filter.Name]
		}
		if override == nil {
			override = virtualHostOverride[filter.Name]
		}

		// Determine the effective disabled state of the filter. The base
		// configuration's disabled state is used unless an override is present.
		// If an override is present, the filter is disabled if the override is
		// a DisabledFilterConfig.
		disabled := filter.Disabled
		if override != nil {
			_, disabled = override.(httpfilter.DisabledFilterConfig)
		}

		if disabled {
			if r.logger.V(2) {
				r.logger.Infof("Filter %q has been disabled.", filter.Name)
			}
			continue
		}

		builder, ok := filter.Filter.(httpfilter.ClientFilterBuilder)
		if !ok {
			// Should not happen if it passed xdsClient validation.
			return nil, fmt.Errorf("filter %q does not support use in client", filter.Name)
		}

		clientFilter := r.getOrCreateClientFilter(builder, newClientFilterKey(&filter))
		i, err := clientFilter.BuildClientInterceptor(filter.Config, override)
		if err != nil {
			return nil, fmt.Errorf("failed to build client interceptor for filter %q: %v", filter.Name, err)
		}
		if i != nil {
			interceptors = append(interceptors, i)
		}
	}

	return &interceptorList{interceptors: interceptors}, nil
}

// interceptorList is a client interceptor that contains a list of client
// interceptors to execute in order.
type interceptorList struct {
	interceptors []httpfilter.ClientInterceptor
}

func (il *interceptorList) NewStream(ctx context.Context, ri iresolver.RPCInfo, newStream func(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStream, error), opts ...grpc.CallOption) (grpc.ClientStream, error) {
	for idx := len(il.interceptors) - 1; idx >= 0; idx-- {
		ns := newStream
		i := il.interceptors[idx]
		newStream = func(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStream, error) {
			return i.NewStream(ctx, ri, ns, opts...)
		}
	}
	return newStream(ctx, opts...)
}

func (il *interceptorList) Close() {
	for _, i := range il.interceptors {
		i.Close()
	}
}

// getOrCreateClientFilter retrieves an existing client filter from the
// httpFilters map or creates a new one if it doesn't exist. It uses the filter
// builder to create a new client filter and stores it in the httpFilters map
// for future use.
//
// Only executed in the context of a serializer callback.
func (r *xdsResolver) getOrCreateClientFilter(builder httpfilter.ClientFilterBuilder, key clientFilterKey) httpfilter.ClientFilter {
	clientFilter, ok := r.httpFilters[key]
	if ok {
		return clientFilter
	}

	cf := builder.BuildClientFilter(httpfilter.ClientFilterOptions{
		FilterName:      key.name,
		MetricsRecorder: r.metricsRecorder,
		Target:          r.target,
	})
	r.httpFilters[key] = cf
	return cf
}

// newClientFilterKey generates a key for the given filter using the filter name
// and type URLs. This is used for storing ClientFilters in a map.
func newClientFilterKey(f *xdsresource.HTTPFilter) clientFilterKey {
	typeURLs := slices.Clone(f.Filter.TypeURLs())
	slices.Sort(typeURLs)
	return clientFilterKey{
		name:     f.Name,
		typeURLs: strings.Join(typeURLs, ":"),
	}
}

func (f *clientFilterKey) String() string {
	return f.name + ":" + f.typeURLs
}

type clientFilterKey struct {
	name     string
	typeURLs string
}