Merge branch 'master' into release/2024.1

CHANGELOG.rst

@@ -18,15 +18,22 @@ Added
 - Handled ``kytos/mef_eline.uni_active_updated`` event.
 - Handled ``kytos/mef_eline.deployed`` event.
 - Handled ``kytos/mef_eline.(failover_link_down|failover_old_path|failover_deployed)`` events.
-- Added UI for telemetry_int.
+- Added UI for telemetry_int to list EVCs and to configure proxy ports
+- Included the endpoints to create, update, delete and list proxy_port metadata, and updated OpenAPI spec. These endpoints should be used to manage the proxy_port metadata instead of directly on topology endpoints since these endpoints provide extra validations.
 
 Changed
 =======
 - Only raise ``FlowsNotFound`` when an EVC is active and flows aren't found. Update status and status_reason accordingly too when installing flows.
+- Validate to support only a single proxy port per UNI for now.
+
+Removed
+=======
+- Removed client side batching with ``BATCH_INTERVAL`` and ``BATCH_SIZE``, now replaced with pacing in ``flow_manager``
 
 Fixed
 =====
 - Only redeploy if INT has been enabled before
+- Fixed batched flows slicing
 
 [2023.2.0] - 2024-02-16
 ***********************

exceptions.py

@@ -53,6 +53,11 @@ class ProxyPortSameSourceIntraEVC(ProxyPortError):
     """
 
 
+class ProxyPortShared(ProxyPortError):
+    """ProxyPortShared. A shared proxy port isn't supported for now.
+    Each uni should have its own proxy port"""
+
+
 class EVCHasNoINT(EVCError):
     """Exception in case the EVC doesn't have INT enabled."""
 

kytos_api_helper.py

@@ -166,3 +166,54 @@ async def add_evcs_metadata(
             f"Failed to add_evc_metadata for EVC ids {list(evcs.keys())} "
             f"status code {response.status_code}, response text: {response.text}"
         )
+
+
+@retry(
+    stop=stop_after_attempt(5),
+    wait=wait_combine(wait_fixed(3), wait_random(min=2, max=7)),
+    before_sleep=before_sleep,
+    retry=retry_if_exception_type(httpx.RequestError),
+)
+async def add_proxy_port_metadata(intf_id: str, port_no: int) -> dict:
+    """Add proxy_port metadata."""
+    async with httpx.AsyncClient(base_url=settings.topology_url) as client:
+        response = await client.post(
+            f"/interfaces/{intf_id}/metadata",
+            timeout=10,
+            json={"proxy_port": port_no},
+        )
+        if response.is_success:
+            return response.json()
+        if response.status_code == 404:
+            raise ValueError(f"interface_id {intf_id} not found")
+        if response.is_server_error:
+            raise httpx.RequestError(response.text)
+        raise UnrecoverableError(
+            f"Failed to add_proxy_port {port_no} metadata for intf_id {intf_id} "
+            f"status code {response.status_code}, response text: {response.text}"
+        )
+
+
+@retry(
+    stop=stop_after_attempt(5),
+    wait=wait_combine(wait_fixed(3), wait_random(min=2, max=7)),
+    before_sleep=before_sleep,
+    retry=retry_if_exception_type(httpx.RequestError),
+)
+async def delete_proxy_port_metadata(intf_id: str) -> dict:
+    """Delete proxy_port metadata."""
+    async with httpx.AsyncClient(base_url=settings.topology_url) as client:
+        response = await client.delete(
+            f"/interfaces/{intf_id}/metadata/proxy_port",
+            timeout=10,
+        )
+        if response.is_success:
+            return response.json()
+        if response.status_code == 404:
+            raise ValueError(f"interface_id {intf_id} or metadata proxy_port not found")
+        if response.is_server_error:
+            raise httpx.RequestError(response.text)
+        raise UnrecoverableError(
+            f"Failed to delete_proxy_port metadata for intf_id {intf_id} "
+            f"status code {response.status_code}, response text: {response.text}"
+        )

