Commit 4375de80 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Compute component: 

- Migrated to use new generic gRPC servicer
- Migrated to use new generic Rest servicer
- Migrated to use new settings framework
- Minor code formatting/styling
parent 38659ca6

src/compute/Config.py

+1 −15
Original line number Diff line number Diff line
@@ -12,23 +12,9 @@ 
# See the License for the specific language governing permissions and

# limitations under the License.



import logging

from werkzeug.security import generate_password_hash



# General settings

LOG_LEVEL = logging.WARNING



# gRPC settings

GRPC_SERVICE_PORT = 9090

GRPC_MAX_WORKERS  = 10

GRPC_GRACE_PERIOD = 60



# REST-API settings

RESTAPI_SERVICE_PORT = 8080

RESTAPI_BASE_URL = '/restconf/data'

# REST-API users

RESTAPI_USERS = {   # TODO: implement a database of credentials and permissions

    'admin': generate_password_hash('admin'),

}
 


# Prometheus settings

METRICS_PORT = 9192

src/compute/client/ComputeClient.py

+22 −17
Original line number Diff line number Diff line
@@ -13,7 +13,10 @@ 
# 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 compute.proto.compute_pb2_grpc import ComputeServiceStub

from compute.proto.context_pb2 import (

    AuthenticationResult, Empty, Service, ServiceId, ServiceIdList, ServiceStatus, TeraFlowController)
@@ -24,8 +27,10 @@ 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 ComputeClient:

    def __init__(self, address, port):

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

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

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

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

        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
@@ -37,55 +42,55 @@ class ComputeClient: 
        self.stub = ComputeServiceStub(self.channel)



    def close(self):

        if(self.channel is not None): self.channel.close()

        if self.channel is not None: self.channel.close()

        self.channel = None

        self.stub = None



    @RETRY_DECORATOR

    def CheckCredentials(self, request : TeraFlowController) -> AuthenticationResult:

        LOGGER.debug('CheckCredentials request: {:s}'.format(str(request)))

        LOGGER.debug('CheckCredentials request: {:s}'.format(grpc_message_to_json_string(request)))

        response = self.stub.CheckCredentials(request)

        LOGGER.debug('CheckCredentials result: {:s}'.format(str(response)))

        LOGGER.debug('CheckCredentials result: {:s}'.format(grpc_message_to_json_string(response)))

        return response



    @RETRY_DECORATOR

    def GetConnectivityServiceStatus(self, request : ServiceId) -> ServiceStatus:

        LOGGER.debug('GetConnectivityServiceStatus request: {:s}'.format(str(request)))

        LOGGER.debug('GetConnectivityServiceStatus request: {:s}'.format(grpc_message_to_json_string(request)))

        response = self.stub.GetConnectivityServiceStatus(request)

        LOGGER.debug('GetConnectivityServiceStatus result: {:s}'.format(str(response)))

        LOGGER.debug('GetConnectivityServiceStatus result: {:s}'.format(grpc_message_to_json_string(response)))

        return response



    @RETRY_DECORATOR

    def CreateConnectivityService(self, request : Service) -> ServiceId:

        LOGGER.debug('CreateConnectivityService request: {:s}'.format(str(request)))

        LOGGER.debug('CreateConnectivityService request: {:s}'.format(grpc_message_to_json_string(request)))

        response = self.stub.CreateConnectivityService(request)

        LOGGER.debug('CreateConnectivityService result: {:s}'.format(str(response)))

        LOGGER.debug('CreateConnectivityService result: {:s}'.format(grpc_message_to_json_string(response)))

        return response



    @RETRY_DECORATOR

    def EditConnectivityService(self, request : Service) -> ServiceId:

        LOGGER.debug('EditConnectivityService request: {:s}'.format(str(request)))

        LOGGER.debug('EditConnectivityService request: {:s}'.format(grpc_message_to_json_string(request)))

        response = self.stub.EditConnectivityService(request)

        LOGGER.debug('EditConnectivityService result: {:s}'.format(str(response)))

        LOGGER.debug('EditConnectivityService result: {:s}'.format(grpc_message_to_json_string(response)))

        return response



    @RETRY_DECORATOR

    def DeleteConnectivityService(self, request : Service) -> Empty:

        LOGGER.debug('DeleteConnectivityService request: {:s}'.format(str(request)))

        LOGGER.debug('DeleteConnectivityService request: {:s}'.format(grpc_message_to_json_string(request)))

        response = self.stub.DeleteConnectivityService(request)

        LOGGER.debug('DeleteConnectivityService result: {:s}'.format(str(response)))

        LOGGER.debug('DeleteConnectivityService result: {:s}'.format(grpc_message_to_json_string(response)))

        return response



    @RETRY_DECORATOR

    def GetAllActiveConnectivityServices(self, request : Empty) -> ServiceIdList:

        LOGGER.debug('GetAllActiveConnectivityServices request: {:s}'.format(str(request)))

        LOGGER.debug('GetAllActiveConnectivityServices request: {:s}'.format(grpc_message_to_json_string(request)))

        response = self.stub.GetAllActiveConnectivityServices(request)

        LOGGER.debug('GetAllActiveConnectivityServices result: {:s}'.format(str(response)))

        LOGGER.debug('GetAllActiveConnectivityServices result: {:s}'.format(grpc_message_to_json_string(response)))

        return response



    @RETRY_DECORATOR

    def ClearAllConnectivityServices(self, request : Empty) -> Empty:

        LOGGER.debug('ClearAllConnectivityServices request: {:s}'.format(str(request)))

        LOGGER.debug('ClearAllConnectivityServices request: {:s}'.format(grpc_message_to_json_string(request)))

        response = self.stub.ClearAllConnectivityServices(request)

        LOGGER.debug('ClearAllConnectivityServices result: {:s}'.format(str(response)))

        LOGGER.debug('ClearAllConnectivityServices result: {:s}'.format(grpc_message_to_json_string(response)))

        return response

