mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-10-12 21:45:41 +03:00
feat(api): return node and add wait option on node lifecycle calls
This commit is contained in:
parent
2fb7056cb9
commit
6a416d1a2a
@ -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
|
||||
@ -326,46 +330,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:
|
||||
await node.start(data=start_data)
|
||||
except HTTPException as e:
|
||||
if not e.status_code == status.HTTP_405_METHOD_NOT_ALLOWED:
|
||||
raise
|
||||
return await _run_lifecycle(node, lambda: node.start(data=start_data), "started", wait, timeout)
|
||||
|
||||
|
||||
@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:
|
||||
await node.stop()
|
||||
except HTTPException as e:
|
||||
if not e.status_code == status.HTTP_405_METHOD_NOT_ALLOWED:
|
||||
raise
|
||||
return await _run_lifecycle(node, lambda: node.stop(strict=True), "stopped", wait, timeout)
|
||||
|
||||
|
||||
@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.
|
||||
|
||||
@ -375,24 +421,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:
|
||||
await node.reload()
|
||||
except HTTPException as e:
|
||||
if not e.status_code == status.HTTP_405_METHOD_NOT_ALLOWED:
|
||||
raise
|
||||
return await _run_lifecycle(node, node.reload, "started", wait, timeout)
|
||||
|
||||
|
||||
@router.post(
|
||||
|
||||
@ -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:
|
||||
# For IOU: we need to send the licence everytime we start a node
|
||||
if self.node_type == "iou":
|
||||
license_check = self._project.controller.iou_license.get("license_check", True)
|
||||
iourc_content = self._project.controller.iou_license.get("iourc_content", None)
|
||||
await self.post(
|
||||
"/start", timeout=240, data={"license_check": license_check, "iourc_content": iourc_content}
|
||||
)
|
||||
else:
|
||||
await self.post("/start", data=data, timeout=240)
|
||||
except asyncio.TimeoutError:
|
||||
raise ControllerTimeoutError(f"Timeout when starting {self._name}")
|
||||
async with self._transitioning("starting"):
|
||||
try:
|
||||
# For IOU: we need to send the licence everytime we start a node
|
||||
if self.node_type == "iou":
|
||||
license_check = self._project.controller.iou_license.get("license_check", True)
|
||||
iourc_content = self._project.controller.iou_license.get("iourc_content", None)
|
||||
await self.post(
|
||||
"/start", timeout=240, data={"license_check": license_check, "iourc_content": iourc_content}
|
||||
)
|
||||
else:
|
||||
await self.post("/start", data=data, timeout=240)
|
||||
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:
|
||||
await self.post("/stop", timeout=240, dont_connect=True)
|
||||
# We don't care if a node is down at this step
|
||||
except (ComputeError, ControllerError):
|
||||
pass
|
||||
except asyncio.TimeoutError:
|
||||
raise ControllerTimeoutError(f"Timeout when stopping {self._name}")
|
||||
stopped = False
|
||||
async with self._transitioning("stopping"):
|
||||
try:
|
||||
await self.post("/stop", timeout=240, dont_connect=True)
|
||||
stopped = True
|
||||
except (ComputeError, ControllerError):
|
||||
# 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],
|
||||
|
||||
@ -36,6 +36,8 @@ class NodeStatus(str, Enum):
|
||||
stopped = "stopped"
|
||||
started = "started"
|
||||
suspended = "suspended"
|
||||
starting = "starting"
|
||||
stopping = "stopping"
|
||||
|
||||
|
||||
class CustomAdapter(BaseModel):
|
||||
|
||||
@ -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
|
||||
@ -244,7 +245,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
|
||||
@ -252,6 +254,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(
|
||||
@ -260,7 +340,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
|
||||
@ -290,7 +371,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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user