diff options
Diffstat (limited to 'src/services/api')
| -rw-r--r-- | src/services/api/background.py | 179 | ||||
| -rw-r--r-- | src/services/api/rest/routers.py | 128 |
2 files changed, 287 insertions, 20 deletions
diff --git a/src/services/api/background.py b/src/services/api/background.py new file mode 100644 index 000000000..2b25af307 --- /dev/null +++ b/src/services/api/background.py @@ -0,0 +1,179 @@ +# Copyright VyOS maintainers and contributors <maintainers@vyos.io> +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public License +# along with this library. If not, see <http://www.gnu.org/licenses/>. + +import time +import functools +from collections import deque +from enum import Enum +from threading import Lock +from typing import Any +from typing import Callable +from typing import Optional +from uuid import uuid4 + +from fastapi import BackgroundTasks +from pydantic import BaseModel +from pydantic import StrictStr +from pydantic import StrictInt + + +def _ts(): + """Return current Unix timestamp (seconds since epoch)""" + return int(time.time()) + + +class BackgroundOpStatus(str, Enum): + queued = 'queued' + running = 'running' + succeeded = 'succeeded' + failed = 'failed' + + @property + def is_completed(self): + """True if the operation is in a terminal state (succeeded/failed)""" + return self in (BackgroundOpStatus.succeeded, BackgroundOpStatus.failed) + + +class BackgroundOpRecord(BaseModel): + """Metadata and outcome for a single background operation""" + + op_id: StrictStr + created_at: StrictInt + started_at: Optional[StrictInt] = None + finished_at: Optional[StrictInt] = None + status: BackgroundOpStatus = BackgroundOpStatus.queued + result: Optional[Any] = None + error: Optional[StrictStr] = None + + +class BackgroundOpError(Exception): + """Raised when a background operation cannot be enqueued/executed""" + + pass + + +class BackgroundOpManager: + """ + In-memory FIFO operation queue. + + Uses BackgroundTasks to schedule a `drain()` call after the response, + so `enqueue()` is fast and non-blocking for the client. + """ + + DEFAULT_MAX_QUEUE_SIZE = 128 + + def __init__(self, max_queue_size: int = DEFAULT_MAX_QUEUE_SIZE): + # max number of queued (pending) operations allowed at a time + self._max_queue_size = max_queue_size + + # FIFO queue of operation IDs waiting to be executed + self._queue = deque() + self._jobs = {} + self._workers = {} + + # protects _queue/_jobs/_workers/_drain_scheduled from concurrent access + self._mx = Lock() + + # whether a drain task has already been scheduled via BackgroundTasks + self._drain_scheduled = False + + def enqueue( + self, + background_tasks: BackgroundTasks, + func: Callable, + *args, + **kwargs, + ) -> BackgroundOpRecord: + """Enqueue a function for background execution and return its record""" + + assert isinstance(background_tasks, BackgroundTasks) + assert callable(func), '`func` argument should be function or lambda' + + with self._mx: + if len(self._queue) >= self._max_queue_size: + raise BackgroundOpError( + f'Background operation queue is full ({self._max_queue_size})' + ) + + op_id = str(uuid4()) + record = BackgroundOpRecord(op_id=op_id, created_at=_ts()) + + self._jobs[op_id] = record + # store the callable for later execution (outside the lock) + self._workers[op_id] = functools.partial(func, *args, **kwargs) + self._queue.append(op_id) + + if not self._drain_scheduled: + # schedule a single drain() call after the current response + background_tasks.add_task(self.drain) + self._drain_scheduled = True + + # Best-effort pruning: keep history bounded by dropping oldest completed records + if len(self._jobs) > self._max_queue_size: + oldest = min(self._jobs.values(), key=lambda record: record.created_at) + if oldest.status.is_completed: + del self._jobs[oldest.op_id] + + return record + + def drain(self): + """Run queued operations sequentially until the queue is empty""" + + while True: + with self._mx: + if not self._queue: + # allow future enqueue() calls to schedule the next drain() + self._drain_scheduled = False + return + + op_id = self._queue.popleft() + record = self._jobs[op_id] + func = self._workers.pop(op_id) + + record.status = BackgroundOpStatus.running + record.started_at = _ts() + + # execute outside the lock to avoid blocking enqueues/status reads + result = error = status = None + try: + result = func() + except Exception as e: # noqa: BLE001 + status = BackgroundOpStatus.failed + error = str(e) + else: + status = BackgroundOpStatus.succeeded + + with self._mx: + record.result = result + record.error = error + record.status = status + record.finished_at = _ts() + + def get_record(self, op_id: str) -> BackgroundOpRecord | None: + """Return a deep copy of a single record""" + + with self._mx: + record = self._jobs.get(op_id) + return record.copy(deep=True) if record else None + + def get_records(self) -> list: + """Return deep copies of all records, sorted oldest-first by created_at""" + + with self._mx: + records = [record.copy(deep=True) for record in self._jobs.values()] + + # stable-ish ordering (oldest first) + records.sort(key=lambda record: record.created_at) + return records diff --git a/src/services/api/rest/routers.py b/src/services/api/rest/routers.py index 438b3a23e..c8102c980 100644 --- a/src/services/api/rest/routers.py +++ b/src/services/api/rest/routers.py @@ -28,6 +28,7 @@ from typing import Callable from typing import TYPE_CHECKING from fastapi import Depends +from fastapi import Query from fastapi import Request from fastapi import Response from fastapi import HTTPException @@ -46,6 +47,8 @@ from vyos.configtree import ConfigTree from vyos.configdiff import get_config_diff from vyos.configsession import ConfigSessionError +from ..background import BackgroundOpManager +from ..background import BackgroundOpError from ..session import SessionState from .models import success from .models import error @@ -230,7 +233,7 @@ class MultipartRequest(Request): 400, f"Malformed command '{0}': 'path' field must be a list of strings", ) - if endpoint in ('/configure'): + if endpoint in ('/configure',): if not c['path']: self.form_err = ( 400, @@ -241,7 +244,7 @@ class MultipartRequest(Request): 400, f"Malformed command '{c}': 'value' field must be a string", ) - if endpoint in ('/configure-section'): + if endpoint in ('/configure-section',): if 'section' not in c and 'config' not in c: self.form_err = ( 400, @@ -293,6 +296,10 @@ router = APIRouter( self_ref_msg = 'Requested HTTP API server configuration change; commit will be called in the background' +# Global background-op manager used by the REST API to run long config commits after the response +background_op_manager = BackgroundOpManager() + + def call_commit(s: SessionState): try: s.session.commit() @@ -342,7 +349,7 @@ def run_commit_confirm(s: SessionState): del env['IN_COMMIT_CONFIRM'] -async def _configure_op( +def _execute_configure_op( data: Union[ ConfirmModel, ConfigureModel, @@ -351,12 +358,15 @@ async def _configure_op( ConfigSectionListModel, ConfigSectionTreeModel, ], - _request: Request, - background_tasks: BackgroundTasks, + background_tasks: BackgroundTasks | None = None, ): # pylint: disable=too-many-branches,too-many-locals,too-many-nested-blocks,too-many-statements # pylint: disable=consider-using-with + # True when invoked by the background operation + # runner (no FastAPI BackgroundTasks context passed in) + is_background_job = background_tasks is None + state = SessionState() session = state.session env = session.get_session_env() @@ -385,6 +395,8 @@ async def _configure_op( try: for c in data: op = c.op + op_error = ConfigSessionError(f"'{op}' is not a valid operation") + if not isinstance(c, (ConfirmModel, BaseConfigSectionTreeModel)): path = c.path @@ -392,7 +404,7 @@ async def _configure_op( if op == 'confirm': msg = session.confirm() else: - raise ConfigSessionError(f"'{op}' is not a valid operation") + raise op_error elif isinstance(c, BaseConfigureModel): if c.value: @@ -422,7 +434,7 @@ async def _configure_op( elif op == 'comment': session.comment(path, value=value) else: - raise ConfigSessionError(f"'{op}' is not a valid operation") + raise op_error elif isinstance(c, BaseConfigSectionModel): if op == 'set': @@ -430,7 +442,7 @@ async def _configure_op( elif op == 'load': session.load_section(path, section) else: - raise ConfigSessionError(f"'{op}' is not a valid operation") + raise op_error elif isinstance(c, BaseConfigSectionTreeModel): if op == 'set': @@ -438,7 +450,7 @@ async def _configure_op( elif op == 'load': session.load_section_tree(mask, config) else: - raise ConfigSessionError(f"'{op}' is not a valid operation") + raise op_error # end for config = Config(session_env=env) @@ -448,20 +460,29 @@ async def _configure_op( if not d.is_node_changed(['service', 'https']): if confirm_time: - out, err = await run_in_threadpool(run_commit_confirm, state) + out, err = run_commit_confirm(state) if err: raise err msg = msg + out if msg else out else: - out, err = await run_in_threadpool(run_commit, state) + out, err = run_commit(state) if err: raise err msg = msg + out if msg else out else: - if confirm_time: - background_tasks.add_task(call_commit_confirm, state) + if is_background_job: + # If already running as a background job, commit synchronously here + if confirm_time: + call_commit_confirm(state) + else: + call_commit(state) else: - background_tasks.add_task(call_commit, state) + # Otherwise schedule the commit to run after the HTTP response + if confirm_time: + background_tasks.add_task(call_commit_confirm, state) + else: + background_tasks.add_task(call_commit, state) + out = self_ref_msg msg = msg + out if msg else out @@ -484,12 +505,62 @@ async def _configure_op( del env['IN_COMMIT_CONFIRM'] lock.release() + # Background jobs return raw success text or raise on failure; + # the API wrapper formats HTTP responses and returns it + if is_background_job: + if status == 200: + return msg + else: + raise RuntimeError(error_msg) + if status != 200: return error(status, error_msg) return success(msg) +async def _configure_op( + data: Union[ + ConfirmModel, + ConfigureModel, + ConfigureListModel, + ConfigSectionModel, + ConfigSectionListModel, + ConfigSectionTreeModel, + ], + background_tasks: BackgroundTasks, + in_background: bool = False, +): + """ + API wrapper for configure operations. + + If `in_background=True`: enqueue the whole configure + workflow and return an operation record immediately. + Otherwise: run the configure workflow in a threadpool + and return the normal API response. + """ + + if in_background: + try: + # Enqueue and return an operation handle that + # can be polled via `/retrieve/background-operations` + record = background_op_manager.enqueue( + background_tasks, + _execute_configure_op, + data, + ) + except BackgroundOpError as e: + return error(500, str(e)) + + return success({'operation': record.model_dump()}) + + return await run_in_threadpool( + _execute_configure_op, + data, + background_tasks=background_tasks, + ) + + def create_path_import_pki_no_prompt(path): correct_paths = ['ca', 'certificate', 'key-pair'] if path[1] not in correct_paths: @@ -504,10 +575,9 @@ async def configure_op( data: Union[ConfigureModel, ConfigureListModel, ConfirmModel], request: Request, background_tasks: BackgroundTasks, + in_background: bool = Query(False), ): - out = await _configure_op(data, request, background_tasks) - - return out + return await _configure_op(data, background_tasks, in_background) @router.post('/configure-section') @@ -515,10 +585,9 @@ async def configure_section_op( data: Union[ConfigSectionModel, ConfigSectionListModel, ConfigSectionTreeModel], request: Request, background_tasks: BackgroundTasks, + in_background: bool = Query(False), ): - out = await _configure_op(data, request, background_tasks) - - return out + return await _configure_op(data, background_tasks, in_background) @router.post('/retrieve') @@ -565,6 +634,25 @@ async def retrieve_op(data: RetrieveModel): return success(res) +@router.post('/retrieve/background-operations') +async def retrieve_background_operations( + op_id: str = Query(None), +): + if op_id: + # Return only that record + record = background_op_manager.get_record(op_id) + records = [record] if record else [] + else: + # Return the full in-memory operation history (oldest first) + records = background_op_manager.get_records() + + result = { + 'operations': [record.model_dump() for record in records], + } + + return success(result) + + @router.post('/config-file') async def config_file_op(data: ConfigFileModel, background_tasks: BackgroundTasks): state = SessionState() |