Commit 810d16cf authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Service component: 

- Migrated to use new generic gRPC servicer
- Migrated to use new generic Rest servicer
- Migrated to use new settings framework
- Migrated tests to use new generic servicers and mock's
- Extracted common unitary test functionalities into separate code files
- Minor code styling/formatting
parent 0ed8cfee

src/service/Config.py

+0 −19
Original line number Diff line number Diff line
@@ -12,22 +12,3 @@ 
# See the License for the specific language governing permissions and

# limitations under the License.

import logging



# General settings

LOG_LEVEL = logging.WARNING



# gRPC settings

GRPC_SERVICE_PORT = 3030

GRPC_MAX_WORKERS  = 10

GRPC_GRACE_PERIOD = 60



# Prometheus settings

METRICS_PORT = 9192



# Dependency micro-service connection settings

CONTEXT_SERVICE_HOST = '127.0.0.1'

CONTEXT_SERVICE_PORT = 1010



DEVICE_SERVICE_HOST = '127.0.0.1'

DEVICE_SERVICE_PORT = 2020

src/service/client/ServiceClient.py

+7 −3
Original line number Diff line number Diff line
@@ -13,6 +13,8 @@ 
# limitations under the License.



import grpc, logging

from common.Constants import ServiceNameEnum

from common.Settings import get_service_host, get_service_port_grpc

from common.tools.client.RetryDecorator import retry, delay_exponential

from common.tools.grpc.Tools import grpc_message_to_json_string

