grpc: add health checks to python services (#28)

also converted line endings for recommendationservice/requirements.txt from
dos to unix.

kubernetes-manifests/emailservice.yaml

@@ -30,12 +30,12 @@ spec:
         - containerPort: 8080
         readinessProbe:
           periodSeconds: 5
-          tcpSocket:
+          exec:
-            port: 8080
+            command: ["/bin/grpc_health_probe", "-addr=:8080"]
         livenessProbe:
           periodSeconds: 5
-          tcpSocket:
+          exec:
-            port: 8080
+            command: ["/bin/grpc_health_probe", "-addr=:8080"]
         resources:
           requests:
             cpu: 100m

kubernetes-manifests/recommendationservice.yaml

@@ -30,12 +30,12 @@ spec:
         - containerPort: 8080
         readinessProbe:
           periodSeconds: 5
-          tcpSocket:
+          exec:
-            port: 8080
+            command: ["/bin/grpc_health_probe", "-addr=:8080"]
         livenessProbe:
           periodSeconds: 5
-          tcpSocket:
+          exec:
-            port: 8080
+            command: ["/bin/grpc_health_probe", "-addr=:8080"]
         env:
         - name: PRODUCT_CATALOG_SERVICE_ADDR
           value: "productcatalogservice:3550"

src/emailservice/Dockerfile

@@ -1,6 +1,13 @@
+
+
 # Use the grpc.io provided Python image as the base image
 FROM grpc/python:1.0
 
+# download the grpc health probe
+RUN GRPC_HEALTH_PROBE_VERSION=v0.1.0-alpha.1 && \
+    wget -qO/bin/grpc_health_probe https://github.com/grpc-ecosystem/grpc-health-probe/releases/download/${GRPC_HEALTH_PROBE_VERSION}/grpc_health_probe-linux-amd64 && \
+    chmod +x /bin/grpc_health_probe
+
 # show python logs as they occur
 ENV PYTHONUNBUFFERED=0
 

src/emailservice/email_server.py

@@ -19,13 +19,14 @@ import argparse
 import os
 import sys
 import time
-
+import grpc
 from jinja2 import Environment, FileSystemLoader, select_autoescape, TemplateError
 from google.api_core.exceptions import GoogleAPICallError
-import grpc
 
 import demo_pb2
 import demo_pb2_grpc
+from grpc_health.v1 import health_pb2
+from grpc_health.v1 import health_pb2_grpc
 
 # from opencensus.trace.ext.grpc import server_interceptor
 # from opencensus.trace.samplers import always_on
@@ -56,7 +57,12 @@ env = Environment(
 )
 template = env.get_template('confirmation.html')
 
-class EmailService(demo_pb2_grpc.EmailServiceServicer):
+class BaseEmailService(demo_pb2_grpc.EmailServiceServicer):
+  def Check(self, request, context):
+    return health_pb2.HealthCheckResponse(
+      status=health_pb2.HealthCheckResponse.SERVING)
+
+class EmailService(BaseEmailService):
   def __init__(self):
     raise Exception('cloud mail client not implemented')
     super().__init__()
@@ -79,7 +85,6 @@ class EmailService(demo_pb2_grpc.EmailServiceServicer):
         "html_body": content
       }
     )
-    
     print("Message sent: {}".format(response.rfc822_message_id))
 
   def SendOrderConfirmation(self, request, context):
@@ -104,18 +109,30 @@ class EmailService(demo_pb2_grpc.EmailServiceServicer):
 
     return demo_pb2.Empty()
 
-class DummyEmailService(demo_pb2_grpc.EmailServiceServicer):
+class DummyEmailService(BaseEmailService):
   def SendOrderConfirmation(self, request, context):
     print('A request to send order confirmation email to {} has been received.'.format(request.email))
     return demo_pb2.Empty()
 
+class HealthCheck():
+  def Check(self, request, context):
+    return health_pb2.HealthCheckResponse(
+      status=health_pb2.HealthCheckResponse.SERVING)
+
 def start(dummy_mode):
   server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))#, interceptors=(tracer_interceptor,))
+  service = None
   if dummy_mode:
-    demo_pb2_grpc.add_EmailServiceServicer_to_server(DummyEmailService(), server)
+    service = DummyEmailService()
   else:
