Merge branch... (c76c4385) · Commits · TFS / nsc

src/api/handlers/restconf_handler.py

+49 −2

Changes for src/api/handlers/restconf_handler.py: 49 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -63,6 +63,9 @@ from src.database.telemetry_client_db import (
		from src.database.telemetry_client_db import (
		delete_subscription as _real_delete_subscription,
		)
		from src.database.telemetry_client_db import (
		delete_telemetry as _real_delete_telemetry,
		)
		from src.database.telemetry_client_db import (
		get_all_clients as _real_get_all_clients,
		)
		@@ -75,6 +78,9 @@ from src.database.telemetry_client_db import (
		from src.database.telemetry_client_db import (
		get_subscription as _real_get_subscription,
		)
		from src.database.telemetry_client_db import (
		get_telemetry_history as _real_get_telemetry_history,
		)
		from src.database.telemetry_client_db import (
		upsert_subscription as _real_upsert_subscription,
		)
		@@ -604,6 +610,7 @@ class RestconfHandler:

		delete_ds_fn = _dep("delete_data_store", _real_delete_data_store)
		delete_ds_fn(xpath)
		_dep("delete_telemetry", _real_delete_telemetry)(slice_id)
		logger.info("Slice %s removed successfully", slice_id)
		return {}, 204

		@@ -638,6 +645,7 @@ class RestconfHandler:
		del_svc_by_slice(slice_item.get("id"))
		except Exception as exc:
		logger.debug("No service_db entries for slice %s: %s", slice_item.get("id"), exc)
		_dep("delete_telemetry", _real_delete_telemetry)(slice_item.get("id"))

		if current_app.config["TFS_L2VPN_SUPPORT"]:
		if hasattr(self.slice_service, "tfs_l2vpn_delete"):
		@@ -944,10 +952,24 @@ class RestconfHandler:
		# RESTCONF Telemetry Operations: Metrics & SSE Streaming
		# -------------------------------------------------------------------------

		def get_telemetry(self, slice_id: str \| None = None) -> tuple[dict[str, Any], int]:
		"""Retrieve monitoring telemetry metrics for one or all slices."""
		def get_telemetry(
		self,
		slice_id: str \| None = None,
		from_ts: float \| None = None,
		to_ts: float \| None = None,
		limit: int \| None = None,
		) -> tuple[dict[str, Any], int]:
		"""Retrieve monitoring telemetry metrics for one or all slices.

		Without from_ts/to_ts/limit, the latest telemetry is computed live and stored.
		With any of them, the stored telemetry history is returned instead.
		"""
		logger.debug("Getting telemetry for slice_id: %s", slice_id)
		if from_ts is not None and to_ts is not None and from_ts > to_ts:
		return send_response(False, code=400, message="'from' must be lower than or equal to 'to'")
		try:
		if from_ts is not None or to_ts is not None or limit is not None:
		return self._get_telemetry_history(slice_id, from_ts, to_ts, limit)
		if slice_id is not None:
		return self._get_single_slice_telemetry(slice_id)
		return self._get_all_slices_telemetry()
		@@ -956,6 +978,31 @@ class RestconfHandler:
		except Exception as exc:
		return send_response(False, code=500, message=str(exc))

		def _get_telemetry_history(
		self,
		slice_id: str \| None,
		from_ts: float \| None,
		to_ts: float \| None,
		limit: int \| None,
		) -> tuple[dict[str, Any], int]:
		"""Fetch stored telemetry history for one slice or for all slices."""
		get_history_fn = _dep("get_telemetry_history", _real_get_telemetry_history)
		if slice_id is not None:
		xpath = f"/ietf-network-slice-service:network-slice-services/slice-service[id='{slice_id}']"
		get_ds_fn = _dep("get_data_store", _real_get_data_store)
		if not get_ds_fn(xpath):
		raise ValueError(f"There is no slice with id '{slice_id}' registered")
		history = get_history_fn(slice_id, from_ts, to_ts, limit)
		return {"slice_id": slice_id, "telemetry_history": history}, 200

		# limit applies per slice: fetch everything in range, group, then keep the N most recent of each
		history_by_slice: dict[str, list[dict[str, Any]]] = {}
		for sample in get_history_fn(None, from_ts, to_ts, None):
		history_by_slice.setdefault(sample["slice_id"], []).append(sample)
		if limit is not None:
		history_by_slice = {sid: samples[-limit:] for sid, samples in history_by_slice.items()}
		return history_by_slice, 200

		def _get_single_slice_telemetry(self, slice_id: str) -> tuple[dict[str, Any], int]:
		"""Fetch telemetry for a single slice."""
		xpath = f"/ietf-network-slice-service:network-slice-services/slice-service[id='{slice_id}']"

src/api/main.py

+4 −0

Changes for src/api/main.py: 4 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -68,10 +68,12 @@ from src.database.telemetry_client_db import (
		delete_all_subscriptions,
		delete_client,
		delete_subscription,
		delete_telemetry,
		get_all_clients,
		get_client,
		get_client_subscriptions,
		get_subscription,
		get_telemetry_history,
		upsert_subscription,
		)
		from src.realizer.restconf.connectors.tfs_connector import (
		@@ -127,6 +129,7 @@ __all__ = [
		"delete_data",
		"delete_data_store",
		"delete_subscription",
		"delete_telemetry",
		"find_slice_for_alert",
		"get_all_clients",
		"get_all_data",
		@@ -138,6 +141,7 @@ __all__ = [
		"get_service_db_data",
		"get_slice_id_by_subscription",
		"get_subscription",
		"get_telemetry_history",
		"normalize_libyang_data",
		"parse_alert_notification",
		"safe_get",

src/database/telemetry_client_db.py

+81 −16

Changes for src/database/telemetry_client_db.py: 81 added lines, 16 removed lines.
Original line number	Diff line number	Diff line
		@@ -22,6 +22,7 @@ from typing import Any
		logger = logging.getLogger(__name__)

		DB_NAME: str = "telemetry_client.db"
		MAX_TELEMETRY_HISTORY: int = 1000


		def init_db() -> None:
		@@ -44,12 +45,17 @@ def init_db() -> None:
		""")
		cursor.execute("""
		CREATE TABLE IF NOT EXISTS telemetry (
		slice_id TEXT PRIMARY KEY,
		id INTEGER PRIMARY KEY AUTOINCREMENT,
		slice_id TEXT NOT NULL,
		telemetry_list TEXT NOT NULL,
		slo_sle_compliance TEXT NOT NULL,
		timestamp REAL NOT NULL
		)
		""")
		cursor.execute("""
		CREATE INDEX IF NOT EXISTS idx_telemetry_slice_ts
		ON telemetry (slice_id, timestamp)
		""")
		conn.commit()


		@@ -177,13 +183,23 @@ def delete_all_subscriptions(client_id: str) -> None:
		conn.commit()


		def upsert_telemetry(
		def _row_to_telemetry(row: tuple[Any, ...]) -> dict[str, Any]:
		"""Convert a telemetry row (slice_id, telemetry_list, slo_sle_compliance, timestamp) to a dict."""
		return {
		"slice_id": row[0],
		"telemetry_list": json.loads(row[1]),
		"slo_sle_compliance": json.loads(row[2]),
		"timestamp": row[3],
		}


		def insert_telemetry(
		slice_id: str,
		telemetry_list: dict[str, Any],
		slo_sle_compliance: dict[str, Any],
		timestamp: float,
		) -> None:
		"""Insert or update telemetry data for a slice."""
		"""Append a telemetry sample for a slice, keeping only the latest MAX_TELEMETRY_HISTORY samples."""
		with get_conn() as conn:
		cursor = conn.cursor()
		cursor.execute("""
		@@ -194,42 +210,91 @@ def upsert_telemetry(
		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,
		))
		cursor.execute("""
		DELETE FROM telemetry
		WHERE slice_id = ? AND id NOT IN (
		SELECT id FROM telemetry
		WHERE slice_id = ?
		ORDER BY timestamp DESC, id DESC
		LIMIT ?
		)
		""", (slice_id, slice_id, MAX_TELEMETRY_HISTORY))
		conn.commit()


		def get_telemetry(slice_id: str) -> dict[str, Any]:
		"""Retrieve telemetry metrics for a slice."""
		"""Retrieve the latest telemetry sample for a slice."""
		with get_conn() as conn:
		cursor = conn.cursor()
		cursor.execute("""
		SELECT slice_id, telemetry_list, slo_sle_compliance, timestamp
		FROM telemetry
		WHERE slice_id = ?
		ORDER BY timestamp DESC, id DESC
		LIMIT 1
		""", (slice_id,))
		row = cursor.fetchone()
		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],
		}
		return _row_to_telemetry(row)


		def get_telemetry_history(
		slice_id: str \| None = None,
		from_ts: float \| None = None,
		to_ts: float \| None = None,
		limit: int \| None = None,
		) -> list[dict[str, Any]]:
		"""
		Retrieve stored telemetry samples ordered by ascending timestamp.

		Args:
		slice_id: Slice to filter by. If None, samples of all slices are returned.
		from_ts: Inclusive lower bound (Unix epoch seconds).
		to_ts: Inclusive upper bound (Unix epoch seconds).
		limit: Return only the N most recent samples within the range.
		"""
		conditions: list[str] = []
		params: list[Any] = []
		if slice_id is not None:
		conditions.append("slice_id = ?")
		params.append(slice_id)
		if from_ts is not None:
		conditions.append("timestamp >= ?")
		params.append(from_ts)
		if to_ts is not None:
		conditions.append("timestamp <= ?")
		params.append(to_ts)
		where = f"WHERE {' AND '.join(conditions)}" if conditions else ""

		query = f"""
		SELECT id, slice_id, telemetry_list, slo_sle_compliance, timestamp
		FROM telemetry
		{where}
		ORDER BY timestamp DESC, id DESC
		"""
		if limit is not None:
		query += " LIMIT ?"
		params.append(limit)

		with get_conn() as conn:
		cursor = conn.cursor()
		cursor.execute(f"""
		SELECT slice_id, telemetry_list, slo_sle_compliance, timestamp
		FROM ({query})
		ORDER BY timestamp ASC, id ASC
		""", params)
		return [_row_to_telemetry(row) for row in cursor.fetchall()]


		def delete_telemetry(slice_id: str) -> None:
		"""Delete telemetry record for a slice."""
		"""Delete all telemetry samples for a slice."""
		with get_conn() as conn:
		cursor = conn.cursor()
		cursor.execute("DELETE FROM telemetry WHERE slice_id = ?", (slice_id,))

src/mapper/aggregate_monitoring.py

+2 −2

Changes for src/mapper/aggregate_monitoring.py: 2 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -3,7 +3,7 @@ from typing import Any

		from flask import current_app

		from src.database.telemetry_client_db import upsert_telemetry
		from src.database.telemetry_client_db import insert_telemetry
		from src.utils.safe_get import safe_get


		@@ -64,7 +64,7 @@ def aggregate_monitoring(slice_id: str, slo_sle_template: dict[str, Any] \| None)

		current_timestamp = time.time()

		upsert_telemetry(
		insert_telemetry(
		slice_id=slice_id,
		telemetry_list=telemetry_json,
		slo_sle_compliance=compliance_json,

src/tests/test_api.py

+53 −0

Changes for src/tests/test_api.py: 53 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -63,6 +63,13 @@ def push_flask_context(flask_app):
		yield


		@pytest.fixture(autouse=True)
		def mock_delete_telemetry():
		"""Avoid touching the real telemetry database when slices are deleted."""
		with patch("src.api.main.delete_telemetry") as mock_delete:
		yield mock_delete


		@pytest.fixture
		def temp_db(tmp_path):
		"""Fixture to create and cleanup a test database using SQLite instead of JSON."""
		@@ -512,6 +519,52 @@ class TestTelemetryStreams:
		res, code = api.get_telemetry("slice-many-sdps")
		assert code == 500

		def test_get_telemetry_history(self, controller_with_mocked_db):
		api = Api(controller_with_mocked_db)
		samples = [
		{"slice_id": "s1", "telemetry_list": {}, "slo_sle_compliance": {}, "timestamp": 1.0},
		{"slice_id": "s2", "telemetry_list": {}, "slo_sle_compliance": {}, "timestamp": 2.0},
		{"slice_id": "s1", "telemetry_list": {}, "slo_sle_compliance": {}, "timestamp": 3.0},
		]

		# Single slice: limit alone triggers history mode and skips live monitoring
		with (
		patch("src.api.main.get_data_store", return_value={"s1": {}}),
		patch("src.api.main.get_telemetry_history", return_value=samples[:1]) as mock_history,
		patch.object(api.slice_service, "monitoring") as mock_monitoring,
		):
		res, code = api.get_telemetry("s1", limit=10)
		assert code == 200
		assert res == {"slice_id": "s1", "telemetry_history": samples[:1]}
		mock_history.assert_called_once_with("s1", None, None, 10)
		mock_monitoring.assert_not_called()

		# Single slice not registered -> 404
		with patch("src.api.main.get_data_store", return_value=None):
		_, code = api.get_telemetry("missing", from_ts=0.0)
		assert code == 404

		# from > to -> 400
		_, code = api.get_telemetry("s1", from_ts=10.0, to_ts=5.0)
		assert code == 400

		# All slices: grouped by slice, limit applied per slice keeping the most recent
		with patch("src.api.main.get_telemetry_history", return_value=samples) as mock_history:
		res, code = api.get_telemetry(from_ts=0.0, to_ts=5.0, limit=1)
		assert code == 200
		assert res == {"s1": [samples[2]], "s2": [samples[1]]}
		mock_history.assert_called_once_with(None, 0.0, 5.0, None)

		def test_delete_slice_removes_telemetry(self, controller_with_mocked_db, mock_delete_telemetry):
		api = Api(controller_with_mocked_db)
		with (
		patch("src.api.main.get_data_store", return_value={"s1": {}}),
		patch("src.api.main.delete_data_store", return_value=True),
		):
		_, code = api.delete_slice_services("s1")
		assert code == 204
		mock_delete_telemetry.assert_called_once_with("s1")


		class TestApiFullCoverage:
		"""Additional comprehensive unit tests for src/api/main.py methods."""