Draining stuff

Author: simmsb

intent_deployer/links.py

@@ -8,13 +8,14 @@
 
 
 class Links:
-    hosts: dict[str, str]
+    switch_links: list[tuple[str, str]]
+    host_links: list[tuple[str, str]]
 
     def __init__(self, links, hosts):
         self.switch_links = [
             (
                 Switches.name_from_id(l["src"]["device"]),
-                Switches.name_from_id(l["src"]["device"]),
+                Switches.name_from_id(l["dst"]["device"]),
             )
             for l in links["links"]
         ]

intent_deployer/network_state.py

@@ -9,10 +9,11 @@
 from intent_deployer.hosts import Hosts
 from intent_deployer.links import Links
 from intent_deployer.switches import Switches
+from intents.utils import OnosRoute
 
 
 @dataclass
-class Intent:
+class IntentPath:
     src: str
     dst: str
     switch_path: list[str]
@@ -21,23 +22,49 @@ class Intent:
     def path(self):
         return [self.src, *self.switch_path, self.dst]
 
-    def as_tuple(self) -> tuple[tuple[str, str], Intent]:
+    def as_tuple(self) -> tuple[tuple[str, str], IntentPath]:
         return ((self.src, self.dst), self)
 
     @classmethod
-    def parse(cls, data: dict[str, Any]) -> Intent:
-        src, *switch_path, dst = cast(list[str], data["paths"][0]["path"])
+    def parse(cls, data: dict[str, Any]) -> IntentPath:
+        src, *switch_path, dst = cast(list[str], data["path"])
         src = Hosts.name_from_mac(src.removesuffix("/None"))
         dst = Hosts.name_from_mac(dst.removesuffix("/None"))
         switch_path = [Switches.name_from_id(s) for s in switch_path]
 
         return cls(src, dst, switch_path)
 
+    def as_onos_route(self, hosts: Hosts, switches: Switches) -> OnosRoute:
+        src, dst = hosts.get_host_id(self.src), hosts.get_host_id(self.dst)
+        switch_path = cast(list[str], [switches.get_switch_id(s) for s in self.switch_path])
+
+        return OnosRoute(key=f"{src}{dst}", route=[src, *switch_path, dst])
+
+
+@dataclass
+class Intent:
+    """A point-to-point intent or whatever"""
+
+    paths: list[IntentPath]
+
+    @classmethod
+    def parse(cls, data: dict[str, Any]) -> Intent:
+        paths = [IntentPath.parse(x) for x in data["paths"]]
+
+        return cls(paths)
+
+    def as_onos_intent(self, hosts: Hosts, switches: Switches) -> list[OnosRoute]:
+        return [p.as_onos_route(hosts, switches) for p in self.paths]
+
+
+# TODO: keep track of bandwidth
+# TODO: move drain part of this to the drain intent?
+
 
 @dataclass
 class Topology:
     network: networkx.Graph
-    intents: dict[tuple[str, str], Intent]
+    intents: list[Intent]
 
     @classmethod
     async def from_current_state(cls) -> Topology:
@@ -51,8 +78,43 @@ async def from_current_state(cls) -> Topology:
                 json={"api_key": config.api_key},
             )
             resp.raise_for_status()
-            intents = dict(
-                Intent.parse(i).as_tuple() for i in resp.json()["routingList"]
-            )
+            intents = [Intent.parse(i) for i in resp.json()["routingList"]]
 
         return cls(network, intents)
+
+    @staticmethod
+    def _perform_reroute(
+        network: networkx.Graph, path: list[str], drain_node: str
+    ) -> IntentPath:
+        """Reroute a `path` to avoid a node.
+
+        This currently splits the path on the `drain_node` we want to drain from, then
+        finds the shortest path between the two sides avoiding `drain_node`
+
+        `network` should be the network *without* `drain_node`.
+        """
+        node_idx = path.index(drain_node)
+        lhs, rhs = path[:node_idx], path[node_idx + 1 :]
+
+        # TODO: we need to detect when it's impossible to reroute, or if a host
+        # will be dropped, etc and warn/cancel the intent
+
+        reroute_path = networkx.shortest_path(network, lhs[-1], rhs[0])
+        reroute_path = cast(list[str], reroute_path)
+        new_path = lhs[:-1] + reroute_path + rhs[1:]
+        return IntentPath(src=lhs[0], dst=rhs[-1], switch_path=new_path[1:-1])
+
+    def drain_node(self, node: str) -> Intent:
+        """Construct an intent that drains traffic from a node."""
+
+        without_node: networkx.Graph = self.network.copy()
+        without_node.remove_node(node)
+
+        new_intent_paths = [
+            Topology._perform_reroute(without_node, intent_path.path, node)
+            for intent in self.intents
+            for intent_path in intent.paths
+            if node in intent_path.path
+        ]
+
+        return Intent(paths=new_intent_paths)

