Commit 1fab7278 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Simap Connector: 

- Store subscriptions in persistent CRDB database
- gRPC Server methods to establish/delete subscriptions
parent 7757182e

src/simap_connector/client/SimapConnectorClient.py

+19 −5
Original line number Diff line number Diff line
@@ -14,12 +14,12 @@ 


import grpc, logging

from common.Constants import ServiceNameEnum

from common.proto.context_pb2 import Empty

#from common.proto.e2eorchestrator_pb2_grpc import E2EOrchestratorServiceStub

from common.Settings import get_service_host, get_service_port_grpc

from common.proto.context_pb2 import Empty

from common.proto.simap_connector_pb2 import Subscription, SubscriptionId

from common.proto.simap_connector_pb2_grpc import SimapConnectorServiceStub

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

#from common.tools.grpc.Tools import grpc_message_to_json

#from common.proto.e2eorchestrator_pb2 import E2EOrchestratorRequest, E2EOrchestratorReply

from common.tools.grpc.Tools import grpc_message_to_json_string



LOGGER = logging.getLogger(__name__)

MAX_RETRIES = 15
@@ -42,9 +42,23 @@ class SimapConnectorClient: 


    def connect(self):

        self.channel = grpc.insecure_channel(self.endpoint)

        #self.stub = E2EOrchestratorServiceStub(self.channel)

        self.stub = SimapConnectorServiceStub(self.channel)



    def close(self):

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

        self.channel = None

        self.stub = None



    @RETRY_DECORATOR

    def EstablishSubscription(self, request : Subscription) -> SubscriptionId:

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

        response = self.stub.EstablishSubscription(request)

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

        return response



    @RETRY_DECORATOR

    def DeleteSubscription(self, request : SubscriptionId) -> Empty:

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

        response = self.stub.DeleteSubscription(request)

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

        return response

src/simap_connector/service/SimapConnectorService.py

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

# limitations under the License.





import logging, sqlalchemy

from common.Constants import ServiceNameEnum

from common.Settings import get_service_port_grpc

from common.proto.simap_connector_pb2 import DESCRIPTOR as SIMAP_CONNECTOR_DESCRIPTOR

from common.proto.simap_connector_pb2_grpc import add_SimapConnectorServiceServicer_to_server

from common.tools.service.GenericGrpcService import GenericGrpcService

from .SimapConnectorServiceServicerImpl import SimapConnectorServiceServicerImpl





LOGGER = logging.getLogger(__name__)





class SimapConnectorService(GenericGrpcService):

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

    def __init__(

        self, db_engine : sqlalchemy.engine.Engine, cls_name : str = __name__

    ) -> None:

        port = get_service_port_grpc(ServiceNameEnum.SIMAP_CONNECTOR)

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

        self.simap_connector_servicer = SimapConnectorServiceServicerImpl(db_engine)



    def install_servicers(self):

        pass
 
        add_SimapConnectorServiceServicer_to_server(self.simap_connector_servicer, self.server)

        self.add_reflection_service_name(SIMAP_CONNECTOR_DESCRIPTOR, 'SimapConnectorService')

src/simap_connector/service/SimapConnectorServiceServicerImpl.py

0 → 100644
+43 −0
Original line number Diff line number Diff line 
# Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)

#

# 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, sqlalchemy

from common.proto.context_pb2 import Empty

from common.proto.simap_connector_pb2 import Subscription, SubscriptionId

from common.proto.simap_connector_pb2_grpc import SimapConnectorServiceServicer

from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method

from .database.Subscription import subscription_set, subscription_delete





LOGGER = logging.getLogger(__name__)



METRICS_POOL = MetricsPool('SimapConnector', 'RPC')





