Commit 302a63b8 authored by Waleed Akbar's avatar Waleed Akbar
Browse files

Merge branch... 

Merge branch 'feat/362-cttc-update-simap-server-to-support-influxdb-connection' of ssh://labs.etsi.org:29419/tfs/controller into feat/369-cttc-mwc26-f5g-simap-demo
parents ee1b4511 983c0282

.gitlab-ci.yml

+61 −67
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.

# # 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.





# stages of the cicd pipeline

stages:

  #- dependencies

  - build

  - unit_test

  - end2end_test

# # stages of the cicd pipeline

# stages:

#   #- dependencies

#   - build

#   - unit_test

#   - end2end_test



workflow:

  rules:

    - if: '$CI_PIPELINE_SOURCE == "merge_request_event" && ($CI_MERGE_REQUEST_TARGET_BRANCH_NAME == "develop" || $CI_MERGE_REQUEST_TARGET_BRANCH_NAME == $CI_DEFAULT_BRANCH)'

      when: always

    - when: never

# # include the individual .gitlab-ci.yml of each micro-service and tests

# include:

#   #- local: '/manifests/.gitlab-ci.yml'

#   - local: '/src/monitoring/.gitlab-ci.yml'

#   - local: '/src/nbi/.gitlab-ci.yml'

#   - local: '/src/context/.gitlab-ci.yml'

#   - local: '/src/device/.gitlab-ci.yml'

#   - local: '/src/service/.gitlab-ci.yml'

#   - local: '/src/qkd_app/.gitlab-ci.yml'

#   - local: '/src/dbscanserving/.gitlab-ci.yml'

#   - local: '/src/opticalattackmitigator/.gitlab-ci.yml'

#   - local: '/src/opticalattackdetector/.gitlab-ci.yml'

#   - local: '/src/opticalattackmanager/.gitlab-ci.yml'

#   - local: '/src/opticalcontroller/.gitlab-ci.yml'

#   - local: '/src/ztp/.gitlab-ci.yml'

#   - local: '/src/policy/.gitlab-ci.yml'

#   - local: '/src/automation/.gitlab-ci.yml'

#   - local: '/src/forecaster/.gitlab-ci.yml'

#   #- local: '/src/webui/.gitlab-ci.yml'

#   #- local: '/src/l3_distributedattackdetector/.gitlab-ci.yml'

#   #- local: '/src/l3_centralizedattackdetector/.gitlab-ci.yml'

#   #- local: '/src/l3_attackmitigator/.gitlab-ci.yml'

#   - local: '/src/slice/.gitlab-ci.yml'

#   #- local: '/src/interdomain/.gitlab-ci.yml'

#   - local: '/src/pathcomp/.gitlab-ci.yml'

#   #- local: '/src/dlt/.gitlab-ci.yml'

#   - local: '/src/load_generator/.gitlab-ci.yml'

#   - local: '/src/bgpls_speaker/.gitlab-ci.yml'

#   - local: '/src/kpi_manager/.gitlab-ci.yml'

#   - local: '/src/kpi_value_api/.gitlab-ci.yml'

#   #- local: '/src/kpi_value_writer/.gitlab-ci.yml'

#   #- local: '/src/telemetry/.gitlab-ci.yml'

#   - local: '/src/analytics/.gitlab-ci.yml'

#   - local: '/src/qos_profile/.gitlab-ci.yml'

#   - local: '/src/vnt_manager/.gitlab-ci.yml'

#   - local: '/src/e2e_orchestrator/.gitlab-ci.yml'

#   - local: '/src/ztp_server/.gitlab-ci.yml'

#   - local: '/src/osm_client/.gitlab-ci.yml'

#   - local: '/src/simap_connector/.gitlab-ci.yml'

#   - local: '/src/pluggables/.gitlab-ci.yml'



# include the individual .gitlab-ci.yml of each micro-service and tests

