Merge pull request #10 from grpc-ecosystem/feature/client-metrics

add Client-Side monitoring.

Author: Michal Witkowski

README.md

@@ -5,7 +5,7 @@
 [![GoDoc](http://img.shields.io/badge/GoDoc-Reference-blue.svg)](https://godoc.org/github.com/grpc-ecosystem/go-grpc-prometheus)
 [![Apache 2.0 License](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](LICENSE)
 
-[Prometheus](https://prometheus.io/) monitoring for your [gRPC Go](https://github.com/grpc/grpc-go) servers.
+[Prometheus](https://prometheus.io/) monitoring for your [gRPC Go](https://github.com/grpc/grpc-go) servers and clients.
 
 A sister implementation for [gRPC Java](https://github.com/grpc/grpc-java) (same metrics, same semantics) is in [grpc-ecosystem/java-grpc-prometheus](https://github.com/grpc-ecosystem/java-grpc-prometheus).
 
@@ -19,6 +19,10 @@ To use Interceptors in chains, please see [`go-grpc-middleware`](https://github.
 
 ## Usage
 
+There are two types of interceptors: client-side and server-side. This package provides monitoring Interceptors for both.
+
+### Server-side
+
 ```go
 import "github.com/grpc-ecosystem/go-grpc-prometheus"
 ...
@@ -27,17 +31,35 @@ import "github.com/grpc-ecosystem/go-grpc-prometheus"
         grpc.StreamInterceptor(grpc_prometheus.StreamServerInterceptor),
         grpc.UnaryInterceptor(grpc_prometheus.UnaryServerInterceptor),
     )
+    // Register your gRPC service implementations.
+    myservice.RegisterMyServiceServer(s.server, &myServiceImpl{})
+    // After all your registrations, make sure all of the Prometheus metrics are initialized.
+    grpc_prometheus.Register(myServer)
     // Register Prometheus metrics handler.    
     http.Handle("/metrics", prometheus.Handler())
 ...
 ```
 
-# Metrics
+### Client-side
 
+```go
+import "github.com/grpc-ecosystem/go-grpc-prometheus"
+...
+   clientConn, err = grpc.Dial(
+       address,
+		   grpc.WithUnaryInterceptor(UnaryClientInterceptor),
+		   grpc.WithStreamInterceptor(StreamClientInterceptor)
+   )
+   client = pb_testproto.NewTestServiceClient(clientConn)
+   resp, err := client.PingEmpty(s.ctx, &myservice.Request{Msg: "hello"})
+...
+```
+
+# Metrics
 
 ## Labels
 
-All server-side metrics start with `grpc_server` as Prometheus subsystem name. Similarly all methods
+All server-side metrics start with `grpc_server` as Prometheus subsystem name. All client-side metrics start with `grpc_client`. Both of them have mirror-concepts. Similarly all methods
 contain the same rich labels:
 
   * `grpc_service` - the [gRPC service](http://www.grpc.io/docs/#defining-a-service) name, which is the combination of protobuf `package` and
@@ -65,9 +87,11 @@ Additionally for completed RPCs, the following labels are used:
 
 ## Counters
 
-The counters and their up to date documentation is in [server_reporter.go](server_reporter.go) and
+The counters and their up to date documentation is in [server_reporter.go](server_reporter.go) and [client_reporter.go](client_reporter.go) 
 the respective Prometheus handler (usually `/metrics`). 
 
+For the purpose of this documentation we will only discuss `grpc_server` metrics. The `grpc_client` ones contain mirror concepts.
+
 For simplicity, let's assume we're tracking a single server-side RPC call of [`mwitkow.testproto.TestService`](examples/testproto/test.proto),
 calling the method `PingList`. The call succeeds and returns 20 messages in the stream.
 
@@ -214,8 +238,7 @@ e.g. "less than 1% of requests are slower than 250ms".
 
 ## Status
 
-This code has been in an upstream [pull request](https://github.com/grpc/grpc-go/pull/299) since August 2015. It has 
-served as the basis for monitoring of production gRPC micro services at [Improbable](https://improbable.io) since then.
+This code has been used since August 2015 as the basis for monitoring of *production* gRPC micro services  at [Improbable](https://improbable.io).
 
 ## License
 

client.go

@@ -0,0 +1,72 @@
+// Copyright 2016 Michal Witkowski. All Rights Reserved.
+// See LICENSE for licensing terms.
+
+// gRPC Prometheus monitoring interceptors for client-side gRPC.
+
+package grpc_prometheus
+
+import (
+	"io"
+
+	"golang.org/x/net/context"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+)
+
+// UnaryClientInterceptor is a gRPC client-side interceptor that provides Prometheus monitoring for Unary RPCs.
+func UnaryClientInterceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
+	monitor := newClientReporter(Unary, method)
+	monitor.SentMessage()
+	err := invoker(ctx, method, req, reply, cc, opts...)
+	if err != nil {
+		monitor.ReceivedMessage()
+	}
+	monitor.Handled(grpc.Code(err))
+	return err
+}
+
+// StreamServerInterceptor is a gRPC client-side interceptor that provides Prometheus monitoring for Streaming RPCs.
+func StreamClientInterceptor(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
+	monitor := newClientReporter(clientStreamType(desc), method)
+	clientStream, err := streamer(ctx, desc, cc, method, opts...)
+	if err != nil {
+		monitor.Handled(grpc.Code(err))
+		return nil, err
+	}
+	return &monitoredClientStream{clientStream, monitor}, nil
+}
+
+func clientStreamType(desc *grpc.StreamDesc) grpcType {
+	if desc.ClientStreams && !desc.ServerStreams {
+		return ClientStream
+	} else if !desc.ClientStreams && desc.ServerStreams {
+		return ServerStream
+	}
+	return BidiStream
+}
+
+// monitoredClientStream wraps grpc.ClientStream allowing each Sent/Recv of message to increment counters.
+type monitoredClientStream struct {
+	grpc.ClientStream
+	monitor *clientReporter
+}
+
+func (s *monitoredClientStream) SendMsg(m interface{}) error {
+	err := s.ClientStream.SendMsg(m)
+	if err == nil {
+		s.monitor.SentMessage()
+	}
+	return err
+}
+
+func (s *monitoredClientStream) RecvMsg(m interface{}) error {
+	err := s.ClientStream.RecvMsg(m)
+	if err == nil {
+		s.monitor.ReceivedMessage()
+	} else if err == io.EOF {
+		s.monitor.Handled(codes.OK)
+	} else {
+		s.monitor.Handled(grpc.Code(err))
+	}
+	return err
+}

client_reporter.go

@@ -0,0 +1,111 @@
+// Copyright 2016 Michal Witkowski. All Rights Reserved.
+// See LICENSE for licensing terms.
+
+package grpc_prometheus
+
+import (
+	"time"
+
+	"google.golang.org/grpc/codes"
+
+	prom "github.com/prometheus/client_golang/prometheus"
+)
+
+var (
+	clientStartedCounter = prom.NewCounterVec(
+		prom.CounterOpts{
+			Namespace: "grpc",
+			Subsystem: "client",
+			Name:      "started_total",
+			Help:      "Total number of RPCs started on the client.",
+		}, []string{"grpc_type", "grpc_service", "grpc_method"})
+
+	clientHandledCounter = prom.NewCounterVec(
+		prom.CounterOpts{
+			Namespace: "grpc",
+			Subsystem: "client",
+			Name:      "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(
+		prom.CounterOpts{
+			Namespace: "grpc",
+			Subsystem: "client",
+			Name:      "msg_received_total",
+			Help:      "Total number of RPC stream messages received by the client.",
+		}, []string{"grpc_type", "grpc_service", "grpc_method"})
+
+	clientStreamMsgSent = prom.NewCounterVec(
+		prom.CounterOpts{
+			Namespace: "grpc",
+			Subsystem: "client",
+			Name:      "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{
+		Namespace: "grpc",
+		Subsystem: "client",
+		Name:      "handling_seconds",
+		Help:      "Histogram of response latency (seconds) of the gRPC until it is finished by the application.",
+		Buckets:   prom.DefBuckets,
+	}
+	clientHandledHistogram *prom.HistogramVec
+)
+
+func init() {
+	prom.MustRegister(clientStartedCounter)
+	prom.MustRegister(clientHandledCounter)
+	prom.MustRegister(clientStreamMsgReceived)
+	prom.MustRegister(clientStreamMsgSent)
+}
+
+// EnableClientHandlingTimeHistogram turns on recording of handling time of RPCs.
+// Histogram metrics can be very expensive for Prometheus to retain and query.
+func EnableClientHandlingTimeHistogram(opts ...HistogramOption) {
+	for _, o := range opts {
+		o(&clientHandledHistogramOpts)
+	}
+	if !clientHandledHistogramEnabled {
+		clientHandledHistogram = prom.NewHistogramVec(
+			clientHandledHistogramOpts,
+			[]string{"grpc_type", "grpc_service", "grpc_method"},
+		)
+		prom.Register(clientHandledHistogram)
+	}
+	clientHandledHistogramEnabled = true
+}
+
+type clientReporter struct {
+	rpcType     grpcType
+	serviceName string
+	methodName  string
+	startTime   time.Time
+}
+
+func newClientReporter(rpcType grpcType, fullMethod string) *clientReporter {
+	r := &clientReporter{rpcType: rpcType}
+	if clientHandledHistogramEnabled {
+		r.startTime = time.Now()
+	}
+	r.serviceName, r.methodName = splitMethodName(fullMethod)
+	clientStartedCounter.WithLabelValues(string(r.rpcType), r.serviceName, r.methodName).Inc()
+	return r
+}
+
+func (r *clientReporter) ReceivedMessage() {
+	clientStreamMsgReceived.WithLabelValues(string(r.rpcType), r.serviceName, r.methodName).Inc()
+}
+
+func (r *clientReporter) SentMessage() {
+	clientStreamMsgSent.WithLabelValues(string(r.rpcType), r.serviceName, r.methodName).Inc()
+}
+
+func (r *clientReporter) Handled(code codes.Code) {
+	clientHandledCounter.WithLabelValues(string(r.rpcType), r.serviceName, r.methodName, code.String()).Inc()
+	if clientHandledHistogramEnabled {
+		clientHandledHistogram.WithLabelValues(string(r.rpcType), r.serviceName, r.methodName).Observe(time.Since(r.startTime).Seconds())
+	}
+}