Commit 5dc77fe4 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Common framework: 

- defined unified constants for all micro-services
- defined common settings getter methods
- defined generic gRPC servicer
- defined generic Rest servicer
- defined common MockServicer for Monitoring component
- removed old common/tests MockService class; to be replaced by new generic gRPC servicer
- minor code formatting/styling
parent 0eaab40b

src/common/Constants.py

+59 −0
Original line number Diff line number Diff line
@@ -12,5 +12,64 @@ 
# See the License for the specific language governing permissions and

# limitations under the License.



import logging

from enum import Enum



# Default logging level

DEFAULT_LOG_LEVEL = logging.WARNING



# Default gRPC server settings

DEFAULT_GRPC_BIND_ADDRESS = '0.0.0.0'

DEFAULT_GRPC_MAX_WORKERS  = 10

DEFAULT_GRPC_GRACE_PERIOD = 60



# Default HTTP server settings

DEFAULT_HTTP_BIND_ADDRESS = '0.0.0.0'



# Default Prometheus settings

DEFAULT_METRICS_PORT = 9192



# Default context and topology UUIDs

DEFAULT_CONTEXT_UUID = 'admin'

DEFAULT_TOPOLOGY_UUID = 'admin'



# Default service names

class ServiceNameEnum(Enum):

    CONTEXT       = 'context'

    DEVICE        = 'device'

    SERVICE       = 'service'

    SLICE         = 'slice'

    AUTOMATION    = 'automation'

    POLICY        = 'policy'

    MONITORING    = 'monitoring'

    DLT           = 'dlt'

    COMPUTE       = 'compute'

    CYBERSECURITY = 'cybersecurity'

    INTERDOMAIN   = 'interdomain'



# Default gRPC service ports

DEFAULT_SERVICE_GRPC_PORTS = {

    ServiceNameEnum.CONTEXT      .value :  1010,

    ServiceNameEnum.DEVICE       .value :  2020,

    ServiceNameEnum.SERVICE      .value :  3030,

    ServiceNameEnum.SLICE        .value :  4040,

    ServiceNameEnum.AUTOMATION   .value :  5050,

    ServiceNameEnum.POLICY       .value :  6060,

    ServiceNameEnum.MONITORING   .value :  7070,

    ServiceNameEnum.DLT          .value :  8080,

    ServiceNameEnum.COMPUTE      .value :  9090,

    ServiceNameEnum.CYBERSECURITY.value : 10000,

    ServiceNameEnum.INTERDOMAIN  .value : 10010,

}



# Default HTTP/REST-API service ports

DEFAULT_SERVICE_HTTP_PORTS = {

    ServiceNameEnum.CONTEXT   .value : 8080,

    ServiceNameEnum.COMPUTE   .value : 8080,

}



# Default HTTP/REST-API service base URLs

DEFAULT_SERVICE_HTTP_BASEURLS = {

    ServiceNameEnum.CONTEXT   .value : '/api',

    ServiceNameEnum.COMPUTE   .value : '/restconf/data',

}

src/common/Settings.py

+61 −2
Original line number Diff line number Diff line
@@ -14,15 +14,33 @@ 


import logging, os, time

from typing import List

from common.Constants import (

    DEFAULT_GRPC_BIND_ADDRESS, DEFAULT_GRPC_GRACE_PERIOD, DEFAULT_GRPC_MAX_WORKERS, DEFAULT_HTTP_BIND_ADDRESS,

    DEFAULT_LOG_LEVEL, DEFAULT_METRICS_PORT, DEFAULT_SERVICE_GRPC_PORTS, DEFAULT_SERVICE_HTTP_BASEURLS,

    DEFAULT_SERVICE_HTTP_PORTS, ServiceNameEnum

)



LOGGER = logging.getLogger(__name__)



DEFAULT_RESTART_DELAY = 5.0 # seconds



ENVVAR_KUBERNETES_PORT            = 'KUBERNETES_PORT'

ENVVAR_GRPC_BIND_ADDRESS          = 'GRPC_BIND_ADDRESS'

ENVVAR_GRPC_MAX_WORKERS           = 'GRPC_MAX_WORKERS'

ENVVAR_GRPC_GRACE_PERIOD          = 'GRPC_GRACE_PERIOD'

