Merge pull request #2952 from markparonyan/api-issue-02-rp8mx3

feat(api): return node and add wait option on node lifecycle calls

gns3server/api/routes/controller/nodes.py

@@ -34,7 +34,11 @@ from fastapi.routing import APIRoute
 from gns3server import schemas
 from gns3server.config import Config
 from gns3server.controller import Controller
-from gns3server.controller.controller_error import ControllerBadRequestError, ControllerForbiddenError
+from gns3server.controller.controller_error import (
+    ControllerBadRequestError,
+    ControllerForbiddenError,
+    ControllerTimeoutError,
+)
 from gns3server.controller.node import Node
 from gns3server.controller.project import Project
 from gns3server.db.repositories.rbac import RbacRepository
@@ -363,46 +367,88 @@ async def duplicate_node(duplicate_data: schemas.NodeDuplicate, node: Node = Dep
     return new_node.asdict()
 
 
+LIFECYCLE_RESPONSES: dict[int | str, dict[str, Any]] = {
+    status.HTTP_408_REQUEST_TIMEOUT: {"model": schemas.ErrorMessage, "description": "Timeout waiting for the node"},
+}
+
+
+async def _run_lifecycle(
+    node: Node, action: Callable, target: str, wait: bool, timeout: int, tolerate_unsupported: bool = True
+) -> dict:
+    async def _run():
+        try:
+            await action()
+        except HTTPException as e:
+            if not tolerate_unsupported or not e.status_code == status.HTTP_405_METHOD_NOT_ALLOWED:
+                raise
+            return
+        if wait and not node.is_always_running():
+            await node.wait_for_status(target)
+
+    if not wait:
+        await _run()
+        return node.asdict()
+    try:
+        await asyncio.wait_for(_run(), timeout)
+    except asyncio.TimeoutError:
+        raise ControllerTimeoutError(f"Timeout when waiting for {node.name} to be {target}")
+    return node.asdict()
+
+
 @router.post(
-    "/{node_id}/start", status_code=status.HTTP_204_NO_CONTENT, dependencies=[Depends(has_privilege("Node.PowerMgmt"))]
+    "/{node_id}/start",
+    response_model=schemas.Node,
+    responses=LIFECYCLE_RESPONSES,
+    dependencies=[Depends(has_privilege("Node.PowerMgmt"))],
 )
-async def start_node(start_data: Optional[dict] = None, node: Node = Depends(dep_node)) -> None:
+async def start_node(
+    start_data: Optional[dict] = None,
+    wait: bool = Query(False, description="Return only when the node status is 'started'"),
+    timeout: int = Query(240, ge=1, le=3600, description="Seconds to wait when 'wait' is true"),
+    node: Node = Depends(dep_node),
+) -> dict:
     """
     Start a node.
 
     Required privilege: Node.PowerMgmt
     """
 
-    try:
+    return await _run_lifecycle(node, lambda: node.start(data=start_data), "started", wait, timeout)
-        await node.start(data=start_data)
-    except HTTPException as e:
-        if not e.status_code == status.HTTP_405_METHOD_NOT_ALLOWED:
-            raise
 
 
 @router.post(
-    "/{node_id}/stop", status_code=status.HTTP_204_NO_CONTENT, dependencies=[Depends(has_privilege("Node.PowerMgmt"))]
+    "/{node_id}/stop",
+    response_model=schemas.Node,
+    responses=LIFECYCLE_RESPONSES,
+    dependencies=[Depends(has_privilege("Node.PowerMgmt"))],
 )
-async def stop_node(node: Node = Depends(dep_node)) -> None:
+async def stop_node(
+    wait: bool = Query(False, description="Return only when the node status is 'stopped'"),
+    timeout: int = Query(240, ge=1, le=3600, description="Seconds to wait when 'wait' is true"),
+    node: Node = Depends(dep_node),
+) -> dict:
     """
     Stop a node.
 
+    Errors reported while stopping are returned to the caller.
+
     Required privilege: Node.PowerMgmt
     """
 
-    try:
+    return await _run_lifecycle(node, lambda: node.stop(strict=True), "stopped", wait, timeout)
-        await node.stop()
-    except HTTPException as e:
-        if not e.status_code == status.HTTP_405_METHOD_NOT_ALLOWED:
-            raise
 
 
 @router.post(
     "/{node_id}/suspend",
-    status_code=status.HTTP_204_NO_CONTENT,
+    response_model=schemas.Node,
+    responses=LIFECYCLE_RESPONSES,
     dependencies=[Depends(has_privilege("Node.PowerMgmt"))],
 )
-async def suspend_node(node: Node = Depends(dep_node)) -> None:
+async def suspend_node(
+    wait: bool = Query(False, description="Return only when the node status is 'suspended'"),
+    timeout: int = Query(240, ge=1, le=3600, description="Seconds to wait when 'wait' is true"),
+    node: Node = Depends(dep_node),
+) -> dict:
     """
     Suspend a node.
 
@@ -412,24 +458,27 @@ async def suspend_node(node: Node = Depends(dep_node)) -> None:
     Required privilege: Node.PowerMgmt
     """
 
-    await node.suspend()
+    return await _run_lifecycle(node, node.suspend, "suspended", wait, timeout, tolerate_unsupported=False)
 
 
 @router.post(
-    "/{node_id}/reload", status_code=status.HTTP_204_NO_CONTENT, dependencies=[Depends(has_privilege("Node.PowerMgmt"))]
+    "/{node_id}/reload",
+    response_model=schemas.Node,
+    responses=LIFECYCLE_RESPONSES,
+    dependencies=[Depends(has_privilege("Node.PowerMgmt"))],
 )
-async def reload_node(node: Node = Depends(dep_node)) -> None:
+async def reload_node(
+    wait: bool = Query(False, description="Return only when the node status is 'started'"),
+    timeout: int = Query(240, ge=1, le=3600, description="Seconds to wait when 'wait' is true"),
+    node: Node = Depends(dep_node),
+) -> dict:
     """
     Reload a node.
 
     Required privilege: Node.PowerMgmt
     """
 
-    try:
+    return await _run_lifecycle(node, node.reload, "started", wait, timeout)
-        await node.reload()
-    except HTTPException as e:
-        if not e.status_code == status.HTTP_405_METHOD_NOT_ALLOWED:
-            raise
 
 
 @router.post(

gns3server/controller/node.py

@@ -16,6 +16,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import asyncio
+import contextlib
 import copy
 import html
 import logging
@@ -134,6 +135,8 @@ class Node:
         self._command_line = None
         self._node_directory = None
         self._status = "stopped"
+        self._status_changed = asyncio.Event()
+        self._transition = None
         self._template_id = template_id
         self._x = 0
         self._y = 0
@@ -785,7 +788,7 @@ class Node:
             elif key == "command_line":
                 self._command_line = value
             elif key == "status":
-                self._status = value
+                self._set_status(value)
             elif key == "console_type":
                 self._console_type = value
             elif key == "aux_type":
@@ -901,6 +904,30 @@ class Node:
     async def destroy(self):
         await self.delete()
 
+    @contextlib.asynccontextmanager
+    async def _transitioning(self, transition):
+        """
+        Report a transitional status while a lifecycle call is in flight
+        """
+
+        self._transition = transition
+        self.project.emit_notification("node.updated", self.asdict())
+        try:
+            yield
+        finally:
+            self._transition = None
+
+    async def wait_for_status(self, status):
+        """
+        Wait until the controller side status of the node matches
+
+        :param status: expected status
+        """
+
+        while self._status != status:
+            self._status_changed.clear()
+            await self._status_changed.wait()
+
     async def start(self, data=None):
         """
         Start a node
@@ -924,32 +951,56 @@ class Node:
                     f"Cannot start node '{self._name}': {len(failed_links)} deferred link(s) "
                     "could not be restored. Please try again."
                 )
-        try:
+        async with self._transitioning("starting"):
-            # For IOU: we need to send the licence everytime we start a node
+            try:
-            if self.node_type == "iou":
+                # For IOU: we need to send the licence everytime we start a node
-                license_check = self._project.controller.iou_license.get("license_check", True)
+                if self.node_type == "iou":
-                iourc_content = self._project.controller.iou_license.get("iourc_content", None)
+                    license_check = self._project.controller.iou_license.get("license_check", True)
-                await self.post(
+                    iourc_content = self._project.controller.iou_license.get("iourc_content", None)
-                    "/start", timeout=240, data={"license_check": license_check, "iourc_content": iourc_content}
+                    await self.post(
-                )
+                        "/start", timeout=240, data={"license_check": license_check, "iourc_content": iourc_content}
-            else:
+                    )
-                await self.post("/start", data=data, timeout=240)
+                else:
-        except asyncio.TimeoutError:
+                    await self.post("/start", data=data, timeout=240)
-            raise ControllerTimeoutError(f"Timeout when starting {self._name}")
+            except asyncio.TimeoutError:
+                raise ControllerTimeoutError(f"Timeout when starting {self._name}")
+        self._confirm_status("started")
 
-    async def stop(self):
+    async def stop(self, strict=False):
         """
         Stop a node
+
+        :param strict: Raise the errors reported by the compute instead of ignoring them
         """
         if self.missing_image:
             return
-        try:
+        stopped = False
-            await self.post("/stop", timeout=240, dont_connect=True)
+        async with self._transitioning("stopping"):
-        # We don't care if a node is down at this step
+            try:
-        except (ComputeError, ControllerError):
+                await self.post("/stop", timeout=240, dont_connect=True)
-            pass
+                stopped = True
-        except asyncio.TimeoutError:
+            except (ComputeError, ControllerError):
-            raise ControllerTimeoutError(f"Timeout when stopping {self._name}")
+                # We don't care if a node is down at this step
+                if strict:
+                    raise
+            except asyncio.TimeoutError:
+                raise ControllerTimeoutError(f"Timeout when stopping {self._name}")
+        if stopped:
+            self._confirm_status("stopped")
+
+    def _set_status(self, status):
+        self._status = status
+        self._status_changed.set()
+
+    def _confirm_status(self, status):
+        """
+        Record the state reached once the compute accepted a lifecycle call,
+        without waiting for the asynchronous compute notification
+        """
+
+        if not self.is_always_running():
+            self._set_status(status)
+        self.project.emit_notification("node.updated", self.asdict())
 
     async def suspend(self):
         """
@@ -958,7 +1009,8 @@ class Node:
         try:
             await self.post("/suspend", timeout=240)
         except asyncio.TimeoutError:
-            raise ControllerTimeoutError(f"Timeout when reloading {self._name}")
+            raise ControllerTimeoutError(f"Timeout when suspending {self._name}")
+        self._confirm_status("suspended")
 
     async def reload(self):
         """
@@ -968,6 +1020,7 @@ class Node:
             await self.post("/reload", timeout=240)
         except asyncio.TimeoutError:
             raise ControllerTimeoutError(f"Timeout when reloading {self._name}")
+        self._confirm_status("started")
 
     async def reset_console(self):
         """
@@ -1280,7 +1333,7 @@ class Node:
         additional_data = {
             "project_id": self._project.id,
             "command_line": self._command_line,
-            "status": self._status,
+            "status": self._transition or self._status,
             "console_host": str(self._compute.console_host),
             "node_directory": self._node_directory,
             "ports": [port.asdict() for port in self.ports],

gns3server/schemas/common.py

@@ -41,6 +41,8 @@ class NodeStatus(str, Enum):
     stopped = "stopped"
     started = "started"
     suspended = "suspended"
+    starting = "starting"
+    stopping = "stopping"
 
 
 class CustomAdapter(BaseModel):

tests/api/routes/controller/test_nodes.py

@@ -34,6 +34,7 @@ from pydantic import SecretStr
 from gns3server.api.routes.controller.nodes import vnc_console, ws_console
 from gns3server.config import Config
 from gns3server.controller.compute import Compute
+from gns3server.controller.controller_error import ComputeError
 from gns3server.controller.node import Node
 from gns3server.controller.project import Project
 from gns3server.services import auth_service
@@ -363,7 +364,8 @@ class TestNodeRoutes:
 
         compute.post = AsyncioMagicMock()
         response = await client.post(app.url_path_for("start_node", project_id=project.id, node_id=node.id), json={})
-        assert response.status_code == status.HTTP_204_NO_CONTENT
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json()["status"] == "started"
 
     async def test_stop_node(
         self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
@@ -371,6 +373,84 @@ class TestNodeRoutes:
 
         compute.post = AsyncioMagicMock()
         response = await client.post(app.url_path_for("stop_node", project_id=project.id, node_id=node.id))
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json()["status"] == "stopped"
+
+    async def test_start_node_wait(
+        self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
+    ) -> None:
+
+        compute.post = AsyncioMagicMock()
+        response = await client.post(
+            app.url_path_for("start_node", project_id=project.id, node_id=node.id) + "?wait=true&timeout=5", json={}
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json()["status"] == "started"
+        response = await client.get(app.url_path_for("get_node", project_id=project.id, node_id=node.id))
+        assert response.json()["status"] == "started"
+
+    async def test_start_node_reports_starting_while_in_flight(
+        self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
+    ) -> None:
+
+        seen = []
+
+        async def post(*args, **kwargs):
+            seen.append(node.asdict()["status"])
+
+        compute.post = post
+        response = await client.post(app.url_path_for("start_node", project_id=project.id, node_id=node.id), json={})
+        assert response.status_code == status.HTTP_200_OK
+        assert seen == ["starting"]
+        assert response.json()["status"] == "started"
+
+    async def test_stop_node_reports_stopping_while_in_flight(
+        self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
+    ) -> None:
+
+        seen = []
+
+        async def post(*args, **kwargs):
+            seen.append(node.asdict()["status"])
+
+        compute.post = post
+        node._status = "started"
+        response = await client.post(app.url_path_for("stop_node", project_id=project.id, node_id=node.id))
+        assert response.status_code == status.HTTP_200_OK
+        assert seen == ["stopping"]
+        assert response.json()["status"] == "stopped"
+
+    async def test_start_node_wait_timeout(
+        self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
+    ) -> None:
+
+        async def post(*args, **kwargs):
+            await asyncio.sleep(10)
+
+        compute.post = post
+        response = await client.post(
+            app.url_path_for("start_node", project_id=project.id, node_id=node.id) + "?wait=true&timeout=1", json={}
+        )
+        assert response.status_code == status.HTTP_408_REQUEST_TIMEOUT
+        assert "message" in response.json()
+        assert node.asdict()["status"] == "stopped"
+
+    async def test_stop_node_reports_compute_error(
+        self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
+    ) -> None:
+
+        compute.post = AsyncioMagicMock(side_effect=ComputeError("compute unreachable"))
+        node._status = "started"
+        response = await client.post(app.url_path_for("stop_node", project_id=project.id, node_id=node.id))
+        assert response.status_code == status.HTTP_409_CONFLICT
+        assert node.asdict()["status"] == "started"
+
+    async def test_stop_all_nodes_tolerates_compute_error(
+        self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
+    ) -> None:
+
+        compute.post = AsyncioMagicMock(side_effect=ComputeError("compute unreachable"))
+        response = await client.post(app.url_path_for("stop_all_nodes", project_id=project.id))
         assert response.status_code == status.HTTP_204_NO_CONTENT
 
     async def test_suspend_node(
@@ -379,7 +459,8 @@ class TestNodeRoutes:
 
         compute.post = AsyncioMagicMock()
         response = await client.post(app.url_path_for("suspend_node", project_id=project.id, node_id=node.id))
-        assert response.status_code == status.HTTP_204_NO_CONTENT
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json()["status"] == "suspended"
 
     async def test_suspend_node_unsupported(
         self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
@@ -409,7 +490,8 @@ class TestNodeRoutes:
 
         compute.post = AsyncioMagicMock()
         response = await client.post(app.url_path_for("reload_node", project_id=project.id, node_id=node.id))
-        assert response.status_code == status.HTTP_204_NO_CONTENT
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json()["status"] == "started"
 
     async def test_isolate_node(
         self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node