Commit e3fc9f2c authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

NBI component: 

- Stabilized support for SocketIO-based websockets next to REST-API endpoints on same server and port
- Added REST-based probe and SocketIO-based heartbeat
- Migrated all NBI plugins to new framework
- Updated existing unitary tests
parent 4c4d91e1

src/nbi/mytest/pytest_code.py

deleted100644 → 0
+0 −105
Original line number Diff line number Diff line 
# Copyright 2022-2024 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 eventlet, eventlet.wsgi, json, logging, os, pytest, requests, threading, time

import websockets.sync.client  # Import synchronous WebSocket client

from nbi.service.NbiApplication import NbiApplication

from nbi.service.rest_server.nbi_plugins import register_restconf

from nbi.service.restapi_resources.health_probes import register_health_probes

from nbi.service.websocket_namespaces.hearthbeat import register_heartbeat







logging.basicConfig(level=logging.DEBUG)

LOGGER = logging.getLogger(__name__)

LOGGER.setLevel(logging.DEBUG)



LOCAL_HOST             = '127.0.0.1'

NBI_SERVICE_PORT       = 18080

NBI_SERVICE_PREFIX_URL = ''

NBI_SERVICE_BASE_URL   = '{:s}:{:d}{:s}'.format(LOCAL_HOST, NBI_SERVICE_PORT, NBI_SERVICE_PREFIX_URL)



class ServerThread(threading.Thread):

    def __init__(self):

        super().__init__(daemon=True)



        self.nbi_app = NbiApplication(base_url=NBI_SERVICE_PREFIX_URL)

        register_health_probes(self.nbi_app)

        register_heartbeat    (self.nbi_app)

        register_restconf     (self.nbi_app)

        self.nbi_app.dump_configuration()



    def run(self):

        try:

            #eventlet.wsgi.server(

            #    eventlet.listen((LOCAL_HOST, NBI_SERVICE_PORT)),

            #    self.nbi_app.get_flask_app(),

            #    debug=True, log_output=True

            #)

            #thread = eventlet.spawn(

            #    self.nbi_app._sio.run, self.nbi_app.get_flask_app(),

            #    host=LOCAL_HOST, port=NBI_SERVICE_PORT,

            #    debug=True, use_reloader=False

            #)

            #thread.wait()

            self.nbi_app._sio.run(

                self.nbi_app.get_flask_app(),

                host=LOCAL_HOST, port=NBI_SERVICE_PORT,

                debug=True, use_reloader=False

            )

        except:

            LOGGER.exception('unhandled')



@pytest.fixture(scope='session')

def nbi_application() -> NbiApplication:

    thread = ServerThread()

    thread.start()

    time.sleep(1)

    yield thread.nbi_app

    thread.join(timeout=1)



def test_restapi_get_healthz(

    nbi_application : NbiApplication    # pylint: disable=redefined-outer-name, unused-argument

) -> None:

    request_url = 'http://' + NBI_SERVICE_BASE_URL + '/healthz'

    LOGGER.warning('Request: GET {:s}'.format(str(request_url)))

    reply = requests.request('get', request_url, timeout=10, allow_redirects=True)

    LOGGER.warning('Reply: {:s}'.format(str(reply.text)))

    assert reply.status_code == requests.codes['OK'], 'Reply failed with status code {:d}'.format(reply.status_code)

    if reply.content and len(reply.content) > 0: return reply.json()



def test_websocket_get_heartbeat(

    nbi_application : NbiApplication    # pylint: disable=redefined-outer-name, unused-argument

) -> None:

    nbi_application.dump_configuration()

    request_url = 'ws://' + NBI_SERVICE_BASE_URL + '/heartbeat'

    LOGGER.warning('Request: WS {:s}'.format(str(request_url)))



    heartbeat_count = 0

    with websockets.sync.client.connect(request_url) as ws:

        while heartbeat_count < 5:

            message = ws.recv()

            LOGGER.warning('Received message: {:s}'.format(str(message)))



            data = json.loads(message)



            # Validate uptime response

            assert "uptime_seconds" in data, "Missing 'uptime_seconds' in response"

            assert isinstance(data["uptime_seconds"], (int, float)), "'uptime_seconds' is not a number"



            heartbeat_count += 1

            LOGGER.warning('--> Heartbeat #{:d}: {:s}'.format(heartbeat_count, str(data)))



    LOGGER.warning('Test completed')

    raise Exception()

src/nbi/service/rest_server/nbi_plugins/tools/__init__.py→src/nbi/run_gunicorn.sh

100644 → 100755
+3 −0
Original line number Diff line number Diff line 
#!/bin/bash

# Copyright 2022-2024 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)

#

# Licensed under the Apache License, Version 2.0 (the "License");
@@ -12,3 +13,5 @@ 
# See the License for the specific language governing permissions and

# limitations under the License.



export FLASK_ENV=development

gunicorn -w 4 --worker-class eventlet -b 0.0.0.0:18080 --log-level DEBUG nbi.service.app:app

run_test_nbi.sh→src/nbi/run_test.sh

+2 −3
Original line number Diff line number Diff line
@@ -13,8 +13,7 @@ 
# See the License for the specific language governing permissions and

# limitations under the License.



export PYTHON_PATH=./src

export LOG_LEVEL=DEBUG

export FLASK_ENV=development



python -m pytest --log-level=DEBUG -o log_cli=true --verbose src/nbi/mytest/pytest_code.py
 
# Add live logs with: -o log_cli=true

