Merge pull request #160 from kytos-ng/epic/vlan_pool

feature: Added endpoints to set and delete `tag_range`

CHANGELOG.rst

@@ -9,9 +9,15 @@ All notable changes to the ``topology`` project will be documented in this file.
 
 Added
 =====
+- Added endpoint ``POST v3/interfaces/{interface_id}/tag_ranges`` to set ``tag_ranges`` to interfaces.
+- Added endpoint ``DELETE v3/interfaces/{interface_id}/tag_ranges`` to delete ``tag_ranges`` from interfaces.
+- Added ``Tag_ranges`` documentation to openapi.yml
+- Added API request POST and DELETE to modify ``Interface.tag_ranges``
+- Added listener for ``kytos/core.interface_tags`` event to save any changes made to ``Interface`` attributes ``tag_ranges`` and ``available_tags``
 
 Deprecated
 ==========
+- Deleted event listener for ``kytos/.*.link_available_tags`` event
 
 Removed
 =======
@@ -25,6 +31,10 @@ Security
 Changed
 =======
 
+General Information
+===================
+- ``scripts/vlan_pool.py`` can be used to change the collection ``interface_details`` to have ``available_tags`` and ``tag_ranges``
+
 [2023.1.0] - 2023-06-26
 ***********************
 

README.rst

@@ -55,7 +55,7 @@ Subscribed
 - ``.*.switch.port.deleted``
 - ``.*.interface.is.nni``
 - ``.*.network_status.updated``
-- ``kytos/.*.link_available_tags``
+- ``kytos/core.interface_tags``
 - ``kytos/maintenance.start_link``
 - ``kytos/maintenance.end_link``
 - ``kytos/maintenance.start_switch``

controllers/__init__.py

@@ -9,11 +9,11 @@
 import pymongo
 from pymongo.collection import ReturnDocument
 from pymongo.errors import AutoReconnect
-from pymongo.operations import UpdateOne
 from tenacity import retry_if_exception_type, stop_after_attempt, wait_random
 
 from kytos.core import log
 from kytos.core.db import Mongo
+from kytos.core.interface import Interface
 from kytos.core.retry import before_sleep, for_all_methods, retries
 from napps.kytos.topology.db.models import (InterfaceDetailDoc, LinkDoc,
                                             SwitchDoc)
@@ -39,7 +39,6 @@ def __init__(self, get_mongo=lambda: Mongo()) -> None:
         self.mongo = get_mongo()
         self.db_client = self.mongo.client
         self.db = self.db_client[self.mongo.db_name]
-        self.interface_details_lock = Lock()
 
     def bootstrap_indexes(self) -> None:
         """Bootstrap all topology related indexes."""
