diff --git a/internal/xds/clients/config.go b/internal/xds/clients/config.go index f106465f646a..aa89d2b1cee6 100644 --- a/internal/xds/clients/config.go +++ b/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) } diff --git a/internal/xds/clients/xdsclient/ads_stream.go b/internal/xds/clients/xdsclient/ads_stream.go index 24e66b834716..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" @@ -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,6 +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.streamEstablished.Store(false) s.onError(err, false) return nil } @@ -248,9 +254,11 @@ func (s *adsStreamImpl) runner(ctx context.Context) { 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() +} diff --git a/internal/xds/clients/xdsclient/authority.go b/internal/xds/clients/xdsclient/authority.go index 3aff20bb43ca..3b5666bf1bb7 100644 --- a/internal/xds/clients/xdsclient/authority.go +++ b/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" + } +} diff --git a/internal/xds/clients/xdsclient/metrics/metrics.go b/internal/xds/clients/xdsclient/metrics/metrics.go index 2e14efb1ae9b..fa1277cf33b6 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 @@ -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 +} diff --git a/internal/xds/clients/xdsclient/test/helpers_test.go b/internal/xds/clients/xdsclient/test/helpers_test.go index 1d6ac9ca7f4a..20225280bd5b 100644 --- a/internal/xds/clients/xdsclient/test/helpers_test.go +++ b/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,13 +280,17 @@ 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{}), } } @@ -292,17 +298,93 @@ 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") + 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) } diff --git a/internal/xds/clients/xdsclient/test/metrics_test.go b/internal/xds/clients/xdsclient/test/metrics_test.go index 8332cc57faef..6df2b1bbf990 100644 --- a/internal/xds/clients/xdsclient/test/metrics_test.go +++ b/internal/xds/clients/xdsclient/test/metrics_test.go @@ -25,6 +25,7 @@ import ( "testing" "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" @@ -96,7 +97,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 @@ -111,7 +112,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. @@ -199,7 +200,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) } } @@ -290,7 +291,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 @@ -308,6 +309,603 @@ 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() + + 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() + + xdsClientConfig := xdsclient.Config{ + 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{}}, + }, + MetricsReporter: tmr, + } + client, err := xdsclient.New(xdsClientConfig) + if err != nil { + t.Fatalf("Failed to create xDS client: %v", err) + } + defer client.Close() + + 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") + } + + 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) + } + + // Now trigger async metrics. + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 1}); err != nil { + t.Fatal(err) + } + + mgmtServer.Stop() + + // 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) + } + + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 0}); err != nil { + t.Fatal(err) + } + + // Verify async metric reporters are unregistered when client closes. + client.Close() + + 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. + 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.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() + + 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() + + xdsClientConfig := xdsclient.Config{ + 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{}}, + }, + 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" + + 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) + } + + if err := tmr.waitForMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { + t.Fatal(err) + } + + // Trigger async metrics. + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{ + Authority: "#old", // Default value for old-style non-xdstp as per gRFC A78 + ResourceType: "ListenerResource", + CacheState: "acked", + Count: 1, + }); err != nil { + t.Fatal(err) + } + + 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) + } + + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{ + Authority: "#old", + ResourceType: "ListenerResource", + CacheState: "nacked_but_cached", + Count: 1, + }); err != nil { + t.Fatal(err) + } +} + +// 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() + + 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() + + xdsClientConfig := xdsclient.Config{ + 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{}}, + }, + MetricsReporter: tmr, + } + client, err := xdsclient.New(xdsClientConfig) + if err != nil { + t.Fatalf("Failed to create xDS client: %v", err) + } + defer client.Close() + + resRequested1 := "res-requested-1" + resRequested2 := "res-requested-2" + resNacked1 := "res-nacked-1" + resNacked2 := "res-nacked-2" + resRemoved := "res-not-exist" + + resList := []string{resRequested1, resRequested2, resNacked1, resNacked2, resRemoved} + 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(resRemoved, "route-config"), + }, + SkipValidation: true, + } + 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 the resource update to be accepted. + 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{}, + SkipValidation: true, + } + if err := mgmtServer.Update(ctx, resourcesEmpty); err != nil { + t.Fatalf("Failed to update management server: %v", err) + } + + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{ + Authority: "#old", + ResourceType: "ListenerResource", + CacheState: "requested", + Count: 2, + }); err != nil { + t.Fatalf("Failed to verify requested count: %v", err) + } + + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{ + Authority: "#old", + ResourceType: "ListenerResource", + CacheState: "nacked", + Count: 2, + }); err != nil { + t.Fatalf("Failed to verify nacked count: %v", err) + } + + // Wait for the xDS client to process the empty update and trigger + // ResourceNotFound for omitted resources immediately. + lw := newListenerWatcher() + client.WatchResource(listenerType.TypeURL, resRemoved, lw) + + // Verify that resources missing from the authoritative update transition to + // the does_not_exist state. + if err := verifyResourceErrorType(ctx, lw.resourceErrCh, xdsresource.ErrorTypeResourceNotFound, ""); err != nil { + t.Fatal(err) + } + + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientResourceStats{ + Authority: "#old", + ResourceType: "ListenerResource", + CacheState: "does_not_exist", + Count: 1, + }); err != nil { + t.Fatalf("Failed to verify does_not_exist count: %v", err) + } +} + +// 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() + + l, err := testutils.LocalTCPListener() + if err != nil { + t.Fatalf("net.Listen() failed: %v", err) + } + lis := testutils.NewRestartableListener(l) + + sendResponse := make(chan struct{}) + streamOpened := make(chan struct{}, 10) + mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{ + Listener: lis, + OnStreamOpen: func(ctx context.Context, _ int64, _ string) error { + select { + case streamOpened <- struct{}{}: + case <-ctx.Done(): + } + return nil + }, + OnStreamResponse: func(ctx context.Context, _ int64, _ *v3discoverypb.DiscoveryRequest, _ *v3discoverypb.DiscoveryResponse) { + select { + case <-sendResponse: + case <-ctx.Done(): + } + }, + }) + nodeID := uuid.New().String() + + resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType} + si := clients.ServerIdentifier{ + ServerURI: mgmtServer.Address, + Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}, + } + // 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 := 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) { + 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{}{}: + case <-ctx.Done(): + } + // Pause stream creation until released by the test. + <-unblock + } + return streamer(ctx, desc, cc, method, cops...) + } + opts = append(opts, grpc.WithStreamInterceptor(interceptor)) + 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}}, + Node: clients.Node{ID: nodeID}, + TransportBuilder: grpctransport.NewBuilder(configs), + ResourceTypes: resourceTypes, + Authorities: map[string]xdsclient.Authority{ + "": {XDSServers: []xdsclient.ServerConfig{}}, + }, + 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() + + 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) + } + + // 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("XDSClientConnected check failed at start - got: %v, want 0", err) + } + + // Verify state transitions to connected (Value: 1) when stream succeeds. + lis.Restart() + + waitForStreamSuccess() + + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 1}); err != nil { + t.Fatalf("XDSClientConnected check failed after 1st NewStream - got: %v, want 1", err) + } + + // 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) + } + + // 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("XDSClientConnected check failed while NewStream is blocked - got: %v, want 1", err) + } + + // 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) + } + + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 0}); err != nil { + 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. + lis.Restart() + + // Clear channels to ensure fresh readings. + for len(streamOpened) > 0 { + <-streamOpened + } + + waitForStreamSuccess() + + tmr.Drain() + tmr.triggerAsyncMetrics() + 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. + close(sendResponse) + + if err := tmr.waitForSpecificMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil { + t.Fatal(err) + } + + tmr.triggerAsyncMetrics() + if err := tmr.waitForSpecificMetric(ctx, &metrics.XDSClientConnected{ServerURI: mgmtServer.Address, Value: 1}); err != nil { + 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. +func (s) TestResourceMetrics_AuthorityOldStyle(t *testing.T) { + mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{}) + nodeID := uuid.New().String() + + serverCfg := xdsclient.ServerConfig{ + ServerIdentifier: clients.ServerIdentifier{ + ServerURI: mgmtServer.Address, + Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}, + }, + } + + tmr := newTestMetricsReporter() + xdsClientConfig := xdsclient.Config{ + 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}}, + }, + 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, newListenerWatcher()) + + 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) + } + + // Verify that the client substitutes empty authority with '#old' in metrics. + 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 b1c6955484dc..ad82e7fd126f 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 { + reporter := &xdsClientMetricReporter{c: client} + client.metricsCleanup = client.metricsReporter.RegisterAsyncReporter(reporter) + } 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,66 @@ func resourceWatchStateForTesting(c *XDSClient, rType ResourceType, resourceName return a.resourceWatchStateForTesting(rType, 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 +} + +// Report implements clients.AsyncReporter. +// This is the entry point invoked by the metrics system during a scrape. +func (r *xdsClientMetricReporter) Report(rec clients.AsyncMetricsRecorder) error { + r.c.reportConnectedState(rec) + r.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.isStreamEstablished() { + 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) { + 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), + }) + } + } + } + } + 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 2b5671ada1e6..23d6227300cd 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: "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"}, + }) + xdsClientResourcesMetric = estats.RegisterInt64AsyncGauge(estats.MetricDescriptor{ + Name: "grpc.xds_client.resources", + 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"}, + }) ) // clientImpl embed xdsclient.XDSClient and implement internal XDSClient @@ -265,3 +279,50 @@ 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 || reporter == 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.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) + } +}