A78 xds gauge metrics with tests

internal/xds/clients/config.go

@@ -109,4 +109,32 @@ 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
+	// 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.
+	// 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.
+	RegisterAsyncReporter(reporter AsyncReporter) func()
+}
+
+// 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 (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 internal/xds/clients/xdsclient/metrics/metrics.go file.
+	ReportMetric(metric any)
 }

internal/xds/clients/xdsclient/ads_stream.go

@@ -22,6 +22,7 @@ import (
 	"context"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"google.golang.org/grpc/grpclog"
@@ -111,8 +112,12 @@ type adsStreamImpl struct {
 	// 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.
 	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.
@@ -241,16 +246,19 @@ 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.streamEstablished.Store(false)
 			s.onError(err, false)
 			return nil
 		}
 		if s.logger.V(2) {
 			s.logger.Infof("ADS stream created")
 		}
 
-		s.mu.Lock()
-		s.firstRequest = true
-		s.mu.Unlock()
+		s.firstRequest.Store(true)
+		if !s.firstStreamCreated.Load() {
+			s.streamEstablished.Store(true)
+			s.firstStreamCreated.Store(true)
+		}
 
 		// Ensure that the most recently created stream is pushed on the
 		// channel for the `send` goroutine to consume.
@@ -383,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
 	}
 
@@ -402,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))
@@ -437,10 +445,14 @@ func (s *adsStreamImpl) recv(stream clients.Stream) bool {
 
 		resources, url, version, nonce, err := s.recvMessage(stream)
 		if err != nil {
+			if !msgReceived {
+				s.streamEstablished.Store(false)
+			}
 			s.onError(err, msgReceived)
 			s.logger.Warningf("ADS stream closed: %v", err)
 			return msgReceived
 		}
+		s.streamEstablished.Store(true)
 		msgReceived = true
 
 		// Invoke the onResponse event handler to parse the incoming message and
@@ -715,3 +727,7 @@ func (fc *adsFlowControl) wait() bool {
 
 	return fc.stopped
 }
+
+func (s *adsStreamImpl) isStreamEstablished() bool {
+	return s.streamEstablished.Load()
+}

internal/xds/clients/xdsclient/authority.go

@@ -934,3 +934,56 @@ 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) {
+		summary := make(map[string]map[string]int)
+		for rType, resourceMap := range a.resources {
+			typeName := rType.TypeName
+			if _, ok := summary[typeName]; !ok {
+				summary[typeName] = make(map[string]int)
+			}
+			for _, state := range resourceMap {
+				s := cacheState(state)
+				summary[typeName][s]++
+			}
+		}
+
+		ret <- summary
+	}
+	a.xdsClientSerializer.ScheduleOr(op, func() {
+		ret <- nil
+	})
+
+	return <-ret
+}
+
+// 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"
+	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"
+	}
+}

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
@@ -40,3 +43,22 @@ type ResourceUpdateInvalid struct {
 type ServerFailure struct {
 	ServerURI string
 }
+
+// XDSClientConnected reports the connectivity state of the ADS stream.
+// 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.
+// For label definitions, see gRFC A78.
+// 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
+	CacheState   string
+	Count        int64
+}

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

@@ -25,12 +25,14 @@ 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/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"
@@ -278,31 +280,111 @@ 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{}
 }
 
 // newTestMetricsReporter returns a new testMetricsReporter.
 func newTestMetricsReporter() *testMetricsReporter {
 	return &testMetricsReporter{
-		metricsCh: testutils.NewChannelWithSize(1),
+		metricsCh:      buffer.NewUnbounded(),
+		asyncReporters: make(map[clients.AsyncReporter]struct{}),
 	}
 }
 
 // waitForMetric waits for a metric to be recorded and verifies that the
 // 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")
+	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())
 	}
-	if diff := cmp.Diff(got, metricsDataWant); diff != "" {
-		return fmt.Errorf("received unexpected metrics value (-got, +want): %v", diff)
+}
+
+// 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 {
+		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())
+		}
+	}
+}
+
+// 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
+		}
 	}
-	return nil
 }
 
 // ReportMetric sends the metrics data to the metricsCh channel.
 func (r *testMetricsReporter) ReportMetric(m any) {
-	r.metricsCh.Replace(m)
+	r.metricsCh.Put(m)
+}
+
+func (r *testMetricsReporter) RegisterAsyncReporter(reporter clients.AsyncReporter) func() {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.asyncReporters[reporter] = struct{}{}
+	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()
+	for reporter := range r.asyncReporters {
+		reporter.Report(r)
+	}
+}
+
+func (r *testMetricsReporter) numAsyncReporters() int {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return len(r.asyncReporters)
 }