python -m pytest --log-level=DEBUG --verbose nbi/tests/test_nbi.py

src/nbi/service/NbiApplication.py

+24 −59
Original line number Diff line number Diff line
@@ -14,7 +14,7 @@ 




import logging, time

from typing import Any, Optional

from typing import Any, List, Optional, Tuple

from flask import Flask, request

from flask_restful import Api, Resource

from flask_socketio import Namespace, SocketIO
@@ -39,28 +39,25 @@ class NbiApplication: 
        self._app.config['SECRET_KEY'] = 'secret!'

        self._app.after_request(log_request)

        self._api = Api(self._app, prefix=base_url)

        #websocket_path = '/'.join([base_url.rstrip('/'), 'websocket'])

        #self._sio = SocketIO(self._app, path=base_url, cors_allowed_origins="*", logger=True, engineio_logger=True)

        self._sio = SocketIO(self._app, cors_allowed_origins="*", logger=True, engineio_logger=True)



        @self._sio.on_error_default  # handles all namespaces without an explicit error handler

        def default_error_handler(e):

            LOGGER.error('[default_error_handler] e={:s}'.format(str(e)))

        #socketio_path = '/'.join([base_url.rstrip('/'), 'socket.io'])

        self._sio = SocketIO(

            self._app, cors_allowed_origins='*', async_mode='eventlet',

            #path=socketio_path,

            logger=True, engineio_logger=True

        )



    def add_rest_api_resource(self, resource_class : Resource, *urls, **kwargs) -> None:

        self._api.add_resource(resource_class, *urls, **kwargs)



    def add_websocket_namespace(self, namespace_class : Namespace, namespace_url : str) -> None:

        LOGGER.warning('[add_websocket_namespace] (before) self._sio.server={:s}'.format(str(self._sio.server)))

        LOGGER.warning('[add_websocket_namespace] (before) self._sio.server.namespace_handlers={:s}'.format(str(self._sio.server.namespace_handlers)))

        LOGGER.warning('[add_websocket_namespace] (before) self._sio.namespace_handlers={:s}'.format(str(self._sio.namespace_handlers)))

        self._sio.on_namespace(namespace_class(namespace_url))

        LOGGER.warning('[add_websocket_namespace] (after) self._sio.server={:s}'.format(str(self._sio.server)))

        LOGGER.warning('[add_websocket_namespace] (after) self._sio.server.namespace_handlers={:s}'.format(str(self._sio.server.namespace_handlers)))

        LOGGER.warning('[add_websocket_namespace] (after) self._sio.namespace_handlers={:s}'.format(str(self._sio.namespace_handlers)))

    def add_rest_api_resources(self, resources : List[Tuple[Resource, str, str]]) -> None:

        for endpoint_name, resource_class, resource_url in resources:

            self.add_rest_api_resource(resource_class, resource_url, endpoint=endpoint_name)



    def add_websocket_namespace(self, namespace : Namespace) -> None:

        self._sio.on_namespace(namespace)



    def websocket_emit_message(

        self, event : str, *args : Any, namespace : str = "/", to : Optional[str] = None

        self, event : str, *args : Any, namespace : str = '/', to : Optional[str] = None

    ) -> None:

        self._sio.emit(event, *args, namespace=namespace, to=to)
@@ -76,46 +73,14 @@ class NbiApplication: 
        for rule in self._app.url_map.iter_rules():

            LOGGER.debug(' - {:s}'.format(str(rule)))



        LOGGER.debug('Configured WebSocket Namespaces:')

        for namespace in self._sio.server.handlers.keys():

            LOGGER.debug(' (server) - {:s}'.format(str(namespace)))



        # TODO: find a way to report configured namespaces, for some reason,

        # those data structures become emptied when SocketIO server starts.

        LOGGER.debug('Configured SocketIO/WebSocket Namespaces:')

        LOGGER.debug('  WARNING: Might report an empty list of namespaces even when')

        LOGGER.debug('           they are properly configured. To be fixed.')

        for handler in self._sio.handlers:

            LOGGER.debug(' - {:s}'.format(str(handler)))

        for namespace in self._sio.namespace_handlers:

            LOGGER.debug(' (ns_hdls) - {:s}'.format(str(namespace)))



    def run_standalone(

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

        debug : bool = False, use_reloader : bool = False

    ) -> None:

        # Run method used when started in a standalone mode, i.e., outside gunicorn or

        # similar WSGI HTTP servers. Otherwise, use mechanism defined by the used

        # WSGI HTTP server.



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



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

        if self.base_url is not None:

            endpoint = '/'.join([endpoint.rstrip('/'), self.base_url])



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

        self._sio.run(

            self._app, host=bind_address, port=bind_port,

            debug=debug, use_reloader=use_reloader

        )



    def start_test_thread(

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

        debug : bool = False, use_reloader : bool = False

    ) -> None:

        # NOTE: To be used for testing purposes with pytest

        # Stop the thread through nbi_app.stop_test_thread()

        self._thread = self._sio.start_background_task(

            self._sio.run, self._app, host=bind_address, port=bind_port,

            debug=debug, use_reloader=use_reloader

        )



    def stop_test_thread(self):

        # NOTE: To be used for testing purposes with pytest

        # Start the thread through nbi_app.start_test_thread(...)

        if self._thread is None: return

        self._thread.join()
 
            LOGGER.debug(' - {:s}'.format(str(namespace)))

        for namespace in self._sio.server.handlers:

            LOGGER.debug(' - {:s}'.format(str(namespace)))