feat: add alert persistence and API endpoints for managing TAPI notifications (300c6deb) · Commits · TFS / nsc

app.py

+2 −0

Original line number	Diff line number	Diff line
		@@ -28,6 +28,7 @@ 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
		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
		@@ -60,6 +61,7 @@ def create_app():
		init_slice()
		init_service()
		init_telemetry()
		init_alert()
		app = Flask(__name__)
		app = create_config(app)
		# app.logger.setLevel(logging.INFO)

src/api/main.py

+87 −0

Original line number	Diff line number	Diff line
		@@ -20,6 +20,7 @@ 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
		import src.database.alert_db as alert_db
		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
		@@ -27,6 +28,8 @@ 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):
		@@ -66,6 +69,90 @@ 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
		if context and isinstance(context, list):
		notification = context[0].get("tapi-notification:notification", {})
		alert_id = notification.get("uuid")

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

		alert_db.save_alert(alert_id, alert_data)
		return send_response(
		True,
		code=201,
		message="Alert 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/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()

swagger/E2E_namespace.py

+110 −1

Original line number	Diff line number	Diff line
		@@ -15,7 +15,7 @@
		# This file is an original contribution from Telefonica Innovación Digital S.L.

		from flask import request
		from flask_restx import Namespace, Resource, reqparse
		from flask_restx import Namespace, Resource, fields, reqparse
		from src.main import NSController
		from src.api.main import Api
		import json
		@@ -34,6 +34,39 @@ gpp_network_slice_request_model = create_gpp_nrm_28541_model(e2e_ns)

		slice_ddbb_model, slice_response_model = create_ietf_network_slice_nbi_yang_model(e2e_ns)

		# Alert Data Model
		alarm_info_model = e2e_ns.model("AlarmInfo", {
		"is-alarmed": fields.Boolean(description="Is alarmed", required=True),
		"perceived-severity": fields.String(description="Perceived severity", required=True),
		"probable-cause": fields.String(description="Probable cause", required=True),
		"threshold-low-breached": fields.Float(description="Threshold low breached", required=False),
		"threshold-high-breached": fields.Float(description="Threshold high breached", required=False)
		})

		additional_info_model = e2e_ns.model("AdditionalInfo", {
		"service-id": fields.String(description="Service ID", required=False),
		"kpi-name": fields.String(description="KPI name", required=False),
		"measured-value": fields.Float(description="Measured value", required=False)
		})

		tapi_notification_model = e2e_ns.model("TapiNotification", {
		"uuid": fields.String(description="UUID", required=True),
		"notification-type": fields.String(description="Notification type", required=True),
		"event-time-stamp": fields.String(description="Event timestamp", required=True),
		"target-object-type": fields.String(description="Target object type", required=False),
		"layer-protocol-name": fields.String(description="Layer protocol name", required=False),
		"alarm-info": fields.Nested(alarm_info_model, required=False),
		"additional-info": fields.Nested(additional_info_model, required=False)
		})

		tapi_notification_item_model = e2e_ns.model("TapiNotificationItem", {
		"tapi-notification:notification": fields.Nested(tapi_notification_model, required=True)
		})

		alert_model = e2e_ns.model("Alert", {
		"tapi-notification:notification-context": fields.List(fields.Nested(tapi_notification_item_model), required=True)
		})

		upload_parser = reqparse.RequestParser()
		upload_parser.add_argument('file', location='files', type='FileStorage', help="File to upload")
		upload_parser.add_argument('json_data', location='form', help="JSON Data in string format")
		@@ -151,3 +184,79 @@ class E2ESlice(Resource):
		controller = NSController(controller_type="E2E")
		data, code = Api(controller).modify_flow(slice_id, json_data)
		return data, code


		@e2e_ns.route("/alert")
		class E2EAlertList(Resource):
		@e2e_ns.doc(summary="Return all alerts", description="Returns all alerts received.")
		@e2e_ns.response(200, "Alerts returned")
		@e2e_ns.response(500, "Internal server error")
		def get(self):
		"""Retrieve all alerts"""
		controller = NSController(controller_type="E2E")
		data, code = Api(controller).get_alerts()
		return data, code

		@e2e_ns.doc(summary="Receive alerts", description="This endpoint allows E2E to receive alerts.")
		@e2e_ns.response(201, "Alert saved successfully")
		@e2e_ns.response(400, "Invalid alert format")
		@e2e_ns.response(500, "Internal server error")
		@e2e_ns.expect(alert_model, validate=True)
		def post(self):
		"""Receive an alert"""
		json_data = request.get_json()
		if not json_data:
		return {
		"success": False,
		"data": None,
		"error": "No alert data sent"
		}, 400
		controller = NSController(controller_type="E2E")
		data, code = Api(controller).receive_alert(json_data)
		return data, code

		@e2e_ns.doc(summary="Delete all alerts", description="Deletes all alerts.")
		@e2e_ns.response(204, "All alerts deleted successfully.")
		@e2e_ns.response(500, "Internal server error")
		def delete(self):
		"""Delete all alerts"""
		controller = NSController(controller_type="E2E")
		data, code = Api(controller).delete_alerts()
		return data, code


		@e2e_ns.route("/alert/<string:alert_id>")
		@e2e_ns.doc(params={"alert_id": "The UUID of the alert to retrieve, modify, or delete"})
		class E2EAlertDetail(Resource):
		@e2e_ns.doc(summary="Return a specific alert", description="Returns specific information related to an alert by providing its UUID")
		@e2e_ns.response(200, "Alert returned")
		@e2e_ns.response(404, "Alert not found.")
		@e2e_ns.response(500, "Internal server error")
		def get(self, alert_id):
		"""Retrieve a specific alert"""
		controller = NSController(controller_type="E2E")
		data, code = Api(controller).get_alerts(alert_id)
		return data, code

		@e2e_ns.doc(summary="Delete a specific alert", description="Deletes a specific alert based on the provided `alert_id`.")
		@e2e_ns.response(204, "Alert deleted successfully.")
		@e2e_ns.response(404, "Alert not found.")
		@e2e_ns.response(500, "Internal server error")
		def delete(self, alert_id):
		"""Delete an alert"""
		controller = NSController(controller_type="E2E")
		data, code = Api(controller).delete_alerts(alert_id)
		return data, code

		@e2e_ns.expect(alert_model, validate=True)
		@e2e_ns.doc(summary="Modify a specific alert", description="Modifies a specific alert and returns the updated alert data")
		@e2e_ns.response(200, "Alert modified")
		@e2e_ns.response(404, "Alert not found.")
		@e2e_ns.response(500, "Internal server error")
		def put(self, alert_id):
		"""Modify an alert"""
		json_data = request.get_json()
		controller = NSController(controller_type="E2E")
		data, code = Api(controller).modify_alert(alert_id, json_data)
		return data, code