Feature/client (#9)

* add client support

* add client support

* add client support

* add client support

Author: zhyon404

README.md

@@ -41,5 +41,3 @@ Note: the `grpc_*` metrics will just show commented out (with their descriptions
 ## TODO
 
 - add stream support
-- add client metrics
-- add example

examples/helloworld/hello_world_client.py

@@ -0,0 +1,38 @@
+from __future__ import print_function
+import logging
+import time
+
+import grpc
+
+import helloworld_pb2
+import helloworld_pb2_grpc
+
+import prometheus_client
+from python_grpc_prometheus.prometheus_client_interceptor import PromClientInterceptor
+
+_ONE_DAY_IN_SECONDS = 60 * 60 * 24
+
+
+def run():
+    interceptor = PromClientInterceptor()
+    with grpc.insecure_channel('localhost:50051') as channel:
+        intercept_channel = grpc.intercept_channel(channel,
+                                                   interceptor)
+        stub = helloworld_pb2_grpc.GreeterStub(intercept_channel)
+        response = stub.SayHello.future(helloworld_pb2.HelloRequest(name='you'))
+        response=response.result()
+    print("Greeter client received: " + response.message)
+
+
+if __name__ == '__main__':
+    logging.basicConfig()
+    try:
+        run()
+    except Exception as e:
+        print(e)
+    prometheus_client.start_http_server(8001)
+    try:
+        while True:
+            time.sleep(_ONE_DAY_IN_SECONDS)
+    except KeyboardInterrupt:
+        pass

examples/helloworld/hello_world_server.py

@@ -0,0 +1,39 @@
+from concurrent import futures
+import time
+import logging
+
+import grpc
+
+import helloworld_pb2
+import helloworld_pb2_grpc
+
+import prometheus_client
+from python_grpc_prometheus.prometheus_server_interceptor import PromServerInterceptor
+
+_ONE_DAY_IN_SECONDS = 60 * 60 * 24
+
+
+class Greeter(helloworld_pb2_grpc.GreeterServicer):
+
+    def SayHello(self, request, context):
+        return helloworld_pb2.HelloReply(message='Hello, %s!' % request.name)
+
+
+def serve():
+    # Add the required interceptor(s) where you create your grpc server, e.g.
+    psi = PromServerInterceptor()
+    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10), interceptors=(psi,))
+    helloworld_pb2_grpc.add_GreeterServicer_to_server(Greeter(), server)
+    server.add_insecure_port('[::]:50051')
+    server.start()
+    prometheus_client.start_http_server(8000)
+    try:
+        while True:
+            time.sleep(_ONE_DAY_IN_SECONDS)
+    except KeyboardInterrupt:
+        server.stop(0)
+
+
+if __name__ == '__main__':
+    logging.basicConfig()
+    serve()

examples/helloworld/helloworld_pb2.py

@@ -0,0 +1,46 @@
+# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
+import grpc
+
+import helloworld_pb2 as helloworld__pb2
+
+
+class GreeterStub(object):
+  """The greeting service definition.
+  """
+
+  def __init__(self, channel):
+    """Constructor.
+
+    Args:
+      channel: A grpc.Channel.
+    """
+    self.SayHello = channel.unary_unary(
+        '/helloworld.Greeter/SayHello',
+        request_serializer=helloworld__pb2.HelloRequest.SerializeToString,
+        response_deserializer=helloworld__pb2.HelloReply.FromString,
+        )
+
+
+class GreeterServicer(object):
+  """The greeting service definition.
+  """
+
+  def SayHello(self, request, context):
+    """Sends a greeting
+    """
+    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+    context.set_details('Method not implemented!')
+    raise NotImplementedError('Method not implemented!')
+
+
+def add_GreeterServicer_to_server(servicer, server):
+  rpc_method_handlers = {
+      'SayHello': grpc.unary_unary_rpc_method_handler(
+          servicer.SayHello,
+          request_deserializer=helloworld__pb2.HelloRequest.FromString,
+          response_serializer=helloworld__pb2.HelloReply.SerializeToString,
+      ),
+  }
+  generic_handler = grpc.method_handlers_generic_handler(
+      'helloworld.Greeter', rpc_method_handlers)
+  server.add_generic_rpc_handlers((generic_handler,))

examples/helloworld/helloworld_pb2_grpc.py

