Merge pull request #5711 from xiang90/client_bytes

*: add client network metrics

Documentation/metrics.md

@@ -71,6 +71,8 @@ All these metrics are prefixed with `etcd_network_`
 | peer_sent_bytes_total           | The total number of bytes sent to the peer with ID `To`.         | Counter(To)   |
 | peer_received_bytes_total       | The total number of bytes received from the peer with ID `From`. | Counter(From) |
 | peer_round_trip_time_seconds    | Round-Trip-Time histogram between peers.                         | Histogram(To) |
+| client_grpc_sent_bytes_total    | The total number of bytes sent to grpc clients.                  | Counter   |
+| client_grpc_received_bytes_total| The total number of bytes received to grpc clients.              | Counter   |
 
 `peer_sent_bytes_total` counts the total number of bytes sent to a specific peer. Usually the leader member sends more data than other members since it is responsible for transmitting replicated data.
 

etcdserver/api/v3rpc/codec.go

@@ -0,0 +1,34 @@
+// Copyright 2016 The etcd Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package v3rpc
+
+import "github.com/gogo/protobuf/proto"
+
+type codec struct{}
+
+func (c *codec) Marshal(v interface{}) ([]byte, error) {
+	b, err := proto.Marshal(v.(proto.Message))
+	sentBytes.Add(float64(len(b)))
+	return b, err
+}
+
+func (c *codec) Unmarshal(data []byte, v interface{}) error {
+	receivedBytes.Add(float64(len(data)))
+	return proto.Unmarshal(data, v.(proto.Message))
+}
+
+func (c *codec) String() string {
+	return "proto"
+}

etcdserver/api/v3rpc/grpc.go

@@ -26,6 +26,7 @@ import (
 
 func Server(s *etcdserver.EtcdServer, tls *tls.Config) *grpc.Server {
 	var opts []grpc.ServerOption
+	opts = append(opts, grpc.CustomCodec(&codec{}))
 	if tls != nil {
 		opts = append(opts, grpc.Creds(credentials.NewTLS(tls)))
 	}

etcdserver/api/v3rpc/metrics.go

@@ -41,10 +41,27 @@ var (
 			Help:      "Bucketed histogram of processing time (s) of handled unary (non-stream) requests.",
 			Buckets:   prometheus.ExponentialBuckets(0.0005, 2, 13),
 		}, []string{"grpc_service", "grpc_method"})
+
+	sentBytes = prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace: "etcd",
+		Subsystem: "network",
+		Name:      "client_grpc_sent_bytes_total",
+		Help:      "The total number of bytes sent to grpc clients.",
+	})
+
+	receivedBytes = prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace: "etcd",
+		Subsystem: "network",
+		Name:      "client_grpc_received_bytes_total",
+		Help:      "The total number of bytes received from grpc clients.",
+	})
 )
 
 func init() {
 	prometheus.MustRegister(receivedCounter)
 	prometheus.MustRegister(failedCounter)
 	prometheus.MustRegister(handlingDuration)
+
+	prometheus.MustRegister(sentBytes)
+	prometheus.MustRegister(receivedBytes)
 }