recommendation_server.py

#!/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.

import os
import random
import time
import datetime
import uuid
import traceback
from concurrent import futures

import googlecloudprofiler
from google.auth.exceptions import DefaultCredentialsError
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 opentelemetry import trace
from opentelemetry.instrumentation.grpc import GrpcInstrumentorClient, GrpcInstrumentorServer
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

from logger import getJSONLogger
logger = getJSONLogger('recommendationservice-server')

RECOMMENDATIONSERVICE = "recommendationservice"
PRODUCTCATALOGSERVICE = "productcatalogservice"

# NOTE: logLevel must be a GELF valid severity value (WARN or ERROR), INFO if not specified
def emitLog(event, logLevel):
  timestamp = str(datetime.datetime.now().isoformat())[:-3] + 'Z'

  if logLevel == "ERROR":
    logger.error(timestamp + " - ERROR - " + RECOMMENDATIONSERVICE + " - " + event)
  elif logLevel == "WARN":
    logger.warn(timestamp + " - WARN - " + RECOMMENDATIONSERVICE + " - " + event)
  else:
    logger.info(timestamp + " - INFO - " + RECOMMENDATIONSERVICE + " - " + event)

def initStackdriverProfiling():
  project_id = None
  try:
    project_id = os.environ["GCP_PROJECT_ID"]
  except KeyError:
    # Environment variable not set
    pass

  for retry in range(1,4):
    try:
      if project_id:
        googlecloudprofiler.start(service='recommendation_server', service_version='1.0.0', verbose=0, project_id=project_id)
      else:
        googlecloudprofiler.start(service='recommendation_server', service_version='1.0.0', verbose=0)
      logger.info("Successfully started Stackdriver Profiler.")
      return
    except (BaseException) as exc:
      logger.info("Unable to start Stackdriver Profiler Python agent. " + str(exc))
      if (retry < 4):
        logger.info("Sleeping %d seconds to retry Stackdriver Profiler agent initialization"%(retry*10))
        time.sleep (1)
      else:
        logger.warning("Could not initialize Stackdriver Profiler after retrying, giving up")
  return

class RecommendationService(demo_pb2_grpc.RecommendationServiceServicer):
    def ListRecommendations(self, request, context):
        metadict = dict(context.invocation_metadata())
        emitLog("Received request from " + metadict['servicename'] + " (request_id: " + metadict['requestid'] + ")", "INFO")
        
        # adding metadata for gRPC towards ProductCatalogService
        newRequestid = '{0}'.format(uuid.uuid4())
        metadata = (('requestid', newRequestid), ('servicename', RECOMMENDATIONSERVICE))
        
        try:
          max_responses = 5
          emitLog("Sending message to " + PRODUCTCATALOGSERVICE + " (request_id: " + newRequestid + ")", "INFO")
          
          # fetch list of products from product catalog stub
          cat_response = product_catalog_stub.ListProducts(demo_pb2.Empty(), timeout=1, metadata=metadata)
          
          emitLog("Receiving answer from " + PRODUCTCATALOGSERVICE + " (request_id: " + newRequestid + ")", "INFO")
          
          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)
          # sample list of indicies to return
          indices = random.sample(range(num_products), num_return)
          # fetch product ids from indices
          prod_list = [filtered_products[i] for i in indices]
          logger.info("[Recv ListRecommendations] product_ids={}".format(prod_list))
          # build and return response
          response = demo_pb2.ListRecommendationsResponse()
          response.product_ids.extend(prod_list)  
          event = "Answered request from " + metadict['servicename'] + " (request_id: " + metadict['requestid'] + ")"
          emitLog(event, "INFO")
          return response
        
        except grpc.RpcError as e:
          if e.code() == grpc.StatusCode.UNAVAILABLE or e.code() == grpc.StatusCode.DEADLINE_EXCEEDED:
            emitLog("Failing to contact " + PRODUCTCATALOGSERVICE + " (request_id: " + newRequestid + "). Root cause: (" + e.details() + ")", "ERROR")
          else:
            emitLog("Error response received from " + PRODUCTCATALOGSERVICE + " (request_id: " + newRequestid + ")", "ERROR")
          return None

    def Check(self, request, context):
        return health_pb2.HealthCheckResponse(
            status=health_pb2.HealthCheckResponse.SERVING)

    def Watch(self, request, context):
        return health_pb2.HealthCheckResponse(
            status=health_pb2.HealthCheckResponse.UNIMPLEMENTED)


if __name__ == "__main__":
    logger.info("initializing recommendationservice")

    try:
      if "DISABLE_PROFILER" in os.environ:
        raise KeyError()
      else:
        logger.info("Profiler enabled.")
        initStackdriverProfiling()
    except KeyError:
        logger.info("Profiler disabled.")

    try:
      grpc_client_instrumentor = GrpcInstrumentorClient()
      grpc_client_instrumentor.instrument()
      grpc_server_instrumentor = GrpcInstrumentorServer()
      grpc_server_instrumentor.instrument()
      if os.environ["ENABLE_TRACING"] == "1":
        trace.set_tracer_provider(TracerProvider())
        otel_endpoint = os.getenv("COLLECTOR_SERVICE_ADDR", "localhost:4317")
        trace.get_tracer_provider().add_span_processor(
          BatchSpanProcessor(
              OTLPSpanExporter(
              endpoint = otel_endpoint,
              insecure = True
            )
          )
        )
    except (KeyError, DefaultCredentialsError):
        logger.info("Tracing disabled.")
    except Exception as e:
        logger.warn(f"Exception on Cloud Trace setup: {traceback.format_exc()}, tracing disabled.") 

    port = os.environ.get('PORT', "8080")
    catalog_addr = os.environ.get('PRODUCT_CATALOG_SERVICE_ADDR', '')
    if catalog_addr == "":
        raise Exception('PRODUCT_CATALOG_SERVICE_ADDR environment variable not set')
    logger.info("product catalog address: " + catalog_addr)
    channel = grpc.insecure_channel(catalog_addr)
    product_catalog_stub = demo_pb2_grpc.ProductCatalogServiceStub(channel)

    # create gRPC server
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))

    # add class to gRPC server
    service = RecommendationService()
    demo_pb2_grpc.add_RecommendationServiceServicer_to_server(service, server)
    health_pb2_grpc.add_HealthServicer_to_server(service, server)

    # start server
    logger.info("listening on port: " + port)
    server.add_insecure_port('[::]:'+port)
    server.start()

    # keep alive
    try:
         while True:
            time.sleep(10000)
    except KeyboardInterrupt:
            server.stop(0)