grpc-ecosystemgh-60 shared monitor struct, metric names adjusted to b…

…e more consistent, CounterOption and HistogramOption replaced by CollectorOption, clientReporter private API
piotrkowalczuk · Oct 21, 2018 · 85d9d85 · 85d9d85
1 parent 93bf462
commit 85d9d85
Show file tree

Hide file tree

Showing 10 changed files with 472 additions and 275 deletions.
diff --git a/client.go b/client.go
@@ -5,35 +5,31 @@
 
 package grpc_prometheus
 
-import (
-	prom "github.com/prometheus/client_golang/prometheus"
-)
-
 var (
-	// DefaultClientMetrics is the default instance of ClientMetrics. It is
-	// intended to be used in conjunction the default Prometheus metrics
-	// registry.
-	DefaultClientMetrics = NewClientMetrics()
-
-	// UnaryClientInterceptor is a gRPC client-side interceptor that provides Prometheus monitoring for Unary RPCs.
-	UnaryClientInterceptor = DefaultClientMetrics.UnaryClientInterceptor()
-
-	// StreamClientInterceptor is a gRPC client-side interceptor that provides Prometheus monitoring for Streaming RPCs.
-	StreamClientInterceptor = DefaultClientMetrics.StreamClientInterceptor()
+//// DefaultClientMetrics is the default instance of ClientMetrics. It is
+//// intended to be used in conjunction the default Prometheus monitor
+//// registry.
+//DefaultClientMetrics = NewClientMetrics()
+//
+//// UnaryInterceptor is a gRPC client-side interceptor that provides Prometheus monitoring for Unary RPCs.
+//UnaryClientInterceptor = DefaultClientMetrics.UnaryInterceptor()
+//
+//// StreamInterceptor is a gRPC client-side interceptor that provides Prometheus monitoring for Streaming RPCs.
+//StreamClientInterceptor = DefaultClientMetrics.StreamInterceptor()
 )
 
 func init() {
-	prom.MustRegister(DefaultClientMetrics.clientStartedCounter)
-	prom.MustRegister(DefaultClientMetrics.clientHandledCounter)
-	prom.MustRegister(DefaultClientMetrics.clientStreamMsgReceived)
-	prom.MustRegister(DefaultClientMetrics.clientStreamMsgSent)
+	//prom.MustRegister(DefaultClientMetrics)
+	//prom.MustRegister(DefaultClientMetrics.clientHandledCounter)
+	//prom.MustRegister(DefaultClientMetrics.clientStreamMsgReceived)
+	//prom.MustRegister(DefaultClientMetrics.clientStreamMsgSent)
 }
 
 // EnableClientHandlingTimeHistogram turns on recording of handling time of
-// RPCs. Histogram metrics can be very expensive for Prometheus to retain and
+// RPCs. Histogram monitor can be very expensive for Prometheus to retain and
 // query. This function acts on the DefaultClientMetrics variable and the
-// default Prometheus metrics registry.
-func EnableClientHandlingTimeHistogram(opts ...HistogramOption) {
-	DefaultClientMetrics.EnableClientHandlingTimeHistogram(opts...)
-	prom.Register(DefaultClientMetrics.clientHandledHistogram)
-}
+// default Prometheus monitor registry.
+//func EnableClientHandlingTimeHistogram(opts ...HistogramCollectorOption) {
+//DefaultClientMetrics.EnableClientHandlingTimeHistogram(opts...)
+//prom.Register(DefaultClientMetrics.clientHandledHistogram)
+//}
diff --git a/client_metrics.go b/client_metrics.go
@@ -10,126 +10,57 @@ import (
 	"google.golang.org/grpc/status"
 )
 
-// ClientMetrics represents a collection of metrics to be registered on a
-// Prometheus metrics registry for a gRPC client.
+// ClientMetrics represents a collection of monitor to be registered on a
+// Prometheus monitor registry for a gRPC client.
 type ClientMetrics struct {
-	clientStartedCounter          *prom.CounterVec
-	clientHandledCounter          *prom.CounterVec
-	clientStreamMsgReceived       *prom.CounterVec
-	clientStreamMsgSent           *prom.CounterVec
-	clientHandledHistogramEnabled bool
-	clientHandledHistogramOpts    prom.HistogramOpts
-	clientHandledHistogram        *prom.HistogramVec
+	*clientMonitor
 }
 
 // NewClientMetrics returns a ClientMetrics object. Use a new instance of
-// ClientMetrics when not using the default Prometheus metrics registry, for
-// example when wanting to control which metrics are added to a registry as
-// opposed to automatically adding metrics via init functions.
-func NewClientMetrics(counterOpts ...CounterOption) *ClientMetrics {
-	opts := counterOptions(counterOpts)
-	return &ClientMetrics{
-		clientStartedCounter: prom.NewCounterVec(
-			opts.apply(prom.CounterOpts{
-				Name: "grpc_client_started_total",
-				Help: "Total number of RPCs started on the client.",
-			}), []string{"grpc_type", "grpc_service", "grpc_method"}),
-
-		clientHandledCounter: prom.NewCounterVec(
-			opts.apply(prom.CounterOpts{
-				Name: "grpc_client_handled_total",
-				Help: "Total number of RPCs completed by the client, regardless of success or failure.",
-			}), []string{"grpc_type", "grpc_service", "grpc_method", "grpc_code"}),
-
-		clientStreamMsgReceived: prom.NewCounterVec(
-			opts.apply(prom.CounterOpts{
-				Name: "grpc_client_msg_received_total",
-				Help: "Total number of RPC stream messages received by the client.",
-			}), []string{"grpc_type", "grpc_service", "grpc_method"}),
+// ClientMetrics when not using the default Prometheus monitor registry, for
+// example when wanting to control which monitor are added to a registry as
+// opposed to automatically adding monitor via init functions.
+func NewClientMetrics(counterOpts ...CollectorOption) *ClientMetrics {
+	opts := counterOptions(counterOpts).apply(prom.Opts{
+		Namespace: namespace,
+		Subsystem: "client",
+	})
 
-		clientStreamMsgSent: prom.NewCounterVec(
-			opts.apply(prom.CounterOpts{
-				Name: "grpc_client_msg_sent_total",
-				Help: "Total number of gRPC stream messages sent by the client.",
-			}), []string{"grpc_type", "grpc_service", "grpc_method"}),
-
-		clientHandledHistogramEnabled: false,
-		clientHandledHistogramOpts: prom.HistogramOpts{
-			Name:    "grpc_client_handling_seconds",
-			Help:    "Histogram of response latency (seconds) of the gRPC until it is finished by the application.",
-			Buckets: prom.DefBuckets,
-		},
-		clientHandledHistogram: nil,
-	}
-}
-
-// Describe sends the super-set of all possible descriptors of metrics
-// collected by this Collector to the provided channel and returns once
-// the last descriptor has been sent.
-func (m *ClientMetrics) Describe(ch chan<- *prom.Desc) {
-	m.clientStartedCounter.Describe(ch)
-	m.clientHandledCounter.Describe(ch)
-	m.clientStreamMsgReceived.Describe(ch)
-	m.clientStreamMsgSent.Describe(ch)
-	if m.clientHandledHistogramEnabled {
-		m.clientHandledHistogram.Describe(ch)
-	}
-}
-
-// Collect is called by the Prometheus registry when collecting
-// metrics. The implementation sends each collected metric via the
-// provided channel and returns once the last metric has been sent.
-func (m *ClientMetrics) Collect(ch chan<- prom.Metric) {
-	m.clientStartedCounter.Collect(ch)
-	m.clientHandledCounter.Collect(ch)
-	m.clientStreamMsgReceived.Collect(ch)
-	m.clientStreamMsgSent.Collect(ch)
-	if m.clientHandledHistogramEnabled {
-		m.clientHandledHistogram.Collect(ch)
-	}
-}
-
-// EnableClientHandlingTimeHistogram turns on recording of handling time of RPCs.
-// Histogram metrics can be very expensive for Prometheus to retain and query.
-func (m *ClientMetrics) EnableClientHandlingTimeHistogram(opts ...HistogramOption) {
-	for _, o := range opts {
-		o(&m.clientHandledHistogramOpts)
-	}
-	if !m.clientHandledHistogramEnabled {
-		m.clientHandledHistogram = prom.NewHistogramVec(
-			m.clientHandledHistogramOpts,
-			[]string{"grpc_type", "grpc_service", "grpc_method"},
-		)
+	return &ClientMetrics{
+		clientMonitor: initClientMonitor(opts),
 	}
-	m.clientHandledHistogramEnabled = true
 }
 
-// UnaryClientInterceptor is a gRPC client-side interceptor that provides Prometheus monitoring for Unary RPCs.
-func (m *ClientMetrics) UnaryClientInterceptor() func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
+// UnaryInterceptor is a gRPC client-side interceptor that provides Prometheus monitoring for Unary RPCs.
+func (m *ClientMetrics) UnaryInterceptor() grpc.UnaryClientInterceptor {
 	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
-		monitor := newClientReporter(m, Unary, method)
-		monitor.SentMessage()
+		rep := newClientReporter(m.clientMonitor, Unary, method)
+		rep.outgoingRequest()
+		rep.outgoingMessage() // TODO: necessary?
+
 		err := invoker(ctx, method, req, reply, cc, opts...)
 		if err != nil {
-			monitor.ReceivedMessage()
+			rep.incomingMessage()
 		}
 		st, _ := status.FromError(err)
-		monitor.Handled(st.Code())
+		rep.incomingResponse(st.Code())
 		return err
 	}
 }
 
-// StreamClientInterceptor is a gRPC client-side interceptor that provides Prometheus monitoring for Streaming RPCs.
-func (m *ClientMetrics) StreamClientInterceptor() func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
+// StreamInterceptor is a gRPC client-side interceptor that provides Prometheus monitoring for Streaming RPCs.
+func (m *ClientMetrics) StreamInterceptor() grpc.StreamClientInterceptor {
 	return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
-		monitor := newClientReporter(m, clientStreamType(desc), method)
+		rep := newClientReporter(m.clientMonitor, clientStreamType(desc), method)
+		rep.outgoingRequest()
+
 		clientStream, err := streamer(ctx, desc, cc, method, opts...)
 		if err != nil {
 			st, _ := status.FromError(err)
-			monitor.Handled(st.Code())
+			rep.incomingResponse(st.Code())
 			return nil, err
 		}
-		return &monitoredClientStream{clientStream, monitor}, nil
+		return &monitoredClientStream{clientStream, rep}, nil
 	}
 }
 