@@ -0,0 +1,28 @@
+from prometheus_client import Counter
+from prometheus_client import Histogram
+
+
+CLIENT_STARTED_COUNTER = Counter(
+    'grpc_client_started_total',
+    'Total number of RPCs started on the client.',
+    ["grpc_type", "grpc_service", "grpc_method"])
+
+CLIENT_HANDLED_COUNTER = Counter(
+    'grpc_client_handled_total',
+    'Total number of RPCs completed by the client, regardless of success or failure.',
+    ["grpc_type", "grpc_service", "grpc_method", "grpc_code"])
+
+CLIENT_HANDLED_LATENCY_SECONDS = Histogram(
+    'grpc_client_handling_seconds',
+    'Histogram of response latency (seconds) of the gRPC until it is finished by the application.',
+    ["grpc_type", "grpc_service", "grpc_method"])
+
+CLIENT_MSG_RECEIVED_TOTAL = Counter(
+    'grpc_client_msg_received_total',
+    'Total number of RPC stream messages received by the client.',
+    ["grpc_type", "grpc_service", "grpc_method"])
+
+CLIENT_MSG_SENT_TOTAL = Counter(
+    'grpc_client_msg_sent_total',
+    'Total number of gRPC stream messages sent by the client.',
+    ["grpc_type", "grpc_service", "grpc_method"])

python_grpc_prometheus/client_metrics.py

@@ -0,0 +1,94 @@
+import grpc
+
+import time
+
+from python_grpc_prometheus.client_metrics import (CLIENT_HANDLED_LATENCY_SECONDS,
+                                                   CLIENT_HANDLED_COUNTER,
+                                                   CLIENT_STARTED_COUNTER,
+                                                   CLIENT_MSG_RECEIVED_TOTAL,
+                                                   CLIENT_MSG_SENT_TOTAL)
+
+from python_grpc_prometheus import util
+from python_grpc_prometheus.util import split_call_details
+from python_grpc_prometheus.util import code_to_string
+
+_ = CLIENT_MSG_RECEIVED_TOTAL
+
+
+class PromClientInterceptor(grpc.UnaryUnaryClientInterceptor, grpc.UnaryStreamClientInterceptor,
+                            grpc.StreamUnaryClientInterceptor,
+                            grpc.StreamStreamClientInterceptor):
+
+    @staticmethod
+    def _callback(grpc_type, grpc_service, grpc_method, start):
+        def callback(future_response):
+            exception = future_response.exception()
+            code = code_to_string(grpc.StatusCode.OK)
+            if exception is not None:
+                if isinstance(exception, grpc.Call):
+                    code = code_to_string(exception.code())
+                else:
+                    code = code_to_string(grpc.StatusCode.UNKNOWN)
+
+            CLIENT_HANDLED_COUNTER.labels(
+                grpc_type=grpc_type,
+                grpc_service=grpc_service,
+                grpc_method=grpc_method,
+                grpc_code=code
+            ).inc()
+
+            CLIENT_HANDLED_LATENCY_SECONDS.labels(
+                grpc_type=grpc_type,
+                grpc_service=grpc_service,
+                grpc_method=grpc_method).observe(max(time.time() - start, 0))
+
+        return callback
+
+    def intercept_unary_unary(self, continuation, client_call_details, request):
+        grpc_service, grpc_method, ok = split_call_details(client_call_details, 3)
+        if not ok:
+            return continuation(client_call_details, request)
+
+        start = time.time()
+        grpc_type = util.Unary
+        CLIENT_STARTED_COUNTER.labels(
+            grpc_type=grpc_type,
+            grpc_service=grpc_service,
+            grpc_method=grpc_method).inc()
+        CLIENT_MSG_SENT_TOTAL.labels(
+            grpc_type=grpc_type,
+            grpc_service=grpc_service,
+            grpc_method=grpc_method).inc()
+
+        try:
+            response = continuation(client_call_details, request)
+            response.add_done_callback(self._callback(util.Unary, grpc_service, grpc_method, start))
+        except grpc.RpcError as e:
+            code = code_to_string(grpc.StatusCode.UNKNOWN)
+            if isinstance(e, grpc.Call):
+                code = code_to_string(e.code())
+            CLIENT_HANDLED_COUNTER.labels(
+                grpc_type=grpc_type,
+                grpc_service=grpc_service,
+                grpc_method=grpc_method,
+                grpc_code=code
+            ).inc()
+
+            CLIENT_HANDLED_LATENCY_SECONDS.labels(
+                grpc_type=grpc_type,
+                grpc_service=grpc_service,
+                grpc_method=grpc_method).observe(max(time.time() - start, 0))
+            raise
+        return response
+
+    def intercept_unary_stream(self, continuation, client_call_details, request):
+        response = continuation(client_call_details, request)
+        return response
+
+    def intercept_stream_unary(self, continuation, client_call_details, request_iterator):
+        response = continuation(client_call_details, request_iterator)
+        return response
+
+    def intercept_stream_stream(self, continuation, client_call_details, request_iterator):
+        response = continuation(client_call_details, request_iterator)
+        return response