include:

  #- local: '/manifests/.gitlab-ci.yml'

  - local: '/src/monitoring/.gitlab-ci.yml'

  - local: '/src/nbi/.gitlab-ci.yml'

  - local: '/src/context/.gitlab-ci.yml'

  - local: '/src/device/.gitlab-ci.yml'

  - local: '/src/service/.gitlab-ci.yml'

  - local: '/src/qkd_app/.gitlab-ci.yml'

  - local: '/src/dbscanserving/.gitlab-ci.yml'

  - local: '/src/opticalattackmitigator/.gitlab-ci.yml'

  - local: '/src/opticalattackdetector/.gitlab-ci.yml'

  - local: '/src/opticalattackmanager/.gitlab-ci.yml'

  - local: '/src/opticalcontroller/.gitlab-ci.yml'

  - local: '/src/ztp/.gitlab-ci.yml'

  - local: '/src/policy/.gitlab-ci.yml'

  - local: '/src/automation/.gitlab-ci.yml'

  - local: '/src/forecaster/.gitlab-ci.yml'

  #- local: '/src/webui/.gitlab-ci.yml'

  #- local: '/src/l3_distributedattackdetector/.gitlab-ci.yml'

  #- local: '/src/l3_centralizedattackdetector/.gitlab-ci.yml'

  #- local: '/src/l3_attackmitigator/.gitlab-ci.yml'

  - local: '/src/slice/.gitlab-ci.yml'

  #- local: '/src/interdomain/.gitlab-ci.yml'

  - local: '/src/pathcomp/.gitlab-ci.yml'

  #- local: '/src/dlt/.gitlab-ci.yml'

  - local: '/src/load_generator/.gitlab-ci.yml'

  - local: '/src/bgpls_speaker/.gitlab-ci.yml'

  - local: '/src/kpi_manager/.gitlab-ci.yml'

  - local: '/src/kpi_value_api/.gitlab-ci.yml'

  #- local: '/src/kpi_value_writer/.gitlab-ci.yml'

  #- local: '/src/telemetry/.gitlab-ci.yml'

  - local: '/src/analytics/.gitlab-ci.yml'

  - local: '/src/qos_profile/.gitlab-ci.yml'

  - local: '/src/vnt_manager/.gitlab-ci.yml'

  - local: '/src/e2e_orchestrator/.gitlab-ci.yml'

  - local: '/src/ztp_server/.gitlab-ci.yml'

  - local: '/src/osm_client/.gitlab-ci.yml'

  - local: '/src/simap_connector/.gitlab-ci.yml'

  - local: '/src/pluggables/.gitlab-ci.yml'



  # This should be last one: end-to-end integration tests

  - local: '/src/tests/.gitlab-ci.yml'
 
#   # This should be last one: end-to-end integration tests

#   - local: '/src/tests/.gitlab-ci.yml'

src/common/tools/rest_conf/server/requirements.in

+1 −0
Original line number Diff line number Diff line
@@ -24,3 +24,4 @@ libyang==2.8.4 
pyopenssl==23.0.0

requests==2.27.1

werkzeug==2.3.7

influxdb3-python==0.16.0

src/common/tools/rest_conf/server/restconf_server/Config.py

+6 −0
Original line number Diff line number Diff line
@@ -20,3 +20,9 @@ RESTCONF_PREFIX = os.environ.get('RESTCONF_PREFIX', '/restconf' ) 
YANG_SEARCH_PATH = os.environ.get('YANG_SEARCH_PATH', './yang'        )

STARTUP_FILE     = os.environ.get('STARTUP_FILE',     './startup.json')

SECRET_KEY       = os.environ.get('SECRET_KEY',       secrets.token_hex(64))



# InfluxDB Configuration

INFLUXDB_HOST     = os.environ.get('INFLUXDB_HOST',     'localhost'      )

INFLUXDB_PORT     = int(os.environ.get('INFLUXDB_PORT', '8181')          )

INFLUXDB_DATABASE = os.environ.get('INFLUXDB_DATABASE', 'simap_telemetry')

INFLUXDB_TOKEN    = os.environ.get('INFLUXDB_TOKEN',    'apiv3_xSq6xD1wBvZ21Pc3uPnqqmgLeSn-kZbz1y4S4u8GtSbJebEly8wl9WUjGJ0Ja_bQuwKB_lpDcHrNJUBCDJMSCw')

src/common/tools/rest_conf/server/restconf_server/app.py

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





import logging

from .callbacks import CallbackOnLinkTelemetry, CallbackOnNodeTelemetry

from .Config import INFLUXDB_HOST, INFLUXDB_PORT, INFLUXDB_TOKEN, INFLUXDB_DATABASE

from .influxdb_client import SimapInfluxDBClient

from .RestConfServerApplication import RestConfServerApplication

import logging