@@ -273,38 +272,30 @@ def bulk_delete_link_metadata_key(
         return self.db.links.update_many({"_id": {"$in": link_ids}},
                                          update_expr)
 
-    def bulk_upsert_interface_details(
-        self, ids_details: List[Tuple[str, dict]]
+    def upsert_interface_details(
+        self,
+        id_: str,
+        available_tags: dict[str, list[list[int]]],
+        tag_ranges: dict[str, list[list[int]]]
     ) -> Optional[dict]:
         """Update or insert interfaces details."""
         utc_now = datetime.utcnow()
-        ops = []
-        for _id, detail_dict in ids_details:
-            ops.append(
-                UpdateOne(
-                    {"_id": _id},
-                    {
-                        "$set": InterfaceDetailDoc(
-                            **{
-                                **detail_dict,
-                                **{
-                                    "updated_at": utc_now,
-                                    "_id": _id,
-                                },
-                            }
-                        ).dict(exclude={"inserted_at"}),
-                        "$setOnInsert": {"inserted_at": utc_now},
-                    },
-                    upsert=True,
-                ),
-            )
-
-        with self.interface_details_lock:
-            with self.db_client.start_session() as session:
-                with session.start_transaction():
-                    return self.db.interface_details.bulk_write(
-                        ops, ordered=False, session=session
-                    )
+        model = InterfaceDetailDoc(**{
+                "_id": id_,
+                "available_tags": available_tags,
+                "tag_ranges": tag_ranges,
+                "updated_at": utc_now
+        }).dict(exclude={"inserted_at"})
+        updated = self.db.interface_details.find_one_and_update(
+            {"_id": id_},
+            {
+                "$set": model,
+                "$setOnInsert": {"inserted_at": utc_now},
+            },
+            return_document=ReturnDocument.AFTER,
+            upsert=True,
+        )
+        return updated
 
     def get_interfaces_details(
         self, interface_ids: List[str]

db/models/__init__.py

@@ -3,7 +3,7 @@
 # pylint: disable=no-name-in-module
 
 from datetime import datetime
-from typing import List, Optional
+from typing import Dict, List, Optional
 
 from pydantic import BaseModel, Field, conlist, validator
 
@@ -124,4 +124,5 @@ def projection() -> dict:
 class InterfaceDetailDoc(DocumentBaseModel):
     """InterfaceDetail DB Document Model."""
 
-    available_vlans: List[int]
+    available_tags: Dict[str, List[List[int]]]
+    tag_ranges: Dict[str, List[List[int]]]

main.py

@@ -3,7 +3,7 @@
 Manage the network topology
 """
 # pylint: disable=wrong-import-order
-
+import pathlib
 import time
 from collections import defaultdict
 from datetime import timezone
@@ -12,8 +12,10 @@
 
 from kytos.core import KytosEvent, KytosNApp, log, rest
 from kytos.core.common import EntityStatus
-from kytos.core.exceptions import KytosLinkCreationError
-from kytos.core.helpers import listen_to, now
+from kytos.core.exceptions import (KytosLinkCreationError,
+                                   KytosSetTagRangeError,
+                                   KytosTagtypeNotSupported)
+from kytos.core.helpers import listen_to, load_spec, now, validate_openapi
 from kytos.core.interface import Interface
 from kytos.core.link import Link
 from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
@@ -34,6 +36,8 @@ class Main(KytosNApp):  # pylint: disable=too-many-public-methods
     This class is the entry point for this napp.
     """
 
+    spec = load_spec(pathlib.Path(__file__).parent / "openapi.yml")
+
     def setup(self):
         """Initialize the NApp's links list."""
         self.links = {}
@@ -46,6 +50,7 @@ def setup(self):
         # to keep track of potential unorded scheduled interface events
         self._intfs_lock = defaultdict(Lock)
         self._intfs_updated_at = {}
+        self._intfs_tags_updated_at = {}
         self.link_up = set()
         self.link_status_lock = Lock()
         self.topo_controller = self.get_topo_controller()
@@ -203,7 +208,7 @@ def _load_switch(self, switch_id, switch_att):
         intf_ids = [v["id"] for v in switch_att.get("interfaces", {}).values()]
         intf_details = self.topo_controller.get_interfaces_details(intf_ids)
         with self._links_lock:
-            self.load_interfaces_available_tags(switch, intf_details)
+            self.load_interfaces_tags_values(switch, intf_details)
 
     # pylint: disable=attribute-defined-outside-init
     def load_topology(self):
@@ -481,6 +486,95 @@ def delete_interface_metadata(self, request: Request) -> JSONResponse:
         self.notify_metadata_changes(interface, 'removed')
         return JSONResponse("Operation successful")
 
+    @staticmethod
+    def map_singular_values(tag_range):
+        """Change integer or singular interger list to
+        list[int, int] when necessary"""
+        if isinstance(tag_range, int):
+            tag_range = [tag_range] * 2
+        elif len(tag_range) == 1:
+            tag_range = [tag_range[0]] * 2
+        return tag_range
+
+    def _get_tag_ranges(self, content: dict):
+        """Get tag_ranges and check validity:
+        - It should be ordered
+        - Not unnecessary partition (eg. [[10,20],[20,30]])
+        - Singular intergers are changed to ranges (eg. [10] to [[10, 10]])
+        The ranges are understood as [inclusive, inclusive]"""
+        ranges = content["tag_ranges"]
+        if len(ranges) < 1:
+            detail = "tag_ranges is empty"
+            raise HTTPException(400, detail=detail)
+        last_tag = 0
+        ranges_n = len(ranges)
+        for i in range(0, ranges_n):
+            ranges[i] = self.map_singular_values(ranges[i])
+            if ranges[i][0] > ranges[i][1]:
+                detail = f"The range {ranges[i]} is not ordered"
+                raise HTTPException(400, detail=detail)
+            if last_tag and last_tag > ranges[i][0]:
+                detail = f"tag_ranges is not ordered. {last_tag}"\
+                         f" is higher than {ranges[i][0]}"
+                raise HTTPException(400, detail=detail)
+            if last_tag and last_tag == ranges[i][0] - 1:
+                detail = f"tag_ranges has an unnecessary partition. "\
+                         f"{last_tag} is before to {ranges[i][0]}"
+                raise HTTPException(400, detail=detail)
+            if last_tag and last_tag == ranges[i][0]:
+                detail = f"tag_ranges has repetition. {ranges[i-1]}"\
+                         f" have same values as {ranges[i]}"
+                raise HTTPException(400, detail=detail)
+            last_tag = ranges[i][1]
+        if ranges[-1][1] > 4095:
+            detail = "Maximum value for tag_ranges is 4095"
+            raise HTTPException(400, detail=detail)
+        if ranges[0][0] < 1:
+            detail = "Minimum value for tag_ranges is 1"
+            raise HTTPException(400, detail=detail)
+        return ranges
+
+    @rest('v3/interfaces/{interface_id}/tag_ranges', methods=['POST'])
+    @validate_openapi(spec)
+    def set_tag_range(self, request: Request) -> JSONResponse:
+        """Set tag range"""
+        content_type_json_or_415(request)
+        content = get_json_or_400(request, self.controller.loop)
+        tag_type = content.get("tag_type")
+        ranges = self._get_tag_ranges(content)
+        interface_id = request.path_params["interface_id"]
+        interface = self.controller.get_interface_by_id(interface_id)
+        if not interface:
+            raise HTTPException(404, detail="Interface not found")
+        try:
+            interface.set_tag_ranges(ranges, tag_type)
+            self.handle_on_interface_tags(interface)
+        except KytosSetTagRangeError as err:
+            detail = f"The new tag_ranges cannot be applied {err}"
+            raise HTTPException(400, detail=detail)
+        except KytosTagtypeNotSupported as err:
+            detail = f"Error with tag_type. {err}"
+            raise HTTPException(400, detail=detail)
+        return JSONResponse("Operation Successful", status_code=200)
+
+    @rest('v3/interfaces/{interface_id}/tag_ranges', methods=['DELETE'])
+    @validate_openapi(spec)
+    def delete_tag_range(self, request: Request) -> JSONResponse:
+        """Set tag_range from tag_type to default value [1, 4095]"""
+        interface_id = request.path_params["interface_id"]
+        params = request.query_params
+        tag_type = params.get("tag_type", 'vlan')
+        interface = self.controller.get_interface_by_id(interface_id)
+        if not interface:
+            raise HTTPException(404, detail="Interface not found")
+        try:
+            interface.remove_tag_ranges(tag_type)
+            self.handle_on_interface_tags(interface)
+        except KytosTagtypeNotSupported as err:
+            detail = f"Error with tag_type. {err}"
+            raise HTTPException(400, detail=detail)
+        return JSONResponse("Operation Successful", status_code=200)
+
     # Link related methods
     @rest('v3/links')
     def get_links(self, _request: Request) -> JSONResponse:
@@ -629,27 +723,25 @@ def handle_link_liveness_disabled(self, interfaces) -> None:
         for link in links.values():
             self.notify_link_status_change(link, reason="liveness_disabled")
 
-    @listen_to("kytos/.*.link_available_tags")
-    def on_link_available_tags(self, event):
-        """Handle on_link_available_tags."""
-        with self._links_lock:
-            self.handle_on_link_available_tags(event.content.get("link"))
-
-    def handle_on_link_available_tags(self, link):
-        """Handle on_link_available_tags."""
-        if link.id not in self.links:
-            return
-        endpoint_a = self.links[link.id].endpoint_a
-        endpoint_b = self.links[link.id].endpoint_b
-        values_a = [tag.value for tag in endpoint_a.available_tags]
-        values_b = [tag.value for tag in endpoint_b.available_tags]
-        ids_details = [
-            (endpoint_a.id, {"_id": endpoint_a.id,
-                             "available_vlans": values_a}),
-            (endpoint_b.id, {"_id": endpoint_b.id,
-                             "available_vlans": values_b})
-        ]
-        self.topo_controller.bulk_upsert_interface_details(ids_details)
+    @listen_to("kytos/core.interface_tags")
+    def on_interface_tags(self, event):
+        """Handle on_interface_tags."""
+        interface = event.content['interface']
+        with self._intfs_lock[interface.id]:
+            if (
+                interface.id in self._intfs_tags_updated_at
+                and self._intfs_tags_updated_at[interface.id] > event.timestamp
+            ):
+                return
+            self._intfs_tags_updated_at[interface.id] = event.timestamp
+        self.handle_on_interface_tags(interface)
+
+    def handle_on_interface_tags(self, interface):
+        """Update interface details"""
+        intf_id = interface.id
+        self.topo_controller.upsert_interface_details(
+            intf_id, interface.available_tags, interface.tag_ranges
+        )
 
     @listen_to('.*.switch.(new|reconnected)')
     def on_new_switch(self, event):
@@ -1050,21 +1142,24 @@ def notify_port_created(self, event):
         self.controller.buffers.app.put(event)
 
     @staticmethod
-    def load_interfaces_available_tags(switch: Switch,
-                                       interfaces_details: List[dict]) -> None:
+    def load_interfaces_tags_values(switch: Switch,
+                                    interfaces_details: List[dict]) -> None:
         """Load interfaces available tags (vlans)."""
         if not interfaces_details:
             return
         for interface_details in interfaces_details:
-            available_vlans = interface_details["available_vlans"]
-            if not available_vlans:
+            available_tags = interface_details['available_tags']
+            if not available_tags:
                 continue
             log.debug(f"Interface id {interface_details['id']} loading "
-                      f"{len(interface_details['available_vlans'])} "
+                      f"{len(available_tags)} "
                       "available tags")
             port_number = int(interface_details["id"].split(":")[-1])
             interface = switch.interfaces[port_number]
-            interface.set_available_tags(interface_details['available_vlans'])
+            interface.set_available_tags_tag_ranges(
+                available_tags,
+                interface_details['tag_ranges']
+            )
 
     @listen_to('topology.interruption.start')
     def on_interruption_start(self, event: KytosEvent):