Function runDisconnectScenario

stats/opentelemetry/e2e_test.go:2231–2344

runDisconnectScenario sets up a functional xDS environment with OpenTelemetry enabled. It performs the following steps: 1. Starts a backend and configures an xDS management server. 2. Establishes a gRPC connection using a custom dialer to capture the underlying transport. 3. Performs a warm-up RPC t

(t *testing.T, name, wantLabel string, action func(*stubserver.StubServer, *grpc.ClientConn, func(error)))

// 4. Executes the provided 'action' callback to trigger a specific disconnection event.
// 5. Polls the OpenTelemetry metric reader to verify the 'grpc.disconnect_error' label matches wantLabel.
func runDisconnectScenario(t *testing.T, name, wantLabel string, action func(*stubserver.StubServer, *grpc.ClientConn, func(error))) {
	backend := stubserver.StartTestService(t, nil)
	port := itestutils.ParsePort(t, backend.Address)
	defer backend.Stop()

	mgmtServer, nodeID, _, xdsResolver := setup.ManagementServerAndResolver(t)

	serviceName := "service-" + name
	clusterName := "cluster-" + serviceName
	resources := e2e.DefaultClientResources(e2e.ResourceParams{
		DialTarget: serviceName,
		NodeID:     nodeID,
		Host:       "localhost",
		Port:       port,
		SecLevel:   e2e.SecurityLevelNone,
	})

	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()

	if err := mgmtServer.Update(ctx, resources); err != nil {
		t.Fatalf("Failed to update xDS resources: %v", err)
	}

	// Setup Telemetry
	reader := metric.NewManualReader()
	provider := metric.NewMeterProvider(metric.WithReader(reader))
	mo := opentelemetry.MetricsOptions{
		MeterProvider:  provider,
		Metrics:        opentelemetry.DefaultMetrics().Add("grpc.subchannel.disconnections"),
		OptionalLabels: []string{"grpc.lb.locality", "grpc.lb.backend_service", "grpc.disconnect_error"},
	}

	connCh := make(chan *errorConn, 1)
	dialer := func(ctx context.Context, addr string) (net.Conn, error) {
		conn, err := (&net.Dialer{}).DialContext(ctx, "tcp", addr)
		if err != nil {
			return nil, err
		}
		activeConn := &errorConn{Conn: conn}
		select {
		case connCh <- activeConn:
		case <-ctx.Done():
			activeConn.Conn.Close()
			return nil, ctx.Err()
		}
		return activeConn, nil
	}

	target := fmt.Sprintf("xds:///%s", serviceName)
	dopts := []grpc.DialOption{
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithResolvers(xdsResolver),
		grpc.WithContextDialer(dialer),
		opentelemetry.DialOption(opentelemetry.Options{MetricsOptions: mo}),
	}
	cc, err := grpc.NewClient(target, dopts...)
	if err != nil {