(t *testing.T)
| 61 | } |
| 62 | |
| 63 | func TestLoad_ConcurrentNodes(t *testing.T) { |
| 64 | baseGoroutines := runtime.NumGoroutine() |
| 65 | t.Logf("baseline goroutines before test: %d", baseGoroutines) |
| 66 | |
| 67 | lis := bufconn.Listen(bufSize) |
| 68 | hub := nodehub.New(nodehub.Options{ |
| 69 | PeerExtractor: mdPeerExtractor, |
| 70 | PushHandler: nodehub.NoopPushHandler{}, |
| 71 | DeadConnectionTimeout: 60 * time.Second, |
| 72 | ReaperInterval: 10 * time.Second, |
| 73 | }) |
| 74 | srv := grpc.NewServer() |
| 75 | nodev1.RegisterNodeAgentServer(srv, hub) |
| 76 | go func() { _ = srv.Serve(lis) }() |
| 77 | defer srv.Stop() |
| 78 | |
| 79 | hubCtx, hubCancel := context.WithCancel(context.Background()) |
| 80 | defer hubCancel() |
| 81 | go hub.RunReaper(hubCtx) |
| 82 | |
| 83 | dialer := func(ctx context.Context, _ string) (net.Conn, error) { |
| 84 | return lis.DialContext(ctx) |
| 85 | } |
| 86 | |
| 87 | nodeIDInterceptor := func(nodeID string) grpc.DialOption { |
| 88 | return grpc.WithStreamInterceptor(func( |
| 89 | ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, |
| 90 | method string, streamer grpc.Streamer, opts ...grpc.CallOption, |
| 91 | ) (grpc.ClientStream, error) { |
| 92 | ctx = metadata.AppendToOutgoingContext(ctx, "x-node-id", nodeID) |
| 93 | return streamer(ctx, desc, cc, method, opts...) |
| 94 | }) |
| 95 | } |
| 96 | |
| 97 | agentCtx, agentCancel := context.WithCancel(context.Background()) |
| 98 | var agentWG sync.WaitGroup |
| 99 | |
| 100 | startErrs := make(chan error, numNodes) |
| 101 | for i := 0; i < numNodes; i++ { |
| 102 | nodeID := fmt.Sprintf("node-%d", i) |
| 103 | agentWG.Add(1) |
| 104 | go func() { |
| 105 | defer agentWG.Done() |
| 106 | cfg := nodeagent.Config{ |
| 107 | NodeID: nodeID, |
| 108 | ServerAddr: "passthrough:///bufnet", |
| 109 | Dispatcher: nopDispatcher{}, |
| 110 | HelloProvider: nodeagent.DefaultHelloProvider(nodeID, nil), |
| 111 | ReconnectBackoff: []time.Duration{200 * time.Millisecond}, |
| 112 | KeepaliveTime: 30 * time.Second, |
| 113 | KeepaliveTimeout: 10 * time.Second, |
| 114 | Dialer: func(_ context.Context) (*grpc.ClientConn, error) { |
| 115 | return grpc.NewClient("passthrough:///bufnet", |
| 116 | grpc.WithContextDialer(dialer), |
| 117 | grpc.WithTransportCredentials(insecure.NewCredentials()), |
| 118 | nodeIDInterceptor(nodeID), |
| 119 | ) |
| 120 | }, |
nothing calls this directly
no test coverage detected