Merge branch 'feat/37-tid-new-alert-manager-for-e2e-optical-slices' of... (30b378f1) · Commits · TFS / nsc

app.py

+2 −0

Original line number	Diff line number	Diff line
		@@ -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.alert_db import init_db as init_alert

		# Paths that do not require authentication (Swagger UI and its static assets)
		# /nsc → Swagger UI
		@@ -58,6 +59,7 @@ def create_app():
		"""Create Flask application with configured API and namespaces."""
		init_slice()
		init_service()
		init_alert()
		app = Flask(__name__)
		app = create_config(app)
		CORS(app)

src/api/main.py

+225 −0

Original line number	Diff line number	Diff line
		@@ -22,6 +22,8 @@ from src.database.service_db import delete_by_slice_id, get_data_by_slice_id
		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
		import src.database.alert_db as alert_db


		class Api:
		def __init__(self, slice_service):
		@@ -61,6 +63,229 @@ class Api:
		# Handle unexpected errors
		return send_response(False, code=500, message=str(e))

		def receive_alert(self, alert_data):
		"""
		Receive and process an alert.

		Args:
		alert_data (dict): The alert payload

		Returns:
		Result of the operation
		"""
		try:
		logging.info(f"Alert received: {alert_data}")
		# Extract uuid if exists
		context = alert_data.get("tapi-notification:notification-context", [])
		alert_id = None
		service_id = None
		subscription_id = None
		if context and isinstance(context, list):
		notification = context[0].get("tapi-notification:notification", {})
		alert_id = notification.get("uuid")
		subscription_id = alert_id
		additional_info = notification.get("additional-info", {})
		service_id = additional_info.get("service-id")

		if not alert_id:
		return send_response(False, code=400, message="UUID not found in alert data")

		# Save alert to DB
		alert_db.save_alert(alert_id, alert_data)

		# Process intent modification based on alert
		slice_id = None
		slice_info = None

		logging.info(f"Looking up intent for subscription_id: {subscription_id}, service_id: {service_id}")

		if subscription_id:
		try:
		import src.database.db as db
		mapped_slice_id = db.get_slice_id_by_subscription(subscription_id)
		if mapped_slice_id:
		slice_id = mapped_slice_id
		logging.info(f"Found slice_id {slice_id} mapped to subscription_id {subscription_id}")
		except Exception as e:
		logging.info(f"Subscription mapping lookup failed: {e}")

		if not slice_id and service_id:
		# 1. Try to find slice_id from service_db
		try:
		import src.database.service_db as service_db
		service_info = service_db.get_data(service_id)
		slice_id = service_info.get("slice_id")
		logging.info(f"Found slice_id {slice_id} in service_db for service_id {service_id}")
		except Exception as e:
		logging.info(f"service_db lookup failed: {e}")
		slice_id = service_id

		if slice_id:
		# 2. Try to get slice data from db
		try:
		import src.database.db as db
		slice_info = db.get_data(slice_id)
		logging.info(f"Found slice_info in db by slice_id {slice_id}")
		except Exception as e:
		logging.info(f"db lookup by slice_id {slice_id} failed: {e}")

		# Fallback: if not found by ID, look up any existing slice in the DB
		if not slice_info:
		try:
		import src.database.db as db
		slices = db.get_all_data()
		logging.info(f"Slices in db: {[s.get('slice_id') for s in slices]}")
		for s in slices:
		if s.get("slice_id") == slice_id:
		slice_info = s
		break
		if not slice_info and slices:
		# Default to the first/only slice if there's any
		slice_info = slices[0]
		logging.info(f"Defaulted to first slice from db: {slice_info.get('slice_id')}")
		except Exception as e:
		logging.info(f"db get_all_data lookup failed: {e}")

		# Fallback: read from /home/llmserver/tfs-nsc/intent.json if DB is empty
		if not slice_info:
		import os
		import json
		fallback_path = "/home/llmserver/tfs-nsc/intent.json"
		if os.path.exists(fallback_path):
		try:
		with open(fallback_path, "r") as f:
		intent_data = json.load(f)
		slice_info = {"slice_id": slice_id or "slice", "intent": intent_data}
		logging.info("Loaded fallback intent from intent.json")
		except Exception as e:
		logging.error(f"Failed to read fallback intent.json: {e}")

		if slice_info:
		intent = slice_info.get("intent")
		curr_slice_id = slice_info.get("slice_id")
		logging.info(f"Processing intent for slice {curr_slice_id}")

		if intent:
		nss = intent.get("ietf-network-slice-service:network-slice-services", {})
		slice_services = nss.get("slice-service", [])
		modified = False

		for service in slice_services:
		# Get all SDP IDs in this slice service
		sdp_list = service.get("sdps", {}).get("sdp", [])
		sdp_ids = [sdp.get("id") for sdp in sdp_list if sdp.get("id")]

		connection_groups = service.get("connection-groups", {}).get("connection-group", [])
		for cg in connection_groups:
		connectivity_constructs = cg.get("connectivity-construct", [])
		for cc in connectivity_constructs:
		p2mp_sender = cc.get("p2mp-sender-sdp")
		p2mp_receivers = cc.get("p2mp-receiver-sdp", [])

		logging.info(f"sdp_ids: {sdp_ids}, p2mp_sender: {p2mp_sender}, p2mp_receivers: {p2mp_receivers}")

		# Find alternative receiver endpoints
		other_endpoints = [
		sdp_id for sdp_id in sdp_ids
		if sdp_id != p2mp_sender and sdp_id not in p2mp_receivers
		]

		old_receiver = None
		new_receiver = None

		if other_endpoints and p2mp_receivers:
		if len(p2mp_receivers) >= 2:
		old_receiver = p2mp_receivers[1]
		new_receiver = other_endpoints[0]
		cc["p2mp-receiver-sdp"] = [p2mp_receivers[0], new_receiver]
		else:
		old_receiver = p2mp_receivers[0]
		new_receiver = other_endpoints[0]
		cc["p2mp-receiver-sdp"] = [new_receiver]
		modified = True

		# Log ORIGEN and DESTINO
		logging.info(f"ORIGEN: {p2mp_sender} DESTINO: {new_receiver}")
		print(f"ORIGEN: {p2mp_sender} DESTINO: {new_receiver}", flush=True)
		else:
		logging.warning("No alternative receiver endpoints found to swap.")

		if p2mp_receivers:
		old_service_id = f"{p2mp_sender}_to_{','.join(p2mp_receivers)}"
		else:
		old_service_id = None

		if modified:
		# Re-apply the modified intent via slice_service.nsc
		try:
		self.slice_service.nsc(intent, curr_slice_id, old_service_id=old_service_id)
		logging.info(f"Slice {curr_slice_id} updated successfully following alert.")
		except Exception as e:
		logging.error(f"Failed to update slice configuration: {e}")
		else:
		logging.warning("No slice intent found to process alert.")

		return send_response(
		True,
		code=201,
		message="Alert processed and saved successfully",
		data=alert_data
		)
		except Exception as e:
		return send_response(False, code=500, message=str(e))

		def get_alerts(self, alert_id=None):
		"""
		Retrieve alert(s).
		"""
		try:
		if alert_id:
		try:
		data = alert_db.get_alert(alert_id)
		return data, 200
		except ValueError as e:
		return send_response(False, code=404, message=str(e))
		else:
		data = alert_db.get_all_alerts()
		return data, 200
		except Exception as e:
		return send_response(False, code=500, message=str(e))

		def modify_alert(self, alert_id, alert_data):
		"""
		Modify/update an alert.
		"""
		try:
		try:
		alert_db.update_alert(alert_id, alert_data)
		return send_response(
		True,
		code=200,
		message="Alert updated successfully",
		data=alert_data
		)
		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_alerts(self, alert_id=None):
		"""
		Delete alert(s).
		"""
		try:
		if alert_id:
		try:
		alert_db.delete_alert(alert_id)
		return {}, 204
		except ValueError as e:
		return send_response(False, code=404, message=str(e))
		else:
		alert_db.delete_all_alerts()
		return {}, 204
		except Exception as e:
		return send_response(False, code=500, message=str(e))

		def get_flows(self,slice_id=None):
		"""
		Retrieve transport network slice information.

src/config/config.py

+2 −0

Original line number	Diff line number	Diff line
		@@ -46,6 +46,8 @@ def create_config(app: Flask):
		app.config["PCE_EXTERNAL"] = os.getenv("PCE_EXTERNAL", "false").lower() == "true"
		app.config["HRAT_IP"] = os.getenv("HRAT_IP", "192.168.1.143")
		app.config["E2E_OPTICAL_IP"] = os.getenv("E2E_OPTICAL_IP", "127.0.0.1")
		app.config["SUBSCRIBE_ALERTS"] = os.getenv("SUBSCRIBE_ALERTS", "false").lower() == "true"
		app.config["SUBSCRIBE_ALERTS_URL"] = os.getenv("SUBSCRIBE_ALERTS_URL", "")

		# Realizer
		app.config["DUMMY_MODE"] = os.getenv("DUMMY_MODE", "true").lower() == "true"

src/database/alert_db.py

0 → 100644

+96 −0

Original line number	Diff line number	Diff line
		# 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.

		import sqlite3
		import json
		import logging

		DB_NAME = "alert.db"

		def init_db():
		"""Initialize the SQLite database for alerts."""
		conn = sqlite3.connect(DB_NAME)
		cursor = conn.cursor()
		cursor.execute("""
		CREATE TABLE IF NOT EXISTS alert (
		alert_id TEXT PRIMARY KEY,
		data TEXT NOT NULL
		)
		""")
		conn.commit()
		conn.close()

		def save_alert(alert_id: str, data_dict: dict):
		"""Save an alert to the database."""
		data_str = json.dumps(data_dict)
		conn = sqlite3.connect(DB_NAME)
		cursor = conn.cursor()
		try:
		cursor.execute("INSERT OR REPLACE INTO alert (alert_id, data) VALUES (?, ?)", (alert_id, data_str))
		conn.commit()
		except sqlite3.Error as e:
		raise ValueError(f"Database error: {e}")
		finally:
		conn.close()

		def update_alert(alert_id: str, data_dict: dict):
		"""Update an existing alert."""
		data_str = json.dumps(data_dict)
		conn = sqlite3.connect(DB_NAME)
		cursor = conn.cursor()
		cursor.execute("UPDATE alert SET data = ? WHERE alert_id = ?", (data_str, alert_id))
		if cursor.rowcount == 0:
		conn.close()
		raise ValueError(f"No alert found with ID '{alert_id}' to update.")
		conn.commit()
		conn.close()

		def delete_alert(alert_id: str):
		"""Delete a specific alert from the database."""
		conn = sqlite3.connect(DB_NAME)
		cursor = conn.cursor()
		cursor.execute("DELETE FROM alert WHERE alert_id = ?", (alert_id,))
		if cursor.rowcount == 0:
		conn.close()
		raise ValueError(f"No alert found with ID '{alert_id}' to delete.")
		conn.commit()
		conn.close()

		def get_alert(alert_id: str) -> dict:
		"""Retrieve a specific alert from the database."""
		conn = sqlite3.connect(DB_NAME)
		cursor = conn.cursor()
		cursor.execute("SELECT data FROM alert WHERE alert_id = ?", (alert_id,))
		row = cursor.fetchone()
		conn.close()
		if row:
		return json.loads(row[0])
		raise ValueError(f"No alert found with ID '{alert_id}'.")

		def get_all_alerts() -> list:
		"""Retrieve all alerts from the database."""
		conn = sqlite3.connect(DB_NAME)
		cursor = conn.cursor()
		cursor.execute("SELECT data FROM alert")
		rows = cursor.fetchall()
		conn.close()
		return [json.loads(row[0]) for row in rows]

		def delete_all_alerts():
		"""Delete all alerts from the database."""
		conn = sqlite3.connect(DB_NAME)
		cursor = conn.cursor()
		cursor.execute("DELETE FROM alert")
		conn.commit()
		conn.close()

src/database/db.py

+25 −0

Original line number	Diff line number	Diff line
		@@ -31,8 +31,33 @@ def init_db():
		controller TEXT NOT NULL
		)
		""")
		cursor.execute("""
		CREATE TABLE IF NOT EXISTS subscription (
		subscription_id TEXT PRIMARY KEY,
		slice_id TEXT NOT NULL
		)
		""")
		conn.commit()
		conn.close()

		def save_subscription(subscription_id: str, slice_id: str):
		conn = sqlite3.connect(DB_NAME)
		cursor = conn.cursor()
		try:
		cursor.execute("INSERT OR REPLACE INTO subscription (subscription_id, slice_id) VALUES (?, ?)", (subscription_id, slice_id))
		conn.commit()
		finally:
		conn.close()

		def get_slice_id_by_subscription(subscription_id: str) -> str:
		conn = sqlite3.connect(DB_NAME)
		cursor = conn.cursor()
		cursor.execute("SELECT slice_id FROM subscription WHERE subscription_id = ?", (subscription_id,))
		row = cursor.fetchone()
		conn.close()
		if row:
		return row[0]
		return None

		# Save data to the database
		def save_data(slice_id: str, intent_dict: dict, controller: str):