logging.basicConfig(
@@ -29,6 +32,29 @@ rcs_app.register_host_meta() 
rcs_app.register_restconf()

LOGGER.info('All connectors registered')



# Initialize InfluxDB client and register telemetry callbacks

try:

    LOGGER.info('Initializing InfluxDB client (host=%s, port=%d, db=%s)...', INFLUXDB_HOST, INFLUXDB_PORT, INFLUXDB_DATABASE)

    influx_client = SimapInfluxDBClient(

        host     = INFLUXDB_HOST,

        port     = INFLUXDB_PORT,

        token    = INFLUXDB_TOKEN,

        database = INFLUXDB_DATABASE

    )

except Exception as e:

    LOGGER.error('Failed to initialize InfluxDB client: %s', e)

    influx_client = None



if influx_client is not None and influx_client.is_connected():

    try:

        rcs_app.callback_dispatcher.register(CallbackOnLinkTelemetry(influx_client))

        rcs_app.callback_dispatcher.register(CallbackOnNodeTelemetry(influx_client))

        LOGGER.info('Telemetry callbacks registered')

    except Exception as e:

        LOGGER.error('Failed to register telemetry callbacks: %s', e)

else:

    LOGGER.warning('InfluxDB client not connected, telemetry callbacks disabled.')



rcs_app.dump_configuration()

app = rcs_app.get_flask_app()

src/common/tools/rest_conf/server/restconf_server/callbacks/TelemetryCallbacks.py

0 → 100644
+270 −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.



"""

Telemetry callbacks for writing SIMAP telemetry to InfluxDB.

"""



import json

import logging

import re

from typing import Any, Dict, Optional, Union



from ..Callbacks import _Callback

from ..influxdb_client import SimapInfluxDBClient





LOGGER = logging.getLogger(__name__)





def _extract_telemetry_from_nested(data: Dict, link_id: str = None, node_id: str = None) -> Optional[Dict]:

    """

    Extract simap-telemetry data from various nested JSON structures returned by YangHandler.

    

    The YangHandler.update() returns different structures depending on context:

    - {"ietf-network:network": [{"network-id": "te", "ietf-network-topology:link": [{"link-id": "L1", "simap-telemetry:simap-telemetry": {...}}]}]}

    - {"ietf-network-topology:link": [{"link-id": "L1", "simap-telemetry:simap-telemetry": {...}}]}

    - {"simap-telemetry:simap-telemetry": {...}}

    - {"ietf-network:node": [...]} for nodes

    

    Args:

        data: The JSON dict to search

        link_id: If searching for link telemetry, the link ID to match

        node_id: If searching for node telemetry, the node ID to match

    

    Returns:

        The telemetry dict or None if not found

    """

    if not isinstance(data, dict):

        return None

    

    # Direct telemetry at top level

    if 'simap-telemetry:simap-telemetry' in data:

        return data['simap-telemetry:simap-telemetry']

    

    # Check for direct telemetry fields (already unwrapped)

    if 'bandwidth-utilization' in data or 'latency' in data or 'cpu-utilization' in data:

        return data

    

    # Search in ietf-network:network array

    if 'ietf-network:network' in data:

        for network in data['ietf-network:network']:

            if isinstance(network, dict):

                # Look for links

                if link_id and 'ietf-network-topology:link' in network:

                    for link in network['ietf-network-topology:link']:

                        if isinstance(link, dict) and link.get('link-id') == link_id:

                            if 'simap-telemetry:simap-telemetry' in link:

                                return link['simap-telemetry:simap-telemetry']

                # Look for nodes

                if node_id and 'node' in network:

                    for node in network['node']:

                        if isinstance(node, dict) and node.get('node-id') == node_id:

                            if 'simap-telemetry:simap-telemetry' in node:

                                return node['simap-telemetry:simap-telemetry']

    

    # Search in ietf-network-topology:link array

    if 'ietf-network-topology:link' in data:

        for link in data['ietf-network-topology:link']:

            if isinstance(link, dict):

                if link_id is None or link.get('link-id') == link_id:

                    if 'simap-telemetry:simap-telemetry' in link:

                        return link['simap-telemetry:simap-telemetry']

    

    # Search in ietf-network:node or node array

    for node_key in ['ietf-network:node', 'node']:

        if node_key in data:

            for node in data[node_key]:

                if isinstance(node, dict):

                    if node_id is None or node.get('node-id') == node_id:

                        if 'simap-telemetry:simap-telemetry' in node:

                            return node['simap-telemetry:simap-telemetry']

    

    return None





class CallbackOnLinkTelemetry(_Callback):

    """

    Callback triggered when link telemetry data is updated via RESTCONF.

    Writes telemetry data to InfluxDB.

    """



    # Pattern matches:

    # /restconf/data/ietf-network:networks/network=<network_id>/ietf-network-topology:link=<link_id>/simap-telemetry:simap-telemetry

    PATTERN = (

        r'/restconf/data/ietf-network:networks'

        r'/network=(?P<network_id>[^/]+)'

        r'/ietf-network-topology:link=(?P<link_id>[^/]+)'

        r'/simap-telemetry:simap-telemetry'

    )



    def __init__(self, influx_client: SimapInfluxDBClient) -> None:

        """

        Initialize the callback with an InfluxDB client.



        Args:

            influx_client: SimapInfluxDBClient instance for writing telemetry

        """

        super().__init__(self.PATTERN)

        self._influx_client = influx_client



    def execute(

        self,

        match: re.Match,

        path: str,

        old_data: Optional[Union[Dict, str]],

        new_data: Optional[Union[Dict, str]]

    ) -> bool:

        """

        Execute the callback to write link telemetry to InfluxDB.



        Args:

            match: Regex match object containing network_id and link_id

            path: Original RESTCONF path

            old_data: Previous telemetry data (unused)

            new_data: New telemetry data to write (can be dict or JSON string)



        Returns:

            True to continue executing other callbacks

        """

        if new_data is None:

            LOGGER.debug("Link telemetry deletion, skipping InfluxDB write")

            return True



        # Handle case where new_data is a JSON string instead of a dict

        if isinstance(new_data, str):

            try:

                new_data = json.loads(new_data)

            except json.JSONDecodeError:

                LOGGER.error("Failed to parse new_data as JSON: %s", new_data)

                return True



        network_id = match.group('network_id')

        link_id = match.group('link_id')



        # Extract telemetry fields from nested JSON structure

        # YangHandler returns various nested formats depending on context

        telemetry = _extract_telemetry_from_nested(new_data, link_id=link_id)

        if telemetry is None:

            LOGGER.warning(

                "Could not extract telemetry from data for link=%s: %s",

                link_id, new_data

            )

            return True



        bandwidth_utilization = telemetry.get('bandwidth-utilization', 0)

        latency = telemetry.get('latency', 0)

        related_service_ids = telemetry.get('related-service-ids', None)



        LOGGER.info(

            "Link telemetry callback: network=%s, link=%s, bw=%s, lat=%s",

            network_id, link_id, bandwidth_utilization, latency

        )



        self._influx_client.write_link_telemetry(

            network_id=network_id,

            link_id=link_id,

            bandwidth_utilization=float(bandwidth_utilization),

            latency=float(latency),

            related_service_ids=related_service_ids

        )



        return True  # Continue to other callbacks





class CallbackOnNodeTelemetry(_Callback):

    """

    Callback triggered when node telemetry data is updated via RESTCONF.

    Writes telemetry data to InfluxDB.

    """



    # Pattern matches:

    # /restconf/data/ietf-network:networks/network=<network_id>/node=<node_id>/simap-telemetry:simap-telemetry

    PATTERN = (

        r'/restconf/data/ietf-network:networks'

        r'/network=(?P<network_id>[^/]+)'

        r'/node=(?P<node_id>[^/]+)'

        r'/simap-telemetry:simap-telemetry'

    )



    def __init__(self, influx_client: SimapInfluxDBClient) -> None:

        """

        Initialize the callback with an InfluxDB client.



        Args:

            influx_client: SimapInfluxDBClient instance for writing telemetry

        """

        super().__init__(self.PATTERN)

        self._influx_client = influx_client



    def execute(

        self,

        match: re.Match,

        path: str,

        old_data: Optional[Union[Dict, str]],

        new_data: Optional[Union[Dict, str]]

    ) -> bool:

        """

        Execute the callback to write node telemetry to InfluxDB.



        Args:

            match: Regex match object containing network_id and node_id

            path: Original RESTCONF path

            old_data: Previous telemetry data (unused)

            new_data: New telemetry data to write (can be dict or JSON string)



        Returns:

            True to continue executing other callbacks

        """

        if new_data is None:

            LOGGER.debug("Node telemetry deletion, skipping InfluxDB write")

            return True



        # Handle case where new_data is a JSON string instead of a dict

        if isinstance(new_data, str):

            try:

                new_data = json.loads(new_data)

            except json.JSONDecodeError:

                LOGGER.error("Failed to parse new_data as JSON: %s", new_data)

                return True



        network_id = match.group('network_id')

        node_id = match.group('node_id')



        # Extract telemetry fields from nested JSON structure

        # YangHandler returns various nested formats depending on context

        telemetry = _extract_telemetry_from_nested(new_data, node_id=node_id)

        if telemetry is None:

            LOGGER.warning(

                "Could not extract telemetry from data for node=%s: %s",

                node_id, new_data

            )

            return True



        cpu_utilization = telemetry.get('cpu-utilization', 0)

        related_service_ids = telemetry.get('related-service-ids', None)



        LOGGER.info(

            "Node telemetry callback: network=%s, node=%s, cpu=%s",

            network_id, node_id, cpu_utilization

        )



        self._influx_client.write_node_telemetry(

            network_id=network_id,

            node_id=node_id,

            cpu_utilization=float(cpu_utilization),

            related_service_ids=related_service_ids

        )



        return True  # Continue to other callbacks