ENVVAR_HTTP_BIND_ADDRESS          = 'HTTP_BIND_ADDRESS'

ENVVAR_LOG_LEVEL                  = 'LOG_LEVEL'

ENVVAR_METRICS_PORT               = 'METRICS_PORT'



ENVVAR_SUFIX_SERVICE_BASEURL_HTTP = 'SERVICE_BASEURL_HTTP'

ENVVAR_SUFIX_SERVICE_HOST         = 'SERVICE_HOST'

ENVVAR_SUFIX_SERVICE_PORT_GRPC    = 'SERVICE_PORT_GRPC'

ENVVAR_SUFIX_SERVICE_PORT_HTTP    = 'SERVICE_PORT_HTTP'



def wait_for_environment_variables(

    required_environment_variables : List[str] = [], wait_delay_seconds : float = DEFAULT_RESTART_DELAY

):

    if 'KUBERNETES_PORT' not in os.environ: return # We're not running in Kubernetes, nothing to wait for

    if ENVVAR_KUBERNETES_PORT not in os.environ: return # We're not running in Kubernetes, nothing to wait for

    missing_variables = set(required_environment_variables).difference(set(os.environ.keys()))

    if len(missing_variables) == 0: return # We have all environment variables defined

    msg = 'Variables({:s}) are missing in Environment({:s}), restarting in {:f} seconds...'
@@ -36,4 +54,45 @@ def get_setting(name, **kwargs): 
        value = kwargs['settings'].pop(name, value)

    if value is not None: return value

    if 'default' in kwargs: return kwargs['default']

    raise Exception('Setting({}) not specified in environment or configuration'.format(name))
 
    raise Exception('Setting({:s}) not specified in environment or configuration'.format(str(name)))



def get_env_var_name(service_name : ServiceNameEnum, env_var_group):

    return ('{:s}SERVICE_{:s}'.format(service_name.value, env_var_group)).upper()



def get_service_host(service_name : ServiceNameEnum):

    envvar_name = get_env_var_name(service_name, ENVVAR_SUFIX_SERVICE_HOST)

    default_value = ('{:s}service'.format(service_name.value))

    return get_setting(envvar_name, default=default_value)



def get_service_port_grpc(service_name : ServiceNameEnum):

    envvar_name = get_env_var_name(service_name, ENVVAR_SUFIX_SERVICE_PORT_GRPC)

    default_value = DEFAULT_SERVICE_GRPC_PORTS.get(service_name.value)

    return get_setting(envvar_name, default=default_value)



def get_service_port_http(service_name : ServiceNameEnum):

    envvar_name = get_env_var_name(service_name, ENVVAR_SUFIX_SERVICE_PORT_HTTP)

    default_value = DEFAULT_SERVICE_HTTP_PORTS.get(service_name.value)

    return get_setting(envvar_name, default=default_value)



def get_service_baseurl_http(service_name : ServiceNameEnum):

    envvar_name = get_env_var_name(service_name, ENVVAR_SUFIX_SERVICE_BASEURL_HTTP)

    default_value = DEFAULT_SERVICE_HTTP_BASEURLS.get(service_name.value)

    return get_setting(envvar_name, default=default_value)



def get_log_level():

    return get_setting(ENVVAR_LOG_LEVEL, default=DEFAULT_LOG_LEVEL)



def get_metrics_port():

    return get_setting(ENVVAR_METRICS_PORT, default=DEFAULT_METRICS_PORT)



def get_grpc_bind_address():

    return get_setting(ENVVAR_GRPC_BIND_ADDRESS, default=DEFAULT_GRPC_BIND_ADDRESS)



def get_grpc_max_workers():

    return get_setting(ENVVAR_GRPC_MAX_WORKERS, default=DEFAULT_GRPC_MAX_WORKERS)



def get_grpc_grace_period():

    return get_setting(ENVVAR_GRPC_GRACE_PERIOD, default=DEFAULT_GRPC_GRACE_PERIOD)



def get_http_bind_address():

    return get_setting(ENVVAR_HTTP_BIND_ADDRESS, default=DEFAULT_HTTP_BIND_ADDRESS)

