From ec83a4ef080c5fd77f04eb05d58954b39bc91f6b Mon Sep 17 00:00:00 2001 From: Madhav Bissa Date: Tue, 6 Jan 2026 20:12:14 +0000 Subject: [PATCH 01/10] a78-xds-client-metrics-connected-resources --- internal/xds/clients/config.go | 19 ++ internal/xds/clients/xdsclient/ads_stream.go | 6 + .../xds/clients/xdsclient/metrics/metrics.go | 15 + .../clients/xdsclient/test/helpers_test.go | 43 ++- .../clients/xdsclient/test/metrics_test.go | 312 ++++++++++++++++++ internal/xds/clients/xdsclient/xdsclient.go | 120 +++++++ internal/xds/xdsclient/clientimpl.go | 70 ++++ 7 files changed, 583 insertions(+), 2 deletions(-) diff --git a/internal/xds/clients/config.go b/internal/xds/clients/config.go index f106465f646a..2cc4751f81ef 100644 --- a/internal/xds/clients/config.go +++ b/internal/xds/clients/config.go @@ -109,4 +109,23 @@ type MetricsReporter interface { // Each client will produce different metrics. Please see the client's // documentation for a list of possible metrics events. ReportMetric(metric any) + + // RegisterAsyncReporter registers a reporter to produce metric values for + // only the listed descriptors. The returned function must be called when + // the metrics are no longer needed, which will remove the reporter. + RegisterAsyncReporter(reporter AsyncReporter) func() +} + +// AsyncReporter is an interface for types that record metrics asynchronously. +// Implementations must be concurrent-safe. +type AsyncReporter interface { + // Report records metric values using the provided recorder. + Report(AsyncMetricsRecorder) error +} + +// AsyncMetricsRecorder is a recorder for async metrics. +type AsyncMetricsRecorder interface { + // ReportMetric reports a metric. The metric will be one of the predefined + // set of types in the metrics.go file. + ReportMetric(metric any) } diff --git a/internal/xds/clients/xdsclient/ads_stream.go b/internal/xds/clients/xdsclient/ads_stream.go index 24e66b834716..1c0cdb32eb88 100644 --- a/internal/xds/clients/xdsclient/ads_stream.go +++ b/internal/xds/clients/xdsclient/ads_stream.go @@ -715,3 +715,9 @@ func (fc *adsFlowControl) wait() bool { return fc.stopped } + +func (fc *adsFlowControl) isStopped() bool { + fc.mu.Lock() + defer fc.mu.Unlock() + return fc.stopped +} diff --git a/internal/xds/clients/xdsclient/metrics/metrics.go b/internal/xds/clients/xdsclient/metrics/metrics.go index 2e14efb1ae9b..f347a556e39f 100644 --- a/internal/xds/clients/xdsclient/metrics/metrics.go +++ b/internal/xds/clients/xdsclient/metrics/metrics.go @@ -40,3 +40,18 @@ type ResourceUpdateInvalid struct { type ServerFailure struct { ServerURI string } + +// XDSClientConnected reports the connectivity state of the xDS stream. +// Value is 1 if connected, 0 otherwise. +type XDSClientConnected struct { + ServerURI string + Value int64 +} + +// XDSClientResourceStats reports the number of resources currently cached. +type XDSClientResourceStats struct { + Authority string + ResourceType string + CacheState string + Count int64 +} diff --git a/internal/xds/clients/xdsclient/test/helpers_test.go b/internal/xds/clients/xdsclient/test/helpers_test.go index 1d6ac9ca7f4a..fc76d8466fc4 100644 --- a/internal/xds/clients/xdsclient/test/helpers_test.go +++ b/internal/xds/clients/xdsclient/test/helpers_test.go @@ -25,10 +25,12 @@ import ( "fmt" "strconv" "strings" + "sync" "testing" "time" "google.golang.org/grpc/internal/grpctest" + "google.golang.org/grpc/internal/xds/clients" "google.golang.org/grpc/internal/xds/clients/internal/pretty" "google.golang.org/grpc/internal/xds/clients/internal/testutils" "google.golang.org/grpc/internal/xds/clients/xdsclient" @@ -279,12 +281,16 @@ func buildResourceName(typeName, auth, id string, ctxParams map[string]string) s // have taken place. type testMetricsReporter struct { metricsCh *testutils.Channel + + mu sync.Mutex + asyncReporters map[clients.AsyncReporter]struct{} } // newTestMetricsReporter returns a new testMetricsReporter. func newTestMetricsReporter() *testMetricsReporter { return &testMetricsReporter{ - metricsCh: testutils.NewChannelWithSize(1), + metricsCh: testutils.NewChannelWithSize(50), + asyncReporters: make(map[clients.AsyncReporter]struct{}), } } @@ -302,7 +308,40 @@ func (r *testMetricsReporter) waitForMetric(ctx context.Context, metricsDataWant return nil } +func (r *testMetricsReporter) waitForSpecificMetric(ctx context.Context, metricsDataWant any) error { + for { + got, err := r.metricsCh.Receive(ctx) + if err != nil { + return fmt.Errorf("timeout waiting for metric: %v (want %T)", err, metricsDataWant) + } + if diff := cmp.Diff(got, metricsDataWant); diff == "" { + return nil + } + // Continue if mismatch. + + } +} + // ReportMetric sends the metrics data to the metricsCh channel. func (r *testMetricsReporter) ReportMetric(m any) { - r.metricsCh.Replace(m) + r.metricsCh.Send(m) +} + +func (r *testMetricsReporter) RegisterAsyncReporter(reporter clients.AsyncReporter) func() { + r.mu.Lock() + defer r.mu.Unlock() + r.asyncReporters[reporter] = struct{}{} + return func() { + r.mu.Lock() + defer r.mu.Unlock() + delete(r.asyncReporters, reporter) + } +} + +func (r *testMetricsReporter) triggerAsyncMetrics() { + r.mu.Lock() + defer r.mu.Unlock() + for reporter := range r.asyncReporters { + reporter.Report(r) + } } diff --git a/internal/xds/clients/xdsclient/test/metrics_test.go b/internal/xds/clients/xdsclient/test/metrics_test.go index 8332cc57faef..8bfae5506d34 100644 --- a/internal/xds/clients/xdsclient/test/metrics_test.go +++ b/internal/xds/clients/xdsclient/test/metrics_test.go @@ -22,6 +22,7 @@ import ( "context" "errors" "net" + "sync" "testing" "github.com/google/uuid" @@ -311,3 +312,314 @@ func (s) TestServerFailureMetrics_AfterResponseRecv(t *testing.T) { t.Fatal(err.Error()) } } + +func (s) TestConnectedMetric(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) + defer cancel() + + tmr := newTestMetricsReporter() + l, err := net.Listen("tcp", "localhost:0") + if err != nil { + t.Fatalf("net.Listen() failed: %v", err) + } + + mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{Listener: l}) + 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()}} + xdsClientConfig := xdsclient.Config{ + Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}}, + Node: clients.Node{ID: nodeID}, + TransportBuilder: grpctransport.NewBuilder(configs), + ResourceTypes: resourceTypes, + Authorities: map[string]xdsclient.Authority{ + "": {XDSServers: []xdsclient.ServerConfig{}}, + }, + MetricsReporter: tmr, + } + client, err := xdsclient.New(xdsClientConfig) + if err != nil { + t.Fatalf("Failed to create xDS client: %v", err) + } + defer client.Close() + + // Initial State: Not connected (until watch starts or channel created?). + // The client creates channels on demand. WatchResource triggers channel creation. + + // Watch a resource to trigger connection. + client.WatchResource(listenerType.TypeURL, "foo", noopListenerWatcher{}) + + // Wait for connection. + const listenerName = "test-listener-resource" + 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) + } + client.WatchResource(listenerType.TypeURL, listenerName, noopListenerWatcher{}) + + // Wait for the update to ensure we are connected. + if err := tmr.waitForMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { + t.Fatal(err.Error()) + } + + // Now trigger async metrics. + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 1}); err != nil { + t.Fatal(err.Error()) + } + + // Stop server to force disconnect. + mgmtServer.Stop() + + // Wait a bit for disconnect to be detected. + // xDS client should detect it on stream failure. + // We can wait for ServerFailure metric (synchronous). + if err := tmr.waitForSpecificMetric(ctx, &metrics.ServerFailure{ServerURI: mgmtServer.Address}); err != nil { + t.Fatal(err.Error()) + } +} + +func (s) TestResourceMetrics(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) + defer cancel() + + tmr := newTestMetricsReporter() + l, err := net.Listen("tcp", "localhost:0") + if err != nil { + t.Fatalf("net.Listen() failed: %v", err) + } + + mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{Listener: l}) + 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()}} + xdsClientConfig := xdsclient.Config{ + Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}}, + Node: clients.Node{ID: nodeID}, + TransportBuilder: grpctransport.NewBuilder(configs), + ResourceTypes: resourceTypes, + Authorities: map[string]xdsclient.Authority{ + "": {XDSServers: []xdsclient.ServerConfig{}}, + }, + MetricsReporter: tmr, + } + client, err := xdsclient.New(xdsClientConfig) + if err != nil { + t.Fatalf("Failed to create xDS client: %v", err) + } + defer client.Close() + + const listenerName = "test-listener-resource" + const routeConfigName = "test-route-configuration-resource" + + // Requested state. + client.WatchResource(listenerType.TypeURL, listenerName, noopListenerWatcher{}) + + resources := e2e.UpdateOptions{ + NodeID: nodeID, + Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(listenerName, routeConfigName)}, + SkipValidation: true, + } + if err := mgmtServer.Update(ctx, resources); err != nil { + t.Fatalf("Failed to update management server: %v", err) + } + + // Wait for Valid update. + if err := tmr.waitForMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { + t.Fatal(err.Error()) + } + + // Trigger async metrics. + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{ + Authority: "", // Default authority + ResourceType: "ListenerResource", + CacheState: "acked", + Count: 1, + }); err != nil { + t.Fatal(err.Error()) + } + + // Nacked but cached. + // Update with bad resource. + resources.Listeners[0].ApiListener = nil + if err := mgmtServer.Update(ctx, resources); err != nil { + t.Fatalf("Failed to update management server: %v", err) + } + + // Wait for Invalid update. + if err := tmr.waitForMetric(ctx, &metrics.ResourceUpdateInvalid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { + t.Fatal(err.Error()) + } + + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{ + Authority: "", + ResourceType: "ListenerResource", + CacheState: "nacked_but_cached", + Count: 1, + }); err != nil { + t.Fatal(err.Error()) + } +} + +// TestResourceMetrics_Extended tests multiple resources in different states: +// - 2 in "requested" state +// - 2 in "nacked" state +// - 1 in "does_not_exist" state +func (s) TestResourceMetrics_Extended(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) + defer cancel() + + tmr := newTestMetricsReporter() + l, err := net.Listen("tcp", "localhost:0") + if err != nil { + t.Fatalf("net.Listen() failed: %v", err) + } + + mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{Listener: l}) + 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()}} + xdsClientConfig := xdsclient.Config{ + Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}}, + Node: clients.Node{ID: nodeID}, + TransportBuilder: grpctransport.NewBuilder(configs), + ResourceTypes: resourceTypes, + Authorities: map[string]xdsclient.Authority{ + "": {XDSServers: []xdsclient.ServerConfig{}}, + }, + MetricsReporter: tmr, + } + client, err := xdsclient.New(xdsClientConfig) + if err != nil { + t.Fatalf("Failed to create xDS client: %v", err) + } + defer client.Close() + + // Resource names + resRequested1 := "res-requested-1" + resRequested2 := "res-requested-2" + resNacked1 := "res-nacked-1" + resNacked2 := "res-nacked-2" + resNotExist := "res-not-exist" + + resList := []string{resRequested1, resRequested2, resNacked1, resNacked2, resNotExist} + for _, res := range resList { + client.WatchResource(listenerType.TypeURL, res, noopListenerWatcher{}) + } + + resources := e2e.UpdateOptions{ + NodeID: nodeID, + Listeners: []*v3listenerpb.Listener{ + e2e.DefaultClientListener(resNacked1, "route-config"), + e2e.DefaultClientListener(resNacked2, "route-config"), + e2e.DefaultClientListener(resNotExist, "route-config"), + }, + SkipValidation: true, + } + // Make Nacked resources invalid + resources.Listeners[0].ApiListener = nil + resources.Listeners[1].ApiListener = nil + + if err := mgmtServer.Update(ctx, resources); err != nil { + t.Fatalf("Failed to update management server: %v", err) + } + + // Wait for Res5 to be valid. + if err := tmr.waitForSpecificMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { + t.Fatal(err.Error()) + } + resourcesEmpty := e2e.UpdateOptions{ + NodeID: nodeID, + Listeners: []*v3listenerpb.Listener{}, + SkipValidation: true, + } + if err := mgmtServer.Update(ctx, resourcesEmpty); err != nil { + t.Fatalf("Failed to update management server: %v", err) + } + + // Verify "requested" count 2 + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{ + Authority: "", + ResourceType: "ListenerResource", + CacheState: "requested", + Count: 2, + }); err != nil { + t.Fatalf("Failed to verify requested count: %v", err) + } + + // Verify "nacked" count 2 + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{ + Authority: "", + ResourceType: "ListenerResource", + CacheState: "nacked", + Count: 2, + }); err != nil { + t.Fatalf("Failed to verify nacked count: %v", err) + } + + // Wait for the does_not_exist state. + // We need to wait for the client to process the empty update and mark the resource as removed. + // We use a watcher to detect the specific error. + removed := make(chan struct{}) + var closeOnce sync.Once + client.WatchResource(listenerType.TypeURL, resNotExist, &testWatcher{ + onError: func(err error) { + if xdsresource.ErrType(err) == xdsresource.ErrorTypeResourceNotFound { + closeOnce.Do(func() { close(removed) }) + } + }, + }) + + // Verify "does_not_exist" count 1 + select { + case <-removed: + case <-ctx.Done(): + t.Fatal("timeout waiting for resource removal") + } + + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{ + Authority: "", + ResourceType: "ListenerResource", + CacheState: "does_not_exist", + Count: 1, + }); err != nil { + t.Fatalf("Failed to verify does_not_exist count: %v", err) + } +} + +type testWatcher struct { + onError func(error) +} + +func (w *testWatcher) ResourceChanged(_ xdsclient.ResourceData, onDone func()) { onDone() } +func (w *testWatcher) ResourceError(err error, onDone func()) { + if w.onError != nil { + w.onError(err) + } + onDone() +} +func (w *testWatcher) AmbientError(_ error, onDone func()) { onDone() } diff --git a/internal/xds/clients/xdsclient/xdsclient.go b/internal/xds/clients/xdsclient/xdsclient.go index b1c6955484dc..67f365e1d647 100644 --- a/internal/xds/clients/xdsclient/xdsclient.go +++ b/internal/xds/clients/xdsclient/xdsclient.go @@ -95,6 +95,8 @@ type XDSClient struct { // Once all references to a channel are dropped, the channel is closed. channelsMu sync.Mutex xdsActiveChannels map[ServerConfig]*channelState // Map from server config to in-use xdsChannels. + + metricsCleanup func() } // New returns a new xDS Client configured with the provided config. @@ -114,6 +116,11 @@ func New(config Config) (*XDSClient, error) { if err != nil { return nil, err } + + // Register this client instance as an Async Reporter. + if client.metricsReporter != nil { + client.metricsCleanup = client.metricsReporter.RegisterAsyncReporter(client) + } return client, nil } @@ -171,6 +178,9 @@ func (c *XDSClient) Close() { if c.done.HasFired() { return } + if c.metricsCleanup != nil { + c.metricsCleanup() + } c.done.Fire() c.topLevelAuthority.close() @@ -441,3 +451,113 @@ func resourceWatchStateForTesting(c *XDSClient, rType ResourceType, resourceName return a.resourceWatchStateForTesting(rType, resourceName) } + +// Report implements clients.AsyncReporter. +// This is the entry point invoked by the metrics system during a scrape. +func (c *XDSClient) Report(rec clients.AsyncMetricsRecorder) error { + c.reportConnectedState(rec) + c.reportResourceStats(rec) + return nil +} + +// reportConnectedState handles the "grpc.xds_client.connected" metric. +func (c *XDSClient) reportConnectedState(rec clients.AsyncMetricsRecorder) { + c.channelsMu.Lock() + defer c.channelsMu.Unlock() + + for _, cs := range c.xdsActiveChannels { + val := int64(0) + if cs.channel.ads != nil && cs.channel.ads.fc != nil && !cs.channel.ads.fc.isStopped() { + val = 1 + } + + rec.ReportMetric(&metrics.XDSClientConnected{ + ServerURI: cs.serverConfig.ServerIdentifier.ServerURI, + Value: val, + }) + } +} + +// reportResourceStats handles the "grpc.xds_client.resources" metric. +func (c *XDSClient) reportResourceStats(rec clients.AsyncMetricsRecorder) { + allAuthorities := make([]*authority, 0, len(c.authorities)+1) + if c.topLevelAuthority != nil { + allAuthorities = append(allAuthorities, c.topLevelAuthority) + } + for _, a := range c.authorities { + allAuthorities = append(allAuthorities, a) + } + + for _, a := range allAuthorities { + stats := a.resourceStats() + for typeURL, stateCounts := range stats { + for cacheState, count := range stateCounts { + if count > 0 { + rec.ReportMetric(&metrics.XDSClientResourceStats{ + Authority: a.name, + ResourceType: typeURL, + CacheState: cacheState, + Count: int64(count), + }) + } + } + } + } +} + +func (a *authority) resourceStats() map[string]map[string]int { + // Create a channel to receive the result + 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) + } + for _, state := range resourceMap { + s := getCacheState(state) + summary[rName][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 { + switch r.md.Status { + case xdsresource.ServiceStatusRequested: + return "requested" + + case xdsresource.ServiceStatusNotExist: + return "does_not_exist" + + case xdsresource.ServiceStatusACKed: + return "acked" + + case xdsresource.ServiceStatusNACKed: + // If the status is NACKed, it means the *latest* update failed. + // However, if 'r.cache' is not nil, it means we are still holding onto + // a previously ACKed version of the resource. + if r.cache != nil { + return "nacked_but_cached" + } + return "nacked" + + default: + // Fallback for initialization states + return "requested" + } +} diff --git a/internal/xds/xdsclient/clientimpl.go b/internal/xds/xdsclient/clientimpl.go index b20886470115..ab0256f0080d 100644 --- a/internal/xds/xdsclient/clientimpl.go +++ b/internal/xds/xdsclient/clientimpl.go @@ -75,6 +75,20 @@ var ( Labels: []string{"grpc.target", "grpc.xds.server"}, Default: false, }) + xdsClientConnectedMetric = estats.RegisterInt64AsyncGauge(estats.MetricDescriptor{ + Name: "grpc.xds_client.connected", + Description: "Experimental. 1 if the xDS stream is connected, 0 otherwise.", + Unit: "1", + 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.", + Unit: "{resource}", + Type: estats.MetricTypeIntAsyncGauge, + Labels: []string{"grpc.target", "grpc.xds.authority", "grpc.xds.cache_state", "grpc.xds.resource_type"}, + }) ) // clientImpl embed xdsclient.XDSClient and implement internal XDSClient @@ -263,3 +277,59 @@ func populateGRPCTransportConfigsFromServerConfig(sc *bootstrap.ServerConfig, gr } return nil } + +// RegisterAsyncReporter adapts the generic clients.AsyncReporter to the +// estats.AsyncMetricReporter interface and registers it. +func (mr *metricsReporter) RegisterAsyncReporter(reporter clients.AsyncReporter) func() { + if mr.recorder == nil { + return func() {} + } + + // Define which metrics we intend to report for OTel registration. + descriptors := []estats.AsyncMetric{ + xdsClientConnectedMetric, + xdsClientResourcesMetric, + } + + // Create the callback wrapper. + // This function is invoked by the stats system during a scrape. + cbWrapper := func(rec estats.AsyncMetricsRecorder) error { + wrapper := &asyncMetricsRecorderAdapter{ + target: mr.target, + delegate: rec, + } + reporter.Report(wrapper) + return nil + } + + // Register with the underlying gRPC stats recorder. + return mr.recorder.RegisterAsyncReporter(estats.AsyncMetricReporterFunc(cbWrapper), descriptors...) +} + +// asyncMetricsRecorderAdapter adapts estats.AsyncMetricsRecorder to clients.AsyncMetricsRecorder. +type asyncMetricsRecorderAdapter struct { + target string + delegate estats.AsyncMetricsRecorder +} + +func (a *asyncMetricsRecorderAdapter) ReportMetric(metric any) { + switch m := metric.(type) { + case *metrics.XDSClientConnected: + // Record: grpc.xds_client.connected + // Labels: grpc.target, grpc.xds.authority + a.delegate.RecordInt64AsyncGauge(xdsClientConnectedMetric, m.Value, + a.target, + m.ServerURI, + ) + + case *metrics.XDSClientResourceStats: + // Record: grpc.xds_client.resources + // Labels: grpc.target, grpc.xds.authority, grpc.xds.cache_state, grpc.xds.resource_type + a.delegate.RecordInt64AsyncGauge(xdsClientResourcesMetric, m.Count, + a.target, + m.Authority, + m.CacheState, + m.ResourceType, + ) + } +} From b73b5d22bd808e2e147e72e099b087668c77b0a7 Mon Sep 17 00:00:00 2001 From: Madhav Bissa Date: Wed, 11 Mar 2026 17:22:42 +0000 Subject: [PATCH 02/10] review comments --- internal/xds/clients/config.go | 12 ++- internal/xds/clients/xdsclient/ads_stream.go | 14 ++- internal/xds/clients/xdsclient/authority.go | 51 +++++++++++ .../xds/clients/xdsclient/metrics/metrics.go | 10 ++- .../clients/xdsclient/test/metrics_test.go | 36 +++++++- internal/xds/clients/xdsclient/xdsclient.go | 85 ++++--------------- internal/xds/xdsclient/clientimpl.go | 21 ++--- 7 files changed, 134 insertions(+), 95 deletions(-) diff --git a/internal/xds/clients/config.go b/internal/xds/clients/config.go index 2cc4751f81ef..6152c36ac2a8 100644 --- a/internal/xds/clients/config.go +++ b/internal/xds/clients/config.go @@ -111,12 +111,18 @@ type MetricsReporter interface { ReportMetric(metric any) // RegisterAsyncReporter registers a reporter to produce metric values for - // only the listed descriptors. The returned function must be called when - // the metrics are no longer needed, which will remove the reporter. + // the set of metrics supported by the client. See the metrics sub-package + // 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. + // + // Once the returned cancel function is called, the Report method on the + // registered reporter is guaranteed not to be called again. RegisterAsyncReporter(reporter AsyncReporter) func() } -// AsyncReporter is an interface for types that record metrics asynchronously. +// AsyncReporter records metrics asynchronously. // Implementations must be concurrent-safe. type AsyncReporter interface { // Report records metric values using the provided recorder. diff --git a/internal/xds/clients/xdsclient/ads_stream.go b/internal/xds/clients/xdsclient/ads_stream.go index 1c0cdb32eb88..1d2a5bbc1d5b 100644 --- a/internal/xds/clients/xdsclient/ads_stream.go +++ b/internal/xds/clients/xdsclient/ads_stream.go @@ -112,6 +112,7 @@ type adsStreamImpl struct { mu sync.Mutex resourceTypeState map[ResourceType]*resourceTypeState // Map of resource types to their state. firstRequest bool // False after the first request is sent out. + streamEstablished bool // True after a successful NewStream call. pendingRequests []request // Subscriptions and unsubscriptions are pushed here. } @@ -250,6 +251,7 @@ func (s *adsStreamImpl) runner(ctx context.Context) { s.mu.Lock() s.firstRequest = true + s.streamEstablished = true s.mu.Unlock() // Ensure that the most recently created stream is pushed on the @@ -437,6 +439,9 @@ func (s *adsStreamImpl) recv(stream clients.Stream) bool { resources, url, version, nonce, err := s.recvMessage(stream) if err != nil { + s.mu.Lock() + s.streamEstablished = false + s.mu.Unlock() s.onError(err, msgReceived) s.logger.Warningf("ADS stream closed: %v", err) return msgReceived @@ -716,8 +721,9 @@ func (fc *adsFlowControl) wait() bool { return fc.stopped } -func (fc *adsFlowControl) isStopped() bool { - fc.mu.Lock() - defer fc.mu.Unlock() - return fc.stopped + +func (s *adsStreamImpl) isStreamEstablished() bool { +s.mu.Lock() +defer s.mu.Unlock() +return s.streamEstablished } diff --git a/internal/xds/clients/xdsclient/authority.go b/internal/xds/clients/xdsclient/authority.go index b8cb78fbdfba..f9574c30e6b8 100644 --- a/internal/xds/clients/xdsclient/authority.go +++ b/internal/xds/clients/xdsclient/authority.go @@ -934,3 +934,54 @@ func (a *authority) resourceWatchStateForTesting(rType ResourceType, resourceNam return state, err } + +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) + } + for _, state := range resourceMap { + s := getCacheState(state) + summary[rName][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 { + switch r.md.Status { + case xdsresource.ServiceStatusRequested: + return "requested" + case xdsresource.ServiceStatusNotExist: + return "does_not_exist" + case xdsresource.ServiceStatusACKed: + return "acked" + case xdsresource.ServiceStatusNACKed: + // If the status is NACKed, it means the *latest* update failed. + // However, if 'r.cache' is not nil, it means we are still holding onto + // a previously ACKed version of the resource. + if r.cache != nil { + return "nacked_but_cached" + } + return "nacked" + default: + // Fallback for initialization states + return "requested" + } +} diff --git a/internal/xds/clients/xdsclient/metrics/metrics.go b/internal/xds/clients/xdsclient/metrics/metrics.go index f347a556e39f..d7aaa95053af 100644 --- a/internal/xds/clients/xdsclient/metrics/metrics.go +++ b/internal/xds/clients/xdsclient/metrics/metrics.go @@ -19,6 +19,9 @@ // Package metrics defines all metrics that can be produced by an xDS client. // All calls to the MetricsRecorder by the xDS client will contain a struct // from this package passed by pointer. +// +// For definitions of these metrics and their labels, see gRFC A78: +// https://github.com/grpc/proposal/blob/master/A78-grpc-metrics-wrr-xds.md package metrics // ResourceUpdateValid is a metric to report a valid resource update from @@ -41,14 +44,17 @@ type ServerFailure struct { ServerURI string } -// XDSClientConnected reports the connectivity state of the xDS stream. -// Value is 1 if connected, 0 otherwise. +// 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 type XDSClientConnected struct { ServerURI string Value int64 } // XDSClientResourceStats reports the number of resources currently cached. +// For label definitions, see gRFC A78. +// Labels: grpc.target, grpc.xds.authority, grpc.xds.cache_state, grpc.xds.resource_type type XDSClientResourceStats struct { Authority string ResourceType string diff --git a/internal/xds/clients/xdsclient/test/metrics_test.go b/internal/xds/clients/xdsclient/test/metrics_test.go index 8bfae5506d34..e060b4f98737 100644 --- a/internal/xds/clients/xdsclient/test/metrics_test.go +++ b/internal/xds/clients/xdsclient/test/metrics_test.go @@ -348,8 +348,13 @@ func (s) TestConnectedMetric(t *testing.T) { } defer client.Close() - // Initial State: Not connected (until watch starts or channel created?). - // The client creates channels on demand. WatchResource triggers channel creation. + // Verify initial state: No metrics reported because no stream is established. + tmr.triggerAsyncMetrics() + sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout) + defer sCancel() + if err := tmr.waitForSpecificMetric(sCtx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address}); err == nil { + t.Fatal("XDSClientConnected metric reported before any watch was started") + } // Watch a resource to trigger connection. client.WatchResource(listenerType.TypeURL, "foo", noopListenerWatcher{}) @@ -386,6 +391,33 @@ func (s) TestConnectedMetric(t *testing.T) { if err := tmr.waitForSpecificMetric(ctx, &metrics.ServerFailure{ServerURI: mgmtServer.Address}); err != nil { t.Fatal(err.Error()) } + + // Trigger async metrics again after disconnect to verify Value is 0. + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 0}); err != nil { + t.Fatal(err.Error()) + } + + // Verify unregistration after client close. + client.Close() + + tmr.mu.Lock() + numReporters := len(tmr.asyncReporters) + tmr.mu.Unlock() + if numReporters != 0 { + t.Fatalf("Async reporter not unregistered after client close, count: %d", numReporters) + } + + // Drain the channel of any leftover metrics from previous pulses. + tmr.metricsCh.Drain() + + tmr.triggerAsyncMetrics() + // No metrics should be reported now because there are no reporters. + sCtx2, sCancel2 := context.WithTimeout(ctx, defaultTestShortTimeout) + defer sCancel2() + if _, err := tmr.metricsCh.Receive(sCtx2); err == nil { + t.Fatal("Metrics reported after all reporters were unregistered") + } } func (s) TestResourceMetrics(t *testing.T) { diff --git a/internal/xds/clients/xdsclient/xdsclient.go b/internal/xds/clients/xdsclient/xdsclient.go index 67f365e1d647..7cb54eaacebe 100644 --- a/internal/xds/clients/xdsclient/xdsclient.go +++ b/internal/xds/clients/xdsclient/xdsclient.go @@ -116,10 +116,10 @@ func New(config Config) (*XDSClient, error) { if err != nil { return nil, err } - // Register this client instance as an Async Reporter. if client.metricsReporter != nil { - client.metricsCleanup = client.metricsReporter.RegisterAsyncReporter(client) + reporter := &xdsClientMetricReporter{c: client} + client.metricsCleanup = client.metricsReporter.RegisterAsyncReporter(reporter) } return client, nil } @@ -452,11 +452,15 @@ func resourceWatchStateForTesting(c *XDSClient, rType ResourceType, resourceName } +type xdsClientMetricReporter struct { + c *XDSClient +} + // Report implements clients.AsyncReporter. // This is the entry point invoked by the metrics system during a scrape. -func (c *XDSClient) Report(rec clients.AsyncMetricsRecorder) error { - c.reportConnectedState(rec) - c.reportResourceStats(rec) +func (r *xdsClientMetricReporter) Report(rec clients.AsyncMetricsRecorder) error { + r.c.reportConnectedState(rec) + r.c.reportResourceStats(rec) return nil } @@ -467,7 +471,7 @@ func (c *XDSClient) reportConnectedState(rec clients.AsyncMetricsRecorder) { for _, cs := range c.xdsActiveChannels { val := int64(0) - if cs.channel.ads != nil && cs.channel.ads.fc != nil && !cs.channel.ads.fc.isStopped() { + if cs.channel.ads.isStreamEstablished() { val = 1 } @@ -480,15 +484,7 @@ func (c *XDSClient) reportConnectedState(rec clients.AsyncMetricsRecorder) { // reportResourceStats handles the "grpc.xds_client.resources" metric. func (c *XDSClient) reportResourceStats(rec clients.AsyncMetricsRecorder) { - allAuthorities := make([]*authority, 0, len(c.authorities)+1) - if c.topLevelAuthority != nil { - allAuthorities = append(allAuthorities, c.topLevelAuthority) - } - for _, a := range c.authorities { - allAuthorities = append(allAuthorities, a) - } - - for _, a := range allAuthorities { + reportForAuthority := func(a *authority) { stats := a.resourceStats() for typeURL, stateCounts := range stats { for cacheState, count := range stateCounts { @@ -503,61 +499,12 @@ func (c *XDSClient) reportResourceStats(rec clients.AsyncMetricsRecorder) { } } } -} - -func (a *authority) resourceStats() map[string]map[string]int { - // Create a channel to receive the result - 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) - } - for _, state := range resourceMap { - s := getCacheState(state) - summary[rName][s]++ - } - } - ret <- summary + if c.topLevelAuthority != nil { + reportForAuthority(c.topLevelAuthority) } - - // 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 { - switch r.md.Status { - case xdsresource.ServiceStatusRequested: - return "requested" - - case xdsresource.ServiceStatusNotExist: - return "does_not_exist" - - case xdsresource.ServiceStatusACKed: - return "acked" - - case xdsresource.ServiceStatusNACKed: - // If the status is NACKed, it means the *latest* update failed. - // However, if 'r.cache' is not nil, it means we are still holding onto - // a previously ACKed version of the resource. - if r.cache != nil { - return "nacked_but_cached" - } - return "nacked" - - default: - // Fallback for initialization states - return "requested" + for _, a := range c.authorities { + reportForAuthority(a) } } + diff --git a/internal/xds/xdsclient/clientimpl.go b/internal/xds/xdsclient/clientimpl.go index ab0256f0080d..59d376044164 100644 --- a/internal/xds/xdsclient/clientimpl.go +++ b/internal/xds/xdsclient/clientimpl.go @@ -77,8 +77,8 @@ var ( }) xdsClientConnectedMetric = estats.RegisterInt64AsyncGauge(estats.MetricDescriptor{ Name: "grpc.xds_client.connected", - Description: "Experimental. 1 if the xDS stream is connected, 0 otherwise.", - Unit: "1", + Description: "Experimental. 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"}, }) @@ -281,7 +281,7 @@ func populateGRPCTransportConfigsFromServerConfig(sc *bootstrap.ServerConfig, gr // RegisterAsyncReporter adapts the generic clients.AsyncReporter to the // estats.AsyncMetricReporter interface and registers it. func (mr *metricsReporter) RegisterAsyncReporter(reporter clients.AsyncReporter) func() { - if mr.recorder == nil { + if mr.recorder == nil || reporter == nil { return func() {} } @@ -316,20 +316,11 @@ func (a *asyncMetricsRecorderAdapter) ReportMetric(metric any) { switch m := metric.(type) { case *metrics.XDSClientConnected: // Record: grpc.xds_client.connected - // Labels: grpc.target, grpc.xds.authority - a.delegate.RecordInt64AsyncGauge(xdsClientConnectedMetric, m.Value, - a.target, - m.ServerURI, - ) - + // Labels: grpc.target, grpc.xds.server + a.delegate.RecordInt64AsyncGauge(xdsClientConnectedMetric, m.Value, a.target, m.ServerURI) case *metrics.XDSClientResourceStats: // Record: grpc.xds_client.resources // Labels: grpc.target, grpc.xds.authority, grpc.xds.cache_state, grpc.xds.resource_type - a.delegate.RecordInt64AsyncGauge(xdsClientResourcesMetric, m.Count, - a.target, - m.Authority, - m.CacheState, - m.ResourceType, - ) + a.delegate.RecordInt64AsyncGauge(xdsClientResourcesMetric, m.Count, a.target, m.Authority, m.CacheState, m.ResourceType) } } From 30b9c738ddb2d1598f90691cf379ec0611dbffa4 Mon Sep 17 00:00:00 2001 From: Madhav Bissa Date: Wed, 11 Mar 2026 21:08:23 +0000 Subject: [PATCH 03/10] fixing vet --- internal/xds/clients/config.go | 7 +++---- internal/xds/clients/xdsclient/ads_stream.go | 7 +++---- internal/xds/clients/xdsclient/xdsclient.go | 12 +++++------- 3 files changed, 11 insertions(+), 15 deletions(-) diff --git a/internal/xds/clients/config.go b/internal/xds/clients/config.go index 6152c36ac2a8..2d74966a6773 100644 --- a/internal/xds/clients/config.go +++ b/internal/xds/clients/config.go @@ -112,10 +112,9 @@ type MetricsReporter interface { // RegisterAsyncReporter registers a reporter to produce metric values for // the set of metrics supported by the client. See the metrics sub-package - // 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. + // 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. // // Once the returned cancel function is called, the Report method on the // registered reporter is guaranteed not to be called again. diff --git a/internal/xds/clients/xdsclient/ads_stream.go b/internal/xds/clients/xdsclient/ads_stream.go index 1d2a5bbc1d5b..45e020dd6e8c 100644 --- a/internal/xds/clients/xdsclient/ads_stream.go +++ b/internal/xds/clients/xdsclient/ads_stream.go @@ -721,9 +721,8 @@ func (fc *adsFlowControl) wait() bool { return fc.stopped } - func (s *adsStreamImpl) isStreamEstablished() bool { -s.mu.Lock() -defer s.mu.Unlock() -return s.streamEstablished + s.mu.Lock() + defer s.mu.Unlock() + return s.streamEstablished } diff --git a/internal/xds/clients/xdsclient/xdsclient.go b/internal/xds/clients/xdsclient/xdsclient.go index 7cb54eaacebe..a97a2dd5cf56 100644 --- a/internal/xds/clients/xdsclient/xdsclient.go +++ b/internal/xds/clients/xdsclient/xdsclient.go @@ -484,13 +484,13 @@ func (c *XDSClient) reportConnectedState(rec clients.AsyncMetricsRecorder) { // reportResourceStats handles the "grpc.xds_client.resources" metric. func (c *XDSClient) reportResourceStats(rec clients.AsyncMetricsRecorder) { - reportForAuthority := func(a *authority) { - stats := a.resourceStats() + reportForAuthority := func(auth *authority) { + stats := auth.resourceStats() for typeURL, stateCounts := range stats { for cacheState, count := range stateCounts { if count > 0 { rec.ReportMetric(&metrics.XDSClientResourceStats{ - Authority: a.name, + Authority: auth.name, ResourceType: typeURL, CacheState: cacheState, Count: int64(count), @@ -499,12 +499,10 @@ func (c *XDSClient) reportResourceStats(rec clients.AsyncMetricsRecorder) { } } } - if c.topLevelAuthority != nil { reportForAuthority(c.topLevelAuthority) } - for _, a := range c.authorities { - reportForAuthority(a) + for _, auth := range c.authorities { + reportForAuthority(auth) } } - From 67a79937c8ac7fb50920541344f7986de812a76c Mon Sep 17 00:00:00 2001 From: Madhav Bissa <48023579+mbissa@users.noreply.github.com> Date: Fri, 20 Mar 2026 14:29:05 +0530 Subject: [PATCH 04/10] Apply suggestion from @gemini-code-assist[bot] Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> --- internal/xds/clients/xdsclient/xdsclient.go | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/internal/xds/clients/xdsclient/xdsclient.go b/internal/xds/clients/xdsclient/xdsclient.go index a97a2dd5cf56..2d5cc82784d9 100644 --- a/internal/xds/clients/xdsclient/xdsclient.go +++ b/internal/xds/clients/xdsclient/xdsclient.go @@ -486,12 +486,12 @@ func (c *XDSClient) reportConnectedState(rec clients.AsyncMetricsRecorder) { func (c *XDSClient) reportResourceStats(rec clients.AsyncMetricsRecorder) { reportForAuthority := func(auth *authority) { stats := auth.resourceStats() - for typeURL, stateCounts := range stats { - for cacheState, count := range stateCounts { - if count > 0 { - rec.ReportMetric(&metrics.XDSClientResourceStats{ - Authority: auth.name, - ResourceType: typeURL, + for resourceType, stateCounts := range stats { + for cacheState, count := range stateCounts { + if count > 0 { + rec.ReportMetric(&metrics.XDSClientResourceStats{ + Authority: auth.name, + ResourceType: resourceType, CacheState: cacheState, Count: int64(count), }) From 795c5d197f53799920583b6064e22bb9926533dc Mon Sep 17 00:00:00 2001 From: Madhav Bissa Date: Fri, 20 Mar 2026 07:33:47 +0000 Subject: [PATCH 05/10] review comments to fix metric determination logic --- internal/xds/clients/xdsclient/ads_stream.go | 26 +++- .../xds/clients/xdsclient/metrics/metrics.go | 2 +- .../clients/xdsclient/test/metrics_test.go | 144 ++++++++++++++++++ 3 files changed, 164 insertions(+), 8 deletions(-) diff --git a/internal/xds/clients/xdsclient/ads_stream.go b/internal/xds/clients/xdsclient/ads_stream.go index 45e020dd6e8c..93142064eda3 100644 --- a/internal/xds/clients/xdsclient/ads_stream.go +++ b/internal/xds/clients/xdsclient/ads_stream.go @@ -109,11 +109,12 @@ type adsStreamImpl struct { notifySender chan struct{} // To notify the sending goroutine of a pending request. // Guards access to the below fields (and to the contents of the map). - mu sync.Mutex - resourceTypeState map[ResourceType]*resourceTypeState // Map of resource types to their state. - firstRequest bool // False after the first request is sent out. - streamEstablished bool // True after a successful NewStream call. - pendingRequests []request // Subscriptions and unsubscriptions are pushed here. + mu sync.Mutex + resourceTypeState map[ResourceType]*resourceTypeState // Map of resource types to their state. + firstRequest bool // False after the first request is sent out. + firstStreamCreated bool // Set to true after the very first ADS stream is created. + streamEstablished bool // Set to true when an ADS stream is established and a response is received, except for the very first stream which is set to true immediately. + pendingRequests []request // Subscriptions and unsubscriptions are pushed here. } // adsStreamOpts contains the options for creating a new ADS Stream. @@ -242,6 +243,9 @@ func (s *adsStreamImpl) runner(ctx context.Context) { stream, err := s.transport.NewStream(ctx, "/envoy.service.discovery.v3.AggregatedDiscoveryService/StreamAggregatedResources") if err != nil { s.logger.Warningf("Failed to create a new ADS streaming RPC: %v", err) + s.mu.Lock() + s.streamEstablished = false + s.mu.Unlock() s.onError(err, false) return nil } @@ -251,7 +255,10 @@ func (s *adsStreamImpl) runner(ctx context.Context) { s.mu.Lock() s.firstRequest = true - s.streamEstablished = true + if !s.firstStreamCreated { + s.streamEstablished = true + s.firstStreamCreated = true + } s.mu.Unlock() // Ensure that the most recently created stream is pushed on the @@ -440,12 +447,17 @@ func (s *adsStreamImpl) recv(stream clients.Stream) bool { resources, url, version, nonce, err := s.recvMessage(stream) if err != nil { s.mu.Lock() - s.streamEstablished = false + if !msgReceived { + s.streamEstablished = false + } s.mu.Unlock() s.onError(err, msgReceived) s.logger.Warningf("ADS stream closed: %v", err) return msgReceived } + s.mu.Lock() + s.streamEstablished = true + s.mu.Unlock() msgReceived = true // Invoke the onResponse event handler to parse the incoming message and diff --git a/internal/xds/clients/xdsclient/metrics/metrics.go b/internal/xds/clients/xdsclient/metrics/metrics.go index d7aaa95053af..4a2a5a01c279 100644 --- a/internal/xds/clients/xdsclient/metrics/metrics.go +++ b/internal/xds/clients/xdsclient/metrics/metrics.go @@ -52,7 +52,7 @@ type XDSClientConnected struct { Value int64 } -// XDSClientResourceStats reports the number of resources currently cached. +// 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 type XDSClientResourceStats struct { diff --git a/internal/xds/clients/xdsclient/test/metrics_test.go b/internal/xds/clients/xdsclient/test/metrics_test.go index e060b4f98737..89e259d95788 100644 --- a/internal/xds/clients/xdsclient/test/metrics_test.go +++ b/internal/xds/clients/xdsclient/test/metrics_test.go @@ -24,6 +24,7 @@ import ( "net" "sync" "testing" + "time" "github.com/google/uuid" "google.golang.org/grpc/credentials/insecure" @@ -655,3 +656,146 @@ func (w *testWatcher) ResourceError(err error, onDone func()) { onDone() } func (w *testWatcher) AmbientError(_ error, onDone func()) { onDone() } +func (s) TestConnectedMetric_Reconnection(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) + defer cancel() + + tmr := newTestMetricsReporter() + l, err := testutils.LocalTCPListener() + if err != nil { + t.Fatalf("net.Listen() failed: %v", err) + } + lis := testutils.NewRestartableListener(l) + + // Use a channel to control when the server sends a response. + 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) + return nil + }, + OnStreamRequest: func(streamID int64, req *v3discoverypb.DiscoveryRequest) error { + t.Logf("ADS stream request received, streamID: %d", streamID) + // For all streams, wait until we are told to send a response. + <-sendResponse + return nil + }, + }) + 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()}} + xdsClientConfig := xdsclient.Config{ + Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}}, + Node: clients.Node{ID: nodeID}, + TransportBuilder: grpctransport.NewBuilder(configs), + ResourceTypes: resourceTypes, + Authorities: map[string]xdsclient.Authority{ + "": {XDSServers: []xdsclient.ServerConfig{}}, + }, + 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) + } + defer client.Close() + + const listenerName = "test-listener-resource" + 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) + } + + // Start watching. This will create the channel entry in xdsActiveChannels. + client.WatchResource(listenerType.TypeURL, listenerName, noopListenerWatcher{}) + + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 0}); err != nil { + t.Fatalf("Step 1 failed: Expected XDSClientConnected to be 0 at start, got: %v", err) + } + + // 2. 1st NewStream OK - metric value 1 + lis.Restart() + + // Wait a bit for the stream to be created. + time.Sleep(1 * time.Second) + + tmr.metricsCh.Drain() + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 1}); err != nil { + t.Fatalf("Step 2 failed: Expected XDSClientConnected to be 1 after 1st NewStream, got: %v", err) + } + + // Allow the first response. + close(sendResponse) + if err := tmr.waitForSpecificMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { + t.Fatal(err.Error()) + } + + // 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()) + } + + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 0}); err != nil { + t.Fatalf("Step 3 failed: Expected XDSClientConnected to be 0 after failure, got: %v", err) + } + + // Prepare for the second stream. + sendResponse = make(chan struct{}) + lis.Restart() + + // 4. 2nd NewStream OK - metric value 0 + // Wait for the client to attempt a new stream. + time.Sleep(5 * time.Second) + + tmr.metricsCh.Drain() + tmr.triggerAsyncMetrics() + sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout) + defer sCancel() + got, err := tmr.metricsCh.Receive(sCtx) + if err != nil { + t.Fatalf("Step 4 failed: Timeout waiting for XDSClientConnected metric: %v", err) + } + t.Logf("Got metric: %+v", got) + if m, ok := got.(*metrics.XDSClientConnected); ok { + if m.Value != 0 { + // This is where it will fail with ORIGINAL code. + t.Fatalf("Step 4 failed: Expected XDSClientConnected to be 0 after 2nd NewStream but before response, got: %d", m.Value) + } + } else { + t.Fatalf("Expected XDSClientConnected metric, got: %T", got) + } + + // 5. 1st Response Recv - metric value 1 + close(sendResponse) + + // Wait for the update to ensure we received a response. + if err := tmr.waitForSpecificMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { + t.Fatal(err.Error()) + } + + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 1}); err != nil { + t.Fatalf("Step 5 failed: Expected XDSClientConnected to be 1 after response, got: %v", err) + } +} From f397493d0a6a25c8d69dffe44b74afd6df3fd4c3 Mon Sep 17 00:00:00 2001 From: Madhav Bissa Date: Fri, 20 Mar 2026 09:30:33 +0000 Subject: [PATCH 06/10] review comments --- internal/xds/clients/config.go | 8 +- internal/xds/clients/xdsclient/authority.go | 24 +++--- .../xds/clients/xdsclient/metrics/metrics.go | 2 + .../clients/xdsclient/test/metrics_test.go | 86 ++++++++++++++++--- internal/xds/clients/xdsclient/xdsclient.go | 34 +++++--- internal/xds/xdsclient/clientimpl.go | 4 +- 6 files changed, 117 insertions(+), 41 deletions(-) diff --git a/internal/xds/clients/config.go b/internal/xds/clients/config.go index 2d74966a6773..aa89d2b1cee6 100644 --- a/internal/xds/clients/config.go +++ b/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) } diff --git a/internal/xds/clients/xdsclient/authority.go b/internal/xds/clients/xdsclient/authority.go index f9574c30e6b8..98d3165bf4c6 100644 --- a/internal/xds/clients/xdsclient/authority.go +++ b/internal/xds/clients/xdsclient/authority.go @@ -935,27 +935,29 @@ 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 }) @@ -963,8 +965,8 @@ func (a *authority) resourceStats() map[string]map[string]int { 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" diff --git a/internal/xds/clients/xdsclient/metrics/metrics.go b/internal/xds/clients/xdsclient/metrics/metrics.go index 4a2a5a01c279..a9ae0adb3d14 100644 --- a/internal/xds/clients/xdsclient/metrics/metrics.go +++ b/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 diff --git a/internal/xds/clients/xdsclient/test/metrics_test.go b/internal/xds/clients/xdsclient/test/metrics_test.go index 89e259d95788..9a21be137f54 100644 --- a/internal/xds/clients/xdsclient/test/metrics_test.go +++ b/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) + } +} diff --git a/internal/xds/clients/xdsclient/xdsclient.go b/internal/xds/clients/xdsclient/xdsclient.go index 2d5cc82784d9..ad82e7fd126f 100644 --- a/internal/xds/clients/xdsclient/xdsclient.go +++ b/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,14 +490,18 @@ 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), }) @@ -499,10 +509,8 @@ func (c *XDSClient) reportResourceStats(rec clients.AsyncMetricsRecorder) { } } } - if c.topLevelAuthority != nil { - reportForAuthority(c.topLevelAuthority) - } - for _, auth := range c.authorities { - reportForAuthority(auth) + reportForAuthority(c.topLevelAuthority) + for _, a := range c.authorities { + reportForAuthority(a) } } diff --git a/internal/xds/xdsclient/clientimpl.go b/internal/xds/xdsclient/clientimpl.go index 59d376044164..913899dde693 100644 --- a/internal/xds/xdsclient/clientimpl.go +++ b/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"}, From 7be03bf56223984116db700178db0b460d124367 Mon Sep 17 00:00:00 2001 From: Madhav Bissa Date: Sun, 22 Mar 2026 08:23:17 +0000 Subject: [PATCH 07/10] making the test more verbose and leak proof --- .../clients/xdsclient/test/metrics_test.go | 86 ++++++++++++++----- 1 file changed, 64 insertions(+), 22 deletions(-) diff --git a/internal/xds/clients/xdsclient/test/metrics_test.go b/internal/xds/clients/xdsclient/test/metrics_test.go index 9a21be137f54..7869c39cad9d 100644 --- a/internal/xds/clients/xdsclient/test/metrics_test.go +++ b/internal/xds/clients/xdsclient/test/metrics_test.go @@ -27,6 +27,7 @@ import ( "time" "github.com/google/uuid" + "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/internal/testutils" "google.golang.org/grpc/internal/xds/clients" @@ -675,10 +676,14 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) { return nil }, OnStreamRequest: func(_ int64, _ *v3discoverypb.DiscoveryRequest) error { - // For all streams, wait until we are told to send a response. - <-sendResponse return nil }, + OnStreamResponse: func(ctx context.Context, _ int64, _ *v3discoverypb.DiscoveryRequest, _ *v3discoverypb.DiscoveryResponse) { + select { + case <-sendResponse: + case <-ctx.Done(): + } + }, }) nodeID := uuid.New().String() @@ -687,7 +692,28 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) { ServerURI: mgmtServer.Address, Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}, } - configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}} + blockNextStream := make(chan struct{}, 1) + blocked := make(chan struct{}) + unblock := make(chan struct{}) + + customGRPCNewClient := func(target string, opts ...grpc.DialOption) (*grpc.ClientConn, error) { + interceptor := func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, cops ...grpc.CallOption) (grpc.ClientStream, error) { + select { + case <-blockNextStream: + select { + case blocked <- struct{}{}: + default: + } + <-unblock + default: + } + return streamer(ctx, desc, cc, method, cops...) + } + opts = append(opts, grpc.WithStreamInterceptor(interceptor)) + return grpc.NewClient(target, opts...) + } + + configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle(), GRPCNewClient: customGRPCNewClient}} xdsClientConfig := xdsclient.Config{ Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}}, Node: clients.Node{ID: nodeID}, @@ -730,13 +756,21 @@ 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) - - tmr.metricsCh.Drain() - tmr.triggerAsyncMetrics() - if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 1}); err != nil { - t.Fatalf("Step 2 failed: Expected XDSClientConnected to be 1 after 1st NewStream, got: %v", err) + // Wait for the stream to be created. Try polling the async metric a few times to prevent flakes. + var step2Passed bool + for i := 0; i < 50; i++ { + tmr.metricsCh.Drain() + tmr.triggerAsyncMetrics() + sCtx, sCancel := context.WithTimeout(ctx, 100*time.Millisecond) + err := tmr.waitForSpecificMetric(sCtx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 1}) + sCancel() + if err == nil { + step2Passed = true + break + } + } + if !step2Passed { + t.Fatalf("Step 2 failed: Expected XDSClientConnected to be 1 after 1st NewStream") } // Allow the first response. @@ -745,25 +779,35 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) { t.Fatal(err.Error()) } - // 3. Stream Fails - metric value 0 + // 3. Stream Fails - metric value 1 + blockNextStream <- struct{}{} lis.Stop() - // Wait for disconnect to be detected. + <-blocked + + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 1}); err != nil { + t.Fatalf("Step 3 failed: Expected XDSClientConnected to be 1 while NewStream is blocked, got: %v", err) + } + + // 4. NewStream Fails - metric value 0 + close(unblock) + if err := tmr.waitForSpecificMetric(ctx, &metrics.ServerFailure{ServerURI: mgmtServer.Address}); err != nil { t.Fatal(err.Error()) } tmr.triggerAsyncMetrics() if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 0}); err != nil { - t.Fatalf("Step 3 failed: Expected XDSClientConnected to be 0 after failure, got: %v", err) + t.Fatalf("Step 4 failed: Expected XDSClientConnected to be 0 after NewStream failure, got: %v", err) } - // Prepare for the second stream. + // Prepare for the next stream. sendResponse = make(chan struct{}) + + // 5. NewStream OK - metric value 0 lis.Restart() - // 4. 2nd NewStream OK - metric value 0 - // Wait for the client to attempt a new stream. time.Sleep(5 * time.Second) tmr.metricsCh.Drain() @@ -772,29 +816,27 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) { defer sCancel() got, err := tmr.metricsCh.Receive(sCtx) if err != nil { - t.Fatalf("Step 4 failed: Timeout waiting for XDSClientConnected metric: %v", err) + t.Fatalf("Step 5 failed: Timeout waiting for XDSClientConnected metric: %v", err) } t.Logf("Got metric: %+v", got) if m, ok := got.(*metrics.XDSClientConnected); ok { if m.Value != 0 { - // This is where it will fail with ORIGINAL code. - t.Fatalf("Step 4 failed: Expected XDSClientConnected to be 0 after 2nd NewStream but before response, got: %d", m.Value) + t.Fatalf("Step 5 failed: Expected XDSClientConnected to be 0 after successful NewStream but before response, got: %d", m.Value) } } else { t.Fatalf("Expected XDSClientConnected metric, got: %T", got) } - // 5. 1st Response Recv - metric value 1 + // 6. 1st Response Recv - metric value 1 close(sendResponse) - // Wait for the update to ensure we received a response. if err := tmr.waitForSpecificMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { t.Fatal(err.Error()) } tmr.triggerAsyncMetrics() if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 1}); err != nil { - t.Fatalf("Step 5 failed: Expected XDSClientConnected to be 1 after response, got: %v", err) + t.Fatalf("Step 6 failed: Expected XDSClientConnected to be 1 after response, got: %v", err) } } From 24976f71c577acf20df9e72eeab66075fd566422 Mon Sep 17 00:00:00 2001 From: Madhav Bissa Date: Wed, 25 Mar 2026 16:13:34 +0000 Subject: [PATCH 08/10] review comments --- internal/xds/clients/xdsclient/ads_stream.go | 45 ++++++++----------- .../xds/clients/xdsclient/metrics/metrics.go | 11 +++-- .../clients/xdsclient/test/helpers_test.go | 17 +++++-- internal/xds/xdsclient/clientimpl.go | 2 +- 4 files changed, 38 insertions(+), 37 deletions(-) diff --git a/internal/xds/clients/xdsclient/ads_stream.go b/internal/xds/clients/xdsclient/ads_stream.go index 93142064eda3..95c5d88b6fed 100644 --- a/internal/xds/clients/xdsclient/ads_stream.go +++ b/internal/xds/clients/xdsclient/ads_stream.go @@ -22,6 +22,7 @@ import ( "context" "fmt" "sync" + "sync/atomic" "time" "google.golang.org/grpc/grpclog" @@ -109,12 +110,14 @@ type adsStreamImpl struct { notifySender chan struct{} // To notify the sending goroutine of a pending request. // Guards access to the below fields (and to the contents of the map). - mu sync.Mutex - resourceTypeState map[ResourceType]*resourceTypeState // Map of resource types to their state. - firstRequest bool // False after the first request is sent out. - firstStreamCreated bool // Set to true after the very first ADS stream is created. - streamEstablished bool // Set to true when an ADS stream is established and a response is received, except for the very first stream which is set to true immediately. - pendingRequests []request // Subscriptions and unsubscriptions are pushed here. + mu sync.Mutex + resourceTypeState map[ResourceType]*resourceTypeState // Map of resource types to their state. + pendingRequests []request // Subscriptions and unsubscriptions are pushed here. + + // The following fields are accessed atomically. + firstRequest atomic.Bool // False after the first request is sent out. + firstStreamCreated atomic.Bool // Set to true after the very first ADS stream is created. + streamEstablished atomic.Bool // Set to true when an ADS stream is established and a response is received, except for the very first stream which is set to true immediately. } // adsStreamOpts contains the options for creating a new ADS Stream. @@ -243,9 +246,7 @@ func (s *adsStreamImpl) runner(ctx context.Context) { stream, err := s.transport.NewStream(ctx, "/envoy.service.discovery.v3.AggregatedDiscoveryService/StreamAggregatedResources") if err != nil { s.logger.Warningf("Failed to create a new ADS streaming RPC: %v", err) - s.mu.Lock() - s.streamEstablished = false - s.mu.Unlock() + s.streamEstablished.Store(false) s.onError(err, false) return nil } @@ -253,13 +254,11 @@ func (s *adsStreamImpl) runner(ctx context.Context) { s.logger.Infof("ADS stream created") } - s.mu.Lock() - s.firstRequest = true - if !s.firstStreamCreated { - s.streamEstablished = true - s.firstStreamCreated = true + s.firstRequest.Store(true) + if !s.firstStreamCreated.Load() { + s.streamEstablished.Store(true) + s.firstStreamCreated.Store(true) } - s.mu.Unlock() // Ensure that the most recently created stream is pushed on the // channel for the `send` goroutine to consume. @@ -392,7 +391,7 @@ func (s *adsStreamImpl) sendMessageLocked(stream clients.Stream, names []string, // The xDS protocol only requires that we send the node proto in the first // discovery request on every stream. Sending the node proto in every // request wastes CPU resources on the client and the server. - if s.firstRequest { + if s.firstRequest.Load() { req.Node = s.nodeProto } @@ -411,7 +410,7 @@ func (s *adsStreamImpl) sendMessageLocked(stream clients.Stream, names []string, s.logger.Warningf("Sending ADS request for type %q, resources: %v, version: %q, nonce: %q failed: %v", url, names, version, nonce, err) return err } - s.firstRequest = false + s.firstRequest.Store(false) if s.logger.V(perRPCVerbosityLevel) { s.logger.Infof("ADS request sent: %v", pretty.ToJSON(req)) @@ -446,18 +445,14 @@ func (s *adsStreamImpl) recv(stream clients.Stream) bool { resources, url, version, nonce, err := s.recvMessage(stream) if err != nil { - s.mu.Lock() if !msgReceived { - s.streamEstablished = false + s.streamEstablished.Store(false) } - s.mu.Unlock() s.onError(err, msgReceived) s.logger.Warningf("ADS stream closed: %v", err) return msgReceived } - s.mu.Lock() - s.streamEstablished = true - s.mu.Unlock() + s.streamEstablished.Store(true) msgReceived = true // Invoke the onResponse event handler to parse the incoming message and @@ -734,7 +729,5 @@ func (fc *adsFlowControl) wait() bool { } func (s *adsStreamImpl) isStreamEstablished() bool { - s.mu.Lock() - defer s.mu.Unlock() - return s.streamEstablished + return s.streamEstablished.Load() } diff --git a/internal/xds/clients/xdsclient/metrics/metrics.go b/internal/xds/clients/xdsclient/metrics/metrics.go index a9ae0adb3d14..fa1277cf33b6 100644 --- a/internal/xds/clients/xdsclient/metrics/metrics.go +++ b/internal/xds/clients/xdsclient/metrics/metrics.go @@ -45,18 +45,17 @@ 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 +// Per gRFC A78, Value is 1 if connected, and 0 otherwise. +// This metric provides the labels grpc.target and grpc.xds.server. type XDSClientConnected struct { ServerURI string Value int64 } -// XDSClientResourceStats reports the current cache states of xDS resources +// 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 +// This metric provides the labels grpc.target, grpc.xds.authority, +// grpc.xds.cache_state, and grpc.xds.resource_type. type XDSClientResourceStats struct { Authority string ResourceType string diff --git a/internal/xds/clients/xdsclient/test/helpers_test.go b/internal/xds/clients/xdsclient/test/helpers_test.go index fc76d8466fc4..84573e464e21 100644 --- a/internal/xds/clients/xdsclient/test/helpers_test.go +++ b/internal/xds/clients/xdsclient/test/helpers_test.go @@ -308,6 +308,14 @@ func (r *testMetricsReporter) waitForMetric(ctx context.Context, metricsDataWant return nil } +// waitForSpecificMetric waits for a specific metric to be recorded, ignoring +// and discarding any other metrics received in the meantime. Returns an error +// if the context expires before the requested metric is received. +// +// This is necessary when multiple distinct metrics (e.g., ServerFailure, +// ResourceUpdateValid, and XDSClientConnected) might be emitted asynchronously +// and concurrently. Using waitForMetric would fail the test if an unrelated +// metric happens to be at the front of the channel. func (r *testMetricsReporter) waitForSpecificMetric(ctx context.Context, metricsDataWant any) error { for { got, err := r.metricsCh.Receive(ctx) @@ -317,8 +325,6 @@ func (r *testMetricsReporter) waitForSpecificMetric(ctx context.Context, metrics if diff := cmp.Diff(got, metricsDataWant); diff == "" { return nil } - // Continue if mismatch. - } } @@ -331,13 +337,16 @@ func (r *testMetricsReporter) RegisterAsyncReporter(reporter clients.AsyncReport r.mu.Lock() defer r.mu.Unlock() r.asyncReporters[reporter] = struct{}{} - return func() { + return sync.OnceFunc(func() { r.mu.Lock() defer r.mu.Unlock() delete(r.asyncReporters, reporter) - } + }) } +// triggerAsyncMetrics invokes the Report method on all registered asynchronous +// metric reporters. This allows tests to explicitly compel the emission of +// async gauge metrics at specific precise points in time for verification. func (r *testMetricsReporter) triggerAsyncMetrics() { r.mu.Lock() defer r.mu.Unlock() diff --git a/internal/xds/xdsclient/clientimpl.go b/internal/xds/xdsclient/clientimpl.go index 913899dde693..a32b02375ce6 100644 --- a/internal/xds/xdsclient/clientimpl.go +++ b/internal/xds/xdsclient/clientimpl.go @@ -77,7 +77,7 @@ var ( }) xdsClientConnectedMetric = estats.RegisterInt64AsyncGauge(estats.MetricDescriptor{ Name: "grpc.xds_client.connected", - Description: "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 has a working ADS stream to the server, 0 otherwise.", Unit: "{connected}", Type: estats.MetricTypeIntAsyncGauge, Labels: []string{"grpc.target", "grpc.xds.server"}, From e51f8ef9c544869275acb5f7b12fc82571e39f9a Mon Sep 17 00:00:00 2001 From: Madhav Bissa Date: Tue, 31 Mar 2026 11:02:18 +0000 Subject: [PATCH 09/10] review comments --- .../clients/xdsclient/test/helpers_test.go | 6 + .../clients/xdsclient/test/metrics_test.go | 284 +++++++++--------- 2 files changed, 154 insertions(+), 136 deletions(-) diff --git a/internal/xds/clients/xdsclient/test/helpers_test.go b/internal/xds/clients/xdsclient/test/helpers_test.go index 84573e464e21..77331f85b66c 100644 --- a/internal/xds/clients/xdsclient/test/helpers_test.go +++ b/internal/xds/clients/xdsclient/test/helpers_test.go @@ -354,3 +354,9 @@ func (r *testMetricsReporter) triggerAsyncMetrics() { reporter.Report(r) } } + +func (r *testMetricsReporter) numAsyncReporters() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.asyncReporters) +} diff --git a/internal/xds/clients/xdsclient/test/metrics_test.go b/internal/xds/clients/xdsclient/test/metrics_test.go index 7869c39cad9d..2df581581566 100644 --- a/internal/xds/clients/xdsclient/test/metrics_test.go +++ b/internal/xds/clients/xdsclient/test/metrics_test.go @@ -24,7 +24,6 @@ import ( "net" "sync" "testing" - "time" "github.com/google/uuid" "google.golang.org/grpc" @@ -99,7 +98,7 @@ func (s) TestResourceUpdateMetrics(t *testing.T) { // cause a resource update valid metric to emit eventually. client.WatchResource(listenerType.TypeURL, listenerResourceName, noopListenerWatcher{}) if err := tmr.waitForMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { - t.Fatal(err.Error()) + t.Fatal(err) } // Update management server with a bad update. This should cause a resource @@ -114,7 +113,7 @@ func (s) TestResourceUpdateMetrics(t *testing.T) { t.Fatalf("Failed to update management server with resources: %v, err: %v", resources, err) } if err := tmr.waitForMetric(ctx, &metrics.ResourceUpdateInvalid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { - t.Fatal(err.Error()) + t.Fatal(err) } // Resource update valid metric should have not emitted. @@ -202,7 +201,7 @@ func (s) TestServerFailureMetrics_BeforeResponseRecv(t *testing.T) { lis.Restart() if err := tmr.waitForMetric(ctx, &metrics.ServerFailure{ServerURI: mgmtServer.Address}); err != nil { - t.Fatal(err.Error()) + t.Fatal(err) } } @@ -293,7 +292,7 @@ func (s) TestServerFailureMetrics_AfterResponseRecv(t *testing.T) { // cause a resource update valid metric to emit eventually. client.WatchResource(listenerType.TypeURL, listenerResourceName, noopListenerWatcher{}) if err := tmr.waitForMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { - t.Fatal(err.Error()) + t.Fatal(err) } // When the client sends an ACK, the management server would reply with an @@ -311,10 +310,14 @@ func (s) TestServerFailureMetrics_AfterResponseRecv(t *testing.T) { // successfully. close(streamCreationQuota) if err := tmr.waitForMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { - t.Fatal(err.Error()) + t.Fatal(err) } } +// TestConnectedMetric verifies the "grpc.xds_client.connected" metric state transitions. +// It begins by ensuring no metrics are reported before connection is attempted. +// Then it establishes a connection by watching a valid resource and verifies the connected state pulses to 1. +// Finally, it stops the management server and verifies the state drops back to 0. func (s) TestConnectedMetric(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) defer cancel() @@ -328,17 +331,18 @@ func (s) TestConnectedMetric(t *testing.T) { mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{Listener: l}) 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()}} xdsClientConfig := xdsclient.Config{ - Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}}, - Node: clients.Node{ID: nodeID}, - TransportBuilder: grpctransport.NewBuilder(configs), - ResourceTypes: resourceTypes, + Servers: []xdsclient.ServerConfig{{ + ServerIdentifier: clients.ServerIdentifier{ + ServerURI: mgmtServer.Address, + Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}, + }, + }}, + Node: clients.Node{ID: nodeID}, + TransportBuilder: grpctransport.NewBuilder(map[string]grpctransport.Config{ + "insecure": {Credentials: insecure.NewBundle()}, + }), + ResourceTypes: map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}, Authorities: map[string]xdsclient.Authority{ "": {XDSServers: []xdsclient.ServerConfig{}}, }, @@ -350,7 +354,6 @@ func (s) TestConnectedMetric(t *testing.T) { } defer client.Close() - // Verify initial state: No metrics reported because no stream is established. tmr.triggerAsyncMetrics() sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout) defer sCancel() @@ -358,10 +361,6 @@ func (s) TestConnectedMetric(t *testing.T) { t.Fatal("XDSClientConnected metric reported before any watch was started") } - // Watch a resource to trigger connection. - client.WatchResource(listenerType.TypeURL, "foo", noopListenerWatcher{}) - - // Wait for connection. const listenerName = "test-listener-resource" resources := e2e.UpdateOptions{ NodeID: nodeID, @@ -375,39 +374,32 @@ func (s) TestConnectedMetric(t *testing.T) { // Wait for the update to ensure we are connected. if err := tmr.waitForMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { - t.Fatal(err.Error()) + t.Fatal(err) } // Now trigger async metrics. tmr.triggerAsyncMetrics() if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 1}); err != nil { - t.Fatal(err.Error()) + t.Fatal(err) } - // Stop server to force disconnect. mgmtServer.Stop() - // Wait a bit for disconnect to be detected. - // xDS client should detect it on stream failure. - // We can wait for ServerFailure metric (synchronous). + // Wait for the synchronous server failure metric to confirm disconnect. if err := tmr.waitForSpecificMetric(ctx, &metrics.ServerFailure{ServerURI: mgmtServer.Address}); err != nil { - t.Fatal(err.Error()) + t.Fatal(err) } - // Trigger async metrics again after disconnect to verify Value is 0. tmr.triggerAsyncMetrics() if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 0}); err != nil { - t.Fatal(err.Error()) + t.Fatal(err) } - // Verify unregistration after client close. + // Verify async metric reporters are unregistered when client closes. client.Close() - tmr.mu.Lock() - numReporters := len(tmr.asyncReporters) - tmr.mu.Unlock() - if numReporters != 0 { - t.Fatalf("Async reporter not unregistered after client close, count: %d", numReporters) + if count := tmr.numAsyncReporters(); count != 0 { + t.Fatalf("Async reporter not unregistered after client close, count: %d", count) } // Drain the channel of any leftover metrics from previous pulses. @@ -415,13 +407,17 @@ func (s) TestConnectedMetric(t *testing.T) { tmr.triggerAsyncMetrics() // No metrics should be reported now because there are no reporters. - sCtx2, sCancel2 := context.WithTimeout(ctx, defaultTestShortTimeout) - defer sCancel2() - if _, err := tmr.metricsCh.Receive(sCtx2); err == nil { + sCtx, sCancel = context.WithTimeout(ctx, defaultTestShortTimeout) + defer sCancel() + if _, err := tmr.metricsCh.Receive(sCtx); err == nil { t.Fatal("Metrics reported after all reporters were unregistered") } } +// TestResourceMetrics verifies that the xDS client correctly tracks resource states (acked, nacked_but_cached) +// in "grpc.xds_client.resources" metric. It watches a resource, pushes a valid update from the management +// server, and asserts that the resource transitions to 'acked' state. Then it pushes an invalid update +// and asserts that the resource transitions to 'nacked_but_cached' state. func (s) TestResourceMetrics(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) defer cancel() @@ -435,17 +431,18 @@ func (s) TestResourceMetrics(t *testing.T) { mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{Listener: l}) 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()}} xdsClientConfig := xdsclient.Config{ - Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}}, - Node: clients.Node{ID: nodeID}, - TransportBuilder: grpctransport.NewBuilder(configs), - ResourceTypes: resourceTypes, + Servers: []xdsclient.ServerConfig{{ + ServerIdentifier: clients.ServerIdentifier{ + ServerURI: mgmtServer.Address, + Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}, + }, + }}, + Node: clients.Node{ID: nodeID}, + TransportBuilder: grpctransport.NewBuilder(map[string]grpctransport.Config{ + "insecure": {Credentials: insecure.NewBundle()}, + }), + ResourceTypes: map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}, Authorities: map[string]xdsclient.Authority{ "": {XDSServers: []xdsclient.ServerConfig{}}, }, @@ -460,7 +457,6 @@ func (s) TestResourceMetrics(t *testing.T) { const listenerName = "test-listener-resource" const routeConfigName = "test-route-configuration-resource" - // Requested state. client.WatchResource(listenerType.TypeURL, listenerName, noopListenerWatcher{}) resources := e2e.UpdateOptions{ @@ -472,24 +468,21 @@ func (s) TestResourceMetrics(t *testing.T) { t.Fatalf("Failed to update management server: %v", err) } - // Wait for Valid update. if err := tmr.waitForMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { - t.Fatal(err.Error()) + t.Fatal(err) } // Trigger async metrics. tmr.triggerAsyncMetrics() if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{ - Authority: "#old", // Default authority + Authority: "#old", // Default value for old-style non-xdstp as per gRFC A78 ResourceType: "ListenerResource", CacheState: "acked", Count: 1, }); err != nil { - t.Fatal(err.Error()) + t.Fatal(err) } - // Nacked but cached. - // Update with bad resource. resources.Listeners[0].ApiListener = nil if err := mgmtServer.Update(ctx, resources); err != nil { t.Fatalf("Failed to update management server: %v", err) @@ -497,7 +490,7 @@ func (s) TestResourceMetrics(t *testing.T) { // Wait for Invalid update. if err := tmr.waitForMetric(ctx, &metrics.ResourceUpdateInvalid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { - t.Fatal(err.Error()) + t.Fatal(err) } tmr.triggerAsyncMetrics() @@ -507,14 +500,15 @@ func (s) TestResourceMetrics(t *testing.T) { CacheState: "nacked_but_cached", Count: 1, }); err != nil { - t.Fatal(err.Error()) + t.Fatal(err) } } -// TestResourceMetrics_Extended tests multiple resources in different states: -// - 2 in "requested" state -// - 2 in "nacked" state -// - 1 in "does_not_exist" state +// TestResourceMetrics_Extended verifies complex resource state transitions (requested, does_not_exist, nacked) +// in "grpc.xds_client.resources" metric across multiple resources. It watches several resources, pushes a +// partial update (some valid, some invalid) to transition them to 'acked' and 'nacked' states. Then it pushes +// an empty authoritative update and asserts that active watchers transition back to 'requested' while omitted +// resources transition to 'does_not_exist'. func (s) TestResourceMetrics_Extended(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) defer cancel() @@ -528,17 +522,18 @@ func (s) TestResourceMetrics_Extended(t *testing.T) { mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{Listener: l}) 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()}} xdsClientConfig := xdsclient.Config{ - Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}}, - Node: clients.Node{ID: nodeID}, - TransportBuilder: grpctransport.NewBuilder(configs), - ResourceTypes: resourceTypes, + Servers: []xdsclient.ServerConfig{{ + ServerIdentifier: clients.ServerIdentifier{ + ServerURI: mgmtServer.Address, + Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}, + }, + }}, + Node: clients.Node{ID: nodeID}, + TransportBuilder: grpctransport.NewBuilder(map[string]grpctransport.Config{ + "insecure": {Credentials: insecure.NewBundle()}, + }), + ResourceTypes: map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}, Authorities: map[string]xdsclient.Authority{ "": {XDSServers: []xdsclient.ServerConfig{}}, }, @@ -550,14 +545,13 @@ func (s) TestResourceMetrics_Extended(t *testing.T) { } defer client.Close() - // Resource names resRequested1 := "res-requested-1" resRequested2 := "res-requested-2" resNacked1 := "res-nacked-1" resNacked2 := "res-nacked-2" - resNotExist := "res-not-exist" + resRemoved := "res-not-exist" - resList := []string{resRequested1, resRequested2, resNacked1, resNacked2, resNotExist} + resList := []string{resRequested1, resRequested2, resNacked1, resNacked2, resRemoved} for _, res := range resList { client.WatchResource(listenerType.TypeURL, res, noopListenerWatcher{}) } @@ -567,11 +561,10 @@ func (s) TestResourceMetrics_Extended(t *testing.T) { Listeners: []*v3listenerpb.Listener{ e2e.DefaultClientListener(resNacked1, "route-config"), e2e.DefaultClientListener(resNacked2, "route-config"), - e2e.DefaultClientListener(resNotExist, "route-config"), + e2e.DefaultClientListener(resRemoved, "route-config"), }, SkipValidation: true, } - // Make Nacked resources invalid resources.Listeners[0].ApiListener = nil resources.Listeners[1].ApiListener = nil @@ -579,9 +572,9 @@ func (s) TestResourceMetrics_Extended(t *testing.T) { t.Fatalf("Failed to update management server: %v", err) } - // Wait for Res5 to be valid. + // Wait for the resource update to be accepted. if err := tmr.waitForSpecificMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { - t.Fatal(err.Error()) + t.Fatal(err) } resourcesEmpty := e2e.UpdateOptions{ NodeID: nodeID, @@ -592,7 +585,6 @@ func (s) TestResourceMetrics_Extended(t *testing.T) { t.Fatalf("Failed to update management server: %v", err) } - // Verify "requested" count 2 tmr.triggerAsyncMetrics() if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{ Authority: "#old", @@ -603,7 +595,6 @@ func (s) TestResourceMetrics_Extended(t *testing.T) { t.Fatalf("Failed to verify requested count: %v", err) } - // Verify "nacked" count 2 tmr.triggerAsyncMetrics() if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{ Authority: "#old", @@ -614,12 +605,11 @@ func (s) TestResourceMetrics_Extended(t *testing.T) { t.Fatalf("Failed to verify nacked count: %v", err) } - // Wait for the does_not_exist state. - // We need to wait for the client to process the empty update and mark the resource as removed. - // We use a watcher to detect the specific error. + // Wait for the xDS client to process the empty update and trigger + // ResourceNotFound for omitted resources immediately. removed := make(chan struct{}) var closeOnce sync.Once - client.WatchResource(listenerType.TypeURL, resNotExist, &testWatcher{ + client.WatchResource(listenerType.TypeURL, resRemoved, &testWatcher{ onError: func(err error) { if xdsresource.ErrType(err) == xdsresource.ErrorTypeResourceNotFound { closeOnce.Do(func() { close(removed) }) @@ -627,7 +617,8 @@ func (s) TestResourceMetrics_Extended(t *testing.T) { }, }) - // Verify "does_not_exist" count 1 + // Verify that resources missing from the authoritative update transition to + // the does_not_exist state. select { case <-removed: case <-ctx.Done(): @@ -645,6 +636,9 @@ func (s) TestResourceMetrics_Extended(t *testing.T) { } } +// testWatcher is a test helper that implements the xdsclient resource watcher +// interface. It allows tests to intercept and handle `ResourceError` events +// via the optional `onError` callback, while ignoring other events. type testWatcher struct { onError func(error) } @@ -657,25 +651,30 @@ func (w *testWatcher) ResourceError(err error, onDone func()) { onDone() } func (w *testWatcher) AmbientError(_ error, onDone func()) { onDone() } + +// TestConnectedMetric_Reconnection verifies "grpc.xds_client.connected" metric accuracy during flaky network conditions. +// It uses a restartable listener and a dial interceptor to pause connection attempts. It asserts that the +// connected state pulses to 1 when a stream succeeds, remains 1 immediately after failures before retries, +// transitions to 0 after subsequent failures, and returns to 1 only after successful stream recreation. func (s) TestConnectedMetric_Reconnection(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) defer cancel() - tmr := newTestMetricsReporter() l, err := testutils.LocalTCPListener() if err != nil { t.Fatalf("net.Listen() failed: %v", err) } lis := testutils.NewRestartableListener(l) - // Use a channel to control when the server sends a response. sendResponse := make(chan struct{}) + streamOpened := make(chan struct{}, 10) mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{ Listener: lis, OnStreamOpen: func(_ context.Context, _ int64, _ string) error { - return nil - }, - OnStreamRequest: func(_ int64, _ *v3discoverypb.DiscoveryRequest) error { + select { + case streamOpened <- struct{}{}: + default: + } return nil }, OnStreamResponse: func(ctx context.Context, _ int64, _ *v3discoverypb.DiscoveryRequest, _ *v3discoverypb.DiscoveryResponse) { @@ -692,10 +691,15 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) { ServerURI: mgmtServer.Address, Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}, } + // blockNextStream is used to tell the interceptor to block the next stream creation attempt. blockNextStream := make(chan struct{}, 1) + // blocked is used by the interceptor to signal the test that it has successfully blocked the stream creation attempt. blocked := make(chan struct{}) + // unblock is used by the test to release the interceptor and allow the stream creation to proceed. unblock := make(chan struct{}) + // customGRPCNewClient overrides the gRPC client creation to inject a stream interceptor + // that can block ADS stream creation attempts for testing reconnection behavior. customGRPCNewClient := func(target string, opts ...grpc.DialOption) (*grpc.ClientConn, error) { interceptor := func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, cops ...grpc.CallOption) (grpc.ClientStream, error) { select { @@ -704,6 +708,7 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) { case blocked <- struct{}{}: default: } + // Pause stream creation until released by the test. <-unblock default: } @@ -713,6 +718,7 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) { return grpc.NewClient(target, opts...) } + tmr := newTestMetricsReporter() configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle(), GRPCNewClient: customGRPCNewClient}} xdsClientConfig := xdsclient.Config{ Servers: []xdsclient.ServerConfig{{ServerIdentifier: si}}, @@ -725,8 +731,7 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) { MetricsReporter: tmr, } - // 1. Initial Start - metric value 0 - // Keep the listener stopped initially so NewStream fails/blocks. + // Stop listener to force initial connection failure. lis.Stop() client, err := xdsclient.New(xdsClientConfig) @@ -745,41 +750,35 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) { t.Fatalf("Failed to update management server: %v", err) } - // Start watching. This will create the channel entry in xdsActiveChannels. + // Initiate watch to trigger connection attempt. client.WatchResource(listenerType.TypeURL, listenerName, noopListenerWatcher{}) tmr.triggerAsyncMetrics() if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 0}); err != nil { - t.Fatalf("Step 1 failed: Expected XDSClientConnected to be 0 at start, got: %v", err) + t.Fatalf("Expected XDSClientConnected to be 0 at start, got: %v", err) } - // 2. 1st NewStream OK - metric value 1 + // Verify state transitions to connected (Value: 1) when stream succeeds. lis.Restart() - // Wait for the stream to be created. Try polling the async metric a few times to prevent flakes. - var step2Passed bool - for i := 0; i < 50; i++ { - tmr.metricsCh.Drain() - tmr.triggerAsyncMetrics() - sCtx, sCancel := context.WithTimeout(ctx, 100*time.Millisecond) - err := tmr.waitForSpecificMetric(sCtx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 1}) - sCancel() - if err == nil { - step2Passed = true - break - } + select { + case <-streamOpened: + case <-ctx.Done(): + t.Fatalf("Timeout waiting for stream in Step 2") } - if !step2Passed { - t.Fatalf("Step 2 failed: Expected XDSClientConnected to be 1 after 1st NewStream") + + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 1}); err != nil { + t.Fatalf("Expected XDSClientConnected to be 1 after 1st NewStream, got: %v", err) } - // Allow the first response. + // Push resource update to confirm the connection is functional. close(sendResponse) if err := tmr.waitForSpecificMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { - t.Fatal(err.Error()) + t.Fatal(err) } - // 3. Stream Fails - metric value 1 + // Verify state remains connected (Value: 1) immediately after failure before NewStream retry. blockNextStream <- struct{}{} lis.Stop() @@ -787,28 +786,36 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) { tmr.triggerAsyncMetrics() if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 1}); err != nil { - t.Fatalf("Step 3 failed: Expected XDSClientConnected to be 1 while NewStream is blocked, got: %v", err) + t.Fatalf("Expected XDSClientConnected to be 1 while NewStream is blocked, got: %v", err) } - // 4. NewStream Fails - metric value 0 + // Verify state transitions to disconnected (Value: 0) after NewStream attempt fails. close(unblock) if err := tmr.waitForSpecificMetric(ctx, &metrics.ServerFailure{ServerURI: mgmtServer.Address}); err != nil { - t.Fatal(err.Error()) + t.Fatal(err) } tmr.triggerAsyncMetrics() if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 0}); err != nil { - t.Fatalf("Step 4 failed: Expected XDSClientConnected to be 0 after NewStream failure, got: %v", err) + t.Fatalf("Expected XDSClientConnected to be 0 after NewStream failure, got: %v", err) } - // Prepare for the next stream. sendResponse = make(chan struct{}) - // 5. NewStream OK - metric value 0 + // Verify state remains disconnected (Value: 0) while waiting for the first response on the new stream. lis.Restart() - time.Sleep(5 * time.Second) + // Clear channels to ensure fresh readings. + for len(streamOpened) > 0 { + <-streamOpened + } + + select { + case <-streamOpened: + case <-ctx.Done(): + t.Fatalf("Timeout waiting for stream in Step 5") + } tmr.metricsCh.Drain() tmr.triggerAsyncMetrics() @@ -816,48 +823,52 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) { defer sCancel() got, err := tmr.metricsCh.Receive(sCtx) if err != nil { - t.Fatalf("Step 5 failed: Timeout waiting for XDSClientConnected metric: %v", err) + t.Fatalf("Timeout waiting for XDSClientConnected metric: %v", err) } t.Logf("Got metric: %+v", got) if m, ok := got.(*metrics.XDSClientConnected); ok { if m.Value != 0 { - t.Fatalf("Step 5 failed: Expected XDSClientConnected to be 0 after successful NewStream but before response, got: %d", m.Value) + t.Fatalf("Expected XDSClientConnected to be 0 after successful NewStream but before response, got: %d", m.Value) } } else { t.Fatalf("Expected XDSClientConnected metric, got: %T", got) } - // 6. 1st Response Recv - metric value 1 + // Verify state returns to connected (Value: 1) after receiving the first response on the new stream. close(sendResponse) if err := tmr.waitForSpecificMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { - t.Fatal(err.Error()) + t.Fatal(err) } tmr.triggerAsyncMetrics() if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 1}); err != nil { - t.Fatalf("Step 6 failed: Expected XDSClientConnected to be 1 after response, got: %v", err) + t.Fatalf("Expected XDSClientConnected to be 1 after response, got: %v", err) } } +// TestResourceMetrics_AuthorityOldStyle verifies that the xDS client correctly falls back to '#old' in +// "grpc.xds_client.resources" metrics for legacy authorities. It watches a resource using default authority +// and asserts that the metric is reported under '#old' instead of an empty string. 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"}, + serverCfg := xdsclient.ServerConfig{ + ServerIdentifier: 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, + Servers: []xdsclient.ServerConfig{serverCfg}, + Node: clients.Node{ID: nodeID}, + TransportBuilder: grpctransport.NewBuilder(map[string]grpctransport.Config{ + "insecure": {Credentials: insecure.NewBundle()}, + }), + ResourceTypes: map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}, Authorities: map[string]xdsclient.Authority{ "": {XDSServers: []xdsclient.ServerConfig{serverCfg}}, }, @@ -888,9 +899,10 @@ func (s) TestResourceMetrics_AuthorityOldStyle(t *testing.T) { } if err := tmr.waitForMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { - t.Fatal(err.Error()) + t.Fatal(err) } + // Verify that the client substitutes empty authority with '#old' in metrics. tmr.triggerAsyncMetrics() if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{ Authority: "#old", From 32dc1cbfb2eda816037ca2139706ccbe3cce7b14 Mon Sep 17 00:00:00 2001 From: Madhav Bissa Date: Thu, 2 Apr 2026 23:30:11 +0000 Subject: [PATCH 10/10] review comments --- .../clients/xdsclient/test/helpers_test.go | 60 ++++-- .../clients/xdsclient/test/metrics_test.go | 194 +++++++++--------- 2 files changed, 139 insertions(+), 115 deletions(-) diff --git a/internal/xds/clients/xdsclient/test/helpers_test.go b/internal/xds/clients/xdsclient/test/helpers_test.go index 77331f85b66c..20225280bd5b 100644 --- a/internal/xds/clients/xdsclient/test/helpers_test.go +++ b/internal/xds/clients/xdsclient/test/helpers_test.go @@ -31,8 +31,8 @@ import ( "google.golang.org/grpc/internal/grpctest" "google.golang.org/grpc/internal/xds/clients" + "google.golang.org/grpc/internal/xds/clients/internal/buffer" "google.golang.org/grpc/internal/xds/clients/internal/pretty" - "google.golang.org/grpc/internal/xds/clients/internal/testutils" "google.golang.org/grpc/internal/xds/clients/xdsclient" "google.golang.org/grpc/internal/xds/clients/xdsclient/internal/xdsresource" "google.golang.org/protobuf/proto" @@ -280,7 +280,7 @@ func buildResourceName(typeName, auth, id string, ctxParams map[string]string) s // recording events on channels and provides helpers to check if certain events // have taken place. type testMetricsReporter struct { - metricsCh *testutils.Channel + metricsCh *buffer.Unbounded mu sync.Mutex asyncReporters map[clients.AsyncReporter]struct{} @@ -289,7 +289,7 @@ type testMetricsReporter struct { // newTestMetricsReporter returns a new testMetricsReporter. func newTestMetricsReporter() *testMetricsReporter { return &testMetricsReporter{ - metricsCh: testutils.NewChannelWithSize(50), + metricsCh: buffer.NewUnbounded(), asyncReporters: make(map[clients.AsyncReporter]struct{}), } } @@ -298,14 +298,16 @@ func newTestMetricsReporter() *testMetricsReporter { // recorded metrics data matches the expected metricsDataWant. Returns // an error if failed to wait or received wrong data. func (r *testMetricsReporter) waitForMetric(ctx context.Context, metricsDataWant any) error { - got, err := r.metricsCh.Receive(ctx) - if err != nil { - return fmt.Errorf("timeout waiting for int64Count") - } - if diff := cmp.Diff(got, metricsDataWant); diff != "" { - return fmt.Errorf("received unexpected metrics value (-got, +want): %v", diff) + select { + case got := <-r.metricsCh.Get(): + r.metricsCh.Load() + if diff := cmp.Diff(got, metricsDataWant); diff != "" { + return fmt.Errorf("received unexpected metrics value (-got, +want): %v", diff) + } + return nil + case <-ctx.Done(): + return fmt.Errorf("timeout waiting for metric check: %v", ctx.Err()) } - return nil } // waitForSpecificMetric waits for a specific metric to be recorded, ignoring @@ -318,19 +320,45 @@ func (r *testMetricsReporter) waitForMetric(ctx context.Context, metricsDataWant // metric happens to be at the front of the channel. func (r *testMetricsReporter) waitForSpecificMetric(ctx context.Context, metricsDataWant any) error { for { - got, err := r.metricsCh.Receive(ctx) - if err != nil { - return fmt.Errorf("timeout waiting for metric: %v (want %T)", err, metricsDataWant) + select { + case got := <-r.metricsCh.Get(): + r.metricsCh.Load() + if diff := cmp.Diff(got, metricsDataWant); diff == "" { + return nil + } + case <-ctx.Done(): + return fmt.Errorf("timeout waiting for specific metric: %v", ctx.Err()) } - if diff := cmp.Diff(got, metricsDataWant); diff == "" { - return nil + } +} + +// Receive waits for a metric to be recorded and returns it. Returns an error +// if the context expires before a metric is received. +func (r *testMetricsReporter) Receive(ctx context.Context) (any, error) { + select { + case got := <-r.metricsCh.Get(): + r.metricsCh.Load() + return got, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +// Drain clears all accumulated metrics from the channel. +func (r *testMetricsReporter) Drain() { + for { + select { + case <-r.metricsCh.Get(): + r.metricsCh.Load() + default: + return } } } // ReportMetric sends the metrics data to the metricsCh channel. func (r *testMetricsReporter) ReportMetric(m any) { - r.metricsCh.Send(m) + r.metricsCh.Put(m) } func (r *testMetricsReporter) RegisterAsyncReporter(reporter clients.AsyncReporter) func() { diff --git a/internal/xds/clients/xdsclient/test/metrics_test.go b/internal/xds/clients/xdsclient/test/metrics_test.go index 2df581581566..6df2b1bbf990 100644 --- a/internal/xds/clients/xdsclient/test/metrics_test.go +++ b/internal/xds/clients/xdsclient/test/metrics_test.go @@ -22,7 +22,6 @@ import ( "context" "errors" "net" - "sync" "testing" "github.com/google/uuid" @@ -314,10 +313,11 @@ func (s) TestServerFailureMetrics_AfterResponseRecv(t *testing.T) { } } -// TestConnectedMetric verifies the "grpc.xds_client.connected" metric state transitions. -// It begins by ensuring no metrics are reported before connection is attempted. -// Then it establishes a connection by watching a valid resource and verifies the connected state pulses to 1. -// Finally, it stops the management server and verifies the state drops back to 0. +// TestConnectedMetric verifies the "grpc.xds_client.connected" metric state +// transitions. It begins by ensuring no metrics are reported before connection +// is attempted. Then it establishes a connection by watching a valid resource +// and verifies the connected state pulses to 1. Finally, it stops the +// management server and verifies the state drops back to 0. func (s) TestConnectedMetric(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) defer cancel() @@ -403,21 +403,23 @@ func (s) TestConnectedMetric(t *testing.T) { } // Drain the channel of any leftover metrics from previous pulses. - tmr.metricsCh.Drain() + tmr.Drain() tmr.triggerAsyncMetrics() // No metrics should be reported now because there are no reporters. sCtx, sCancel = context.WithTimeout(ctx, defaultTestShortTimeout) defer sCancel() - if _, err := tmr.metricsCh.Receive(sCtx); err == nil { + if _, err := tmr.Receive(sCtx); err == nil { t.Fatal("Metrics reported after all reporters were unregistered") } } -// TestResourceMetrics verifies that the xDS client correctly tracks resource states (acked, nacked_but_cached) -// in "grpc.xds_client.resources" metric. It watches a resource, pushes a valid update from the management -// server, and asserts that the resource transitions to 'acked' state. Then it pushes an invalid update -// and asserts that the resource transitions to 'nacked_but_cached' state. +// TestResourceMetrics verifies that the xDS client correctly tracks resource +// states (acked, nacked_but_cached) in "grpc.xds_client.resources" metric. It +// watches a resource, pushes a valid update from the management server, and +// asserts that the resource transitions to 'acked' state. Then it pushes an +// invalid update and asserts that the resource transitions to +// 'nacked_but_cached' state. func (s) TestResourceMetrics(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) defer cancel() @@ -504,11 +506,13 @@ func (s) TestResourceMetrics(t *testing.T) { } } -// TestResourceMetrics_Extended verifies complex resource state transitions (requested, does_not_exist, nacked) -// in "grpc.xds_client.resources" metric across multiple resources. It watches several resources, pushes a -// partial update (some valid, some invalid) to transition them to 'acked' and 'nacked' states. Then it pushes -// an empty authoritative update and asserts that active watchers transition back to 'requested' while omitted -// resources transition to 'does_not_exist'. +// TestResourceMetrics_Extended verifies complex resource state transitions +// (requested, does_not_exist, nacked) in "grpc.xds_client.resources" metric +// across multiple resources. It watches several resources, pushes a partial +// update (some valid, some invalid) to transition them to 'acked' and 'nacked' +// states, then management server removes a resource and asserts that active +// watchers transition back to 'requested' while omitted resources transition +// to 'does_not_exist'. func (s) TestResourceMetrics_Extended(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) defer cancel() @@ -576,6 +580,7 @@ func (s) TestResourceMetrics_Extended(t *testing.T) { if err := tmr.waitForSpecificMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { t.Fatal(err) } + // Push an empty update to remove all resources from the management server. resourcesEmpty := e2e.UpdateOptions{ NodeID: nodeID, Listeners: []*v3listenerpb.Listener{}, @@ -607,22 +612,13 @@ func (s) TestResourceMetrics_Extended(t *testing.T) { // Wait for the xDS client to process the empty update and trigger // ResourceNotFound for omitted resources immediately. - removed := make(chan struct{}) - var closeOnce sync.Once - client.WatchResource(listenerType.TypeURL, resRemoved, &testWatcher{ - onError: func(err error) { - if xdsresource.ErrType(err) == xdsresource.ErrorTypeResourceNotFound { - closeOnce.Do(func() { close(removed) }) - } - }, - }) + lw := newListenerWatcher() + client.WatchResource(listenerType.TypeURL, resRemoved, lw) // Verify that resources missing from the authoritative update transition to // the does_not_exist state. - select { - case <-removed: - case <-ctx.Done(): - t.Fatal("timeout waiting for resource removal") + if err := verifyResourceErrorType(ctx, lw.resourceErrCh, xdsresource.ErrorTypeResourceNotFound, ""); err != nil { + t.Fatal(err) } tmr.triggerAsyncMetrics() @@ -636,26 +632,12 @@ func (s) TestResourceMetrics_Extended(t *testing.T) { } } -// testWatcher is a test helper that implements the xdsclient resource watcher -// interface. It allows tests to intercept and handle `ResourceError` events -// via the optional `onError` callback, while ignoring other events. -type testWatcher struct { - onError func(error) -} - -func (w *testWatcher) ResourceChanged(_ xdsclient.ResourceData, onDone func()) { onDone() } -func (w *testWatcher) ResourceError(err error, onDone func()) { - if w.onError != nil { - w.onError(err) - } - onDone() -} -func (w *testWatcher) AmbientError(_ error, onDone func()) { onDone() } - -// TestConnectedMetric_Reconnection verifies "grpc.xds_client.connected" metric accuracy during flaky network conditions. -// It uses a restartable listener and a dial interceptor to pause connection attempts. It asserts that the -// connected state pulses to 1 when a stream succeeds, remains 1 immediately after failures before retries, -// transitions to 0 after subsequent failures, and returns to 1 only after successful stream recreation. +// TestConnectedMetric_Reconnection verifies "grpc.xds_client.connected" metric +// accuracy during flaky network conditions. It uses a restartable listener and +// a dial interceptor to pause connection attempts. It asserts that the +// connected state pulses to 1 when a stream succeeds, remains 1 immediately +// after failures before retries, transitions to 0 after subsequent failures, +// and returns to 1 only after successful stream recreation. func (s) TestConnectedMetric_Reconnection(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) defer cancel() @@ -670,10 +652,10 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) { streamOpened := make(chan struct{}, 10) mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{ Listener: lis, - OnStreamOpen: func(_ context.Context, _ int64, _ string) error { + OnStreamOpen: func(ctx context.Context, _ int64, _ string) error { select { case streamOpened <- struct{}{}: - default: + case <-ctx.Done(): } return nil }, @@ -691,26 +673,37 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) { ServerURI: mgmtServer.Address, Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}, } - // blockNextStream is used to tell the interceptor to block the next stream creation attempt. - blockNextStream := make(chan struct{}, 1) - // blocked is used by the interceptor to signal the test that it has successfully blocked the stream creation attempt. + // streamAttemptCh is used by the interceptor to notify the test that a + // stream is attempting to open. + streamAttemptCh := testutils.NewChannel() + // commandCh is used by the test to tell the interceptor whether to block or pass. + commandCh := testutils.NewChannel() + // blocked is used by the interceptor to signal the test that it has + // successfully blocked the stream creation attempt. blocked := make(chan struct{}) - // unblock is used by the test to release the interceptor and allow the stream creation to proceed. + // unblock is used by the test to release the interceptor and allow the + // stream creation to proceed. unblock := make(chan struct{}) - // customGRPCNewClient overrides the gRPC client creation to inject a stream interceptor - // that can block ADS stream creation attempts for testing reconnection behavior. + // customGRPCNewClient overrides the gRPC client creation to inject a stream + // interceptor that can block ADS stream creation attempts for testing + // reconnection behavior. customGRPCNewClient := func(target string, opts ...grpc.DialOption) (*grpc.ClientConn, error) { interceptor := func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, cops ...grpc.CallOption) (grpc.ClientStream, error) { - select { - case <-blockNextStream: + streamAttemptCh.Send(struct{}{}) // Notify test + + cmd, err := commandCh.Receive(ctx) // Wait for command + if err != nil { + return nil, err + } + + if cmd.(string) == "block" { select { case blocked <- struct{}{}: - default: + case <-ctx.Done(): } // Pause stream creation until released by the test. <-unblock - default: } return streamer(ctx, desc, cc, method, cops...) } @@ -731,6 +724,19 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) { MetricsReporter: tmr, } + waitForStreamSuccess := func() { + for { + select { + case <-streamOpened: + return + case <-streamAttemptCh.C: + commandCh.Send("pass") + case <-ctx.Done(): + t.Fatalf("Timeout waiting for stream success") + } + } + } + // Stop listener to force initial connection failure. lis.Stop() @@ -755,21 +761,17 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) { tmr.triggerAsyncMetrics() if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 0}); err != nil { - t.Fatalf("Expected XDSClientConnected to be 0 at start, got: %v", err) + t.Fatalf("XDSClientConnected check failed at start - got: %v, want 0", err) } // Verify state transitions to connected (Value: 1) when stream succeeds. lis.Restart() - select { - case <-streamOpened: - case <-ctx.Done(): - t.Fatalf("Timeout waiting for stream in Step 2") - } + waitForStreamSuccess() tmr.triggerAsyncMetrics() if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 1}); err != nil { - t.Fatalf("Expected XDSClientConnected to be 1 after 1st NewStream, got: %v", err) + t.Fatalf("XDSClientConnected check failed after 1st NewStream - got: %v, want 1", err) } // Push resource update to confirm the connection is functional. @@ -778,18 +780,24 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) { t.Fatal(err) } - // Verify state remains connected (Value: 1) immediately after failure before NewStream retry. - blockNextStream <- struct{}{} + // Verify state remains connected (Value: 1) immediately after failure before + // NewStream retry. lis.Stop() + if _, err := streamAttemptCh.Receive(ctx); err != nil { + t.Fatalf("Timeout waiting for 2nd stream attempt (block): %v", err) + } + commandCh.Send("block") + <-blocked tmr.triggerAsyncMetrics() if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 1}); err != nil { - t.Fatalf("Expected XDSClientConnected to be 1 while NewStream is blocked, got: %v", err) + t.Fatalf("XDSClientConnected check failed while NewStream is blocked - got: %v, want 1", err) } - // Verify state transitions to disconnected (Value: 0) after NewStream attempt fails. + // Verify state transitions to disconnected (Value: 0) after NewStream attempt + // fails. close(unblock) if err := tmr.waitForSpecificMetric(ctx, &metrics.ServerFailure{ServerURI: mgmtServer.Address}); err != nil { @@ -798,12 +806,13 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) { tmr.triggerAsyncMetrics() if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 0}); err != nil { - t.Fatalf("Expected XDSClientConnected to be 0 after NewStream failure, got: %v", err) + t.Fatalf("XDSClientConnected check failed after NewStream failure - got: %v, want 0", err) } sendResponse = make(chan struct{}) - // Verify state remains disconnected (Value: 0) while waiting for the first response on the new stream. + // Verify state remains disconnected (Value: 0) while waiting for the first + // response on the new stream. lis.Restart() // Clear channels to ensure fresh readings. @@ -811,30 +820,16 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) { <-streamOpened } - select { - case <-streamOpened: - case <-ctx.Done(): - t.Fatalf("Timeout waiting for stream in Step 5") - } + waitForStreamSuccess() - tmr.metricsCh.Drain() + tmr.Drain() tmr.triggerAsyncMetrics() - sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout) - defer sCancel() - got, err := tmr.metricsCh.Receive(sCtx) - if err != nil { - t.Fatalf("Timeout waiting for XDSClientConnected metric: %v", err) - } - t.Logf("Got metric: %+v", got) - if m, ok := got.(*metrics.XDSClientConnected); ok { - if m.Value != 0 { - t.Fatalf("Expected XDSClientConnected to be 0 after successful NewStream but before response, got: %d", m.Value) - } - } else { - t.Fatalf("Expected XDSClientConnected metric, got: %T", got) + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 0}); err != nil { + t.Fatalf("XDSClientConnected check failed after successful NewStream but before response - got: %v, want 0", err) } - // Verify state returns to connected (Value: 1) after receiving the first response on the new stream. + // Verify state returns to connected (Value: 1) after receiving the first + // response on the new stream. close(sendResponse) if err := tmr.waitForSpecificMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { @@ -843,13 +838,14 @@ func (s) TestConnectedMetric_Reconnection(t *testing.T) { tmr.triggerAsyncMetrics() if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 1}); err != nil { - t.Fatalf("Expected XDSClientConnected to be 1 after response, got: %v", err) + t.Fatalf("XDSClientConnected check failed after response- got: %v, want 1", err) } } -// TestResourceMetrics_AuthorityOldStyle verifies that the xDS client correctly falls back to '#old' in -// "grpc.xds_client.resources" metrics for legacy authorities. It watches a resource using default authority -// and asserts that the metric is reported under '#old' instead of an empty string. +// TestResourceMetrics_AuthorityOldStyle verifies that the xDS client correctly +// falls back to '#old' in "grpc.xds_client.resources" metrics for legacy +// authorities. It watches a resource using default authority and asserts that +// the metric is reported under '#old' instead of an empty string. func (s) TestResourceMetrics_AuthorityOldStyle(t *testing.T) { mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{}) nodeID := uuid.New().String() @@ -886,7 +882,7 @@ func (s) TestResourceMetrics_AuthorityOldStyle(t *testing.T) { const listenerName = "test-listener" - client.WatchResource(listenerType.TypeURL, listenerName, &testWatcher{}) + client.WatchResource(listenerType.TypeURL, listenerName, newListenerWatcher()) resources := e2e.UpdateOptions{ NodeID: nodeID,