-    raise Exception('non-dummy mode not implemented')
+    raise Exception('non-dummy mode not implemented yet')
-  server.add_insecure_port('[::]:8080')
+
+  demo_pb2_grpc.add_EmailServiceServicer_to_server(service, server)
+  health_pb2_grpc.add_HealthServicer_to_server(service, server)
+
+  port = os.environ.get('PORT', "8080")
+  print("listening on port: "+port)
+  server.add_insecure_port('[::]:'+port)
   server.start()
   try:
     while True:
@@ -125,5 +142,5 @@ def start(dummy_mode):
 
 
 if __name__ == '__main__':
-  print('Starting the email service in dummy mode.')
+  print('starting the email service in dummy mode.')
   start(dummy_mode = True)

src/emailservice/requirements.txt

@@ -9,6 +9,7 @@ google-cloud-trace==0.19.0
 googleapis-common-protos==1.5.3
 grpc-google-iam-v1==0.11.4
 grpcio==1.12.1
+grpcio-health-checking==1.14.1
 grpcio-tools==1.12.1
 idna==2.7
 Jinja2==2.10

src/recommendationservice/Dockerfile

@@ -3,6 +3,11 @@ FROM grpc/python:1.0
 # show python logs as they occur
 ENV PYTHONUNBUFFERED=0
 
+# download the grpc health probe
+RUN GRPC_HEALTH_PROBE_VERSION=v0.1.0-alpha.1 && \
+    wget -qO/bin/grpc_health_probe https://github.com/grpc-ecosystem/grpc-health-probe/releases/download/${GRPC_HEALTH_PROBE_VERSION}/grpc_health_probe-linux-amd64 && \
+    chmod +x /bin/grpc_health_probe
+
 # get packages
 WORKDIR /recommendationservice
 COPY requirements.txt requirements.txt

src/recommendationservice/demo_pb2_grpc.py

@@ -1,19 +1,3 @@
-#!/usr/bin/python
-#
-# Copyright 2018 Google LLC
-#
-# 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.
-
 # Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
 import grpc
 
@@ -441,11 +425,6 @@ class CheckoutServiceStub(object):
     Args:
       channel: A grpc.Channel.
     """
-    self.CreateOrder = channel.unary_unary(
-        '/hipstershop.CheckoutService/CreateOrder',
-        request_serializer=demo__pb2.CreateOrderRequest.SerializeToString,
-        response_deserializer=demo__pb2.CreateOrderResponse.FromString,
-        )
     self.PlaceOrder = channel.unary_unary(
         '/hipstershop.CheckoutService/PlaceOrder',
         request_serializer=demo__pb2.PlaceOrderRequest.SerializeToString,
@@ -458,13 +437,6 @@ class CheckoutServiceServicer(object):
 
   """
 
-  def CreateOrder(self, request, context):
-    # missing associated documentation comment in .proto file
-    pass
-    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
-    context.set_details('Method not implemented!')
-    raise NotImplementedError('Method not implemented!')
-
   def PlaceOrder(self, request, context):
     # missing associated documentation comment in .proto file
     pass