src/common/tests/MockServicerImpl_Monitoring.py

0 → 100644
+34 −0
Original line number Diff line number Diff line 
# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)

#

# 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 grpc, logging

from queue import Queue

from common.tools.grpc.Tools import grpc_message_to_json_string

from monitoring.proto.context_pb2 import Empty

from monitoring.proto.monitoring_pb2 import Kpi

from monitoring.proto.monitoring_pb2_grpc import MonitoringServiceServicer



LOGGER = logging.getLogger(__name__)



class MockServicerImpl_Monitoring(MonitoringServiceServicer):

    def __init__(self, queue_samples : Queue):

        LOGGER.info('[__init__] Creating Servicer...')

        self.queue_samples = queue_samples

        LOGGER.info('[__init__] Servicer Created')



    def IncludeKpi(self, request : Kpi, context : grpc.ServicerContext) -> Empty:

        LOGGER.info('[IncludeKpi] request={:s}'.format(grpc_message_to_json_string(request)))

        self.queue_samples.put(request)

        return Empty()

src/common/tools/object_factory/Device.py

+1 −1
Original line number Diff line number Diff line
@@ -20,8 +20,8 @@ from context.proto.context_pb2 import DeviceDriverEnum, DeviceOperationalStatusE 


DEVICE_DISABLED = DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_DISABLED



DEVICE_EMUPR_TYPE   = DeviceTypeEnum.EMULATED_PACKET_ROUTER.value

DEVICE_EMUOLS_TYPE  = DeviceTypeEnum.EMULATED_OPTICAL_LINE_SYSTEM.value

DEVICE_EMUPR_TYPE   = DeviceTypeEnum.EMULATED_PACKET_ROUTER.value

DEVICE_EMU_DRIVERS  = [DeviceDriverEnum.DEVICEDRIVER_UNDEFINED]

DEVICE_EMU_ADDRESS  = '127.0.0.1'

DEVICE_EMU_PORT     = '0'

src/common/tests/MockService.py→src/common/tools/service/GenericGrpcService.py

+28 −12
Original line number Diff line number Diff line
@@ -12,20 +12,27 @@ 
# See the License for the specific language governing permissions and

# limitations under the License.



from typing import Optional, Union

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.Settings import get_grpc_bind_address, get_grpc_grace_period, get_grpc_max_workers



GRPC_MAX_WORKERS  = 10

GRPC_GRACE_PERIOD = 60



class MockService:

    def __init__(self, address, port, max_workers=GRPC_MAX_WORKERS, grace_period=GRPC_GRACE_PERIOD, cls_name=__name__):

class GenericGrpcService:

    def __init__(

        self, bind_port : Union[str, int], bind_address : Optional[str] = None, max_workers : Optional[int] = None,

        grace_period : Optional[int] = None, enable_health_servicer : bool = True, cls_name : str = __name__

    ) -> None:

        self.logger = logging.getLogger(cls_name)

        self.address = address

        self.port = port

        self.bind_port = bind_port

        self.bind_address = get_grpc_bind_address() if bind_address is None else bind_address

        self.max_workers = get_grpc_max_workers() if max_workers is None else max_workers

        self.grace_period = get_grpc_grace_period() if grace_period is None else grace_period

        self.enable_health_servicer = enable_health_servicer

        self.endpoint = None

        self.max_workers = max_workers

        self.grace_period = grace_period

        self.health_servicer = None

        self.pool = None

        self.server = None
@@ -33,7 +40,7 @@ class MockService: 
        pass



    def start(self):

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

        self.endpoint = '{:s}:{:s}'.format(str(self.bind_address), str(self.bind_port))

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

            str(self.endpoint), str(self.max_workers)))
@@ -42,14 +49,23 @@ class MockService: 


        self.install_servicers()



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

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

        if self.enable_health_servicer:

            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)



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

        self.endpoint = '{:s}:{:s}'.format(str(self.bind_address), str(self.bind_port))

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

        self.server.start()

        if self.enable_health_servicer:

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



        self.logger.debug('Service started')



    def stop(self):

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

        if self.enable_health_servicer:

            self.health_servicer.enter_graceful_shutdown()

        self.server.stop(self.grace_period)

        self.logger.debug('Service stopped')