Method TestConnectedMetric

internal/xds/clients/xdsclient/test/metrics_test.go:321–415

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 manag

(t *testing.T)

// 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)
	}