class SimapConnectorServiceServicerImpl(SimapConnectorServiceServicer):

    def __init__(self, db_engine : sqlalchemy.engine.Engine) -> None:

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

        self.db_engine = db_engine

        LOGGER.debug('Servicer Created')



    def _get_metrics(self) -> MetricsPool: return METRICS_POOL



    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)

    def EstablishSubscription(self, request : Subscription, context : grpc.ServicerContext) -> SubscriptionId:

        return subscription_set(self.db_engine, request)



    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)

    def DeleteSubscription(self, request : SubscriptionId, context : grpc.ServicerContext) -> Empty:

        return subscription_delete(self.db_engine, request)

src/simap_connector/service/__main__.py

+17 −1
Original line number Diff line number Diff line
@@ -24,6 +24,8 @@ from simap_connector.Config import ( 
    SIMAP_SERVER_SCHEME, SIMAP_SERVER_ADDRESS, SIMAP_SERVER_PORT,

    SIMAP_SERVER_USERNAME, SIMAP_SERVER_PASSWORD,

)

from .database.Engine import Engine

from .database.models._Base import rebuild_database

from .simap_updater.SimapClient import SimapClient

from .simap_updater.SimapUpdater import SimapUpdater

from .telemetry.TelemetryPool import TelemetryPool
@@ -59,8 +61,22 @@ def main(): 
    metrics_port = get_metrics_port()

    start_http_server(metrics_port)



    # Get Database Engine instance and initialize database, if needed

    LOGGER.info('Getting SQLAlchemy DB Engine...')

    db_engine = Engine.get_engine()

    if db_engine is None:

        LOGGER.error('Unable to get SQLAlchemy DB Engine...')

        return -1



    try:

        Engine.create_database(db_engine)

    except: # pylint: disable=bare-except # pragma: no cover

        LOGGER.exception('Failed to check/create the database: {:s}'.format(str(db_engine.url)))



    rebuild_database(db_engine)



    # Starting service

    grpc_service = SimapConnectorService()

    grpc_service = SimapConnectorService(db_engine)

    grpc_service.start()

src/simap_connector/service/database/Engine.py

0 → 100644
+55 −0
Original line number Diff line number Diff line 
# Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)

#

# 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 logging, sqlalchemy, sqlalchemy_utils

from common.Settings import get_setting



LOGGER = logging.getLogger(__name__)



APP_NAME = 'tfs'

ECHO = False # true: dump SQL commands and transactions executed

CRDB_URI_TEMPLATE = 'cockroachdb://{:s}:{:s}@cockroachdb-public.{:s}.svc.cluster.local:{:s}/{:s}?sslmode={:s}'



class Engine:

    @staticmethod

    def get_engine() -> sqlalchemy.engine.Engine:

        crdb_uri = get_setting('CRDB_URI', default=None)

        if crdb_uri is None:

            CRDB_NAMESPACE = get_setting('CRDB_NAMESPACE')

            CRDB_SQL_PORT  = get_setting('CRDB_SQL_PORT')

            CRDB_DATABASE  = get_setting('CRDB_DATABASE')

            CRDB_USERNAME  = get_setting('CRDB_USERNAME')

            CRDB_PASSWORD  = get_setting('CRDB_PASSWORD')

            CRDB_SSLMODE   = get_setting('CRDB_SSLMODE')

            crdb_uri = CRDB_URI_TEMPLATE.format(

                CRDB_USERNAME, CRDB_PASSWORD, CRDB_NAMESPACE, CRDB_SQL_PORT, CRDB_DATABASE, CRDB_SSLMODE)



        try:

            engine = sqlalchemy.create_engine(

                crdb_uri, connect_args={'application_name': APP_NAME}, echo=ECHO, future=True)

        except: # pylint: disable=bare-except # pragma: no cover

            LOGGER.exception('Failed to connect to database: {:s}'.format(str(crdb_uri)))

            return None



        return engine



    @staticmethod

    def create_database(engine : sqlalchemy.engine.Engine) -> None:

        if not sqlalchemy_utils.database_exists(engine.url):

            sqlalchemy_utils.create_database(engine.url)



    @staticmethod

    def drop_database(engine : sqlalchemy.engine.Engine) -> None:

        if sqlalchemy_utils.database_exists(engine.url):

            sqlalchemy_utils.drop_database(engine.url)