from service.proto.context_pb2 import Empty, Service, ServiceId
@@ -24,9 +26,11 @@ DELAY_FUNCTION = delay_exponential(initial=0.01, increment=2.0, maximum=5.0) 
RETRY_DECORATOR = retry(max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect')



class ServiceClient:

    def __init__(self, address, port):

        self.endpoint = '{:s}:{:s}'.format(str(address), str(port))

        LOGGER.debug('Creating channel to {:s}...'.format(self.endpoint))

    def __init__(self, host=None, port=None):

        if not host: host = get_service_host(ServiceNameEnum.SERVICE)

        if not port: port = get_service_port_grpc(ServiceNameEnum.SERVICE)

        self.endpoint = '{:s}:{:s}'.format(str(host), str(port))

        LOGGER.debug('Creating channel to {:s}...'.format(str(self.endpoint)))

        self.channel = None

        self.stub = None

        self.connect()

src/service/service/ServiceService.py

+10 −60
Original line number Diff line number Diff line
@@ -12,72 +12,22 @@ 
# See the License for the specific language governing permissions and

# limitations under the License.



import grpc, logging

from concurrent import futures

from grpc_health.v1.health import HealthServicer, OVERALL_HEALTH

from grpc_health.v1.health_pb2 import HealthCheckResponse

from grpc_health.v1.health_pb2_grpc import add_HealthServicer_to_server

from common.Constants import ServiceNameEnum

from common.Settings import get_service_port_grpc

from common.orm.backend.BackendEnum import BackendEnum

from common.orm.Database import Database

from common.orm.Factory import get_database_backend

from context.client.ContextClient import ContextClient

from device.client.DeviceClient import DeviceClient

from service.Config import GRPC_SERVICE_PORT, GRPC_MAX_WORKERS, GRPC_GRACE_PERIOD

from common.tools.service.GenericGrpcService import GenericGrpcService

from service.proto.service_pb2_grpc import add_ServiceServiceServicer_to_server

from .ServiceServiceServicerImpl import ServiceServiceServicerImpl

from .service_handler_api.ServiceHandlerFactory import ServiceHandlerFactory



BIND_ADDRESS = '0.0.0.0'

LOGGER = logging.getLogger(__name__)

class ServiceService(GenericGrpcService):

    def __init__(self, service_handler_factory : ServiceHandlerFactory, cls_name: str = __name__) -> None:

        port = get_service_port_grpc(ServiceNameEnum.SERVICE)

        super().__init__(port, cls_name=cls_name)

        database = Database(get_database_backend(backend=BackendEnum.INMEMORY))

        self.service_servicer = ServiceServiceServicerImpl(database, service_handler_factory)



class ServiceService:

    def __init__(

        self, context_client : ContextClient, device_client : DeviceClient,

        service_handler_factory : ServiceHandlerFactory,

        address=BIND_ADDRESS, port=GRPC_SERVICE_PORT, max_workers=GRPC_MAX_WORKERS,

        grace_period=GRPC_GRACE_PERIOD):



        self.context_client = context_client

        self.device_client = device_client

        self.service_handler_factory = service_handler_factory

        self.address = address

        self.port = port

        self.endpoint = None

        self.max_workers = max_workers

        self.grace_period = grace_period

        self.service_servicer = None

        self.health_servicer = None

        self.pool = None

        self.server = None



        self.database = Database(get_database_backend(backend=BackendEnum.INMEMORY))



    def start(self):

        self.endpoint = '{:s}:{:s}'.format(str(self.address), str(self.port))

        LOGGER.info('Starting Service (tentative endpoint: {:s}, max_workers: {:s})...'.format(

            str(self.endpoint), str(self.max_workers)))



        self.pool = futures.ThreadPoolExecutor(max_workers=self.max_workers)

        self.server = grpc.server(self.pool) # , interceptors=(tracer_interceptor,))



        self.service_servicer = ServiceServiceServicerImpl(

            self.context_client, self.device_client, self.database, self.service_handler_factory)

    def install_servicers(self):

        add_ServiceServiceServicer_to_server(self.service_servicer, self.server)
 


        self.health_servicer = HealthServicer(

            experimental_non_blocking=True, experimental_thread_pool=futures.ThreadPoolExecutor(max_workers=1))

        add_HealthServicer_to_server(self.health_servicer, self.server)



        port = self.server.add_insecure_port(self.endpoint)

        self.endpoint = '{:s}:{:s}'.format(str(self.address), str(port))

        LOGGER.info('Listening on {:s}...'.format(str(self.endpoint)))

        self.server.start()

        self.health_servicer.set(OVERALL_HEALTH, HealthCheckResponse.SERVING) # pylint: disable=maybe-no-member



        LOGGER.debug('Service started')



    def stop(self):

        LOGGER.debug('Stopping service (grace period {:s} seconds)...'.format(str(self.grace_period)))

        self.health_servicer.enter_graceful_shutdown()

        self.server.stop(self.grace_period)

        LOGGER.debug('Service stopped')

src/service/service/ServiceServiceServicerImpl.py

+3 −6
Original line number Diff line number Diff line
@@ -39,13 +39,10 @@ METHOD_NAMES = ['CreateService', 'UpdateService', 'DeleteService'] 
METRICS = create_metrics(SERVICE_NAME, METHOD_NAMES)



class ServiceServiceServicerImpl(ServiceServiceServicer):

    def __init__(

        self, context_client : ContextClient, device_client : DeviceClient, database : Database,

        service_handler_factory : ServiceHandlerFactory):



    def __init__(self, database : Database, service_handler_factory : ServiceHandlerFactory) -> None:

        LOGGER.debug('Creating Servicer...')

        self.context_client = context_client

        self.device_client = device_client

        self.context_client = ContextClient()

        self.device_client = DeviceClient()

        self.database = database

        self.service_handler_factory = service_handler_factory

        LOGGER.debug('Servicer Created')

src/service/service/__main__.py

+12 −34
Original line number Diff line number Diff line
@@ -14,12 +14,10 @@ 


import logging, signal, sys, threading

from prometheus_client import start_http_server

from common.Settings import get_setting, wait_for_environment_variables

from context.client.ContextClient import ContextClient

from device.client.DeviceClient import DeviceClient

from service.Config import (

    CONTEXT_SERVICE_HOST, CONTEXT_SERVICE_PORT, DEVICE_SERVICE_HOST, DEVICE_SERVICE_PORT, GRPC_SERVICE_PORT,

    GRPC_MAX_WORKERS, GRPC_GRACE_PERIOD, LOG_LEVEL, METRICS_PORT)

from common.Constants import ServiceNameEnum

from common.Settings import (

    ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_GRPC, get_env_var_name, get_log_level, get_metrics_port,

    wait_for_environment_variables)

from .ServiceService import ServiceService

from .service_handler_api.ServiceHandlerFactory import ServiceHandlerFactory

from .service_handlers import SERVICE_HANDLERS
@@ -34,51 +32,31 @@ def signal_handler(signal, frame): # pylint: disable=redefined-outer-name 
def main():

    global LOGGER # pylint: disable=global-statement



    grpc_service_port    = get_setting('SERVICESERVICE_SERVICE_PORT_GRPC', default=GRPC_SERVICE_PORT   )

    max_workers          = get_setting('MAX_WORKERS',                      default=GRPC_MAX_WORKERS    )

    grace_period         = get_setting('GRACE_PERIOD',                     default=GRPC_GRACE_PERIOD   )

    log_level            = get_setting('LOG_LEVEL',                        default=LOG_LEVEL           )

    metrics_port         = get_setting('METRICS_PORT',                     default=METRICS_PORT        )



    log_level = get_log_level()

    logging.basicConfig(level=log_level)

    LOGGER = logging.getLogger(__name__)



    wait_for_environment_variables([

        'CONTEXTSERVICE_SERVICE_HOST', 'CONTEXTSERVICE_SERVICE_PORT_GRPC',

        'DEVICESERVICE_SERVICE_HOST', 'DEVICESERVICE_SERVICE_PORT_GRPC'

        get_env_var_name(ServiceNameEnum.CONTEXT, ENVVAR_SUFIX_SERVICE_HOST     ),

        get_env_var_name(ServiceNameEnum.CONTEXT, ENVVAR_SUFIX_SERVICE_PORT_GRPC),

        get_env_var_name(ServiceNameEnum.DEVICE,  ENVVAR_SUFIX_SERVICE_HOST     ),

        get_env_var_name(ServiceNameEnum.DEVICE,  ENVVAR_SUFIX_SERVICE_PORT_GRPC),

    ])



    context_service_host = get_setting('CONTEXTSERVICE_SERVICE_HOST',      default=CONTEXT_SERVICE_HOST)

    context_service_port = get_setting('CONTEXTSERVICE_SERVICE_PORT_GRPC', default=CONTEXT_SERVICE_PORT)

    device_service_host  = get_setting('DEVICESERVICE_SERVICE_HOST',       default=DEVICE_SERVICE_HOST )

    device_service_port  = get_setting('DEVICESERVICE_SERVICE_PORT_GRPC',  default=DEVICE_SERVICE_PORT )



    signal.signal(signal.SIGINT,  signal_handler)

    signal.signal(signal.SIGTERM, signal_handler)



    LOGGER.info('Starting...')



    # Start metrics server

    metrics_port = get_metrics_port()

    start_http_server(metrics_port)



    # Initialize Context Client

    if context_service_host is None or context_service_port is None:

        raise Exception('Wrong address({:s}):port({:s}) of Context component'.format(

            str(context_service_host), str(context_service_port)))

    context_client = ContextClient(context_service_host, context_service_port)



    # Initialize Device Client

    if device_service_host is None or device_service_port is None:

        raise Exception('Wrong address({:s}):port({:s}) of Device component'.format(

            str(device_service_host), str(device_service_port)))

    device_client = DeviceClient(device_service_host, device_service_port)



    # Initialize ServiceHandler Factory

    service_handler_factory = ServiceHandlerFactory(SERVICE_HANDLERS)



    # Starting service service

    grpc_service = ServiceService(

        context_client, device_client, service_handler_factory, port=grpc_service_port, max_workers=max_workers,

        grace_period=grace_period)

    grpc_service = ServiceService(service_handler_factory)

    grpc_service.start()



    # Wait for Ctrl+C or termination signal