review comments

Author: mbissa

internal/xds/clients/config.go

@@ -115,6 +115,7 @@ type MetricsReporter interface {
 	// for the specific client (e.g. internal/xds/clients/xdsclient/metrics/metrics.go)
 	// for the list of supported metrics. The returned function must be called
 	// when the metrics are no longer needed, which will remove the reporter.
+	// The function is expected to be idempotent.
 	//
 	// Once the returned cancel function is called, the Report method on the
 	// registered reporter is guaranteed not to be called again.
@@ -123,14 +124,17 @@ type MetricsReporter interface {
 
 // AsyncReporter records metrics asynchronously.
 // Implementations must be concurrent-safe.
+// The metric will be recorded once per collection cycle, rather than every time
+// its value changes.
 type AsyncReporter interface {
 	// Report records metric values using the provided recorder.
 	Report(AsyncMetricsRecorder) error
 }
 
-// AsyncMetricsRecorder is a recorder for async metrics.
+// AsyncMetricsRecorder is a recorder for async metrics (i.e the metric will be
+// recorded once per collection cycle, rather than every time its value changes).
 type AsyncMetricsRecorder interface {
 	// ReportMetric reports a metric. The metric will be one of the predefined
-	// set of types in the metrics.go file.
+	// set of types in the internal/xds/clients/xdsclient/metrics/metrics.go file.
 	ReportMetric(metric any)
 }

internal/xds/clients/xdsclient/authority.go

@@ -935,36 +935,38 @@ func (a *authority) resourceWatchStateForTesting(rType ResourceType, resourceNam
 	return state, err
 }
 
+// resourceStats returns a snapshot of the current state of all resources watched
+// by this authority. The return value is a nested map where:
+//   - The outer map's key is the resource type name (e.g., "ListenerResource").
+//   - The inner map's key is the cache state of the resource (e.g., "requested",
+//     "acked", "nacked", "does_not_exist").
+//   - The inner map's value is the total count of resources in that specific state.
 func (a *authority) resourceStats() map[string]map[string]int {
 	ret := make(chan map[string]map[string]int, 1)
 	op := func(context.Context) {
-		// Map: ResourceType (String) -> CacheState (String) -> Count (Int)
 		summary := make(map[string]map[string]int)
 		for rType, resourceMap := range a.resources {
-			rName := rType.TypeName
-			if _, ok := summary[rName]; !ok {
-				summary[rName] = make(map[string]int)
+			typeName := rType.TypeName
+			if _, ok := summary[typeName]; !ok {
+				summary[typeName] = make(map[string]int)
 			}
 			for _, state := range resourceMap {
-				s := getCacheState(state)
-				summary[rName][s]++
+				s := cacheState(state)
+				summary[typeName][s]++
 			}
 		}
 
 		ret <- summary
 	}
-
-	// Schedule the operation.
-	// If the serializer is closed/context canceled, the second func (onFailure) runs.
 	a.xdsClientSerializer.ScheduleOr(op, func() {
 		ret <- nil
 	})
 
 	return <-ret
 }
 
-// getCacheState determines the metrics label string for a given resource state.
-func getCacheState(r *resourceState) string {
+// cacheState determines the metrics label string for a given resource state.
+func cacheState(r *resourceState) string {
 	switch r.md.Status {
 	case xdsresource.ServiceStatusRequested:
 		return "requested"

internal/xds/clients/xdsclient/metrics/metrics.go

@@ -47,6 +47,7 @@ type ServerFailure struct {
 // XDSClientConnected reports the connectivity state of the ADS stream.
 // Per gRFC A78, Value is 1 if connected, 0 otherwise.
 // Labels: grpc.target, grpc.xds.server
+// grpc.target is added by asyncMetricsRecorderAdapter
 type XDSClientConnected struct {
 	ServerURI string
 	Value     int64
@@ -55,6 +56,7 @@ type XDSClientConnected struct {
 // XDSClientResourceStats reports the current cache states of xDS resources
 // For label definitions, see gRFC A78.
 // Labels: grpc.target, grpc.xds.authority, grpc.xds.cache_state, grpc.xds.resource_type
+// grpc.target is added by asyncMetricsRecorderAdapter
 type XDSClientResourceStats struct {
 	Authority    string
 	ResourceType string

internal/xds/clients/xdsclient/test/metrics_test.go

@@ -479,7 +479,7 @@ func (s) TestResourceMetrics(t *testing.T) {
 	// Trigger async metrics.
 	tmr.triggerAsyncMetrics()
 	if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{
-		Authority:    "", // Default authority
+		Authority:    "#old", // Default authority
 		ResourceType: "ListenerResource",
 		CacheState:   "acked",
 		Count:        1,
@@ -501,7 +501,7 @@ func (s) TestResourceMetrics(t *testing.T) {
 
 	tmr.triggerAsyncMetrics()
 	if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{
-		Authority:    "",
+		Authority:    "#old",
 		ResourceType: "ListenerResource",
 		CacheState:   "nacked_but_cached",
 		Count:        1,
@@ -594,7 +594,7 @@ func (s) TestResourceMetrics_Extended(t *testing.T) {
 	// Verify "requested" count 2
 	tmr.triggerAsyncMetrics()
 	if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{
-		Authority:    "",
+		Authority:    "#old",
 		ResourceType: "ListenerResource",
 		CacheState:   "requested",
 		Count:        2,
@@ -605,7 +605,7 @@ func (s) TestResourceMetrics_Extended(t *testing.T) {
 	// Verify "nacked" count 2
 	tmr.triggerAsyncMetrics()
 	if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{
-		Authority:    "",
+		Authority:    "#old",
 		ResourceType: "ListenerResource",
 		CacheState:   "nacked",
 		Count:        2,
@@ -635,7 +635,7 @@ func (s) TestResourceMetrics_Extended(t *testing.T) {
 
 	tmr.triggerAsyncMetrics()
 	if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{
-		Authority:    "",
+		Authority:    "#old",
 		ResourceType: "ListenerResource",
 		CacheState:   "does_not_exist",
 		Count:        1,
@@ -671,12 +671,10 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) {
 	sendResponse := make(chan struct{})
 	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
 		Listener: lis,
-		OnStreamOpen: func(ctx context.Context, streamID int64, typeURL string) error {
-			t.Logf("ADS stream opened, streamID: %d", streamID)
+		OnStreamOpen: func(_ context.Context, _ int64, _ string) error {
 			return nil
 		},
-		OnStreamRequest: func(streamID int64, req *v3discoverypb.DiscoveryRequest) error {
-			t.Logf("ADS stream request received, streamID: %d", streamID)
+		OnStreamRequest: func(_ int64, _ *v3discoverypb.DiscoveryRequest) error {
 			// For all streams, wait until we are told to send a response.
 			<-sendResponse
 			return nil
@@ -700,11 +698,11 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) {
 		},
 		MetricsReporter: tmr,
 	}
-	
+
 	// 1. Initial Start - metric value 0
 	// Keep the listener stopped initially so NewStream fails/blocks.
 	lis.Stop()
-	
+
 	client, err := xdsclient.New(xdsClientConfig)
 	if err != nil {
 		t.Fatalf("Failed to create xDS client: %v", err)
@@ -731,7 +729,7 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) {
 
 	// 2. 1st NewStream OK - metric value 1
 	lis.Restart()
-	
+
 	// Wait a bit for the stream to be created.
 	time.Sleep(1 * time.Second)
 
@@ -749,7 +747,7 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) {
 
 	// 3. Stream Fails - metric value 0
 	lis.Stop()
-	
+
 	// Wait for disconnect to be detected.
 	if err := tmr.waitForSpecificMetric(ctx, &metrics.ServerFailure{ServerURI: mgmtServer.Address}); err != nil {
 		t.Fatal(err.Error())
@@ -799,3 +797,65 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) {
 		t.Fatalf("Step 5 failed: Expected XDSClientConnected to be 1 after response, got: %v", err)
 	}
 }
+
+func (s) TestResourceMetrics_AuthorityOldStyle(t *testing.T) {
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
+	nodeID := uuid.New().String()
+
+	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
+	si := clients.ServerIdentifier{
+		ServerURI:  mgmtServer.Address,
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
+	}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
+	serverCfg := xdsclient.ServerConfig{ServerIdentifier: si}
+
+	tmr := newTestMetricsReporter()
+	xdsClientConfig := xdsclient.Config{
+		Servers:          []xdsclient.ServerConfig{serverCfg},
+		Node:             clients.Node{ID: nodeID},
+		TransportBuilder: grpctransport.NewBuilder(configs),
+		ResourceTypes:    resourceTypes,
+		Authorities: map[string]xdsclient.Authority{
+			"": {XDSServers: []xdsclient.ServerConfig{serverCfg}},
+		},
+		MetricsReporter: tmr,
+	}
+
+	client, err := xdsclient.New(xdsClientConfig)
+	if err != nil {
+		t.Fatalf("Failed to create xDS client: %v", err)
+	}
+	defer client.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	const listenerName = "test-listener"
+
+	client.WatchResource(listenerType.TypeURL, listenerName, &testWatcher{})
+
+	resources := e2e.UpdateOptions{
+		NodeID:         nodeID,
+		Listeners:      []*v3listenerpb.Listener{e2e.DefaultClientListener(listenerName, "route-config")},
+		SkipValidation: true,
+	}
+
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatalf("Failed to update management server: %v", err)
+	}
+
+	if err := tmr.waitForMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil {
+		t.Fatal(err.Error())
+	}
+
+	tmr.triggerAsyncMetrics()
+	if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{
+		Authority:    "#old",
+		ResourceType: "ListenerResource",
+		CacheState:   "acked",
+		Count:        1,
+	}); err != nil {
+		t.Fatalf("Failed to observe grpc.xds.authority '#old' metric substitution: %v", err)
+	}
+}

internal/xds/clients/xdsclient/xdsclient.go

@@ -452,6 +452,12 @@ func resourceWatchStateForTesting(c *XDSClient, rType ResourceType, resourceName
 
 }
 
+// xdsClientMetricReporter is a wrapper around XDSClient used solely for
+// reporting metrics. We create this separate type to implement the
+// clients.AsyncReporter interface, preventing its Report method from
+// becoming part of the public XDSClient API. This is especially important
+// because the AsyncReporter interface is experimental, and we want to
+// avoid coupling experimental changes to the stable XDSClient API.
 type xdsClientMetricReporter struct {
 	c *XDSClient
 }
@@ -484,25 +490,27 @@ func (c *XDSClient) reportConnectedState(rec clients.AsyncMetricsRecorder) {
 
 // reportResourceStats handles the "grpc.xds_client.resources" metric.
 func (c *XDSClient) reportResourceStats(rec clients.AsyncMetricsRecorder) {
-	reportForAuthority := func(auth *authority) {
-		stats := auth.resourceStats()
-        for resourceType, stateCounts := range stats {
-            for cacheState, count := range stateCounts {
-                if count > 0 {
-                    rec.ReportMetric(&metrics.XDSClientResourceStats{
-                        Authority:    auth.name,
-                        ResourceType: resourceType,
+	reportForAuthority := func(a *authority) {
+		stats := a.resourceStats()
+		for resourceType, stateCounts := range stats {
+			for cacheState, count := range stateCounts {
+				if count > 0 {
+					authorityName := a.name
+					if authorityName == "" {
+						authorityName = "#old"
+					}
+					rec.ReportMetric(&metrics.XDSClientResourceStats{
+						Authority:    authorityName,
+						ResourceType: resourceType,
 						CacheState:   cacheState,
 						Count:        int64(count),
 					})
 				}
 			}
 		}
 	}
-	if c.topLevelAuthority != nil {
-		reportForAuthority(c.topLevelAuthority)
-	}
-	for _, auth := range c.authorities {
-		reportForAuthority(auth)
+	reportForAuthority(c.topLevelAuthority)
+	for _, a := range c.authorities {
+		reportForAuthority(a)
 	}
 }

internal/xds/xdsclient/clientimpl.go

@@ -77,14 +77,14 @@ var (
 	})
 	xdsClientConnectedMetric = estats.RegisterInt64AsyncGauge(estats.MetricDescriptor{
 		Name:        "grpc.xds_client.connected",
-		Description: "Experimental. A metric that is 1 if the xDS Client is connected to an xDS server, 0 otherwise.",
+		Description: "A metric that is 1 if the xDS Client is connected to an xDS server, 0 otherwise.",
 		Unit:        "{connected}",
 		Type:        estats.MetricTypeIntAsyncGauge,
 		Labels:      []string{"grpc.target", "grpc.xds.server"},
 	})
 	xdsClientResourcesMetric = estats.RegisterInt64AsyncGauge(estats.MetricDescriptor{
 		Name:        "grpc.xds_client.resources",
-		Description: "Experimental. Number of xDS resources currently cached.",
+		Description: "Counts of xDS resources.",
 		Unit:        "{resource}",
 		Type:        estats.MetricTypeIntAsyncGauge,
 		Labels:      []string{"grpc.target", "grpc.xds.authority", "grpc.xds.cache_state", "grpc.xds.resource_type"},