intent_deployer/notifiers.py

@@ -2,7 +2,7 @@
 
 import logging
 from dataclasses import dataclass
-from typing import Callable, Optional, Protocol, Type
+from typing import Callable, Optional, Protocol, Type, TYPE_CHECKING
 import httpx
 
 from yarl import URL
@@ -11,14 +11,19 @@
 
 log: logging.Logger = logging.getLogger(__name__)
 
-
-class Notifier(Protocol):
-    @classmethod
-    def parse(cls, body: dict) -> Notifier:
-        ...
-
-    async def send(self, msg: str):
-        ...
+if TYPE_CHECKING:
+    # BUG: in py3.9.7 data classes that inherit from typing.Protocol break so
+    # only actually use Protocol for type checking, use a plain class otherwise
+    class Notifier(Protocol):
+        @classmethod
+        def parse(cls, body: dict) -> Notifier:
+            ...
+
+        async def send(self, msg: str):
+            ...
+else:
+    class Notifier:
+        pass
 
 
 _notifiers: dict[str, Type[Notifier]] = {}

intents/drain.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from pydantic.main import BaseModel
+
+from intent_deployer import config
+from intent_deployer.compile import (
+    Context,
+    register_nile_handler,
+    register_dialogflow_handler,
+    register_intent_executor,
+)
+from intent_deployer.deploy import deploy_policy
+from intent_deployer.network_state import Topology
+from intent_deployer.hosts import Hosts
+from intent_deployer.parser import NileIntent
+from intent_deployer.switches import Switches
+from intent_deployer.types import IntentDeployException
+from intents.utils import PushIntentPolicy
+
+
+class DrainIntent(BaseModel):
+    node: str
+
+
+@register_nile_handler("drainIntent")
+def handle_nile_intent(intent: NileIntent) -> DrainIntent:
+    if len(intent.targets) != 1:
+        raise IntentDeployException(
+            "Exactly one target should be specified in the drain intent"
+        )
+
+    node = intent.targets[0].name
+
+    return DrainIntent(node=node)
+
+
+@register_dialogflow_handler("drain")
+def handle_dialogflow_intent(params: dict[str, Any]) -> DrainIntent:
+    return DrainIntent(node=params["node"])
+
+
+@register_intent_executor(DrainIntent, PushIntentPolicy)
+async def handle_drain_intent(ctx: Context, intent: DrainIntent):
+    hosts = await Hosts.from_api()
+    switches = await Switches.from_api()
+    topology = await Topology.from_current_state()
+    drain_intent = topology.drain_node(intent.node)
+
+    await ctx.confirm_if_needed(
+        f"This will drain intents from {intent.node} by rerouting {len(drain_intent.paths)} flows",
+        lambda: deploy_policy(
+            config.ngcdi_url / "push_intent",
+            PushIntentPolicy(
+                api_key=config.api_key,
+                routes=drain_intent.as_onos_intent(hosts, switches),
+            ),
+        ),
+    )

intents/forwarding.py

@@ -1,11 +1,9 @@
 from __future__ import annotations
-from intent_deployer.dialogflow import DialogflowContext
 
 import re
-from typing import Literal, TypedDict, Any
+from typing import Any
 
 from pydantic.main import BaseModel