@@ -151,20 +82,20 @@ type monitoredClientStream struct {
 func (s *monitoredClientStream) SendMsg(m interface{}) error {
 	err := s.ClientStream.SendMsg(m)
 	if err == nil {
-		s.monitor.SentMessage()
+		s.monitor.outgoingMessage()
 	}
 	return err
 }
 
 func (s *monitoredClientStream) RecvMsg(m interface{}) error {
 	err := s.ClientStream.RecvMsg(m)
 	if err == nil {
-		s.monitor.ReceivedMessage()
+		s.monitor.incomingMessage()
 	} else if err == io.EOF {
-		s.monitor.Handled(codes.OK)
+		s.monitor.incomingResponse(codes.OK)
 	} else {
 		st, _ := status.FromError(err)
-		s.monitor.Handled(st.Code())
+		s.monitor.incomingResponse(st.Code())
 	}
 	return err
 }
diff --git a/client_reporter.go b/client_reporter.go
@@ -6,41 +6,49 @@ package grpc_prometheus
 import (
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"google.golang.org/grpc/codes"
 )
 
 type clientReporter struct {
-	metrics     *ClientMetrics
+	monitor     *clientMonitor
 	rpcType     grpcType
 	serviceName string
 	methodName  string
 	startTime   time.Time
+	labels      prometheus.Labels
 }
 
-func newClientReporter(m *ClientMetrics, rpcType grpcType, fullMethod string) *clientReporter {
-	r := &clientReporter{
-		metrics: m,
-		rpcType: rpcType,
+func newClientReporter(m *clientMonitor, rpcType grpcType, fullMethod string) *clientReporter {
+	serviceName, methodName := splitMethodName(fullMethod)
+	return &clientReporter{
+		monitor:   m,
+		startTime: time.Now(),
+		labels: prometheus.Labels{
+			labelService: serviceName,
+			labelMethod:  methodName,
+			labelType:    string(rpcType),
+		},
 	}
-	if r.metrics.clientHandledHistogramEnabled {
-		r.startTime = time.Now()
-	}
-	r.serviceName, r.methodName = splitMethodName(fullMethod)
-	r.metrics.clientStartedCounter.WithLabelValues(string(r.rpcType), r.serviceName, r.methodName).Inc()
-	return r
 }
 
-func (r *clientReporter) ReceivedMessage() {
-	r.metrics.clientStreamMsgReceived.WithLabelValues(string(r.rpcType), r.serviceName, r.methodName).Inc()
+func (r *clientReporter) outgoingRequest() {
+	r.monitor.requestsTotal.With(r.labels).Inc()
+}
+
+func (r *clientReporter) outgoingMessage() {
+	r.monitor.messagesSent.With(r.labels).Inc()
 }
 
-func (r *clientReporter) SentMessage() {
-	r.metrics.clientStreamMsgSent.WithLabelValues(string(r.rpcType), r.serviceName, r.methodName).Inc()
+func (r *clientReporter) incomingMessage() {
+	r.monitor.messagesReceived.With(r.labels).Inc()
 }
 
-func (r *clientReporter) Handled(code codes.Code) {
-	r.metrics.clientHandledCounter.WithLabelValues(string(r.rpcType), r.serviceName, r.methodName, code.String()).Inc()
-	if r.metrics.clientHandledHistogramEnabled {
-		r.metrics.clientHandledHistogram.WithLabelValues(string(r.rpcType), r.serviceName, r.methodName).Observe(time.Since(r.startTime).Seconds())
+func (r *clientReporter) incomingResponse(code codes.Code) {
+	r.labels[labelCode] = code.String()
+	r.monitor.responsesTotal.With(r.labels).Inc()
+	r.monitor.requestDuration.With(r.labels).Observe(time.Since(r.startTime).Seconds())
+	if code != codes.OK {
+		r.monitor.errors.With(r.labels).Inc()
 	}
 }