internal/xds: move the LDS and RDS watchers to dependency manager

internal/grpctest/tlogger.go

@@ -66,10 +66,10 @@ type tLogger struct {
 	v           int
 	initialized bool
 
-	mu     sync.Mutex // guards t, start, and errors
-	t      *testing.T
-	start  time.Time
-	errors map[*regexp.Regexp]int
+	mu    sync.Mutex
+	t     *testing.T
+	start time.Time
+	logs  map[logType]map[*regexp.Regexp]int
 }
 
 func init() {
@@ -87,7 +87,11 @@ func init() {
 		}
 	}
 	// Initialize tLogr with the determined verbosity level.
-	tLogr = &tLogger{errors: make(map[*regexp.Regexp]int), v: vLevel}
+	logsMap := map[logType]map[*regexp.Regexp]int{
+		errorLog:   {},
+		warningLog: {},
+	}
+	tLogr = &tLogger{logs: logsMap, v: vLevel}
 }
 
 // getCallingPrefix returns the <file:line> at the given depth from the stack.
@@ -115,11 +119,14 @@ func (tl *tLogger) log(ltype logType, depth int, format string, args ...any) {
 		switch ltype {
 		case errorLog:
 			// fmt.Sprintln is used rather than fmt.Sprint because tl.Log uses fmt.Sprintln behavior.
-			if tl.expected(fmt.Sprintln(args...)) {
+			if tl.expected(fmt.Sprintln(args...), errorLog) {
 				tl.t.Log(args...)
 			} else {
 				tl.t.Error(args...)
 			}
+		case warningLog:
+			tl.expected(fmt.Sprintln(args...), warningLog)
+			tl.t.Log(args...)
 		case fatalLog:
 			panic(fmt.Sprint(args...))
 		default:
@@ -130,11 +137,14 @@ func (tl *tLogger) log(ltype logType, depth int, format string, args ...any) {
 		format = "%v " + format + "%s"
 		switch ltype {
 		case errorLog:
-			if tl.expected(fmt.Sprintf(format, args...)) {
+			if tl.expected(fmt.Sprintf(format, args...), errorLog) {
 				tl.t.Logf(format, args...)
 			} else {
 				tl.t.Errorf(format, args...)
 			}
+		case warningLog:
+			tl.expected(fmt.Sprintln(args...), warningLog)
+			tl.t.Log(args...)
 		case fatalLog:
 			panic(fmt.Sprintf(format, args...))
 		default:
@@ -154,7 +164,8 @@ func (tl *tLogger) update(t *testing.T) {
 	}
 	tl.t = t
 	tl.start = time.Now()
-	tl.errors = map[*regexp.Regexp]int{}
+	tl.logs[errorLog] = map[*regexp.Regexp]int{}
+	tl.logs[warningLog] = map[*regexp.Regexp]int{}
 }
 
 // ExpectError declares an error to be expected. For the next test, the first
@@ -163,40 +174,56 @@ func (tl *tLogger) update(t *testing.T) {
 // Update(). Note that if an expected error is not encountered, this will cause
 // the test to fail.
 func ExpectError(expr string) {
-	ExpectErrorN(expr, 1)
+	expectLogsN(expr, 1, errorLog)
 }
 
 // ExpectErrorN declares an error to be expected n times.
 func ExpectErrorN(expr string, n int) {
+	expectLogsN(expr, n, errorLog)
+}
+
+// ExpectWarning declares a warning to be expected.
+func ExpectWarning(expr string) {
+	expectLogsN(expr, 1, warningLog)
+}
+
+func expectLogsN(expr string, n int, logType logType) {
 	tLogr.mu.Lock()
 	defer tLogr.mu.Unlock()
 	re, err := regexp.Compile(expr)
 	if err != nil {
 		tLogr.t.Error(err)
 		return
 	}
-	tLogr.errors[re] += n
+	tLogr.logs[logType][re] += n
 }
 
 // endTest checks if expected errors were not encountered.
 func (tl *tLogger) endTest(t *testing.T) {
 	tl.mu.Lock()
 	defer tl.mu.Unlock()
-	for re, count := range tl.errors {
+	for re, count := range tl.logs[errorLog] {
 		if count > 0 {
 			t.Errorf("Expected error '%v' not encountered", re.String())
 		}
 	}
-	tl.errors = map[*regexp.Regexp]int{}
+	for re, count := range tl.logs[warningLog] {
+		if count > 0 {
+			t.Errorf("Expected warning '%v' not encountered", re.String())
+		}
+	}
+	tl.logs[errorLog] = map[*regexp.Regexp]int{}
+	tl.logs[warningLog] = map[*regexp.Regexp]int{}
 }
 
-// expected determines if the error string is protected or not.
-func (tl *tLogger) expected(s string) bool {
-	for re, count := range tl.errors {
+// expected determines if the log string of the particular type is protected or
+// not.
+func (tl *tLogger) expected(s string, logType logType) bool {
+	for re, count := range tl.logs[logType] {
 		if re.FindStringIndex(s) != nil {
-			tl.errors[re]--
+			tl.logs[logType][re]--
 			if count <= 1 {
-				delete(tl.errors, re)
+				delete(tl.logs[logType], re)
 			}
 			return true
 		}

internal/xds/xdsclient/xdsresource/xdsconfig.go

@@ -0,0 +1,113 @@
+/*
+ *
+ * Copyright 2025 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 xdsresource
+
+import "google.golang.org/grpc/resolver"
+
+// XDSConfig holds the complete gRPC client-side xDS configuration containing
+// all necessary resources.
+type XDSConfig struct {
+	// Listener holds the listener configuration. It is guaranteed to be
+	// non-nil.
+	Listener *ListenerUpdate
+
+	// RouteConfig holds the route configuration. It will be populated even if
+	// the route configuration was inlined into the Listener resource. It is
+	// guaranteed to be non-nil.
+	RouteConfig *RouteConfigUpdate
+
+	// VirtualHost is selected from the route configuration whose domain field
+	// offers the best match against the provided dataplane authority. It is
+	// guaranteed to be non-nil.
+	VirtualHost *VirtualHost
+
+	// Clusters is a map from cluster name to its configuration.
+	Clusters map[string]*ClusterResult
+}
+
+// ClusterResult contains a cluster's configuration when a valid resource is
+// received from the management server. It contains an error when:
+//   - an invalid resource is received from the management server and
+//     a valid resource was not already present or
+//   - the cluster resource does not exist on the management server
+type ClusterResult struct {
+	Config ClusterConfig
+	Err    error
+}
+
+// ClusterConfig contains configuration for a single cluster.
+type ClusterConfig struct {
+	// Cluster configuration for the cluster. This field is always set to a
+	// non-nil value.
+	Cluster *ClusterUpdate
+	// EndpointConfig contains endpoint configuration for a leaf cluster. This
+	// field is only set for EDS and LOGICAL_DNS clusters.
+	EndpointConfig *EndpointConfig
+	// AggregateConfig contains configuration for an aggregate cluster. This
+	// field is only set for AGGREGATE clusters.
+	AggregateConfig *AggregateConfig
+}
+
+// AggregateConfig holds the configuration for an aggregate cluster.
+type AggregateConfig struct {
+	// LeafClusters contains a prioritized list of names of the leaf clusters
+	// for the cluster.
+	LeafClusters []string
+}
+
+// EndpointConfig contains configuration corresponding to the endpoints in a
+// cluster. Only one of EDSUpdate or DNSEndpoints will be populated based on the
+// cluster type.
+type EndpointConfig struct {
+	// Endpoint configurartion for the EDS clusters.
+	EDSUpdate *EndpointsUpdate
+	// Endpoint configuration for the LOGICAL_DNS clusters.
+	DNSEndpoints *DNSUpdate
+	// ResolutionNote stores error encountered while obtaining endpoints data for the cluster. It may contain a nil value when a valid endpoint datais received. It contains an error when:
+	//   - an invalid resource is received from the management server or
+	//   - the endpoint resource does not exist on the management server
+	ResolutionNote error
+}
+
+// DNSUpdate represents the result of a DNS resolution, containing a list of
+// discovered endpoints. This is only populated for the LOGICAL_DNS clusters.
+type DNSUpdate struct {
+	// Endpoints is the complete list of endpoints returned by the DNS resolver.
+	Endpoints []resolver.Endpoint
+}
+
+// xdsConfigkey is the type used as the key to store XDSConfig in the Attributes
+// field of resolver.State.
+type xdsConfigkey struct{}
+
+// SetXDSConfig returns a copy of state in which the Attributes field is updated
+// with the XDSConfig.
+func SetXDSConfig(state resolver.State, config *XDSConfig) resolver.State {
+	state.Attributes = state.Attributes.WithValue(xdsConfigkey{}, config)
+	return state
+}
+
+// XDSConfigFromResolverState returns XDSConfig stored as an attribute in the
+// resolver state.
+func XDSConfigFromResolverState(state resolver.State) *XDSConfig {
+	state.Attributes.Value(xdsConfigkey{})
+	if v := state.Attributes.Value(xdsConfigkey{}); v != nil {
+		return v.(*XDSConfig)
+	}
+	return nil
+}

internal/xds/xdsdepmgr/watch_service.go

@@ -0,0 +1,115 @@
+/*
+ *
+ * Copyright 2025 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 xdsdepmgr
+
+import (
+	"sync/atomic"
+
+	"google.golang.org/grpc/internal/xds/xdsclient/xdsresource"
+)
+
+type listenerWatcher struct {
+	resourceName string
+	cancel       func()
+	depMgr       *DependencyManager
+	stopped      atomic.Bool
+}
+
+func newListenerWatcher(resourceName string, depMgr *DependencyManager) *listenerWatcher {
+	lw := &listenerWatcher{resourceName: resourceName, depMgr: depMgr}
+	lw.cancel = xdsresource.WatchListener(depMgr.xdsClient, resourceName, lw)
+	return lw
+}
+
+func (l *listenerWatcher) ResourceChanged(update *xdsresource.ListenerUpdate, onDone func()) {
+	if l.stopped.Load() {
+		onDone()
+		return
+	}
+	l.depMgr.onListenerResourceUpdate(update, onDone)
+}
+
+func (l *listenerWatcher) ResourceError(err error, onDone func()) {
+	if l.stopped.Load() {
+		onDone()
+		return
+	}
+	l.depMgr.onListenerResourceError(err, onDone)
+}
+
+func (l *listenerWatcher) AmbientError(err error, onDone func()) {
+	if l.stopped.Load() {
+		onDone()
+		return
+	}
+	l.depMgr.onListenerResourceAmbientError(err, onDone)
+}
+
+func (l *listenerWatcher) stop() {
+	l.stopped.Store(true)
+	l.cancel()
+	if l.depMgr.logger.V(2) {
+		l.depMgr.logger.Infof("Canceling watch on Listener resource %q", l.resourceName)
+	}
+}
+
+type routeConfigWatcher struct {
+	resourceName string
+	cancel       func()
+	depMgr       *DependencyManager
+	stopped      atomic.Bool
+}
+
+func newRouteConfigWatcher(resourceName string, depMgr *DependencyManager) *routeConfigWatcher {
+	rw := &routeConfigWatcher{resourceName: resourceName, depMgr: depMgr}
+	rw.cancel = xdsresource.WatchRouteConfig(depMgr.xdsClient, resourceName, rw)
+	return rw
+}
+
+func (r *routeConfigWatcher) ResourceChanged(u *xdsresource.RouteConfigUpdate, onDone func()) {
+	if r.stopped.Load() {
+		onDone()
+		return
+	}
+	r.depMgr.onRouteConfigResourceUpdate(r.resourceName, u, onDone)
+}
+
+func (r *routeConfigWatcher) ResourceError(err error, onDone func()) {
+	if r.stopped.Load() {
+		onDone()
+		return
+	}
+	r.depMgr.onRouteConfigResourceError(r.resourceName, err, onDone)
+}
+
+func (r *routeConfigWatcher) AmbientError(err error, onDone func()) {
+	if r.stopped.Load() {
+		onDone()
+		return
+	}
+	r.depMgr.onRouteConfigResourceAmbientError(r.resourceName, err, onDone)
+}
+
+func (r *routeConfigWatcher) stop() {
+	r.stopped.Store(true)
+	r.cancel()
+	if r.depMgr.logger.V(2) {
+		r.depMgr.logger.Infof("Canceling watch on RouteConfiguration resource %q", r.resourceName)
+	}
+}