-import httpx
 
 from intent_deployer import config
 from intent_deployer.compile import (
@@ -19,7 +17,7 @@
 from intent_deployer.parser import NileIntent
 from intent_deployer.switches import Switches
 from intent_deployer.types import IntentDeployException
-from intents.utils import Route, PushIntentPolicy, get_possible_routes
+from intents.utils import OnosRoute, PushIntentPolicy, get_possible_routes
 
 
 class ForwardIntent(BaseModel):
@@ -30,6 +28,7 @@ class ForwardIntent(BaseModel):
 
 ForwardIntent.update_forward_refs()
 
+
 @register_nile_handler("forwardIntent")
 def handle_nile_intent(intent: NileIntent) -> ForwardIntent:
     if len(intent.endpoints) != 1:
@@ -64,9 +63,6 @@ def handle_dialogflow_intent(params: dict[str, Any]) -> ForwardIntent:
 @register_intent_executor(ForwardIntent, PushIntentPolicy)
 async def handle_forward_intent(ctx: Context, intent: ForwardIntent):
 
-    # if isinstance(ctx, DialogflowContext) and (notifier := ctx.get_notifier()) is not None:
-    #     notifier.send("Heya")
-
     hosts = await Hosts.from_api()
     switches = await Switches.from_api()
 
@@ -97,7 +93,7 @@ async def handle_forward_intent(ctx: Context, intent: ForwardIntent):
         lambda: deploy_policy(
             config.ngcdi_url / "push_intent",
             PushIntentPolicy(
-                api_key=config.api_key, routes=[Route(key=src_id + dst_id, route=route)]
+                api_key=config.api_key, routes=[OnosRoute(key=src_id + dst_id, route=route)]
             ),
         ),
     )

intents/upgrade.py

@@ -5,29 +5,21 @@
 from intent_deployer.dialogflow import DialogflowContext
 
 import logging
-import re
-from typing import Literal, TypedDict, Any
+from typing import Any
 
 from pydantic.main import BaseModel
-import httpx
 
 # who designed this library?
 from osmclient.sol005.client import Client as OSMClient
-from osmclient.client import Client as make_osm_client
 
 from intent_deployer import config
 from intent_deployer.compile import (
     Context,
-    register_nile_handler,
     register_dialogflow_handler,
     register_intent_executor,
 )
 from intent_deployer.deploy import deploy_policy
-from intent_deployer.hosts import Hosts
-from intent_deployer.parser import NileIntent
-from intent_deployer.switches import Switches
-from intent_deployer.types import IntentDeployException, Policy
-from intents.utils import PushIntentPolicy, Route, get_possible_routes
+from intents.utils import PushIntentPolicy, OnosRoute
 
 log: logging.Logger = logging.getLogger(__name__)
 
@@ -61,7 +53,7 @@ def handle_dialogflow_downgrade_intent(params: dict[str, Any]) -> DowngradeInten
 up_reroute = PushIntentPolicy(
     api_key=config.api_key,
     routes=[
-        Route(
+        OnosRoute(
             key="00:00:00:00:00:06/None00:00:00:00:00:09/None",
             route=[
                 "00:00:00:00:00:06/None",
@@ -77,7 +69,7 @@ def handle_dialogflow_downgrade_intent(params: dict[str, Any]) -> DowngradeInten
 down_reroute = PushIntentPolicy(
     api_key=config.api_key,
     routes=[
-        Route(
+        OnosRoute(
             key="00:00:00:00:00:06/None00:00:00:00:00:09/None",
             route=[
                 "00:00:00:00:00:06/None",
@@ -235,7 +227,7 @@ async def perform_downgrade(ctx: Context):
 
 
 @register_intent_executor(UpgradeIntent)
-async def handle_upgrade_intent(ctx: Context, intent: UpgradeIntent):
+async def handle_upgrade_intent(ctx: Context, _intent: UpgradeIntent):
     src = "h6"
     dst = "h9"
 
@@ -272,7 +264,7 @@ async def handle_upgrade_intent(ctx: Context, intent: UpgradeIntent):
 
 
 @register_intent_executor(DowngradeIntent)
-async def handle_downgrade_intent(ctx: Context, intent: DowngradeIntent):
+async def handle_downgrade_intent(ctx: Context, _intent: DowngradeIntent):
     src = "h6"
     dst = "h9"
 

intents/utils.py

@@ -9,13 +9,14 @@
 from intent_deployer.types import IntentDeployException, Policy
 
 
-class Route(BaseModel):
+class OnosRoute(BaseModel):
     key: str
     route: list[str]
+    weight: int = 1
 
 
 class PushIntentPolicy(Policy):
-    routes: list[Route]
+    routes: list[OnosRoute]
 
 
 class GetRoutes(TypedDict):