@@ -475,11 +447,6 @@ class CheckoutServiceServicer(object):
 
 def add_CheckoutServiceServicer_to_server(servicer, server):
   rpc_method_handlers = {
-      'CreateOrder': grpc.unary_unary_rpc_method_handler(
-          servicer.CreateOrder,
-          request_deserializer=demo__pb2.CreateOrderRequest.FromString,
-          response_serializer=demo__pb2.CreateOrderResponse.SerializeToString,
-      ),
       'PlaceOrder': grpc.unary_unary_rpc_method_handler(
           servicer.PlaceOrder,
           request_deserializer=demo__pb2.PlaceOrderRequest.FromString,
@@ -489,3 +456,47 @@ def add_CheckoutServiceServicer_to_server(servicer, server):
   generic_handler = grpc.method_handlers_generic_handler(
       'hipstershop.CheckoutService', rpc_method_handlers)
   server.add_generic_rpc_handlers((generic_handler,))
+
+
+class AdsServiceStub(object):
+  """------------Ads service------------------
+
+  """
+
+  def __init__(self, channel):
+    """Constructor.
+
+    Args:
+      channel: A grpc.Channel.
+    """
+    self.GetAds = channel.unary_unary(
+        '/hipstershop.AdsService/GetAds',
+        request_serializer=demo__pb2.AdsRequest.SerializeToString,
+        response_deserializer=demo__pb2.AdsResponse.FromString,
+        )
+
+
+class AdsServiceServicer(object):
+  """------------Ads service------------------
+
+  """
+
+  def GetAds(self, request, context):
+    # missing associated documentation comment in .proto file
+    pass
+    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+    context.set_details('Method not implemented!')
+    raise NotImplementedError('Method not implemented!')
+
+
+def add_AdsServiceServicer_to_server(servicer, server):
+  rpc_method_handlers = {
+      'GetAds': grpc.unary_unary_rpc_method_handler(
+          servicer.GetAds,
+          request_deserializer=demo__pb2.AdsRequest.FromString,
+          response_serializer=demo__pb2.AdsResponse.SerializeToString,
+      ),
+  }
+  generic_handler = grpc.method_handlers_generic_handler(
+      'hipstershop.AdsService', rpc_method_handlers)
+  server.add_generic_rpc_handlers((generic_handler,))

src/recommendationservice/recommendation_server.py

@@ -15,16 +15,19 @@
 # limitations under the License.
 
 import grpc
-import demo_pb2
-import demo_pb2_grpc
 from concurrent import futures
 import time
 import traceback
 import random
 import os
-
 import googleclouddebugger
 
+import demo_pb2
+import demo_pb2_grpc
+from grpc_health.v1 import health_pb2
+from grpc_health.v1 import health_pb2_grpc
+
+
 # TODO(morganmclean,ahmetb) tracing currently disabled due to memory leak (see TODO below)
 # from opencensus.trace.ext.grpc import server_interceptor
 # from opencensus.trace.samplers import always_on
@@ -35,7 +38,7 @@ class RecommendationService(demo_pb2_grpc.RecommendationServiceServicer):
     def ListRecommendations(self, request, context):
         max_responses = 5
         # fetch list of products from product catalog stub
-        cat_response = stub.ListProducts(demo_pb2.Empty())
+        cat_response = product_catalog_stub.ListProducts(demo_pb2.Empty())
         product_ids = [x.id for x in cat_response.products]
         filtered_products = list(set(product_ids)-set(request.product_ids))
         num_products = len(filtered_products)
@@ -50,6 +53,11 @@ class RecommendationService(demo_pb2_grpc.RecommendationServiceServicer):
         response.product_ids.extend(prod_list)
         return response
 
+    def Check(self, request, context):
+        return health_pb2.HealthCheckResponse(
+            status=health_pb2.HealthCheckResponse.SERVING)
+
+
 if __name__ == "__main__":
     print("initializing recommendationservice")
 
@@ -78,16 +86,16 @@ if __name__ == "__main__":
     if catalog_addr == "":
         raise Exception('PRODUCT_CATALOG_SERVICE_ADDR environment variable not set')
     print("product catalog address: " + catalog_addr)
-
-    # stub for product catalog service
     channel = grpc.insecure_channel(catalog_addr)
-    stub = demo_pb2_grpc.ProductCatalogServiceStub(channel)
+    product_catalog_stub = demo_pb2_grpc.ProductCatalogServiceStub(channel)
 
     # create gRPC server
     server = grpc.server(futures.ThreadPoolExecutor(max_workers=10)) # ,interceptors=(tracer_interceptor,))
 
     # add class to gRPC server
-    demo_pb2_grpc.add_RecommendationServiceServicer_to_server(RecommendationService(), server)
+    service = RecommendationService()
+    demo_pb2_grpc.add_RecommendationServiceServicer_to_server(service, server)
+    health_pb2_grpc.add_HealthServicer_to_server(service, server)
 
     # start server
     print("listening on port: " + port)

src/recommendationservice/requirements.txt

@@ -12,6 +12,7 @@ google-cloud-trace==0.19.0
 google-python-cloud-debugger==2.8
 googleapis-common-protos==1.5.3
 grpcio==1.13.0
+grpcio-health-checking==1.14.1
 grpcio-tools==1.0.0
 httplib2==0.11.3
 idna==2.7