src/compute/service/ComputeService.py

+9 −48
Original line number Diff line number Diff line
@@ -12,56 +12,17 @@ 
# 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.tools.service.GenericGrpcService import GenericGrpcService

from compute.proto.compute_pb2_grpc import add_ComputeServiceServicer_to_server

from compute.service.ComputeServiceServicerImpl import ComputeServiceServicerImpl

from compute.Config import GRPC_SERVICE_PORT, GRPC_MAX_WORKERS, GRPC_GRACE_PERIOD



BIND_ADDRESS = '0.0.0.0'

LOGGER = logging.getLogger(__name__)



class ComputeService:

    def __init__(self, address=BIND_ADDRESS, port=GRPC_SERVICE_PORT, max_workers=GRPC_MAX_WORKERS,

                 grace_period=GRPC_GRACE_PERIOD):

        self.address = address

        self.port = port

        self.endpoint = None

        self.max_workers = max_workers

        self.grace_period = grace_period

        self.compute_servicer = None

        self.health_servicer = None

        self.pool = None

        self.server = None



    def start(self):

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

        LOGGER.debug('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,))



class ComputeService(GenericGrpcService):

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

        port = get_service_port_grpc(ServiceNameEnum.COMPUTE)

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

        self.compute_servicer = ComputeServiceServicerImpl()

        add_ComputeServiceServicer_to_server(self.compute_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')
 
    def install_servicers(self):

        add_ComputeServiceServicer_to_server(self.compute_servicer, self.server)

src/compute/service/__main__.py

+10 −16
Original line number Diff line number Diff line
@@ -14,10 +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 compute.Config import (

    GRPC_SERVICE_PORT, GRPC_MAX_WORKERS, GRPC_GRACE_PERIOD, LOG_LEVEL, RESTAPI_SERVICE_PORT, RESTAPI_BASE_URL,

    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 .ComputeService import ComputeService

from .rest_server.RestServer import RestServer

from .rest_server.nbi_plugins.ietf_l2vpn import register_ietf_l2vpn
@@ -32,20 +32,13 @@ def signal_handler(signal, frame): # pylint: disable=redefined-outer-name 
def main():

    global LOGGER # pylint: disable=global-statement



    grpc_service_port    = get_setting('COMPUTESERVICE_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           )

    restapi_service_port = get_setting('RESTAPI_SERVICE_PORT',             default=RESTAPI_SERVICE_PORT)

    restapi_base_url     = get_setting('RESTAPI_BASE_URL',                 default=RESTAPI_BASE_URL    )

    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',

        'SERVICESERVICE_SERVICE_HOST', 'SERVICESERVICE_SERVICE_PORT_GRPC'

        get_env_var_name(ServiceNameEnum.CONTEXT, ENVVAR_SUFIX_SERVICE_HOST     ),

        get_env_var_name(ServiceNameEnum.CONTEXT, ENVVAR_SUFIX_SERVICE_PORT_GRPC),

    ])



    signal.signal(signal.SIGINT,  signal_handler)
@@ -54,13 +47,14 @@ def main(): 
    LOGGER.info('Starting...')



    # Start metrics server

    metrics_port = get_metrics_port()

    start_http_server(metrics_port)



    # Starting compute service

    grpc_service = ComputeService(port=grpc_service_port, max_workers=max_workers, grace_period=grace_period)

    grpc_service = ComputeService()

    grpc_service.start()



    rest_server = RestServer(port=restapi_service_port, base_url=restapi_base_url)

    rest_server = RestServer()

    register_ietf_l2vpn(rest_server)

    rest_server.start()

src/compute/service/rest_server/RestServer.py

+9 −42
Original line number Diff line number Diff line
@@ -12,45 +12,12 @@ 
# See the License for the specific language governing permissions and

# limitations under the License.



import logging, threading, time

from flask import Flask, request

from flask_restful import Api, Resource

from werkzeug.serving import make_server

from compute.Config import RESTAPI_BASE_URL, RESTAPI_SERVICE_PORT



logging.getLogger('werkzeug').setLevel(logging.WARNING)



BIND_ADDRESS = '0.0.0.0'

LOGGER = logging.getLogger(__name__)



def log_request(response):

    timestamp = time.strftime('[%Y-%b-%d %H:%M]')

    LOGGER.info('%s %s %s %s %s', timestamp, request.remote_addr, request.method, request.full_path, response.status)

    return response



class RestServer(threading.Thread):

    def __init__(self, host=BIND_ADDRESS, port=RESTAPI_SERVICE_PORT, base_url=RESTAPI_BASE_URL):

        threading.Thread.__init__(self, daemon=True)

        self.host = host

        self.port = port

        self.base_url = base_url

        self.srv = None

        self.ctx = None

        self.app = Flask(__name__)

        self.app.after_request(log_request)

        self.api = Api(self.app, prefix=self.base_url)



    def add_resource(self, resource : Resource, *urls, **kwargs):

        self.api.add_resource(resource, *urls, **kwargs)



    def run(self):

        self.srv = make_server(self.host, self.port, self.app, threaded=True)

        self.ctx = self.app.app_context()

        self.ctx.push()



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

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

        self.srv.serve_forever()



    def shutdown(self):

        self.srv.shutdown()
 
from common.Constants import ServiceNameEnum

from common.Settings import get_service_baseurl_http, get_service_port_http

from common.tools.service.GenericRestServer import GenericRestServer



class RestServer(GenericRestServer):

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

        bind_port = get_service_port_http(ServiceNameEnum.COMPUTE)

        base_url = get_service_baseurl_http(ServiceNameEnum.COMPUTE)

        super().__init__(bind_port, base_url, cls_name=cls_name)