2018-06-14 19:30:50 +00:00
|
|
|
import grpc
|
|
|
|
import demo_pb2
|
|
|
|
import demo_pb2_grpc
|
2018-06-14 19:45:44 +00:00
|
|
|
from concurrent import futures
|
|
|
|
import time
|
2018-06-14 23:10:11 +00:00
|
|
|
import random
|
2018-06-21 05:37:56 +00:00
|
|
|
import os
|
2018-06-14 19:30:50 +00:00
|
|
|
|
2018-07-16 21:00:42 +00:00
|
|
|
import googleclouddebugger
|
2018-07-16 18:25:10 +00:00
|
|
|
|
2018-07-18 04:01:21 +00:00
|
|
|
# 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
|
|
|
|
# from opencensus.trace.exporters import stackdriver_exporter
|
|
|
|
# from opencensus.trace.exporters import print_exporter
|
|
|
|
|
2018-06-14 19:45:44 +00:00
|
|
|
class RecommendationService(demo_pb2_grpc.RecommendationServiceServicer):
|
|
|
|
def ListRecommendations(self, request, context):
|
2018-06-21 06:26:32 +00:00
|
|
|
max_responses = 5
|
2018-06-21 06:40:47 +00:00
|
|
|
# fetch list of products from product catalog stub
|
2018-06-21 06:26:32 +00:00
|
|
|
cat_response = stub.ListProducts(demo_pb2.Empty())
|
2018-06-26 21:01:05 +00:00
|
|
|
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)
|
|
|
|
num_return = min(max_responses, num_products)
|
2018-06-21 06:40:47 +00:00
|
|
|
# sample list of indicies to return
|
2018-06-26 21:01:05 +00:00
|
|
|
indices = random.sample(range(num_products), num_return)
|
2018-06-21 06:40:47 +00:00
|
|
|
# fetch product ids from indices
|
2018-06-26 21:01:05 +00:00
|
|
|
prod_list = [filtered_products[i] for i in indices]
|
2018-07-18 04:01:21 +00:00
|
|
|
print("[Recv ListRecommendations] product_ids={}".format(prod_list))
|
2018-06-21 06:40:47 +00:00
|
|
|
# build and return response
|
2018-06-14 19:45:44 +00:00
|
|
|
response = demo_pb2.ListRecommendationsResponse()
|
2018-06-14 23:10:11 +00:00
|
|
|
response.product_ids.extend(prod_list)
|
2018-06-14 19:45:44 +00:00
|
|
|
return response
|
2018-07-17 15:39:23 +00:00
|
|
|
|
2018-06-14 19:45:44 +00:00
|
|
|
if __name__ == "__main__":
|
2018-07-18 04:01:21 +00:00
|
|
|
print("initializing recommendationservice")
|
|
|
|
|
|
|
|
# TODO(morganmclean,ahmetb) enabling the tracing interceptor/sampler below
|
|
|
|
# causes an unbounded memory leak eventually OOMing the container.
|
|
|
|
# ----
|
|
|
|
# try:
|
|
|
|
# sampler = always_on.AlwaysOnSampler()
|
|
|
|
# exporter = stackdriver_exporter.StackdriverExporter()
|
|
|
|
# tracer_interceptor = server_interceptor.OpenCensusServerInterceptor(sampler, exporter)
|
|
|
|
# except:
|
|
|
|
# tracer_interceptor = server_interceptor.OpenCensusServerInterceptor()
|
2018-07-16 18:25:10 +00:00
|
|
|
|
2018-07-16 21:00:42 +00:00
|
|
|
try:
|
|
|
|
googleclouddebugger.enable(
|
|
|
|
module='recommendationserver',
|
2018-07-16 22:09:51 +00:00
|
|
|
version='1.0.0'
|
2018-07-16 21:00:42 +00:00
|
|
|
)
|
2018-07-18 04:01:21 +00:00
|
|
|
except Exception, err:
|
|
|
|
print("could not enable debugger")
|
|
|
|
traceback.print_exc()
|
2018-07-16 21:00:42 +00:00
|
|
|
pass
|
2018-07-16 18:25:10 +00:00
|
|
|
|
2018-06-21 05:37:56 +00:00
|
|
|
port = os.environ.get('PORT', "8080")
|
2018-06-21 17:22:28 +00:00
|
|
|
catalog_addr = os.environ.get('PRODUCT_CATALOG_SERVICE_ADDR', '')
|
|
|
|
if catalog_addr == "":
|
2018-07-18 04:01:21 +00:00
|
|
|
raise Exception('PRODUCT_CATALOG_SERVICE_ADDR environment variable not set')
|
2018-06-21 05:37:56 +00:00
|
|
|
print("product catalog address: " + catalog_addr)
|
2018-07-18 04:01:21 +00:00
|
|
|
|
2018-06-21 06:26:32 +00:00
|
|
|
# stub for product catalog service
|
|
|
|
channel = grpc.insecure_channel(catalog_addr)
|
|
|
|
stub = demo_pb2_grpc.ProductCatalogServiceStub(channel)
|
2018-07-18 04:01:21 +00:00
|
|
|
|
2018-06-14 19:45:44 +00:00
|
|
|
# create gRPC server
|
2018-07-18 04:01:21 +00:00
|
|
|
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10)) # ,interceptors=(tracer_interceptor,))
|
|
|
|
|
2018-06-14 19:45:44 +00:00
|
|
|
# add class to gRPC server
|
|
|
|
demo_pb2_grpc.add_RecommendationServiceServicer_to_server(RecommendationService(), server)
|
2018-07-18 04:01:21 +00:00
|
|
|
|
2018-06-14 19:45:44 +00:00
|
|
|
# start server
|
2018-07-18 04:01:21 +00:00
|
|
|
print("listening on port: " + port)
|
2018-06-14 22:52:41 +00:00
|
|
|
server.add_insecure_port('[::]:'+port)
|
2018-06-14 19:45:44 +00:00
|
|
|
server.start()
|
2018-07-18 04:01:21 +00:00
|
|
|
|
2018-06-14 19:45:44 +00:00
|
|
|
# keep alive
|
|
|
|
try:
|
|
|
|
while True:
|
2018-06-14 22:31:06 +00:00
|
|
|
time.sleep(10000)
|
2018-06-14 19:45:44 +00:00
|
|
|
except KeyboardInterrupt:
|
2018-07-16 22:09:51 +00:00
|
|
|
server.stop(0)
|