main.py

@@ -13,6 +13,7 @@
 from tenacity import RetryError
 
 from kytos.core import KytosEvent, KytosNApp, log, rest
+from kytos.core.common import EntityStatus
 from kytos.core.helpers import alisten_to, avalidate_openapi_request, load_spec
 from kytos.core.rest_api import HTTPException, JSONResponse, Request, aget_json_or_400
 
@@ -22,15 +23,15 @@
     EVCHasNoINT,
     EVCNotFound,
     FlowsNotFound,
+    ProxyPortError,
     ProxyPortNotFound,
     ProxyPortSameSourceIntraEVC,
+    ProxyPortShared,
     ProxyPortStatusNotUP,
     UnrecoverableError,
 )
 from .managers.int import INTManager
 
-# pylint: disable=fixme
-
 
 class Main(KytosNApp):
     """Main class of kytos/telemetry NApp.
@@ -115,7 +116,12 @@ async def enable_telemetry(self, request: Request) -> JSONResponse:
             await self.int_manager.enable_int(evcs, force)
         except (EVCNotFound, FlowsNotFound, ProxyPortNotFound) as exc:
             raise HTTPException(404, detail=str(exc))
-        except (EVCHasINT, ProxyPortStatusNotUP, ProxyPortSameSourceIntraEVC) as exc:
+        except (
+            EVCHasINT,
+            ProxyPortStatusNotUP,
+            ProxyPortSameSourceIntraEVC,
+            ProxyPortShared,
+        ) as exc:
             raise HTTPException(409, detail=str(exc))
         except RetryError as exc:
             exc_error = str(exc.last_attempt.exception())
@@ -232,7 +238,7 @@ async def redeploy_telemetry(self, request: Request) -> JSONResponse:
             await self.int_manager.redeploy_int(evcs)
         except (EVCNotFound, FlowsNotFound, ProxyPortNotFound) as exc:
             raise HTTPException(404, detail=str(exc))
-        except (EVCHasNoINT, ProxyPortSameSourceIntraEVC) as exc:
+        except (EVCHasNoINT, ProxyPortSameSourceIntraEVC, ProxyPortShared) as exc:
             raise HTTPException(409, detail=str(exc))
         except RetryError as exc:
             exc_error = str(exc.last_attempt.exception())
@@ -291,6 +297,101 @@ async def evc_compare(self, _request: Request) -> JSONResponse:
         ]
         return JSONResponse(response)
 
+    @rest("v1/uni/{interface_id}/proxy_port", methods=["DELETE"])
+    async def delete_proxy_port_metadata(self, request: Request) -> JSONResponse:
+        """Delete proxy port metadata."""
+        intf_id = request.path_params["interface_id"]
+        intf = self.controller.get_interface_by_id(intf_id)
+        if not intf:
+            raise HTTPException(404, detail=f"Interface id {intf_id} not found")
+        if "proxy_port" not in intf.metadata:
+            return JSONResponse("Operation successful")
+
+        qparams = request.query_params
+        force = qparams.get("force", "false").lower() == "true"
+
+        try:
+            pp = self.int_manager.srcs_pp[self.int_manager.unis_src[intf_id]]
+            if pp.evc_ids and not force:
+                return JSONResponse(
+                    {
+                        "status_code": 409,
+                        "code": 409,
+                        "description": f"{pp} is in use on {len(pp.evc_ids)} EVCs",
+                        "evc_ids": sorted(pp.evc_ids),
+                    },
+                    status_code=409,
+                )
+        except KeyError:
+            pass
+
+        try:
+            await api.delete_proxy_port_metadata(intf_id)
+            return JSONResponse("Operation successful")
+        except ValueError as exc:
+            raise HTTPException(404, detail=str(exc))
+        except UnrecoverableError as exc:
+            raise HTTPException(500, detail=str(exc))
+
+    @rest("v1/uni/{interface_id}/proxy_port/{port_number:int}", methods=["POST"])
+    async def add_proxy_port_metadata(self, request: Request) -> JSONResponse:
+        """Add proxy port metadata."""
+        intf_id = request.path_params["interface_id"]
+        port_no = request.path_params["port_number"]
+        qparams = request.query_params
+        if not (intf := self.controller.get_interface_by_id(intf_id)):
+            raise HTTPException(404, detail=f"Interface id {intf_id} not found")
+        if "proxy_port" in intf.metadata and intf.metadata["proxy_port"] == port_no:
+            return JSONResponse("Operation successful")
+
+        force = qparams.get("force", "false").lower() == "true"
+        try:
+            pp = self.int_manager.get_proxy_port_or_raise(intf_id, "no_evc_id", port_no)
+            if pp.status != EntityStatus.UP and not force:
+                raise HTTPException(409, detail=f"{pp} status isn't UP")
+            self.int_manager._validate_new_dedicated_proxy_port(intf, port_no)
+        except ProxyPortShared as exc:
+            raise HTTPException(409, detail=exc.message)
+        except ProxyPortError as exc:
+            raise HTTPException(404, detail=exc.message)
+
+        try:
+            await api.add_proxy_port_metadata(intf_id, port_no)
+            return JSONResponse("Operation successful")
+        except ValueError as exc:
+            raise HTTPException(404, detail=str(exc))
+        except UnrecoverableError as exc:
+            raise HTTPException(500, detail=str(exc))
+
+    @rest("v1/uni/proxy_port")
+    async def list_uni_proxy_ports(self, _request: Request) -> JSONResponse:
+        """List configured UNI proxy ports."""
+        interfaces_proxy_ports = []
+        for switch in self.controller.switches.copy().values():
+            for intf in switch.interfaces.copy().values():
+                if "proxy_port" in intf.metadata:
+                    payload = {
+                        "uni": {
+                            "id": intf.id,
+                            "status": intf.status.value,
+                            "status_reason": sorted(intf.status_reason),
+                        },
+                        "proxy_port": {
+                            "port_number": intf.metadata["proxy_port"],
+                            "status": "DOWN",
+                            "status_reason": [],
+                        },
+                    }
+                    try:
+                        pp = self.int_manager.get_proxy_port_or_raise(
+                            intf.id, "no_evc_id"
+                        )
+                        payload["proxy_port"]["status"] = pp.status.value
+                    except ProxyPortError as exc:
+                        payload["proxy_port"]["status_reason"] = [exc.message]
+                    interfaces_proxy_ports.append(payload)
+        return JSONResponse(interfaces_proxy_ports)
+
     @alisten_to("kytos/mef_eline.evcs_loaded")
     async def on_mef_eline_evcs_loaded(self, event: KytosEvent) -> None:
         """Handle kytos/mef_eline.evcs_loaded."""
@@ -543,24 +644,3 @@ async def on_intf_metadata_removed(self, event: KytosEvent) -> None:
     async def on_intf_metadata_added(self, event: KytosEvent) -> None:
         """On interface metadata added."""
         await self.int_manager.handle_pp_metadata_added(event.content["interface"])
-
-    # Event-driven methods: future
-    def listen_for_new_evcs(self):
-        """Change newly created EVC to INT-enabled EVC based on the metadata field
-        (future)"""
-        pass
-
-    def listen_for_evc_change(self):
-        """Change newly created EVC to INT-enabled EVC based on the
-        metadata field (future)"""
-        pass
-
-    def listen_for_path_changes(self):
-        """Change EVC's new path to INT-enabled EVC based on the metadata field
-        when there is a path change. (future)"""
-        pass
-
-    def listen_for_topology_changes(self):
-        """If the topology changes, make sure it is not the loop ports.
-        If so, update proxy ports"""
-        pass