Function TestConcurrentGrpcCalls

integration/grpc_server_test.go:143–243

(t *testing.T)

}

func TestConcurrentGrpcCalls(t *testing.T) {
	cfg := server.Config{}
	(&cfg).RegisterFlags(flag.NewFlagSet("fake", flag.ContinueOnError))

	tc := map[string]struct {
		cfg      server.Config
		register func(s *grpc.Server)
		validate func(t testing.TB, con *grpc.ClientConn)
	}{
		"distributor": {
			cfg: cfg,
			register: func(s *grpc.Server) {
				d := &mockGprcServer{}
				distributorpb.RegisterDistributorServer(s, d)
			},
			validate: func(t testing.TB, conn *grpc.ClientConn) {
				client := distributorpb.NewDistributorClient(conn)
				wg := sync.WaitGroup{}
				n := 10000
				wg.Add(n)
				for i := 0; i < n; i++ {
					go func(i int) {
						defer wg.Done()
						ctx := context.Background()
						ctx = metadata.NewOutgoingContext(ctx, metadata.MD{"i": []string{strconv.Itoa(i)}})
						_, err := client.Push(ctx, createRequest(i))
						require.NoError(t, err)
					}(i)
				}

				wg.Wait()
			},
		},
		"distributor push stream": {
			cfg: cfg,
			register: func(s *grpc.Server) {
				d := &mockGprcServer{}
				ingester_client.RegisterIngesterServer(s, d)
			},
			validate: func(t testing.TB, conn *grpc.ClientConn) {
				ctx := context.Background()
				client := ingester_client.NewIngesterClient(conn)
				wg := sync.WaitGroup{}
				n := 10000
				wg.Add(n)
				for i := 0; i < n; i++ {
					go func(i int) {
						defer wg.Done()
						stream, err := client.PushStream(ctx)
						require.NoError(t, err)

						ctx = metadata.NewOutgoingContext(ctx, metadata.MD{"i": []string{strconv.Itoa(i)}})
						err = stream.Send(&cortexpb.StreamWriteRequest{TenantID: strconv.Itoa(i), Request: createRequest(i)})
						require.NoError(t, err)
						_, err = stream.Recv()
						require.NoError(t, err)
						//err = stream.Send(&cortexpb.StreamWriteRequest{"i", createRequest(i + 1)})
						//require.NoError(t, err)