Function TestGrpcStatsStreaming

middleware/grpc_stats_test.go:136–264

(t *testing.T)

}

func TestGrpcStatsStreaming(t *testing.T) {
	reg := prometheus.NewRegistry()

	received := promauto.With(reg).NewHistogramVec(prometheus.HistogramOpts{
		Name:    "received_payload_bytes",
		Help:    "Size of received gRPC messages",
		Buckets: BodySizeBuckets,
	}, []string{"method", "route"})

	sent := promauto.With(reg).NewHistogramVec(prometheus.HistogramOpts{
		Name:    "sent_payload_bytes",
		Help:    "Size of sent gRPC",
		Buckets: BodySizeBuckets,
	}, []string{"method", "route"})

	inflightRequests := promauto.With(reg).NewGaugeVec(prometheus.GaugeOpts{
		Name: "inflight_requests",
		Help: "Current number of inflight requests.",
	}, []string{"method", "route"})

	stats := NewStatsHandler(reg, received, sent, inflightRequests, true)

	serv := grpc.NewServer(grpc.StatsHandler(stats), grpc.MaxSendMsgSize(10e6), grpc.MaxRecvMsgSize(10e6))
	defer serv.GracefulStop()

	listener, err := net.Listen("tcp", "localhost:0")
	require.NoError(t, err)

	middleware_test.RegisterEchoServerServer(serv, &halfEcho{log: t.Log})

	go func() {
		require.NoError(t, serv.Serve(listener))
	}()

	conn, err := grpc.NewClient(listener.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(10e6), grpc.MaxCallSendMsgSize(10e6)))
	require.NoError(t, err)
	defer func() {
		require.NoError(t, conn.Close())
	}()

	fc := middleware_test.NewEchoServerClient(conn)

	s, err := fc.Process(context.Background())
	require.NoError(t, err)

	for ix := 0; ix < 5; ix++ {
		msg := &middleware_test.Msg{
			Body: []byte(generateString((ix + 1) * 1024 * 1024)),
		}

		t.Log("Client Sending", msg.Size())
		err = s.Send(msg)
		require.NoError(t, err)

		_, err := s.Recv()
		require.NoError(t, err)

		err = testutil.GatherAndCompare(reg, bytes.NewBufferString(`