Function TestGrpcStatsMaxStreams

middleware/grpc_stats_test.go:309–396

(t *testing.T)

}

func TestGrpcStatsMaxStreams(t *testing.T) {
	const (
		waitTime = 1 * time.Second
		sleep    = waitTime / 10
	)
	reg := prometheus.NewRegistry()

	waitAndExpectMaxStreams := func(expected int) {
		var err error
		for endTime := time.Now().Add(waitTime); time.Now().Before(endTime); {
			err = testutil.GatherAndCompare(reg, bytes.NewBufferString(fmt.Sprintf(`
			# HELP grpc_concurrent_streams_by_conn_max The current number of concurrent streams in the connection with the most concurrent streams.
			# TYPE grpc_concurrent_streams_by_conn_max gauge
			grpc_concurrent_streams_by_conn_max{} %d
		`, expected)), "grpc_concurrent_streams_by_conn_max")
			if err == nil {
				break
			}
			time.Sleep(sleep)
		}
		require.NoError(t, err)
	}

	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.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))
	}()

	// Create a connection with 10 streams
	ctx, cancel1 := context.WithCancel(context.Background())
	defer cancel1()
	conn1 := launchConnWithStreams(t, ctx, listener, 10)