Implement some form of state tracking

Author: simmsb

app.py

@@ -8,6 +8,7 @@
 from fastapi.exceptions import HTTPException, RequestValidationError
 from fastapi.params import Depends
 from fastapi.security import HTTPBasic
+from pydantic import UUID4
 from pydantic.main import BaseModel
 from starlette.background import BackgroundTasks
 from starlette.responses import JSONResponse
@@ -32,11 +33,15 @@
 from intent_deployer.config import get_settings
 from intent_deployer.deploy import deploy_policy
 from intent_deployer.dialogflow import DialogflowContext
+from intent_deployer.state_track import ActiveIntent, list_intents, lookup_intent, new_status_tracker
 from intent_deployer.switches import Switches
 from intent_deployer.types import (
     APIResponse,
     APIResponseStatus,
+    CreatedIntentResponse,
+    GetIntentResponse,
     IntentDeployException,
+    ListIntentResponse,
     NileIntentRequest,
 )
 from intents.utils import PushIntentPolicy
@@ -207,20 +212,25 @@ async def dialogflow_webhook(
     @app.post(
         f"/{intent.__name__}",
         name=f"invoke_{intent.__name__}",
-        response_model=APIResponse,
+        response_model=CreatedIntentResponse,
         responses={400: {"model": APIResponse}, 422: {"model": APIResponse}},
     )
     async def invoke_intent(body: intent, background_tasks: BackgroundTasks):
+        ctx = None
         try:
-            ctx = Context(background_tasks)
+            tracked_intent = new_status_tracker(intent.__name__, body)
+            ctx = Context(background_tasks, tracked_intent)
             await handle_intent(ctx, body)
 
-            return APIResponse(status=APIResponseStatus(code=200, details="success"))
+            return CreatedIntentResponse(status=APIResponseStatus(code=200, details="success"),
+                                         intent_id=tracked_intent.id)
         except IntentDeployException as e:
             log.exception(e)
+            if ctx: ctx.push_status("failure", str(e))
             return format_json_error(400, f"Could not deploy intent. \n{e}")
         except Exception as e:
             log.exception(e)
+            if ctx: ctx.push_status("failure", str(e))
             return format_json_error(400, f"Could not deploy intent. internal error")
 
 
@@ -234,3 +244,19 @@ async def alertmanager_webhook(
     await handle_alert(body)
 
     return APIResponse(status=APIResponseStatus(code=200, details="success"))
+
+@app.get("/intents", response_model=ListIntentResponse)
+async def list_intents_endpoint():
+    return ListIntentResponse(status=APIResponseStatus(code=200, details="success"),
+                              intents=list_intents())
+
+@app.get("/intents/{intent_id}", response_model=GetIntentResponse,
+         responses={404: {"model": APIResponse}})
+async def get_intent_endpoint(intent_id: UUID4):
+    intent_status = lookup_intent(intent_id)
+
+    if intent_status:
+        return GetIntentResponse(status=APIResponseStatus(code=200, details="success"),
+                                 intent=intent_status)
+
+    return format_json_error(404, f"Could not find an intent with id: {intent_id}")

intent_deployer/compile.py

@@ -13,6 +13,7 @@
 from fastapi import BackgroundTasks
 
 from intent_deployer.parser import NileIntent
+from intent_deployer.state_track import ActiveIntent, new_status_tracker
 from intent_deployer.types import (
     IntentDeployException,
     Policy,
@@ -27,6 +28,11 @@
 @dataclass
 class Context:
     background_tasks: BackgroundTasks
+    status: Optional[ActiveIntent]
+
+    def push_status(self, state: str, extra: Any = None):
+        assert self.status
+        self.status.push_status(state, extra)
 
     async def confirm_if_needed(
         self, msg: str, on_confirm: Callable[[], Coroutine[None, None, None]]
@@ -129,15 +135,17 @@ def wrapper(handler: AlertExecutor) -> AlertExecutor:
     return wrapper
 
 
-async def handle_nile(ctx: Context, intent: NileIntent):
-    handler = nile_handlers.get(intent.name)
+async def handle_nile(ctx: Context, nile_intent: NileIntent):
+    handler = nile_handlers.get(nile_intent.name)
     if handler is None:
         raise IntentDeployException(
-            f'No known nile handler for the intent type "{intent.name}"',
+            f'No known nile handler for the intent type "{nile_intent.name}"',
             f"Possible intents types are: {list(nile_handlers.keys())}",
         )
 
-    intent = handler(intent)
+    intent = handler(nile_intent)
+    tracked_intent_status = new_status_tracker(nile_intent.name, intent)
+    ctx.status = tracked_intent_status
     await handle_intent(ctx, intent)
 
 
@@ -150,6 +158,8 @@ async def handle_dialogflow(ctx: Context, type_: str, params: dict[str, Any]):
         )
 
     intent = handler(params)
+    tracked_intent_status = new_status_tracker(type_, intent)
+    ctx.status = tracked_intent_status
     await handle_intent(ctx, intent)
 
 

intent_deployer/dialogflow.py

@@ -65,7 +65,7 @@ def get_or_create(cls, background_tasks, session, *args, **kwargs):
             instance.background_tasks = background_tasks
             return instance
 
-        instance = cls(background_tasks, session, *args, **kwargs)
+        instance = cls(background_tasks, None, session, *args, **kwargs)
         cls.instances[session] = instance
         return instance
 
@@ -124,11 +124,14 @@ async def inner():
                     await notifier.send("Intent deployed")
                 except IntentDeployException as e:
                     log.exception(e)
+                    self.push_status("failure", str(e))
                     await notifier.send(f"Could not deploy that intent: {e}")
                 except Exception as e:
                     log.exception(e)
+                    self.push_status("failure", str(e))
                     await notifier.send(f"Could not deploy that intent, internal error")
             else:
+                self.push_status("cancelled")
                 await notifier.send("Cancelled")
 
         self.background_tasks.add_task(inner)

intent_deployer/state_track.py

@@ -0,0 +1,52 @@
+from datetime import datetime
+from typing import Any, Optional
+from uuid import uuid4
+
+from pydantic import UUID4
+from pydantic.main import BaseModel
+
+class IntentStatus(BaseModel):
+    """Represents the status an intent was in, such as: starting, testing, completed."""
+
+    state: str
+    extra: Any
+    start_time: datetime
+    end_time: Optional[datetime] = None
+
+# I can't think of a good name for this right now, basically this is something
+# that stores the current state of the intent
+
+class ActiveIntent(BaseModel):
+    id: UUID4
+    configuration: BaseModel
+    name: str
+    statuses: list[IntentStatus]
+
+    def push_status(self, state: str, extra: Any = None):
+        now = datetime.utcnow()
+
+        if self.statuses:
+            self.statuses[-1].end_time = now
+
+        self.statuses.append(IntentStatus(state=state, extra=extra, start_time=now))
+
+    def current_status(self) -> Optional[IntentStatus]:
+        return self.statuses[-1] if self.statuses else None
+
+
+__active_intents: dict[UUID4, ActiveIntent] = {}
+
+def new_status_tracker(name: str, configuration: BaseModel) -> ActiveIntent:
+    id_ = uuid4()
+
+    active_intent = ActiveIntent(id=id_, name=name, configuration=configuration, statuses=[])
+
+    __active_intents[id_] = active_intent
+
+    return active_intent
+
+def lookup_intent(id: UUID4) -> Optional[ActiveIntent]:
+    return __active_intents.get(id)
+
+def list_intents() -> list[ActiveIntent]:
+    return list(__active_intents.values())

intent_deployer/types.py

@@ -3,7 +3,9 @@
 from typing import Generic, Literal, TypeVar
 from pydantic.config import BaseConfig
 from pydantic.generics import GenericModel
-from pydantic import BaseModel
+from pydantic import UUID4, BaseModel
+
+from intent_deployer.state_track import ActiveIntent
 
 
 class APIResponseStatus(BaseModel):
@@ -15,6 +17,15 @@ class APIResponse(BaseModel):
     status: APIResponseStatus
 
 
+class CreatedIntentResponse(APIResponse):
+    intent_id: UUID4
+
+class ListIntentResponse(APIResponse):
+    intents: list[ActiveIntent]
+
+class GetIntentResponse(APIResponse):
+    intent: ActiveIntent
+
 class NileIntentRequest(BaseModel):
     type: Literal["nile"]
     intent: str

intents/drain.py

@@ -45,7 +45,8 @@ def handle_dialogflow_intent(params: dict[str, Any]) -> DrainIntent:
 
 @register_intent_executor(DrainIntent, PushIntentPolicy)
 async def handle_drain_intent(ctx: Context, intent: DrainIntent):
-    await raise_if_spp()
+    await raise_if_spp(ctx)
+    ctx.push_status("starting")
 
     hosts = await Hosts.from_api()
     switches = await Switches.from_api()
@@ -69,17 +70,20 @@ async def drain_intent_test_followup():
             if notifier is not None:
                 await notifier.send("Testing configuration, please wait")
 
+        ctx.push_status("testing")
         runner = NEATRunner.generate_mtv_config(new_topology)
         success = await runner.run()
 
         if success:
+            ctx.push_status("applying")
             if isinstance(ctx, DialogflowContext):
                 notifier = ctx.get_notifier()
                 if notifier is not None:
                     await notifier.send("Tests passed, applying intent")
             await add_intents(drain_intents, hosts, switches)
         if not success:
             raise IntentDeployException("After testing the new configuration the tests did not pass, abandoning")
+        ctx.push_status("complete")
 
     await ctx.confirm_if_needed(
         f"This will drain intents from {intent.node} by rerouting {len(drain_intents)} flows",

intents/forwarding.py

@@ -68,7 +68,8 @@ def handle_dialogflow_intent(params: dict[str, Any]) -> ForwardIntent:
 
 @register_intent_executor(ForwardIntent)
 async def handle_forward_intent(ctx: Context, intent: ForwardIntent):
-    await raise_if_spp()
+    await raise_if_spp(ctx)
+    ctx.push_status("starting")
 
     hosts = await Hosts.from_api()
     switches = await Switches.from_api()

intents/rebalance.py

@@ -33,7 +33,8 @@ def handle_dialogflow_intent(_params: dict[str, Any]) -> RebalanceIntent:
 
 @register_intent_executor(RebalanceIntent, PushIntentPolicy)
 async def handle_drain_intent(ctx: Context, intent: RebalanceIntent):
-    await raise_if_spp()
+    await raise_if_spp(ctx)
+    ctx.push_status("starting")
 
     hosts = await Hosts.from_api()
     switches = await Switches.from_api()
@@ -52,10 +53,12 @@ async def rebalance_intent_test_followup():
             if notifier is not None:
                 await notifier.send("Testing configuration, please wait")
 
+        ctx.push_status("testing")
         runner = NEATRunner.generate_mtv_config(new_topology)
         success = await runner.run()
 
         if success:
+            ctx.push_status("applying")
             if isinstance(ctx, DialogflowContext):
                 notifier = ctx.get_notifier()
                 if notifier is not None:
@@ -65,6 +68,7 @@ async def rebalance_intent_test_followup():
             raise IntentDeployException(
                 "After testing the new configuration the tests did not pass, abandoning"
             )
+        ctx.push_status("complete")
 
     await ctx.confirm_if_needed(
         f"This will reroute all intents",

intents/upgrade.py

@@ -90,7 +90,8 @@ def handle_dialogflow_downgrade_intent(params: dict[str, Any]) -> DowngradeInten
 
 @register_intent_executor(UpgradeIntent)
 async def handle_upgrade_intent(ctx: Context, intent: UpgradeIntent):
-    await raise_if_spp()
+    await raise_if_spp(ctx)
+    ctx.push_status("starting")
 
     topology = await Topology.from_current_state([])
 
@@ -101,6 +102,7 @@ async def handle_upgrade_intent(ctx: Context, intent: UpgradeIntent):
         )
 
     async def upgrade_intent_test_followup():
+        ctx.push_status("applying")
         with Telnet("172.17.0.1", 7123) as tn:
             x = tn.read_until(b"_________", timeout=3)
             print(x.decode(), end="")
@@ -113,6 +115,7 @@ async def upgrade_intent_test_followup():
             notifier = ctx.get_notifier()
             if notifier is not None:
                 await notifier.send("Reconfigured router!")
+        ctx.push_status("complete")
 
         # if isinstance(ctx, DialogflowContext):
         #     notifier = ctx.get_notifier()

intents/utils.py

@@ -6,6 +6,7 @@
 import httpx
 
 from intent_deployer import config
+from intent_deployer.compile import Context
 from intent_deployer.network_state import is_spp
 from intent_deployer.types import IntentDeployException, Policy
 
@@ -44,6 +45,7 @@ async def get_possible_routes(src: str, dst: str) -> list[list[str]]:
 
     return list(data["routes"].values())
 
-async def raise_if_spp():
+async def raise_if_spp(ctx: Context):
     if await is_spp():
+        ctx.push_status("failed", "spp enabled")
         raise IntentDeployException("SPP Active, no modification allowed")