diff --git a/app.py b/app.py index d3f1706c0634f1a795ff0788719d762c7f8cd5e6..2812a13314d7f3c3187411d8b323ad1d692e4c0b 100644 --- a/app.py +++ b/app.py @@ -27,6 +27,7 @@ from src.webui.gui import gui_bp from src.config.config import create_config from src.database.db import init_db as init_slice from src.database.service_db import init_db as init_service +from src.database.telemetry_client_db import init_db as init_telemetry # Paths that do not require authentication (Swagger UI and its static assets) # /nsc → Swagger UI @@ -58,8 +59,10 @@ def create_app(): """Create Flask application with configured API and namespaces.""" init_slice() init_service() + init_telemetry() app = Flask(__name__) app = create_config(app) + # app.logger.setLevel(logging.INFO) CORS(app) # Configure logging to provide clear and informative log messages @@ -104,4 +107,4 @@ def create_app(): if __name__ == "__main__": app = create_app() - app.run(host="0.0.0.0", port=NSC_PORT, debug=True) \ No newline at end of file + app.run(host="0.0.0.0", port=NSC_PORT, debug=True, threaded=True) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 6fde3e4a66ba00ce1e8bbe8c8595ab07a1fede6d..d8af18ec090402e269acf63416a3335cdb3e6d45 100644 --- a/requirements.txt +++ b/requirements.txt @@ -14,7 +14,7 @@ # This file is an original contribution from Telefonica Innovación Digital S.L. -Flask +flask[async] flask-cors flask-restx netmiko @@ -25,4 +25,4 @@ coverage pytest libyang==3.3.0 sysrepo==1.7.6 - +aiohttp \ No newline at end of file diff --git a/src/api/main.py b/src/api/main.py index eae484a4cb98666230147a149b0dd5f2235da326..0685911ecf837d995c4d20d9a37ff01758cbfcea 100644 --- a/src/api/main.py +++ b/src/api/main.py @@ -15,18 +15,23 @@ # This file is an original contribution from Telefonica Innovación Digital S.L. from src.utils.send_response import send_response -import logging +import logging, json, time, requests, traceback, asyncio, aiohttp from flask import current_app from src.database.db import get_data, delete_data, get_all_data, delete_all_data from src.database.service_db import delete_by_slice_id, get_data_by_slice_id +from src.database.telemetry_client_db import create_client, get_client, get_all_clients, delete_client, delete_all_clients, upsert_subscription, get_subscription, get_client_subscriptions, delete_subscription, delete_all_subscriptions from src.realizer.tfs.helpers.tfs_connector import tfs_connector from src.utils.safe_get import safe_get from src.database.sysrepo_store import get_data_store, create_data_store, delete_data_store, update_data_store, normalize_libyang_data +from typing import Dict, Tuple +from src.realizer.restconf.connectors.tfs_connector import tfs_connector as tfs_restconf_connector +from src.planner.shortest_path import get_shortest_path class Api: + def __init__(self, slice_service): self.slice_service = slice_service - + def add_flow(self, intent): """ Create a new transport network slice. @@ -744,4 +749,262 @@ class Api: except ValueError as e: return send_response(False, code=404, message=str(e)) except Exception as e: - return send_response(False, code=500, message=str(e)) \ No newline at end of file + return send_response(False, code=500, message=str(e)) + + # --- CLIENTS --- + + def get_clients(self, client_id=None): + try: + if client_id: + return get_client(client_id), 200 + clients = get_all_clients() + if not clients: + raise ValueError("No clients found") + return clients, 200 + + except ValueError as e: + return send_response(False, code=404, message=str(e)) + except Exception as e: + return send_response(False, code=500, message=str(e)) + + def add_client(self, client_id): + try: + create_client(client_id) + logging.info(f"Client '{client_id}' created successfully") + return send_response( + True, + code=201, + message=f"Client '{client_id}' created successfully", + data={"client_id": client_id} + ) + + except ValueError as e: + return send_response(False, code=409, message=str(e)) + except Exception as e: + return send_response(False, code=500, message=str(e)) + + def delete_clients(self, client_id=None): + try: + if client_id: + delete_client(client_id) + logging.info(f"Client '{client_id}' removed successfully") + else: + delete_all_clients() + logging.info("All clients removed successfully") + return {}, 204 + except ValueError as e: + return send_response(False, code=404, message=str(e)) + except Exception as e: + return send_response(False, code=500, message=str(e)) + + # --- SUBSCRIPTIONS --- + + def get_subscriptions(self, client_id, slice_id = None): + try: + if slice_id is not None: + try: + subscription = get_subscription(client_id, slice_id) + except ValueError: + subscription = None + if not subscription: + raise ValueError(f"Client '{client_id}' has no subscription for slice '{slice_id}'") + + telemetry = self.get_telemetry(slice_id) + + return { + **subscription, + "telemetry": telemetry + }, 200 + + subscriptions = get_client_subscriptions(client_id) + + result = [] + + for sub in subscriptions: + slice_id = sub["slice_id"] + + telemetry = self.get_telemetry(slice_id) + + result.append({ + **sub, + "telemetry": telemetry + }) + + return { + "client_id": client_id, + "subscriptions": result + }, 200 + + except ValueError as e: + return send_response(False, code=404, message=str(e)) + except Exception as e: + return send_response(False, code=500, message=str(e)) + + def add_subscription(self, client_id, slice_id, frequency): + try: + try: + subscription = get_subscription(client_id, slice_id) + except ValueError: + subscription = None + if subscription: + raise ValueError(f"Client '{client_id}' already has a subscription for slice '{slice_id}'") + if not frequency: + raise KeyError("Field 'frequency' is required") + + upsert_subscription(client_id, slice_id, frequency) + logging.info(f"Subscription for slice '{slice_id}' and client '{client_id}' created successfully") + return send_response( + True, + code=201, + message="Subscription successfully created", + data={ + "sliceId": slice_id, + "frequency": frequency + } + ) + except KeyError as e: + return send_response(False, code=400, message=str(e)) + except ValueError as e: + return send_response(False, code=404, message=str(e)) + except Exception as e: + return send_response(False, code=500, message=str(e)) + + def update_subscription(self, client_id, slice_id, frequency): + try: + try: + subscription = get_subscription(client_id, slice_id) + except ValueError: + subscription = None + if not subscription: + raise ValueError(f"Client '{client_id}' has no subscription for slice '{slice_id}'") + if not frequency: + raise KeyError("Field 'frequency' is required") + + upsert_subscription(client_id, slice_id, frequency) + logging.info(f"Subscription for slice '{slice_id}' and client '{client_id}' modified successfully") + return send_response( + True, + code=201, + message="Subscription successfully modified", + data={ + "sliceId": slice_id, + "frequency": frequency + } + ) + except KeyError as e: + return send_response(False, code=400, message=str(e)) + except ValueError as e: + return send_response(False, code=404, message=str(e)) + except Exception as e: + return send_response(False, code=500, message=str(e)) + + def delete_subscriptions(self, client_id, slice_id = None): + try: + subscriptions = get_client_subscriptions(client_id) + if slice_id: + if slice_id not in subscriptions: + raise ValueError(f"Client '{client_id}' has no subscription for slice '{slice_id}'") + delete_subscription(client_id, slice_id) + logging.info(f"Subscription for slice '{slice_id}' and client '{client_id}' removed successfully") + return {}, 204 + delete_all_subscriptions(client_id) + logging.info(f"All subscriptions for client '{client_id}' removed successfully") + return {}, 204 + + except ValueError as e: + return send_response(False, code=404, message=str(e)) + except Exception as e: + return send_response(False, code=500, message=str(e)) + + # --- TELEMETRY --- + + def get_telemetry(self, slice_id = None): + logging.debug(f"Getting telemetry for slice_id: {slice_id}") + try: + if slice_id is not None: + slice_sdps = get_data_store(f"/ietf-network-slice-service:network-slice-services/slice-service[id='{slice_id}']/sdps") + logging.debug(f"SDPs for slice_id '{slice_id}': {slice_sdps}") + if not slice_sdps: + raise ValueError("No SDPs found") + if len(slice_sdps) > 2: + raise Exception(f"Monitoring for more than 2 SDPs is not supported. Found {len(slice_sdps)} SDPs.") + xpath = f"/ietf-network-slice-service:network-slice-services/slice-service[id='{slice_id}']" + existing_slice = get_data_store(xpath) + if not existing_slice: + raise ValueError(f"There is no slice with id '{slice_id}' registered") + template_id = safe_get(existing_slice, ["network-slice-services", "slice-service", slice_id, "slo-sle-template"]) + slo_sle_template = self.get_slo_sle_templates(template_id)[0] + slo_sle_template = safe_get(slo_sle_template, ["network-slice-services", "slo-sle-templates", "slo-sle-template"]) + slo_sle_template = next(iter(slo_sle_template), None) + if not slo_sle_template: + raise ValueError(f"SLO/SLE template '{template_id}' not found for slice '{slice_id}'") + metrics = self.slice_service.monitoring(slice_id, slo_sle_template, slice_sdps) + return metrics, 200 + + telemetry_data = {} + slices_data = self.get_slice_services()[0] + slice_service_list = slices_data["network-slice-services"]["slice-service"] + if isinstance(slice_service_list, dict): + slice_service_list = list(slice_service_list.values()) + + for slice in slice_service_list: + selected_template_id = slice.get("slo-sle-template") + slo_sle_template = self.get_slo_sle_templates(selected_template_id)[0] + slo_sle_template = safe_get(slo_sle_template, ["network-slice-services", "slo-sle-templates", "slo-sle-template"]) + slo_sle_template = next(iter(slo_sle_template), None) + if not slo_sle_template: + raise ValueError(f"SLO/SLE template '{selected_template_id}' not found for slice '{slice['id']}'") + slice_id = slice["id"] + slice_sdps = get_data_store(f"/ietf-network-slice-service:network-slice-services/slice-service[id='{slice_id}']/sdps") + if not slice_sdps: + raise ValueError("No SDPs found") + if len(slice_sdps) > 2: + raise Exception(f"Monitoring for more than 2 SDPs is not supported. Found {len(slice_sdps)} SDPs.") + telemetry_data[slice_id] = self.slice_service.monitoring(slice_id, slo_sle_template, slice_sdps) + return telemetry_data, 200 + + except ValueError as e: + return send_response(False, code=404, message=str(e)) + except Exception as e: + return send_response(False, code=500, message=str(e)) + + def sync_stream(self, async_gen_func, *args, **kwargs): + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + agen = async_gen_func(*args, **kwargs) + + try: + while True: yield loop.run_until_complete(agen.__anext__()) + except StopAsyncIteration: pass + finally: loop.close() + + async def stream_client_subscriptions(self, client_id): + while True: + try: + data, code = self.get_subscriptions(client_id) + if code == 200: + yield f"data: {json.dumps(data)}\n\n" + subs = data.get("subscriptions", []) + freq = max([s["frequency"] for s in subs]) if subs else 5 + await asyncio.sleep(freq) + else: + yield f"event: error\ndata: {json.dumps(data)}\n\n" + break + except Exception as e: + yield f"event: error\ndata: {json.dumps({'error': str(e)})}\n\n" + break + + async def stream_slice_subscription(self, client_id, slice_id): + while True: + try: + data, code = self.get_subscriptions(client_id, slice_id) + if code == 200: + yield f"data: {json.dumps(data)}\n\n" + freq = data.get("frequency", 5) + await asyncio.sleep(freq) + else: + yield f"event: error\ndata: {json.dumps(data)}\n\n" + break + except Exception as e: + yield f"event: error\ndata: {json.dumps({'error': str(e)})}\n\n" + break \ No newline at end of file diff --git a/src/config/.env.example b/src/config/.env.example index 1e45691f81c0bb4b86ca93be34d20586c9527dad..5b30a134097bfe757cecc65f1c0b77287a3576bf 100644 --- a/src/config/.env.example +++ b/src/config/.env.example @@ -42,7 +42,7 @@ E2E_OPTICAL_IP=127.0.0.1 # Realizer # ------------------------- # If true, no config sent to controllers -DUMMY_MODE=false +DUMMY_MODE=true # ------------------------- # Teraflow @@ -66,12 +66,17 @@ TFS_E2E_IP=127.0.0.1 # ------------------------- # Restconf Controller # ------------------------- -RESTCONF_IP=127.0.0.1 +RESTCONF_IP=192.168.27.189 # Options: TFS or IXIA SDN_CONTROLLER_TYPE=TFS # Options: FRR, CISCO DATAPLANE_SUPPORT=CISCO +# ------------------------- +# Monitoring +# ------------------------- +SDN_SUBSCRIPTION_PERIOD=10 + # ------------------------- # WebUI # ------------------------- diff --git a/src/config/config.py b/src/config/config.py index c059ede30bf820c802c0467f50f70a4a9b0351e1..2e5a9aaf534fde593f4a8f645b0ffdf56dec8f01 100644 --- a/src/config/config.py +++ b/src/config/config.py @@ -69,8 +69,13 @@ def create_config(app: Flask): app.config["SDN_CONTROLLER_TYPE"] = os.getenv("SDN_CONTROLLER_TYPE", "TFS") app.config["DATAPLANE_SUPPORT"] = os.getenv("DATAPLANE_SUPPORT", "FRR") + # Monitoring + app.config["SDN_SUBSCRIPTION_PERIOD"] = int(os.getenv("SDN_SUBSCRIPTION_PERIOD", "10")) + app.config["TELEMETRY_CACHE"] = {} + # API app.config["API_USERNAME"] = os.getenv("API_USERNAME", "admin") app.config["API_PASSWORD"] = os.getenv("API_PASSWORD", "admin") return app + diff --git a/src/config/constants.py b/src/config/constants.py index 279d459281d2ba5f1d69344f3b3534f3f4c880c8..7bfeb6900765f68a3f894c297105fbaa1b12eeba 100644 --- a/src/config/constants.py +++ b/src/config/constants.py @@ -28,6 +28,8 @@ DATABASE_PATH = SRC_PATH / "database" CONFIG_PATH = SRC_PATH / "config" NBI_L2_PATH = "restconf/data/ietf-l2vpn-svc:l2vpn-svc/vpn-services" NBI_L3_PATH = "restconf/data/ietf-l3vpn-svc:l3vpn-svc/vpn-services" +NBI_IETF_NETWORKS_PATH = "restconf/data/ietf-network:networks" +NBI_SIMAP_SUSCRIPTION_PATH = "/restconf/operations/subscriptions:establish-subscription" diff --git a/src/database/telemetry_client_db.py b/src/database/telemetry_client_db.py new file mode 100644 index 0000000000000000000000000000000000000000..d9d23528ebafb0e7bc01e337b55e3f97aba6f2c4 --- /dev/null +++ b/src/database/telemetry_client_db.py @@ -0,0 +1,314 @@ +# Copyright 2022-2026 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. + +# This file is an original contribution from Telefonica Innovación Digital S.L. + +import sqlite3, logging, json + +DB_NAME = "telemetry_client.db" + + +def init_db(): + """ + Initialize database and create telemetry_client table if not exists. + """ + conn = sqlite3.connect(DB_NAME) + cursor = conn.cursor() + + cursor.execute(""" + CREATE TABLE IF NOT EXISTS client ( + client_id TEXT PRIMARY KEY + ) + """) + + cursor.execute(""" + CREATE TABLE IF NOT EXISTS client_slice_subscription ( + client_id TEXT, + slice_id TEXT, + frequency INTEGER NOT NULL, + + PRIMARY KEY (client_id, slice_id), + + FOREIGN KEY (client_id) REFERENCES client(client_id) ON DELETE CASCADE + ) + """) + + cursor.execute(""" + CREATE TABLE IF NOT EXISTS telemetry ( + slice_id TEXT PRIMARY KEY, + + telemetry_list TEXT NOT NULL, + slo_sle_compliance TEXT NOT NULL, + + timestamp REAL NOT NULL + ) + """) + + conn.commit() + conn.close() + +def get_conn(): + conn = sqlite3.connect(DB_NAME) + conn.execute("PRAGMA foreign_keys = ON") + return conn + +def create_client(client_id: str): + conn = get_conn() + cursor = conn.cursor() + + try: + cursor.execute( + "INSERT INTO client (client_id) VALUES (?)", + (client_id,) + ) + conn.commit() + + except sqlite3.IntegrityError: + raise ValueError(f"Client '{client_id}' already exists") + + finally: + conn.close() + +def delete_client(client_id: str): + """ + Delete client. + """ + conn = get_conn() + cursor = conn.cursor() + + cursor.execute( + "DELETE FROM client WHERE client_id = ?", + (client_id,) + ) + + if cursor.rowcount == 0: + raise ValueError(f"No client found with id '{client_id}' to delete.") + + logging.debug(f"Client '{client_id}' deleted.") + + conn.commit() + conn.close() + + +def get_client(client_id: str) -> dict: + conn = get_conn() + cursor = conn.cursor() + + cursor.execute( + "SELECT * FROM client WHERE client_id = ?", + (client_id,) + ) + + row = cursor.fetchone() + + if not row: + conn.close() + raise ValueError(f"No client found with id '{client_id}'.") + + result = { + "client_id": row[0] + } + + conn.close() + + return result + + +def get_all_clients() -> list[dict]: + conn = get_conn() + cursor = conn.cursor() + + cursor.execute("SELECT client_id FROM client") + + rows = cursor.fetchall() + + conn.close() + + return [ + { + "client_id": row[0] + } + for row in rows + ] + + +def delete_all_clients(): + conn = get_conn() + cursor = conn.cursor() + + cursor.execute("DELETE FROM client") + + conn.commit() + conn.close() + + logging.debug("All clients deleted.") + +def upsert_subscription(client_id: str, slice_id: str, frequency: int): + conn = get_conn() + cursor = conn.cursor() + + cursor.execute(""" + INSERT INTO client_slice_subscription (client_id, slice_id, frequency) + VALUES (?, ?, ?) + ON CONFLICT(client_id, slice_id) + DO UPDATE SET frequency = excluded.frequency + """, (client_id, slice_id, frequency)) + + conn.commit() + conn.close() + +def get_subscription(client_id: str, slice_id: str) -> dict: + conn = get_conn() + cursor = conn.cursor() + + cursor.execute(""" + SELECT client_id, slice_id, frequency + FROM client_slice_subscription + WHERE client_id = ? AND slice_id = ? + """, (client_id, slice_id)) + + row = cursor.fetchone() + conn.close() + + if not row: + raise ValueError("Subscription not found") + + return { + "client_id": row[0], + "slice_id": row[1], + "frequency": row[2] + } + +def get_client_subscriptions(client_id: str) -> list[dict]: + conn = get_conn() + cursor = conn.cursor() + + cursor.execute(""" + SELECT slice_id, frequency + FROM client_slice_subscription + WHERE client_id = ? + """, (client_id,)) + + rows = cursor.fetchall() + conn.close() + + return [ + { + "slice_id": r[0], + "frequency": r[1] + } + for r in rows + ] + +def delete_subscription(client_id: str, slice_id: str): + conn = get_conn() + cursor = conn.cursor() + + cursor.execute(""" + DELETE FROM client_slice_subscription + WHERE client_id = ? AND slice_id = ? + """, (client_id, slice_id)) + + if cursor.rowcount == 0: + raise ValueError("Subscription not found") + + conn.commit() + conn.close() + +def delete_all_subscriptions(client_id: str): + conn = get_conn() + cursor = conn.cursor() + + cursor.execute(""" + DELETE FROM client_slice_subscription + WHERE client_id = ? + """, (client_id,)) + + conn.commit() + conn.close() + +def upsert_telemetry( + slice_id: str, + telemetry_list: dict, + slo_sle_compliance: dict, + timestamp: float +): + conn = get_conn() + cursor = conn.cursor() + + cursor.execute(""" + INSERT INTO telemetry ( + slice_id, + telemetry_list, + slo_sle_compliance, + timestamp + ) + VALUES (?, ?, ?, ?) + + ON CONFLICT(slice_id) + DO UPDATE SET + telemetry_list = excluded.telemetry_list, + slo_sle_compliance = excluded.slo_sle_compliance, + timestamp = excluded.timestamp + """, ( + slice_id, + json.dumps(telemetry_list), + json.dumps(slo_sle_compliance), + timestamp + )) + + conn.commit() + conn.close() + +def get_telemetry(slice_id: str): + conn = get_conn() + cursor = conn.cursor() + + cursor.execute(""" + SELECT + slice_id, + telemetry_list, + slo_sle_compliance, + timestamp + FROM telemetry + WHERE slice_id = ? + """, (slice_id,)) + + row = cursor.fetchone() + + conn.close() + + if not row: + raise ValueError( + f"No telemetry found for slice '{slice_id}'" + ) + + return { + "slice_id": row[0], + "telemetry_list": json.loads(row[1]), + "slo_sle_compliance": json.loads(row[2]), + "timestamp": row[3] + } + +def delete_telemetry(slice_id: str): + conn = get_conn() + cursor = conn.cursor() + + cursor.execute(""" + DELETE FROM telemetry + WHERE slice_id = ? + """, (slice_id,)) + + conn.commit() + conn.close() \ No newline at end of file diff --git a/src/main.py b/src/main.py index ba6418701dccac1a5fce83719e274f623ce38b97..e11ff27932b880dd876076dc8564375af7a2c3f8 100644 --- a/src/main.py +++ b/src/main.py @@ -22,6 +22,7 @@ from src.nbi_processor.main import nbi_processor from src.database.store_data import store_data from src.mapper.main import mapper from src.realizer.main import realizer +from src.planner.shortest_path import get_shortest_path from src.realizer.send_controller import send_controller class NSController: @@ -95,7 +96,7 @@ class NSController: ietf_intents = nbi_processor(intent_json) for intent in ietf_intents: - logging.info(intent) + logging.debug(intent) # Mapper services, rules = mapper(intent, controller_type=self.controller_type) logging.debug(f"Services: {services}") @@ -129,4 +130,38 @@ class NSController: return { "slices": self.response, "setup_time": setup_time - } \ No newline at end of file + } + + def monitoring(self, slice_id, slo_sle_template, sdps): + """ + Monitor the status of a specific network slice. + + Workflow: + 1. Validate the provided slice_id + 2. Request the planner the shortest path for the slice + 3. Request the realizer to retrieve the metrics for the links of the specified slice + 4. Realizer retrieves the metrics from the controller for the specified slice + 5. Request mapper to aggregate the metrics and store them in the database + 6. Return the monitoring information for the specified slice + + Args: + slice_id (str): The identifier of the network slice to monitor + slo_sle_template (dict): The SLO-SLE template for the network slice + sdps (list): The SDPs for the network slice + + Returns: + dict: Contains monitoring information for the specified slice + """ + payload = { + "slice_id": slice_id, + "slo_sle_template": slo_sle_template, + "sdps": sdps + } + + # Request the realizer to retrieve the metrics for the links of the specified slice + realizer(payload, action="MONITOR", controller_type=self.controller_type) + + # Request the mapper to aggregate the metrics and store them in the database + metrics = mapper(payload, action="MONITOR") + + return metrics \ No newline at end of file diff --git a/src/mapper/aggregate_monitoring.py b/src/mapper/aggregate_monitoring.py new file mode 100644 index 0000000000000000000000000000000000000000..91d3c0f0cccd79addda2241c0fe0f8edbf4dcb10 --- /dev/null +++ b/src/mapper/aggregate_monitoring.py @@ -0,0 +1,64 @@ +import logging +import time +from flask import current_app +from src.database.telemetry_client_db import get_telemetry, upsert_telemetry +from src.utils.safe_get import safe_get + + +def aggregate_monitoring(slice_id, slo_sle_template): + """ + Aggregates monitoring data for a given slice + + Args: + slice_id (str): The identifier of the network slice. + slo_sle_template (dict): The SLO-SLE template for the network slice. + """ + + if slice_id in current_app.config["TELEMETRY_CACHE"]: + slos = safe_get(slo_sle_template, ["slo-policy", "metric-bound"]) + slo_bandwidth = [ s["bound"] for s in slos if s.get("metric-type", "") == "ietf-network-slice-service:two-way-bandwidth" ][0] if slos else 0 + slo_latency = [ s["bound"] for s in slos if s.get("metric-type", "") == "ietf-network-slice-service:two-way-delay-maximum" ][0] if slos else "N/A" + + links = current_app.config["TELEMETRY_CACHE"][slice_id] + + if not links: + raise Exception(f"No telemetry data available for slice '{slice_id}'") + + bandwidth = min([link["bandwidth"] for _, link in links.items()]) + latency = sum([link["latency"] for _, link in links.items()]) + + violated_metrics = [] + + if slo_latency != "N/A" and latency > float(slo_latency): + violated_metrics.append("latency") + + if slo_bandwidth != 0 and bandwidth < float(slo_bandwidth): + violated_metrics.append("bandwidth") + + telemetry_json = { + "latency": latency, + "bandwidth": bandwidth + } + + compliance_json = { + "is_compliant": len(violated_metrics) == 0, + "violated_metrics": violated_metrics + } + + current_timestamp = time.time() + + upsert_telemetry( + slice_id=slice_id, + telemetry_list=telemetry_json, + slo_sle_compliance=compliance_json, + timestamp=current_timestamp + ) + + return { + "slice_id": slice_id, + "telemetry_list": telemetry_json, + "slo_sle_compliance": compliance_json, + "timestamp": current_timestamp + } + else: + raise Exception(f"Telemetry cache for slice '{slice_id}' is not initialized. Please make sure the SDN controller is sending telemetry data for this slice.") \ No newline at end of file diff --git a/src/mapper/main.py b/src/mapper/main.py index 6a3b160fcb65b49cbc86c79dec97cf2f9ab2da44..3c1e943e103f62eb0a4a210650b8ff4ed83f1bc4 100644 --- a/src/mapper/main.py +++ b/src/mapper/main.py @@ -20,12 +20,13 @@ from src.utils.safe_get import safe_get from .slo_viability import slo_viability from .get_service_template import get_service_template from .process_connnectivity import process_connectivity +from .aggregate_monitoring import aggregate_monitoring from src.realizer.main import realizer from flask import current_app from src.database.sysrepo_store import get_data_store, create_data_store, delete_data_store, update_data_store, normalize_libyang_data from src.database.service_db import save_data, update_data -def mapper(ietf_intent, controller_type="TFS"): +def mapper(payload, controller_type="TFS", action="CREATE"): """ Map an IETF network slice intent to the most suitable Network Resource Partition (NRP). @@ -44,128 +45,137 @@ def mapper(ietf_intent, controller_type="TFS"): dict or None: Optimal path if planner is enabled; otherwise, None. """ optimal_path = None - services = [ietf_intent] - - if current_app.config["NRP_ENABLED"]: - # Retrieve NRP view - nrp_view = realizer(None, True, "READ") - - # Extract Service Level Objectives (SLOs) from the intent - slos = ietf_intent["ietf-network-slice-service:network-slice-services"]["slo-sle-templates"]["slo-sle-template"][0]["slo-policy"]["metric-bound"] - if slos: - # Find candidate NRPs that can meet the SLO requirements - candidates = [ - (nrp, slo_viability(slos, nrp)[1]) - for nrp in nrp_view - if slo_viability(slos, nrp)[0] and nrp["available"] - ] - logging.debug(f"Candidates: {candidates}") - - # Select the best NRP based on candidates - best_nrp = max(candidates, key=lambda x: x[1])[0] if candidates else None - logging.debug(f"Best NRP: {best_nrp}") - - if best_nrp: - best_nrp["slices"].append(ietf_intent["ietf-network-slice-service:network-slice-services"]["slice-service"][0]["id"]) - # Update NRP view - realizer(ietf_intent, True, "UPDATE") - # TODO Here we should put how the slice is attached to an already created nrp - else: - # Request the controller to create a new NRP that meets the SLOs - answer = realizer(ietf_intent, True, "CREATE", best_nrp) - if not answer: - logging.error("Slice rejected due to lack of NRPs") - return None - # TODO Here we should put how the slice is attached to the new nrp - - if current_app.config["PLANNER_ENABLED"]: - optimal_path = Planner().planner(ietf_intent, current_app.config["PLANNER_TYPE"]) - logging.debug(f"Optimal path: {optimal_path}") - - if controller_type == "RESTCONF": - # Initialize available templates - templates = get_data_store("/ietf-network-slice-service:network-slice-services/slo-sle-templates/slo-sle-template") - normalized_templates = normalize_libyang_data(templates) - logging.debug(f"Normalized templates: {normalized_templates}") - available_templates = safe_get(normalized_templates, ["slo-sle-templates", "slo-sle-template"]) or [] + services = None + + if action == "CREATE": + ietf_intent = payload + services = [ietf_intent] + if current_app.config["NRP_ENABLED"]: + # Retrieve NRP view + nrp_view = realizer(None, True, "READ") + + # Extract Service Level Objectives (SLOs) from the intent + slos = ietf_intent["ietf-network-slice-service:network-slice-services"]["slo-sle-templates"]["slo-sle-template"][0]["slo-policy"]["metric-bound"] + if slos: + # Find candidate NRPs that can meet the SLO requirements + candidates = [ + (nrp, slo_viability(slos, nrp)[1]) + for nrp in nrp_view + if slo_viability(slos, nrp)[0] and nrp["available"] + ] + logging.debug(f"Candidates: {candidates}") + + # Select the best NRP based on candidates + best_nrp = max(candidates, key=lambda x: x[1])[0] if candidates else None + logging.debug(f"Best NRP: {best_nrp}") + + if best_nrp: + best_nrp["slices"].append(ietf_intent["ietf-network-slice-service:network-slice-services"]["slice-service"][0]["id"]) + # Update NRP view + realizer(ietf_intent, True, "UPDATE") + # TODO Here we should put how the slice is attached to an already created nrp + else: + # Request the controller to create a new NRP that meets the SLOs + answer = realizer(ietf_intent, True, "CREATE", best_nrp) + if not answer: + logging.error("Slice rejected due to lack of NRPs") + return None + # TODO Here we should put how the slice is attached to the new nrp + + if current_app.config["PLANNER_ENABLED"]: + optimal_path = Planner().planner(ietf_intent, current_app.config["PLANNER_TYPE"]) + logging.debug(f"Optimal path: {optimal_path}") - # Add templates from intent - for template in safe_get(ietf_intent, ["ietf-network-slice-service:network-slice-services","slo-sle-templates", "slo-sle-template"]): - available_templates.append(template) - logging.debug(f"Available templates: {available_templates}") - - services = [] - - # Process each slice service - for slice_service in safe_get(ietf_intent, ["ietf-network-slice-service:network-slice-services", "slice-service"]): - service_id = safe_get(slice_service, ["id"]) - logging.debug(f"Service ID: {service_id}") - - way = safe_get(slice_service, ['service-tags', 'tag-type', 0, 'tag-type-value', 0]) - logging.debug(f"Way: {way}") + if controller_type == "RESTCONF": + # Initialize available templates + templates = get_data_store("/ietf-network-slice-service:network-slice-services/slo-sle-templates/slo-sle-template") + normalized_templates = normalize_libyang_data(templates) + logging.debug(f"Normalized templates: {normalized_templates}") + available_templates = safe_get(normalized_templates, ["slo-sle-templates", "slo-sle-template"]) or [] - # Get service-level template - service_template = get_service_template(slice_service, available_templates) - logging.debug(f"Service Template: {service_template}") + # Add templates from intent + for template in safe_get(ietf_intent, ["ietf-network-slice-service:network-slice-services","slo-sle-templates", "slo-sle-template"]): + available_templates.append(template) + logging.debug(f"Available templates: {available_templates}") - # Process connection groups - for connection_group in safe_get(slice_service, ["connection-groups", "connection-group"]): - connection_group_id = safe_get(connection_group, ['id']) - group_id = f"{service_id}-{connection_group_id}" - logging.debug(f"Group ID: {group_id}") - - # Start with service-level template for this group - template = service_template # Reset template for each connection group - - # Override template if specified at group level - group_template = get_service_template(connection_group, available_templates) - if group_template is not None: - template = group_template - logging.debug(f"Group Template: {template}") + services = [] + + # Process each slice service + for slice_service in safe_get(ietf_intent, ["ietf-network-slice-service:network-slice-services", "slice-service"]): + service_id = safe_get(slice_service, ["id"]) + logging.debug(f"Service ID: {service_id}") + + way = safe_get(slice_service, ['service-tags', 'tag-type', 0, 'tag-type-value', 0]) + logging.debug(f"Way: {way}") - connectivity_type = safe_get(connection_group, ['connectivity-type']) - logging.debug(f"Connectivity Type: {connectivity_type}") + # Get service-level template + service_template = get_service_template(slice_service, available_templates) + logging.debug(f"Service Template: {service_template}") - # Process connectivity constructs - for connectivity_construct in safe_get(connection_group, ["connectivity-construct"]): - connectivity_construct_id = safe_get(connectivity_construct, ['id']) - construct_id = f"{group_id}-{connectivity_construct_id}" - logging.debug(f"Construct ID: {construct_id}") + # Process connection groups + for connection_group in safe_get(slice_service, ["connection-groups", "connection-group"]): + connection_group_id = safe_get(connection_group, ['id']) + group_id = f"{service_id}-{connection_group_id}" + logging.debug(f"Group ID: {group_id}") + + # Start with service-level template for this group + template = service_template # Reset template for each connection group - # Start with group-level template for this construct - final_template = template # Reset template for each connectivity construct + # Override template if specified at group level + group_template = get_service_template(connection_group, available_templates) + if group_template is not None: + template = group_template + logging.debug(f"Group Template: {template}") - # Override template if specified at construct level - construct_template = get_service_template(connectivity_construct, available_templates) - if construct_template is not None: - final_template = construct_template - logging.debug(f"Final Template: {final_template}") + connectivity_type = safe_get(connection_group, ['connectivity-type']) + logging.debug(f"Connectivity Type: {connectivity_type}") - # Process SDPs based on connectivity type - sdps = process_connectivity( - connection_group_id, - connectivity_type, - connectivity_construct, - connectivity_construct_id, - slice_service - ) - logging.debug(f"SDPs: {sdps}") - if sdps: # Only append if SDPs were found - service = { - "id": construct_id, - "template": final_template, - "connectivity_type": connectivity_type, - "sdps": sdps, - "way": way - } - services.append(service) - if not current_app.config["DUMMY_MODE"]: - # Save mapping from service_id to slice_id - save_data(service_id=safe_get(service, ["id"]), slice_id=service_id) - logging.debug(f"Service added: {service}") + # Process connectivity constructs + for connectivity_construct in safe_get(connection_group, ["connectivity-construct"]): + connectivity_construct_id = safe_get(connectivity_construct, ['id']) + construct_id = f"{group_id}-{connectivity_construct_id}" + logging.debug(f"Construct ID: {construct_id}") + + # Start with group-level template for this construct + final_template = template # Reset template for each connectivity construct + + # Override template if specified at construct level + construct_template = get_service_template(connectivity_construct, available_templates) + if construct_template is not None: + final_template = construct_template + logging.debug(f"Final Template: {final_template}") - # Break only for point-to-point - if connectivity_type == "point-to-point": - break + # Process SDPs based on connectivity type + sdps = process_connectivity( + connection_group_id, + connectivity_type, + connectivity_construct, + connectivity_construct_id, + slice_service + ) + logging.debug(f"SDPs: {sdps}") + if sdps: # Only append if SDPs were found + service = { + "id": construct_id, + "template": final_template, + "connectivity_type": connectivity_type, + "sdps": sdps, + "way": way + } + services.append(service) + if not current_app.config["DUMMY_MODE"]: + # Save mapping from service_id to slice_id + save_data(service_id=safe_get(service, ["id"]), slice_id=service_id) + logging.debug(f"Service added: {service}") + + # Break only for point-to-point + if connectivity_type == "point-to-point": + break + elif action == "MONITOR": + logging.debug("Mapper action: MONITOR") + slice_id = payload.get("slice_id", None) + slo_sle_template = payload.get("slo_sle_template", None) + metrics = aggregate_monitoring(slice_id, slo_sle_template) + return metrics return services, optimal_path \ No newline at end of file diff --git a/src/planner/shortest_path.py b/src/planner/shortest_path.py new file mode 100644 index 0000000000000000000000000000000000000000..1784bf183e0df139aafa267ec39fc61c3553d613 --- /dev/null +++ b/src/planner/shortest_path.py @@ -0,0 +1,93 @@ +# Copyright 2022-2026 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. + +# This file is an original contribution from Telefonica Innovación Digital S.L. + +import logging, traceback +from collections import deque +from typing import Dict, Tuple + +import logging, traceback +from collections import deque +from typing import Dict, Tuple + +def normalize_node_id(node_id: str) -> str: + if isinstance(node_id, str) and node_id.startswith("urn:tfs:node:"): + return node_id.split(":")[-1] + return node_id + +def get_shortest_path(network, src_node_id, dst_node_id, directed_graph: bool = False) -> Tuple[Dict[str, any], int]: + try: + source_node_idx = normalize_node_id(src_node_id) + destination_node_idx = normalize_node_id(dst_node_id) + + nodes = network["node"] + links = network["ietf-network-topology:link"] + + # Construir grafo + graph = { + normalize_node_id(node["node-id"]): set() + for node in nodes + } + + for link in links: + src = normalize_node_id(link["source"]["source-node"]) + dst = normalize_node_id(link["destination"]["dest-node"]) + + graph[src].add(dst) + if not directed_graph: + graph[dst].add(src) + + # BFS + visited = {node: False for node in graph} + prev = {node: None for node in graph} + + if source_node_idx not in graph: + return {"message": f"Source node '{source_node_idx}' not found"}, 404 + + if destination_node_idx not in graph: + return {"message": f"Destination node '{destination_node_idx}' not found"}, 404 + + queue = deque([source_node_idx]) + visited[source_node_idx] = True + + while queue: + current = queue.popleft() + + if current == destination_node_idx: + break + + for neighbor in graph[current]: + if not visited[neighbor]: + visited[neighbor] = True + prev[neighbor] = current + queue.append(neighbor) + + if not visited[destination_node_idx]: + return {"message": "No path found"}, 404 + + # Reconstruir camino + path = [] + at = destination_node_idx + + while at is not None: + path.append(at) + at = prev[at] + + path.reverse() + return path, 200 + + except Exception as e: + logging.exception("Error calculating shortest path") + raise \ No newline at end of file diff --git a/src/realizer/get_metrics.py b/src/realizer/get_metrics.py new file mode 100644 index 0000000000000000000000000000000000000000..d204217c737ad8d7f83ac5609d42ec18557093bc --- /dev/null +++ b/src/realizer/get_metrics.py @@ -0,0 +1,35 @@ +import logging +import asyncio +from .restconf.connectors.tfs_connector import tfs_connector +from flask import current_app + +def get_metrics(path, slice_id, controller_type): + if controller_type == "RESTCONF": + links = [] + if slice_id not in current_app.config["TELEMETRY_CACHE"]: + logging.debug(f"Telemetry for slice '{slice_id}' not in cache. Initializing.") + current_app.config["TELEMETRY_CACHE"][slice_id] = {} + for i in range(len(path)-1): + link_id = f"{path[i]}-{path[i+1]}" + links.append(link_id) + current_app.config["TELEMETRY_CACHE"][slice_id][link_id] = { + "timestamp": None, #time.time(), + "bandwidth": 0, + "latency": 0, + "services": [] + } + # Get telemetry only of the links in the shortest path: + logging.debug(f"Starting telemetry streams for slice '{slice_id}'") + connector = tfs_connector() + bg_loop = connector.get_background_loop() + coro = connector.startStreams( + current_app.config["RESTCONF_IP"], + current_app.config["SDN_SUBSCRIPTION_PERIOD"], + links, + slice_id, + current_app.config["TELEMETRY_CACHE"] + ) + future = asyncio.run_coroutine_threadsafe(coro, bg_loop) + future.result() + else: + logging.debug(f"Telemetry for slice '{slice_id}' already in cache. Skipping.") \ No newline at end of file diff --git a/src/realizer/main.py b/src/realizer/main.py index 393d0254f68b02a864e0401659d76bbaa08eb5eb..947276733c5b7b70a60242719cec361b7bfc75f8 100644 --- a/src/realizer/main.py +++ b/src/realizer/main.py @@ -17,9 +17,13 @@ import logging from .select_way import select_way from .nrp_handler import nrp_handler +from .get_metrics import get_metrics from src.utils.safe_get import safe_get +from src.planner.shortest_path import get_shortest_path +from src.realizer.restconf.connectors.tfs_connector import tfs_connector +from flask import current_app -def realizer(service, need_nrp=False, order=None, nrp=None, controller_type=None, response=None, rules = None): +def realizer(payload, need_nrp=False, order=None, nrp=None, controller_type=None, response=None, rules = None, action="CREATE"): """ Manage the slice creation workflow. @@ -39,45 +43,67 @@ def realizer(service, need_nrp=False, order=None, nrp=None, controller_type=None Returns: dict: A realization request for the specified network slice type. """ - if need_nrp: - # Perform NRP-related operations - nrp_view = nrp_handler(order, nrp) - return nrp_view - else: - # Select slice service method - if controller_type == "E2E": - if isinstance(rules, list) and len(rules) > 0: rules = rules[0] - actions = rules.get("actions", []) if (rules and not type(rules)== str) else [] + if action == "CREATE": + service = payload + if need_nrp: + # Perform NRP-related operations + nrp_view = nrp_handler(order, nrp) + return nrp_view + else: + # Select slice service method + if controller_type == "E2E": + if isinstance(rules, list) and len(rules) > 0: rules = rules[0] + actions = rules.get("actions", []) if (rules and not type(rules)== str) else [] - has_transceiver = any(a.get("type", "").startswith("XR_AGENT_ACTIVATE_TRANSCEIVER") for a in actions) - has_optical = any(a.get("type", "").startswith("PROVISION_MEDIA_CHANNEL") for a in actions) - has_l3 = any(a.get("type", "").startswith("CONFIG_VPNL3") for a in actions) - has_l2 = any(a.get("type", "").startswith("CONFIG_VPNL2") for a in actions) + has_transceiver = any(a.get("type", "").startswith("XR_AGENT_ACTIVATE_TRANSCEIVER") for a in actions) + has_optical = any(a.get("type", "").startswith("PROVISION_MEDIA_CHANNEL") for a in actions) + has_l3 = any(a.get("type", "").startswith("CONFIG_VPNL3") for a in actions) + has_l2 = any(a.get("type", "").startswith("CONFIG_VPNL2") for a in actions) - del_transceiver = any(a.get("type", "").startswith("DEACTIVATE_XR_AGENT_TRANSCEIVER") for a in actions) - del_optical = any(a.get("type", "").startswith("DEPROVISION_OPTICAL_RESOURCE") for a in actions) - del_l3 = any(a.get("type", "").startswith("REMOVE_VPNL3") for a in actions) - del_l2 = any(a.get("type", "").startswith("REMOVE_VPNL2") for a in actions) + del_transceiver = any(a.get("type", "").startswith("DEACTIVATE_XR_AGENT_TRANSCEIVER") for a in actions) + del_optical = any(a.get("type", "").startswith("DEPROVISION_OPTICAL_RESOURCE") for a in actions) + del_l3 = any(a.get("type", "").startswith("REMOVE_VPNL3") for a in actions) + del_l2 = any(a.get("type", "").startswith("REMOVE_VPNL2") for a in actions) - if has_transceiver: selected_way = "L3oWDM" - elif has_optical and has_l3: selected_way = "L3oWDM" - elif has_optical and has_l2: selected_way = "L2oWDM" - elif has_optical: selected_way = "OPTIC" - elif has_l3: selected_way = "L3VPN" - elif has_l2: selected_way = "L2VPN" + if has_transceiver: selected_way = "L3oWDM" + elif has_optical and has_l3: selected_way = "L3oWDM" + elif has_optical and has_l2: selected_way = "L2oWDM" + elif has_optical: selected_way = "OPTIC" + elif has_l3: selected_way = "L3VPN" + elif has_l2: selected_way = "L2VPN" - elif del_transceiver: selected_way = "DEL_L3oWDM" - elif del_optical and del_l3: selected_way = "DEL_L3oWDM" - elif del_optical and del_l2: selected_way = "DEL_L2oWDM" - elif del_optical: selected_way = "DEL_OPTIC" - elif del_l3: selected_way = "DEL_L3VPN" - elif del_l2: selected_way = "DEL_L2VPN" + elif del_transceiver: selected_way = "DEL_L3oWDM" + elif del_optical and del_l3: selected_way = "DEL_L3oWDM" + elif del_optical and del_l2: selected_way = "DEL_L2oWDM" + elif del_optical: selected_way = "DEL_OPTIC" + elif del_l3: selected_way = "DEL_L3VPN" + elif del_l2: selected_way = "DEL_L2VPN" + else: + logging.warning("Cannot determine the realization way from rules. Skipping request.") + return None + way = selected_way + else: + way = service.get("way", None) or safe_get(service, ['ietf-network-slice-service:network-slice-services', 'slice-service', 0, 'service-tags', 'tag-type', 0, 'tag-type-value', 0]) + logging.info(f"Selected way: {way}") + request = select_way(controller=controller_type, way=way, ietf_intent=service, response=response, rules = rules) + return request + elif action == "MONITOR": + logging.debug("Realizer action: MONITOR") + slice_id = payload.get("slice_id", None) + sdps = payload.get("sdps", None) + sdps_object = safe_get(sdps, ['network-slice-services', 'slice-service']) + sdps_object = sdps_object[slice_id]['sdps']['sdp'] + sdp_ids = [sdp["id"] for sdp in sdps_object] + # Retrieve Network topology + topology, response = tfs_connector().get_network_topology(current_app.config["RESTCONF_IP"],slice_id) + if response == 200: + logging.debug(f"Retrieved topology for slice '{slice_id}'") + # Get shortest path + path, response = get_shortest_path(topology, sdp_ids[0], sdp_ids[1]) + if response == 200: + logging.debug(f"Retrieved shortest path for slice '{slice_id}': {path}") + get_metrics(path, slice_id, controller_type) else: - logging.warning("Cannot determine the realization way from rules. Skipping request.") - return None - way = selected_way + raise Exception("Error: Shortest path not retrieved ") else: - way = service.get("way", None) or safe_get(service, ['ietf-network-slice-service:network-slice-services', 'slice-service', 0, 'service-tags', 'tag-type', 0, 'tag-type-value', 0]) - logging.info(f"Selected way: {way}") - request = select_way(controller=controller_type, way=way, ietf_intent=service, response=response, rules = rules) - return request + raise Exception("Error: Topology not retrieved") \ No newline at end of file diff --git a/src/realizer/restconf/connectors/tfs_connector.py b/src/realizer/restconf/connectors/tfs_connector.py index a4f213944d9bec4edfab96c9d79925f305208558..863c02c2444924313f4e636c5228dffcebece47d 100644 --- a/src/realizer/restconf/connectors/tfs_connector.py +++ b/src/realizer/restconf/connectors/tfs_connector.py @@ -14,13 +14,38 @@ # This file includes original contributions from Telefonica Innovación Digital S.L. -import logging, requests, json -from src.config.constants import NBI_L2_PATH, NBI_L3_PATH +import logging, requests, json, aiohttp, asyncio, threading +from src.config.constants import NBI_L2_PATH, NBI_L3_PATH, NBI_IETF_NETWORKS_PATH, NBI_SIMAP_SUSCRIPTION_PATH +from typing import Dict, Tuple + +# Temp until moving to .env +SDN_SUBSCRIPTION_PERIOD = 10 # seconds +SDN_SUBS_INACTIVITY_THRESHOLD = 30 # seconds class tfs_connector(): """ Helper class to interact with TeraFlowSDN Northbound Interface (NBI) and WebUI. """ + _loop = None + _thread = None + _session = None + + @classmethod + def get_background_loop(cls): + if cls._loop is None: + cls._loop = asyncio.new_event_loop() + cls._thread = threading.Thread(target=cls._run_loop, args=(cls._loop,), daemon=True) + cls._thread.start() + return cls._loop + + @staticmethod + def _run_loop(loop): + asyncio.set_event_loop(loop) + loop.run_forever() + + def __init__(self): + pass + def webui_post(self, tfs_ip, service): """ Post service descriptor to TFS WebUI. @@ -100,4 +125,141 @@ class tfs_connector(): response.raise_for_status() logging.debug('Service deleted successfully') logging.debug("Http response: %s",response.text) - return response \ No newline at end of file + return response + + def get_network_topology(self, tfs_ip: str, slice_id: str) -> Tuple[Dict[str, any], int]: + user="admin" + password="admin" + url = f'http://{user}:{password}@{tfs_ip}' + url = url + f'/{NBI_IETF_NETWORKS_PATH}' + # MOCKED TOPOLOGY + response = requests.get(url, timeout=60) + response.raise_for_status() + network_raw = response.json() + network = next((n["ietf-network:networks"]["network"][0] for n in network_raw if n["ietf-network:networks"]["network"][0]["network-id"] == "urn:tfs:network:admin"), None) + logging.debug(f"Retrieved topology for slice '{slice_id}': {network}") + + return network, 200 + + # --- SDN STREAMS --- + + async def get_session(self): + # Crea la sesión si no existe en el loop de background + if tfs_connector._session is None or tfs_connector._session.closed: + tfs_connector._session = aiohttp.ClientSession() + return tfs_connector._session + + async def close_session(self): + # Llama a esto al apagar tu app + if tfs_connector._session: + await tfs_connector._session.close() + + def _extract_stream_uri(self, payload, base_url): + if not isinstance(payload, dict): + return None + + if "uri" in payload and isinstance(payload["uri"], str): + uri = payload["uri"].strip() + if uri.startswith(("http://", "https://")): + return uri + if base_url: + return f"{base_url.rstrip('/')}/{uri.lstrip('/')}" + return uri + + for key in ("ietf-subscribed-notifications:subscription-result", "subscription-result", "stream", "ietf-subscribed-notifications:stream"): + if key in payload: + uri = self._extract_stream_uri(payload[key], base_url) + if uri: + return uri + + for value in payload.values(): + if isinstance(value, dict): + uri = self._extract_stream_uri(value, base_url) + if uri: + return uri + + return None + + async def listenStream(self, session, link_id, slice_id, stream_url, telemetry_cache): + logging.debug(f"Listening to stream for link {link_id} in slice {slice_id} at {stream_url}") + try: + async with session.get( + stream_url, + auth=aiohttp.BasicAuth("admin", "admin"), + timeout=None, + headers = { + "Accept": "text/event-stream", + "Cache-Control": "no-cache" + } + ) as resp: + logging.debug(f"Connection established for link {link_id}") + buffer = "" + + async for chunk in resp.content.iter_any(): + buffer += chunk.decode() + + while "\n\n" in buffer: + event_block, buffer = buffer.split("\n\n", 1) + + for line in event_block.split("\n"): + if line.startswith("data:"): + json_str = line.replace("data:", "").strip() + + try: + data = json.loads(json_str) + logging.debug(f"Received data for link {link_id}: {data}") + + telemetry = data["notification"]["push-update"]["datastore-contents"]["simap-telemetry:simap-telemetry"] + bw = telemetry.get("bandwidth-utilization", "N/A") + latency = telemetry.get("latency", "N/A") + telemetry_cache.setdefault(slice_id, {})[link_id] = { + "timestamp": data["notification"]["eventTime"], + "bandwidth": float(bw) if bw != "N/A" else bw, + "latency": float(latency) if latency != "N/A" else latency, + "services": telemetry.get("related-service-ids", []) + } + except json.JSONDecodeError: + logging.error(f"Error parsing JSON for link {link_id}: {json_str}") + except Exception as e: + logging.error(f"Error processing data for link {link_id}: {e}") + except aiohttp.ClientError as e: + logging.error(f"Connection error with stream for link {link_id}: {e}") + except Exception as e: + logging.error(f"Unexpected error in listenStream for link {link_id}: {e}") + + logging.info(f"Stopped listening to stream for link {link_id} in slice {slice_id}") + + async def startStreams(self, tfs_ip, suscription_period, links, slice_id, telemetry_cache): + #async with self.get_session() as session: #timeout = aiohttp.ClientTimeout(total=10) + session = await self.get_session() + tasks = [] + base_url = f'http://{tfs_ip}' + url = f'{base_url}{NBI_SIMAP_SUSCRIPTION_PATH}' + + for link in links: + sub_resp = await session.post(url, json={ + "ietf-subscribed-notifications:input": { + "datastore": "operational", + "ietf-yang-push:datastore-xpath-filter": f"/ietf-network:networks/network=admin/ietf-network-topology:link={link}/simap-telemetry:simap-telemetry", + "ietf-yang-push:periodic": { + "ietf-yang-push:period": suscription_period + } + } + }, + auth=aiohttp.BasicAuth("admin", "admin") + ) + sub_resp.raise_for_status() + sub_data = await sub_resp.json(content_type=None) + stream_url = self._extract_stream_uri(sub_data, base_url) + + if stream_url: + tasks.append( + asyncio.create_task( + self.listenStream(session, link, slice_id, stream_url, telemetry_cache) + ) + ) + logging.debug("Created telemetry listener for link '%s' in slice '%s' at %s", link, slice_id, stream_url) + else: + logging.warning("Subscription for link '%s' did not return a stream URI. Response: %s", link, sub_data) + + logging.debug(f"Telemetry cache for slice '{slice_id}': {telemetry_cache.get(slice_id)}") diff --git a/src/tests/collections/NSC.postman_collection.json b/src/tests/collections/NSC.postman_collection.json new file mode 100644 index 0000000000000000000000000000000000000000..568e796301464af2db5ad93e5f507ad75dc0b055 --- /dev/null +++ b/src/tests/collections/NSC.postman_collection.json @@ -0,0 +1,560 @@ +{ + "info": { + "_postman_id": "420789ec-9203-4f33-9511-2100c7a934f0", + "name": "NSC", + "schema": "https://schema.getpostman.com/json/collection/v2.1.0/collection.json" + }, + "item": [ + { + "name": "Operations", + "item": [ + { + "name": "Clients", + "item": [ + { + "name": "Create Client", + "request": { + "auth": { + "type": "basic", + "basic": [ + { + "key": "password", + "value": "admin", + "type": "string" + }, + { + "key": "username", + "value": "admin", + "type": "string" + } + ] + }, + "method": "POST", + "header": [], + "url": { + "raw": "{{HOST}}/restconf/telemetry/operations/clients/{{CLIENT_ID}}", + "host": [ + "{{HOST}}" + ], + "path": [ + "restconf", + "telemetry", + "operations", + "clients", + "{{CLIENT_ID}}" + ] + } + }, + "response": [] + }, + { + "name": "Get All Clients", + "request": { + "auth": { + "type": "basic", + "basic": [ + { + "key": "password", + "value": "admin", + "type": "string" + }, + { + "key": "username", + "value": "admin", + "type": "string" + } + ] + }, + "method": "GET", + "header": [], + "url": { + "raw": "{{HOST}}/restconf/telemetry/operations/clients", + "host": [ + "{{HOST}}" + ], + "path": [ + "restconf", + "telemetry", + "operations", + "clients" + ] + } + }, + "response": [] + }, + { + "name": "Delete Client", + "request": { + "auth": { + "type": "basic", + "basic": [ + { + "key": "password", + "value": "admin", + "type": "string" + }, + { + "key": "username", + "value": "admin", + "type": "string" + } + ] + }, + "method": "DELETE", + "header": [], + "url": { + "raw": "{{HOST}}/restconf/telemetry/operations/clients/{{CLIENT_ID}}", + "host": [ + "{{HOST}}" + ], + "path": [ + "restconf", + "telemetry", + "operations", + "clients", + "{{CLIENT_ID}}" + ] + } + }, + "response": [] + } + ] + }, + { + "name": "Subscriptions", + "item": [ + { + "name": "Slices", + "item": [ + { + "name": "Create Slice Subscription", + "request": { + "auth": { + "type": "basic", + "basic": [ + { + "key": "password", + "value": "admin", + "type": "string" + }, + { + "key": "username", + "value": "admin", + "type": "string" + } + ] + }, + "method": "POST", + "header": [], + "body": { + "mode": "raw", + "raw": "{\n \"slice_id\": \"{{SLICE_ID}}\",\n \"frequency\": 10\n}", + "options": { + "raw": { + "language": "json" + } + } + }, + "url": { + "raw": "{{HOST}}/restconf/operations/telemetry/subscription/{{CLIENT_ID}}", + "host": [ + "{{HOST}}" + ], + "path": [ + "restconf", + "operations", + "telemetry", + "subscription", + "{{CLIENT_ID}}" + ] + } + }, + "response": [] + }, + { + "name": "Get Slice Subscription", + "request": { + "auth": { + "type": "basic", + "basic": [ + { + "key": "password", + "value": "admin", + "type": "string" + }, + { + "key": "username", + "value": "admin", + "type": "string" + } + ] + }, + "method": "GET", + "header": [], + "url": { + "raw": "{{HOST}}/restconf/operations/telemetry/subscription/{{CLIENT_ID}}/slice/{{SLICE_ID}}", + "host": [ + "{{HOST}}" + ], + "path": [ + "restconf", + "operations", + "telemetry", + "subscription", + "{{CLIENT_ID}}", + "slice", + "{{SLICE_ID}}" + ] + } + }, + "response": [] + }, + { + "name": "Update Slice Subscription", + "request": { + "auth": { + "type": "basic", + "basic": [ + { + "key": "password", + "value": "admin", + "type": "string" + }, + { + "key": "username", + "value": "admin", + "type": "string" + } + ] + }, + "method": "PUT", + "header": [], + "body": { + "mode": "raw", + "raw": "{\n \"frequency\": 5\n}", + "options": { + "raw": { + "language": "json" + } + } + }, + "url": { + "raw": "{{HOST}}/restconf/operations/telemetry/subscription/{{CLIENT_ID}}/slice/{{SLICE_ID}}", + "host": [ + "{{HOST}}" + ], + "path": [ + "restconf", + "operations", + "telemetry", + "subscription", + "{{CLIENT_ID}}", + "slice", + "{{SLICE_ID}}" + ] + } + }, + "response": [] + }, + { + "name": "Delete Slice Subscription", + "request": { + "auth": { + "type": "basic", + "basic": [ + { + "key": "password", + "value": "admin", + "type": "string" + }, + { + "key": "username", + "value": "admin", + "type": "string" + } + ] + }, + "method": "DELETE", + "header": [], + "url": { + "raw": "{{HOST}}/restconf/operations/telemetry/subscription/{{CLIENT_ID}}/slice/{{SLICE_ID}}", + "host": [ + "{{HOST}}" + ], + "path": [ + "restconf", + "operations", + "telemetry", + "subscription", + "{{CLIENT_ID}}", + "slice", + "{{SLICE_ID}}" + ] + } + }, + "response": [] + }, + { + "name": "Stream Telemetry Subscription", + "request": { + "auth": { + "type": "basic", + "basic": [ + { + "key": "password", + "value": "admin", + "type": "string" + }, + { + "key": "username", + "value": "admin", + "type": "string" + } + ] + }, + "method": "GET", + "header": [], + "url": { + "raw": "{{HOST}}/restconf/operations/telemetry/subscription/{{CLIENT_ID}}/slice/{{SLICE_ID}}/stream", + "host": [ + "{{HOST}}" + ], + "path": [ + "restconf", + "operations", + "telemetry", + "subscription", + "{{CLIENT_ID}}", + "slice", + "{{SLICE_ID}}", + "stream" + ] + } + }, + "response": [] + } + ] + }, + { + "name": "All subscriptions of a client", + "request": { + "auth": { + "type": "basic", + "basic": [ + { + "key": "password", + "value": "admin", + "type": "string" + }, + { + "key": "username", + "value": "admin", + "type": "string" + } + ] + }, + "method": "GET", + "header": [], + "url": { + "raw": "{{HOST}}/restconf/operations/telemetry/subscription/{{CLIENT_ID}}", + "host": [ + "{{HOST}}" + ], + "path": [ + "restconf", + "operations", + "telemetry", + "subscription", + "{{CLIENT_ID}}" + ] + } + }, + "response": [] + }, + { + "name": "Delete all subscriptions of a client", + "request": { + "auth": { + "type": "basic", + "basic": [ + { + "key": "password", + "value": "admin", + "type": "string" + }, + { + "key": "username", + "value": "admin", + "type": "string" + } + ] + }, + "method": "DELETE", + "header": [], + "url": { + "raw": "{{HOST}}/restconf/operations/telemetry/subscription/{{CLIENT_ID}}", + "host": [ + "{{HOST}}" + ], + "path": [ + "restconf", + "operations", + "telemetry", + "subscription", + "{{CLIENT_ID}}" + ] + } + }, + "response": [] + } + ] + } + ] + }, + { + "name": "slices", + "item": [ + { + "name": "Create Slice", + "request": { + "auth": { + "type": "basic", + "basic": [ + { + "key": "password", + "value": "admin", + "type": "string" + }, + { + "key": "username", + "value": "admin", + "type": "string" + } + ] + }, + "method": "POST", + "header": [], + "body": {}, + "url": { + "raw": "{{HOST}}/restconf/data/ietf-network-slice-service:network-slice-services", + "host": [ + "{{HOST}}" + ], + "path": [ + "restconf", + "data", + "ietf-network-slice-service:network-slice-services" + ] + } + }, + "response": [] + }, + { + "name": "Get Slices", + "request": { + "auth": { + "type": "basic", + "basic": [ + { + "key": "password", + "value": "admin", + "type": "string" + }, + { + "key": "username", + "value": "admin", + "type": "string" + } + ] + }, + "method": "GET", + "header": [], + "url": { + "raw": "{{HOST}}/restconf/data/ietf-network-slice-service:network-slice-services", + "host": [ + "{{HOST}}" + ], + "path": [ + "restconf", + "data", + "ietf-network-slice-service:network-slice-services" + ] + } + }, + "response": [] + } + ] + }, + { + "name": "Telemetry of All Slices", + "request": { + "auth": { + "type": "basic", + "basic": [ + { + "key": "password", + "value": "admin", + "type": "string" + }, + { + "key": "username", + "value": "admin", + "type": "string" + } + ] + }, + "method": "GET", + "header": [], + "url": { + "raw": "{{HOST}}/restconf/operations/telemetry/slice", + "host": [ + "{{HOST}}" + ], + "path": [ + "restconf", + "operations", + "telemetry", + "slice" + ] + } + }, + "response": [] + }, + { + "name": "Telemetry of a Specific Slice", + "request": { + "auth": { + "type": "basic", + "basic": [ + { + "key": "password", + "value": "admin", + "type": "string" + }, + { + "key": "username", + "value": "admin", + "type": "string" + } + ] + }, + "method": "GET", + "header": [], + "url": { + "raw": "{{HOST}}/restconf/operations/telemetry/slice/{{SLICE_ID}}", + "host": [ + "{{HOST}}" + ], + "path": [ + "restconf", + "operations", + "telemetry", + "slice", + "{{SLICE_ID}}" + ] + } + }, + "response": [] + } + ] +} \ No newline at end of file diff --git a/src/tests/collections/NSC.postman_environment.json b/src/tests/collections/NSC.postman_environment.json new file mode 100644 index 0000000000000000000000000000000000000000..34117f0063162fe508114b7d13c7288f2cb60c89 --- /dev/null +++ b/src/tests/collections/NSC.postman_environment.json @@ -0,0 +1,28 @@ +{ + "id": "fb33daed-2c4b-4c85-a9b2-4d338123ed0b", + "name": "NSC", + "values": [ + { + "key": "HOST", + "value": "http://192.168.27.189:8085/nsc", + "type": "default", + "enabled": true + }, + { + "key": "CLIENT_ID", + "value": "test-client", + "type": "default", + "enabled": true + }, + { + "key": "SLICE_ID", + "value": "simap-slice", + "type": "default", + "enabled": true + } + ], + "color": null, + "_postman_variable_scope": "environment", + "_postman_exported_at": "2026-03-27T10:00:42.609Z", + "_postman_exported_using": "Postman/12.3.5" +} \ No newline at end of file diff --git a/swagger/restconf_namespace.py b/swagger/restconf_namespace.py index 96829472cba6ecc6eb6dbfe9bc1eaa4deec40b6b..079faf04947f392a78b106e2dcab236721081a58 100644 --- a/swagger/restconf_namespace.py +++ b/swagger/restconf_namespace.py @@ -14,8 +14,9 @@ # This file is an original contribution from Telefonica Innovación Digital S.L. -from flask import request -from flask_restx import Namespace, Resource, reqparse +import logging +from flask import request, Response, request, current_app, stream_with_context +from flask_restx import Namespace, Resource, fields from src.main import NSController from src.api.main import Api from swagger.models.create_models_restconf import create_ietf_network_slice_nbi_yang_model @@ -177,7 +178,6 @@ class SdpList(Resource): @restconf_ns.doc(summary="Create Slice Service SDPs") @restconf_ns.expect(sdp_model) - #@restconf_ns.expect(sdp_model) @restconf_ns.response(201, "SDP created", slice_response_model) @restconf_ns.response(409, "SDP already exists") def post(self, slice_service_id): @@ -211,4 +211,184 @@ class Sdp(Resource): @restconf_ns.response(204, "SDP deleted") def delete(self, slice_service_id, sdp_id): controller = NSController(controller_type="RESTCONF") - return Api(controller).delete_sdps(slice_service_id, sdp_id) \ No newline at end of file + return Api(controller).delete_sdps(slice_service_id, sdp_id) + +# ================================================================================================================================================ +@restconf_ns.route("/operations/telemetry/client") +class ClientsList(Resource): + @restconf_ns.doc(description="Gets the list of all registered clients") + @restconf_ns.response(200, "Clients retrieved") + @restconf_ns.response(404, "No clients found") + @restconf_ns.response(500, "Internal server error") + def get(self): + return Api(NSController(controller_type="RESTCONF")).get_clients() + + @restconf_ns.doc(description="Deletes all clients") + @restconf_ns.response(204, "Clients deleted") + @restconf_ns.response(500, "Internal server error") + def delete(self): + return Api(NSController(controller_type="RESTCONF")).delete_clients() + +@restconf_ns.route("/operations/telemetry/client/") +class Clients(Resource): + @restconf_ns.response(200, "Client retrieved") + @restconf_ns.response(404, "No client found") + @restconf_ns.response(500, "Internal server error") + def get(self, client_id): + return Api(NSController(controller_type="RESTCONF")).get_clients(client_id) + + @restconf_ns.doc(description="Creates a new client") + @restconf_ns.response(201, "Client created", model=restconf_ns.model("ClientModel", { + "client_id": fields.String(required=True, description="Client ID"), + })) + @restconf_ns.response(500, "Internal server error") + def post(self, client_id): + return Api(NSController(controller_type="RESTCONF")).add_client(client_id) + + @restconf_ns.doc(description="Deletes a given client") + @restconf_ns.response(204, "Client deleted") + @restconf_ns.response(404, "No client found") + @restconf_ns.response(500, "Internal server error") + def delete(self, client_id): + return Api(NSController(controller_type="RESTCONF")).delete_clients(client_id) + +@restconf_ns.route("/operations/telemetry/subscription/") +class Subscriptions(Resource): + @restconf_ns.doc( + description="Retrieves all telemetry subscriptions of a given client.", + params={"client_id": "Client ID."} + ) + @restconf_ns.response(200, "Subscriptions successfully retrieved") + @restconf_ns.response(500, "Internal server error") + def get(self, client_id): + api = Api(NSController(controller_type="RESTCONF")) + return current_app.ensure_sync(api.get_subscriptions)(client_id) + + @restconf_ns.doc( + description="Creates a new telemetry subscription for a given client.", + params={"client_id": "Client ID."} + ) + @restconf_ns.response(201, "Subscription successfully created") + @restconf_ns.response(400, "Invalid request") + @restconf_ns.response(500, "Internal server error") + @restconf_ns.expect(restconf_ns.model("TelemetrySubscriptionModel", { + "slice_id": fields.String(required=True, description="Network Slice ID"), + "frequency": fields.Integer(required=True, description="Telemetry push frequency in seconds") + }), validate=True) + def post(self, client_id): + data = request.json + slice_id = data.get("slice_id") + frequency = data.get("frequency") + + api = Api(NSController(controller_type="RESTCONF")) + return current_app.ensure_sync(api.add_subscription)( + client_id, + slice_id, + frequency + ) + + @restconf_ns.doc( + description="Deletes all telemetry subscriptions of a given client.", + params={"client_id": "Client ID."} + ) + @restconf_ns.response(204, "Subscriptions successfully deleted") + @restconf_ns.response(500, "Internal server error") + def delete(self, client_id): + return Api(NSController(controller_type="RESTCONF")).delete_subscriptions(client_id) + + +@restconf_ns.route("/operations/telemetry/subscription//slice/") +class SliceSubscriptions(Resource): + @restconf_ns.doc( + description="Retrieves a specific telemetry subscription of a given client.", + params={ + "client_id": "Client ID.", + "slice_id": "Network Slice ID." + } + ) + @restconf_ns.response(200, "Subscription successfully retrieved") + @restconf_ns.response(500, "Internal server error") + def get(self, client_id, slice_id): + api = Api(NSController(controller_type="RESTCONF")) + return current_app.ensure_sync(api.get_subscriptions)(client_id, slice_id) + + @restconf_ns.doc( + description="Updates a telemetry subscription for a given client.", + params={ + "client_id": "Client ID.", + "slice_id": "Network Slice ID." + } + ) + @restconf_ns.response(201, "Subscription successfully updated") + @restconf_ns.response(400, "Invalid request") + @restconf_ns.response(500, "Internal server error") + @restconf_ns.expect(restconf_ns.model("TelemetryUpdateSubscriptionModel", { + "frequency": fields.Integer(required=True, description="Telemetry push frequency in seconds") + }), validate=True) + def put(self, client_id, slice_id): + data = request.json + frequency = data.get("frequency") + + api = Api(NSController(controller_type="RESTCONF")) + return current_app.ensure_sync(api.update_subscription)( + client_id, + slice_id, + frequency + ) + + @restconf_ns.doc( + description="Deletes a specific telemetry subscription of a given client.", + params={ + "client_id": "Client ID.", + "slice_id": "Network Slice ID." + } + ) + @restconf_ns.response(204, "Subscription successfully deleted") + @restconf_ns.response(500, "Internal server error") + def delete(self, client_id, slice_id): + return Api(NSController(controller_type="RESTCONF")).delete_subscriptions(client_id, slice_id) + + +@restconf_ns.route("/operations/telemetry/subscription//slice//stream") +class SliceSubscriptionsStream(Resource): + @restconf_ns.doc( + description="Opens a telemetry stream of a given subscription of a certain client", + params={ + "client_id": "Client ID.", + "slice_id": "Network Slice ID." + } + ) + @restconf_ns.response(200, "Telemetry stream opened") + @restconf_ns.response(500, "Internal server error") + def get(self, client_id, slice_id): + api = Api(NSController(controller_type="RESTCONF")) + return Response( + stream_with_context( + api.sync_stream( + api.stream_slice_subscription, + client_id, + slice_id + ) + ), + mimetype="text/event-stream" + ) + +@restconf_ns.route("/operations/telemetry/slice") +class TelemetryList(Resource): + @restconf_ns.doc(description="Gets latest telemetry of all slices.") + @restconf_ns.response(200, "Telemetry retrieved") + @restconf_ns.response(500, "Internal server error") + def get(self): + controller = NSController(controller_type="RESTCONF") + logging.info("Retrieving latest telemetry for all slices") + return current_app.ensure_sync(Api(controller).get_telemetry)() +@restconf_ns.route("/operations/telemetry/slice/") +class Telemetry(Resource): + @restconf_ns.doc(description="Gets latest telemetry of a certain network slice") + @restconf_ns.response(200, "Telemetry retrieved") + @restconf_ns.response(404, "Slice not found") + @restconf_ns.response(500, "Internal server error") + def get(self, slice_id): + controller = NSController(controller_type="RESTCONF") + logging.info(f"Retrieving latest telemetry for slice '{slice_id}'") + return current_app.ensure_sync(Api(controller).get_telemetry)(slice_id=slice_id) \ No newline at end of file