vbatts
/
microservices-demo
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

recomendationservice: disable tracing b/c memleak 

enabling the tracing interceptor/sampler causes an unbounded memory leak
eventually OOMing the container.

Signed-off-by: Ahmet Alp Balkan <ahmetb@google.com>
This commit is contained in:
Ahmet Alp Balkan 2018-07-17 21:01:21 -07:00
parent 2c8681d324
commit 316db889aa
1 changed files with 29 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
src/recommendationservice/recommendation_server.py
View File

@ -6,13 +6,14 @@ import time
import random import random
import os import os
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
import googleclouddebugger import googleclouddebugger
# 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
class RecommendationService(demo_pb2_grpc.RecommendationServiceServicer): class RecommendationService(demo_pb2_grpc.RecommendationServiceServicer):
def ListRecommendations(self, request, context): def ListRecommendations(self, request, context):
max_responses = 5 max_responses = 5
@ -26,47 +27,56 @@ class RecommendationService(demo_pb2_grpc.RecommendationServiceServicer):
indices = random.sample(range(num_products), num_return) indices = random.sample(range(num_products), num_return)
# fetch product ids from indices # fetch product ids from indices
prod_list = [filtered_products[i] for i in indices] prod_list = [filtered_products[i] for i in indices]
print("handling request: {}".format(prod_list)) print("[Recv ListRecommendations] product_ids={}".format(prod_list))
# build and return response # build and return response
response = demo_pb2.ListRecommendationsResponse() response = demo_pb2.ListRecommendationsResponse()
response.product_ids.extend(prod_list) response.product_ids.extend(prod_list)
return response return response
if __name__ == "__main__": if __name__ == "__main__":
print("starting recommendationservice") print("initializing recommendationservice")
try:
sampler = always_on.AlwaysOnSampler() # TODO(morganmclean,ahmetb) enabling the tracing interceptor/sampler below
exporter = stackdriver_exporter.StackdriverExporter() # causes an unbounded memory leak eventually OOMing the container.
tracer_interceptor = server_interceptor.OpenCensusServerInterceptor(sampler, exporter) # ----
except: # try:
tracer_interceptor = server_interceptor.OpenCensusServerInterceptor() # sampler = always_on.AlwaysOnSampler()
# exporter = stackdriver_exporter.StackdriverExporter()
# tracer_interceptor = server_interceptor.OpenCensusServerInterceptor(sampler, exporter)
# except:
# tracer_interceptor = server_interceptor.OpenCensusServerInterceptor()
try: try:
googleclouddebugger.enable( googleclouddebugger.enable(
module='recommendationserver', module='recommendationserver',
version='1.0.0' version='1.0.0'
) )
except: except Exception, err:
print("could not enable debugger")
traceback.print_exc()
pass pass
# get port from $PORT envar
port = os.environ.get('PORT', "8080") port = os.environ.get('PORT', "8080")
# get product catalog service address from $PRODUCT_CATALOG_SERVICE_ADDR envar
catalog_addr = os.environ.get('PRODUCT_CATALOG_SERVICE_ADDR', '') catalog_addr = os.environ.get('PRODUCT_CATALOG_SERVICE_ADDR', '')
if catalog_addr == "": if catalog_addr == "":
raise Exception('PRODUCT_CATALOG_SERVICE_ADDR environment not set') raise Exception('PRODUCT_CATALOG_SERVICE_ADDR environment variable not set')
print("product catalog address: " + catalog_addr) print("product catalog address: " + catalog_addr)
print("listening on port: " + port)
# stub for product catalog service # stub for product catalog service
channel = grpc.insecure_channel(catalog_addr) channel = grpc.insecure_channel(catalog_addr)
stub = demo_pb2_grpc.ProductCatalogServiceStub(channel) stub = demo_pb2_grpc.ProductCatalogServiceStub(channel)
# create gRPC server # create gRPC server
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10),interceptors=(tracer_interceptor,)) server = grpc.server(futures.ThreadPoolExecutor(max_workers=10)) # ,interceptors=(tracer_interceptor,))
# add class to gRPC server # add class to gRPC server
demo_pb2_grpc.add_RecommendationServiceServicer_to_server(RecommendationService(), server) demo_pb2_grpc.add_RecommendationServiceServicer_to_server(RecommendationService(), server)
# start server # start server
print("listening on port: " + port)
server.add_insecure_port('[::]:'+port) server.add_insecure_port('[::]:'+port)
server.start() server.start()
# keep alive # keep alive
try: try:
while True: while True: