diff --git a/AGENTS.md b/AGENTS.md index 29d70fba5..5e341eb13 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -26,7 +26,7 @@ mypy simplyblock_web simplyblock_cli simplyblock_core # Type check (or: tox -e ## Architecture -Three packages, one entry point: +Three packages, two front-ends (CLI and Web API) over a shared core: | Package | Role | |---------|------| @@ -34,7 +34,12 @@ Three packages, one entry point: | `simplyblock_core/` | Business logic, data models, background services, FDB access | | `simplyblock_web/` | REST API — FastAPI (v2) + Flask (v1) hybrid on a single uvicorn process | -Data flows: **CLI → Web API → Core controllers → FoundationDB**. Storage nodes are reached via JSON-RPC (`rpc_client.py`). +Two front-ends sit on top of the core and **both call it in-process** — the CLI does *not* go through the Web API: + +- **`sbctl` CLI → Core controllers → FoundationDB.** `clibase.py` imports `simplyblock_core` directly (`cluster_ops`, `storage_node_ops`, the `controllers` package, `DBController`) and instantiates `DBController()` itself. +- **Web API → Core controllers → FoundationDB.** `simplyblock_web` is a separate entry point serving remote/programmatic clients over the same core. + +Storage nodes are reached via JSON-RPC (`rpc_client.py`). ## Coding Conventions @@ -109,3 +114,13 @@ Edit only `AGENTS.md` files and `.agents/skills/` contents. Never edit the symli ### Local overrides At every level where an `AGENTS.md` exists, also check for a sibling `AGENTS.local.md`. If present, load it in addition to `AGENTS.md` — its contents extend or override the checked-in instructions. `AGENTS.local.md` is gitignored and intended for per-developer notes that should not be committed. + +## graphify + +This project has a knowledge graph at graphify-out/ with god nodes, community structure, and cross-file relationships. + +Rules: +- For codebase questions, first run `graphify query ""` when graphify-out/graph.json exists. Use `graphify path "" ""` for relationships and `graphify explain ""` for focused concepts. These return a scoped subgraph, usually much smaller than GRAPH_REPORT.md or raw grep output. +- If graphify-out/wiki/index.md exists, use it for broad navigation instead of raw source browsing. +- Read graphify-out/GRAPH_REPORT.md only for broad architecture review or when query/path/explain do not surface enough context. +- After modifying code, run `graphify update .` to keep the graph current (AST-only, no API cost). diff --git a/requirements.txt b/requirements.txt index 6297959f3..b00a1f75d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -23,6 +23,7 @@ sentry-sdk[flask]>=1.40 flask-openapi3 jsonschema fastapi +sse-starlette uvicorn prometheus_api_client paramiko diff --git a/simplyblock_core/cluster_ops.py b/simplyblock_core/cluster_ops.py index 6c86cbf33..89f829f49 100755 --- a/simplyblock_core/cluster_ops.py +++ b/simplyblock_core/cluster_ops.py @@ -35,6 +35,19 @@ db_controller = DBController() + +def watch_clusters(): + """Stream changes across all clusters.""" + return db_controller.watch(Cluster) + + +def watch_cluster(cluster_id): + """Stream changes for a single cluster.""" + return db_controller.watch( + Cluster, + select=lambda models: [c for c in models if c.get_id() == cluster_id], + ) + def _create_update_user(cluster_id, grafana_url, grafana_secret: SecretStr, user_secret: SecretStr, update_secret=False): session = requests.session() session.auth = ("admin", grafana_secret.get_secret_value()) diff --git a/simplyblock_core/controllers/device_controller.py b/simplyblock_core/controllers/device_controller.py index 9ca35de17..cbbca0e10 100644 --- a/simplyblock_core/controllers/device_controller.py +++ b/simplyblock_core/controllers/device_controller.py @@ -5,6 +5,7 @@ from simplyblock_core import distr_controller, utils, storage_node_ops, constants from simplyblock_core.controllers import device_events, tasks_controller from simplyblock_core.db_controller import DBController +from simplyblock_core.models.cluster import Cluster from simplyblock_core.models.nvme_device import NVMeDevice, JMDevice from simplyblock_core.models.storage_node import StorageNode from simplyblock_core.prom_client import PromClient @@ -19,6 +20,39 @@ logger = logging.getLogger() +def _explode_devices(nodes, node_id): + # Devices are embedded in their StorageNode record; expose them as per-device + # entries so the diff yields device-level events. The one place that knows + # devices live inside nodes. + for node in nodes: + if node.get_id() == node_id: + return list(node.nvme_devices) + return [] + + +def watch_devices(cluster_id, node_id): + """Stream device changes for one storage node.""" + db = DBController() + return db.watch( + StorageNode, + select=lambda nodes: _explode_devices(nodes, node_id), + ancestors=[(Cluster, cluster_id), (StorageNode, node_id)], + ) + + +def watch_device(cluster_id, node_id, device_id): + """Stream changes for a single device.""" + db = DBController() + return db.watch( + StorageNode, + select=lambda nodes: [ + device for device in _explode_devices(nodes, node_id) + if device.get_id() == device_id + ], + ancestors=[(Cluster, cluster_id), (StorageNode, node_id)], + ) + + def get_storage_node_by_jm_device(db_controller: DBController, id) -> StorageNode: try: return next( diff --git a/simplyblock_core/controllers/lvol_controller.py b/simplyblock_core/controllers/lvol_controller.py index 9b0a643f6..aafe23f2a 100755 --- a/simplyblock_core/controllers/lvol_controller.py +++ b/simplyblock_core/controllers/lvol_controller.py @@ -28,6 +28,29 @@ logger = utils.get_logger(__name__) +def watch_volumes(cluster_id, pool_id): + """Stream volume changes for one pool (same scope as get_lvols_by_pool_id).""" + db = DBController() + return db.watch( + LVol, + select=lambda models: db.get_lvols_by_pool_id(pool_id, source=models), + ancestors=[(Cluster, cluster_id), (Pool, pool_id)], + ) + + +def watch_volume(cluster_id, pool_id, volume_id): + """Stream changes for a single volume.""" + db = DBController() + return db.watch( + LVol, + select=lambda models: [ + lvol for lvol in db.get_lvols_by_pool_id(pool_id, source=models) + if lvol.get_id() == volume_id + ], + ancestors=[(Cluster, cluster_id), (Pool, pool_id)], + ) + + def _create_crypto_lvol(rpc_client, lvol, cluster): name = lvol.crypto_bdev base_name = f"{lvol.lvs_name}/{lvol.lvol_bdev}" diff --git a/simplyblock_core/controllers/pool_controller.py b/simplyblock_core/controllers/pool_controller.py index 181e0e70e..83f5f0f7b 100644 --- a/simplyblock_core/controllers/pool_controller.py +++ b/simplyblock_core/controllers/pool_controller.py @@ -13,12 +13,33 @@ from simplyblock_core.controllers import pool_events, lvol_controller from simplyblock_core.db_controller import DBController from simplyblock_core.kms import KMSException, create_kms_connection, pool_kek_name +from simplyblock_core.models.cluster import Cluster from simplyblock_core.models.pool import Pool from simplyblock_core.prom_client import PromClient logger = lg.getLogger() +def watch_pools(cluster_id): + """Stream pool changes for one cluster (same scope as get_pools).""" + db = DBController() + return db.watch( + Pool, + select=lambda models: db.get_pools(cluster_id, source=models), + ancestors=[(Cluster, cluster_id)], + ) + + +def watch_pool(cluster_id, pool_id): + """Stream changes for a single pool.""" + db = DBController() + return db.watch( + Pool, + select=lambda models: [pool for pool in models if pool.get_id() == pool_id], + ancestors=[(Cluster, cluster_id)], + ) + + def _generate_string(length): return ''.join(random.SystemRandom().choice( string.ascii_letters + string.digits) for _ in range(length)) diff --git a/simplyblock_core/controllers/snapshot_controller.py b/simplyblock_core/controllers/snapshot_controller.py index a608cebec..b71cb9bf9 100755 --- a/simplyblock_core/controllers/snapshot_controller.py +++ b/simplyblock_core/controllers/snapshot_controller.py @@ -19,6 +19,7 @@ from simplyblock_core.kms._exceptions import KMSException from simplyblock_core.db_controller import DBController from simplyblock_core.models.job_schedule import JobSchedule +from simplyblock_core.models.cluster import Cluster from simplyblock_core.models.pool import Pool from simplyblock_core.models.snapshot import SnapShot from simplyblock_core.models.lvol_model import LVol @@ -27,6 +28,29 @@ logger = lg.getLogger() + +def watch_snapshots(cluster_id, pool_id): + """Stream snapshot changes for one pool (same scope as get_snapshots_by_pool_id).""" + db = DBController() + return db.watch( + SnapShot, + select=lambda models: db.get_snapshots_by_pool_id(pool_id, source=models), + ancestors=[(Cluster, cluster_id), (Pool, pool_id)], + ) + + +def watch_snapshot(cluster_id, pool_id, snapshot_id): + """Stream changes for a single snapshot.""" + db = DBController() + return db.watch( + SnapShot, + select=lambda models: [ + snap for snap in db.get_snapshots_by_pool_id(pool_id, source=models) + if snap.get_id() == snapshot_id + ], + ancestors=[(Cluster, cluster_id), (Pool, pool_id)], + ) + db_controller = DBController() diff --git a/simplyblock_core/controllers/tasks_controller.py b/simplyblock_core/controllers/tasks_controller.py index ece14cccb..75580b8c3 100644 --- a/simplyblock_core/controllers/tasks_controller.py +++ b/simplyblock_core/controllers/tasks_controller.py @@ -14,6 +14,28 @@ logger = logging.getLogger() db = db_controller.DBController() + +def watch_tasks(cluster_id): + """Stream task changes for one cluster (excludes device-migration tasks, + matching the task list endpoint).""" + return db.watch( + JobSchedule, + select=lambda models: [ + task for task in db.get_job_tasks(cluster_id, source=models) + if task.function_name != JobSchedule.FN_DEV_MIG + ], + ancestors=[(Cluster, cluster_id)], + ) + + +def watch_task(cluster_id, task_id): + """Stream changes for a single task.""" + return db.watch( + JobSchedule, + select=lambda models: [task for task in models if task.uuid == task_id], + ancestors=[(Cluster, cluster_id)], + ) + # Identity used for task leases. Hostname (not pid) so a runner that crashes # and restarts on the same host re-claims its own in-flight tasks immediately. _RUNNER_HOST = socket.gethostname() diff --git a/simplyblock_core/db_controller.py b/simplyblock_core/db_controller.py index cf522e1d0..ca713f3a5 100644 --- a/simplyblock_core/db_controller.py +++ b/simplyblock_core/db_controller.py @@ -7,7 +7,7 @@ import fdb from typing import Any, List, Optional -from simplyblock_core import constants +from simplyblock_core import constants, watches from simplyblock_core.models.cluster import Cluster, ClusterAddNodeLock, PortReservation from simplyblock_core.models.events import EventObj from simplyblock_core.models.job_schedule import JobSchedule @@ -56,13 +56,22 @@ def __init__(self): except Exception: logger.exception("FDB initialization failed") + def watch(self, model_cls, *, select=None, ancestors=()): + """Async stream of watch.ChangeEvent batches for a watched model class. + + Thin pass-through to :func:`simplyblock_core.watch.watch` so controllers + reach the watch primitive the same way they reach every other DB op. + """ + from simplyblock_core import watch as _watch + return _watch.watch(model_cls, select=select, ancestors=ancestors) + def get_storage_nodes(self) -> List[StorageNode]: ret = StorageNode().read_from_db(self.kv_store) ret = sorted(ret, key=lambda x: x.create_dt) return ret - def get_storage_nodes_by_cluster_id(self, cluster_id: str) -> List[StorageNode]: - ret = StorageNode().read_from_db(self.kv_store) + def get_storage_nodes_by_cluster_id(self, cluster_id: str, *, source=None) -> List[StorageNode]: + ret = source if source is not None else StorageNode().read_from_db(self.kv_store) nodes = [] for n in ret: if n.cluster_id == cluster_id: @@ -101,15 +110,11 @@ def get_storage_device_by_id(self, id: str) -> NVMeDevice: return device - def get_pools(self, cluster_id: Optional[str] = None) -> List[Pool]: - pools = [] + def get_pools(self, cluster_id: Optional[str] = None, *, source=None) -> List[Pool]: + all_pools = source if source is not None else Pool().read_from_db(self.kv_store) if cluster_id: - for pool in Pool().read_from_db(self.kv_store): - if pool.cluster_id == cluster_id: - pools.append(pool) - else: - pools = Pool().read_from_db(self.kv_store) - return pools + return [pool for pool in all_pools if pool.cluster_id == cluster_id] + return all_pools def get_pool_by_id(self, id: str) -> Pool: pool = single_or_none(Pool().read_from_db(self.kv_store, id)) @@ -123,8 +128,8 @@ def get_pool_by_name(self, name: str) -> Pool: raise KeyError(f'Pool {name} not found') return pool - def get_lvols(self, cluster_id: Optional[str] = None) -> List[LVol]: - lvols = self.get_all_lvols() + def get_lvols(self, cluster_id: Optional[str] = None, *, source=None) -> List[LVol]: + lvols = source if source is not None else self.get_all_lvols() lvols = [lvol for lvol in lvols if lvol.status != LVol.STATUS_DELETED] if not cluster_id: return lvols @@ -155,9 +160,9 @@ def get_lvols_by_node_id(self, node_id: str) -> List[LVol]: lvols.append(lvol) return sorted(lvols, key=lambda x: x.create_dt) - def get_lvols_by_pool_id(self, pool_id: str) -> List[LVol]: + def get_lvols_by_pool_id(self, pool_id: str, *, source=None) -> List[LVol]: lvols = [] - for lvol in self.get_lvols(): + for lvol in self.get_lvols(source=source): if lvol.pool_uuid == pool_id: lvols.append(lvol) return sorted(lvols, key=lambda x: x.create_dt) @@ -170,9 +175,9 @@ def get_hostnames_by_pool_id(self, pool_id: str) -> List[str]: hostnames.append(lv.hostname) return hostnames - def get_snapshots(self, cluster_id: Optional[str] = None) -> List[SnapShot]: + def get_snapshots(self, cluster_id: Optional[str] = None, *, source=None) -> List[SnapShot]: start_time = time.time() - snaps = SnapShot().read_from_db(self.kv_store) + snaps = source if source is not None else SnapShot().read_from_db(self.kv_store) if cluster_id: snaps = [n for n in snaps if n.cluster_id == cluster_id] ret = sorted(snaps, key=lambda x: x.created_at) @@ -283,7 +288,9 @@ def get_device_capacity(self, device, limit=1) -> List[DeviceStatObject]: self.kv_store, id="%s/%s" % (device.cluster_id, device.get_id()), limit=limit, reverse=True) return stats - def get_clusters(self) -> List[Cluster]: + def get_clusters(self, *, source=None) -> List[Cluster]: + if source is not None: + return source return Cluster().read_from_db(self.kv_store) def get_cluster_by_id(self, cluster_id: str) -> Cluster: @@ -299,8 +306,11 @@ def get_port_stats(self, node_id: str, port_id: str, limit: int = 20) -> List[Po def get_events(self, event_id: str = " ", limit: int = 0, reverse: bool = False) -> List[EventObj]: return EventObj().read_from_db(self.kv_store, id=event_id, limit=limit, reverse=reverse) - def get_job_tasks(self, cluster_id: str, reverse: bool = True, limit: int = 0) -> List[JobSchedule]: - ret = JobSchedule().read_from_db(self.kv_store, id=cluster_id, reverse=reverse, limit=limit) + def get_job_tasks(self, cluster_id: str, reverse: bool = True, limit: int = 0, *, source=None) -> List[JobSchedule]: + if source is not None: + ret = [t for t in source if t.cluster_id == cluster_id] + else: + ret = JobSchedule().read_from_db(self.kv_store, id=cluster_id, reverse=reverse, limit=limit) return sorted(ret, key=lambda x: x.date) @@ -326,9 +336,9 @@ def get_snapshots_by_node_id(self, node_id: str) -> List[SnapShot]: ret.append(snap) return sorted(ret, key=lambda x: x.create_dt) - def get_snapshots_by_pool_id(self, pool_id: str) -> List[SnapShot]: + def get_snapshots_by_pool_id(self, pool_id: str, *, source=None) -> List[SnapShot]: ret = [] - snaps = self.get_snapshots() + snaps = self.get_snapshots(source=source) for snap in snaps: if snap.pool_uuid == pool_id: ret.append(snap) @@ -675,6 +685,8 @@ def _atomic_update_tx(self, tr, key, model_cls, mutate_fn): if mutate_fn(obj) is False: return obj tr[key] = json.dumps(obj.to_dict(unwrap_secrets=True)).encode() + if getattr(model_cls, '_WATCHED', False): + tr.add(watches.watch_counter_key(model_cls), watches.ONE_LE64) return obj def atomic_update(self, obj, mutate_fn): @@ -900,6 +912,7 @@ def _try_set_node_restarting_tx(self, tr, cluster_id, node_id): prefix = target.get_db_id() data = json.dumps(target.get_clean_dict(unwrap_secrets=True)) tr[prefix.encode()] = data.encode() + tr.add(watches.watch_counter_key(StorageNode), watches.ONE_LE64) return True, None diff --git a/simplyblock_core/models/base_model.py b/simplyblock_core/models/base_model.py index 1e2509a4e..4b898d2db 100644 --- a/simplyblock_core/models/base_model.py +++ b/simplyblock_core/models/base_model.py @@ -9,11 +9,18 @@ from pydantic import SecretBytes, SecretStr +from simplyblock_core import watches + class BaseModel(object): _STATUS_CODE_MAP: dict = {} + # When True, write_to_db()/remove() atomically bump watch_seq/ + # in the same FDB transaction so watchers (SSE API) wake up. Plain class + # attribute, not an annotation: must stay out of get_attrs_map()/to_dict(). + _WATCHED = False + id: str = "" uuid: str = "" name: str = "" @@ -197,14 +204,29 @@ def get_last(self, kv_store): return objects[0] return None + @staticmethod + def _write_tx(tr, key, value, counter_key): + tr.set(key, value) + tr.add(counter_key, watches.ONE_LE64) + + @staticmethod + def _remove_tx(tr, key, counter_key): + tr.clear(key) + tr.add(counter_key, watches.ONE_LE64) + def write_to_db(self, kv_store=None): if not kv_store: from simplyblock_core.db_controller import DBController kv_store = DBController().kv_store try: - prefix = self.get_db_id() - st = json.dumps(self.to_dict(unwrap_secrets=True)) - kv_store.set(prefix.encode(), st.encode()) + key = self.get_db_id().encode() + value = json.dumps(self.to_dict(unwrap_secrets=True)).encode() + if self._WATCHED: + import fdb + fdb.transactional(BaseModel._write_tx)( + kv_store, key, value, watches.watch_counter_key(type(self))) + else: + kv_store.set(key, value) return True except Exception: from simplyblock_core import utils @@ -212,8 +234,12 @@ def write_to_db(self, kv_store=None): exit(1) def remove(self, kv_store): - prefix = self.get_db_id() - return kv_store.clear(prefix.encode()) + key = self.get_db_id().encode() + if not self._WATCHED: + return kv_store.clear(key) + import fdb + return fdb.transactional(BaseModel._remove_tx)( + kv_store, key, watches.watch_counter_key(type(self))) def keys(self): return self.get_attrs_map().keys() diff --git a/simplyblock_core/models/cluster.py b/simplyblock_core/models/cluster.py index f06a4997c..e5d2625c1 100644 --- a/simplyblock_core/models/cluster.py +++ b/simplyblock_core/models/cluster.py @@ -16,6 +16,8 @@ class HashicorpVaultSettings(BaseModel): class Cluster(BaseModel): + _WATCHED = True + STATUS_ACTIVE = "active" STATUS_READONLY = 'read_only' STATUS_INACTIVE = "inactive" diff --git a/simplyblock_core/models/job_schedule.py b/simplyblock_core/models/job_schedule.py index 859f5fcfb..8de7e02a7 100644 --- a/simplyblock_core/models/job_schedule.py +++ b/simplyblock_core/models/job_schedule.py @@ -6,6 +6,8 @@ class JobSchedule(BaseModel): + _WATCHED = True + STATUS_NEW = 'new' STATUS_RUNNING = 'running' STATUS_SUSPENDED = 'suspended' diff --git a/simplyblock_core/models/lvol_model.py b/simplyblock_core/models/lvol_model.py index ea658eb75..65e8880e4 100644 --- a/simplyblock_core/models/lvol_model.py +++ b/simplyblock_core/models/lvol_model.py @@ -7,6 +7,8 @@ class LVol(BaseModel): + _WATCHED = True + STATUS_IN_CREATION = 'in_creation' STATUS_ONLINE = 'online' STATUS_OFFLINE = 'offline' diff --git a/simplyblock_core/models/pool.py b/simplyblock_core/models/pool.py index cf3f19f50..81d77c6f9 100644 --- a/simplyblock_core/models/pool.py +++ b/simplyblock_core/models/pool.py @@ -9,6 +9,8 @@ class Pool(BaseModel): + _WATCHED = True + STATUS_ACTIVE = "active" STATUS_INACTIVE = "inactive" diff --git a/simplyblock_core/models/snapshot.py b/simplyblock_core/models/snapshot.py index 46559e8d0..62002ce1e 100644 --- a/simplyblock_core/models/snapshot.py +++ b/simplyblock_core/models/snapshot.py @@ -6,6 +6,8 @@ class SnapShot(BaseModel): + _WATCHED = True + STATUS_ONLINE = 'online' STATUS_OFFLINE = 'offline' STATUS_IN_DELETION = 'in_deletion' diff --git a/simplyblock_core/models/storage_node.py b/simplyblock_core/models/storage_node.py index 21d22a701..d7c4b76b3 100644 --- a/simplyblock_core/models/storage_node.py +++ b/simplyblock_core/models/storage_node.py @@ -21,6 +21,8 @@ class StorageNode(BaseNodeObject): + _WATCHED = True + # Restart phase constants (per-LVS) RESTART_PHASE_PRE_BLOCK = "pre_block" RESTART_PHASE_BLOCKED = "blocked" diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index 7d19d4b27..2af054c17 100755 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -46,6 +46,27 @@ logger = utils.get_logger(__name__) +def watch_storage_nodes(cluster_id): + """Stream storage-node changes for one cluster (same scope as + get_storage_nodes_by_cluster_id).""" + db = DBController() + return db.watch( + StorageNode, + select=lambda models: db.get_storage_nodes_by_cluster_id(cluster_id, source=models), + ancestors=[(Cluster, cluster_id)], + ) + + +def watch_storage_node(cluster_id, node_id): + """Stream changes for a single storage node.""" + db = DBController() + return db.watch( + StorageNode, + select=lambda models: [node for node in models if node.get_id() == node_id], + ancestors=[(Cluster, cluster_id)], + ) + + class LVSRestartRequiredError(Exception): """Raised when an LVS fails to recover via ``bdev_examine`` during activation-mode recreate. The node's SPDK holds partial state that diff --git a/simplyblock_core/watch.py b/simplyblock_core/watch.py new file mode 100644 index 000000000..539702b99 --- /dev/null +++ b/simplyblock_core/watch.py @@ -0,0 +1,336 @@ +# coding=utf-8 +"""Generic entity-watch primitive for streaming state changes. + +Change signal: every write/remove of a watched model class atomically bumps +``watch_seq/`` in the same FDB transaction as the entity mutation +(see :mod:`simplyblock_core.watches`). One shared :class:`WatchHub` per model +class owns a single FDB watch on that counter key; when it fires, the hub +re-reads the class's entity range once and fans the parsed models out to every +subscriber. Scoping (e.g. volumes of one pool) happens per subscription via a +``select`` callable, so N scoped subscribers cost one watch and one shared range +read per change. + +The hub registers the watch *before* reading the range: any write not visible to +the range read has a commit version at or after the watch's read version and +therefore fires it, so no update is lost. A reconcile timeout re-reads the range +even without a watch fire, bounding staleness for writes from pre-upgrade +components that don't bump counters. + +This module owns the *mechanism* only. It emits typed :class:`ChangeEvent` +batches (``created``/``updated``/``deleted`` over model objects); building wire +DTOs and framing them as SSE is the web layer's concern (``_sse.py``). Keeping +the change-detection here — behind the ``ChangeEvent`` interface — means the +counter-and-re-read backend can later be swapped for a version-index or +change-log backend without touching controllers or the API. +""" + +import asyncio +import json +import logging +import threading +from dataclasses import dataclass +from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Type + +import fdb + +from simplyblock_core import watches + +logger = logging.getLogger(__name__) + +# Upper bound on staleness for entity writes that don't bump the change counter +# (processes from before the counter was introduced). +WATCH_RECONCILE_SEC = 30.0 + +_FDB_RETRY_BACKOFF_SEC = (1.0, 2.0, 4.0) + +_UNSET: Any = object() + +# Filters the shared full model list down to one subscription's scope. Returns +# model objects (typically by reusing a DBController getter with ``source=``). +Select = Callable[[List[Any]], List[Any]] + + +class WatchUnavailable(Exception): + """The watch backend failed irrecoverably; the stream must terminate.""" + + +@dataclass +class ChangeEvent: + """A single entity transition within a watched, scoped set.""" + + kind: str # 'created' | 'updated' | 'deleted' + id: str + # For 'deleted': the entity's current (unfiltered) model if it left the + # scoped set but is still retrievable (e.g. an LVol whose status flipped to + # 'deleted'), else None once it is physically gone. + model: Optional[Any] + + +def diff_by_id( + prev: Dict[str, dict], new: Dict[str, Any], full: Dict[str, Any], +) -> Tuple[List[ChangeEvent], Dict[str, dict]]: + """Diff the scoped set against the previous emission, keyed by ``get_id()``. + + ``prev`` maps id -> the ``to_dict()`` snapshot emitted last time; ``new`` and + ``full`` map id -> model for the current scoped and unscoped sets. Returns + the events (created/updated in ``new`` iteration order, deletions appended) + and the fresh id -> ``to_dict()`` snapshot to carry forward. + """ + events: List[ChangeEvent] = [] + new_dicts: Dict[str, dict] = {} + for entity_id, model in new.items(): + snapshot = model.to_dict() + new_dicts[entity_id] = snapshot + if entity_id not in prev: + events.append(ChangeEvent('created', entity_id, model)) + elif snapshot != prev[entity_id]: + events.append(ChangeEvent('updated', entity_id, model)) + # else unchanged: no event + for entity_id in prev: + if entity_id in new: + continue + # Left the scoped set. If still present in the unscoped set, carry its + # current model so the wire layer can render the final representation; + # otherwise it is physically gone (model=None -> empty representation). + events.append(ChangeEvent('deleted', entity_id, full.get(entity_id))) + return events, new_dicts + + +class _Subscription: + """Coalescing mailbox: written by the hub thread, read by one stream.""" + + def __init__(self, loop: asyncio.AbstractEventLoop, notify: asyncio.Event) -> None: + self._loop = loop + self._notify = notify + self._latest: Any = _UNSET + self._dirty = False + self.failed = False + + def _deliver(self, state: Any) -> None: # event-loop thread + self._latest = state + self._dirty = True + self._notify.set() + + def _fail(self) -> None: # event-loop thread + self.failed = True + self._notify.set() + + def deliver(self, state: Any) -> None: # hub thread + self._loop.call_soon_threadsafe(self._deliver, state) + + def fail(self) -> None: # hub thread + self._loop.call_soon_threadsafe(self._fail) + + def take(self) -> Any: + """Latest state if there is an unconsumed publication, else ``_UNSET``.""" + if not self._dirty: + return _UNSET + self._dirty = False + return self._latest + + +def _watch_tx(tr: Any, key: bytes) -> Any: + return tr.watch(key) + + +class WatchHub: + """Single FDB watch + shared range re-read for one watched model class. + + Delivers ``List[model]`` (the full current class range) to subscribers. + Change detection compares the *raw* parsed records (cheap) so the reconcile + re-read only republishes on a real change; models are constructed only when + something changed and shared across all subscribers. + """ + + def __init__(self, model_cls: Type[Any]) -> None: + instance = model_cls() + self._model_cls = model_cls + self._name = model_cls.__name__ + # Built from parts rather than get_db_id(): a fresh JobSchedule's + # compound get_id() is not empty, which would corrupt the prefix. + self._prefix = f'{instance.object_type}/{instance.name}/'.encode() + self._counter_key = watches.watch_counter_key(model_cls) + self._lock = threading.Lock() + self._subs: set = set() + self._cache: Any = _UNSET # last published List[model] (for late joiners) + self._raw: Any = _UNSET # last raw dict[key, dict] (for change detection) + self._stop = threading.Event() + self._wake = threading.Event() + self._thread: Optional[threading.Thread] = None + + def subscribe(self, loop: asyncio.AbstractEventLoop, notify: asyncio.Event) -> _Subscription: + sub = _Subscription(loop, notify) + with self._lock: + alive = self._thread is not None and self._thread.is_alive() and not self._stop.is_set() + if not alive: + self._cache = _UNSET + self._raw = _UNSET + self._stop = threading.Event() + self._wake = threading.Event() + self._thread = threading.Thread( + target=self._run, args=(self._stop, self._wake), + daemon=True, name=f'watch-{self._name}', + ) + self._thread.start() + self._subs.add(sub) + if self._cache is not _UNSET: + # subscribe() runs on the event loop thread, so deliver directly. + sub._deliver(self._cache) + return sub + + def unsubscribe(self, sub: _Subscription) -> None: + with self._lock: + self._subs.discard(sub) + if not self._subs: + self._stop.set() + self._wake.set() + + def subscriber_count(self) -> int: + with self._lock: + return len(self._subs) + + def _publish(self, raw: Dict[Any, dict], models: List[Any]) -> None: + with self._lock: + self._raw = raw + self._cache = models + subs = list(self._subs) + for sub in subs: + sub.deliver(models) + + def _fail_all(self) -> None: + with self._lock: + subs = list(self._subs) + for sub in subs: + sub.fail() + + def _read_raw(self, kv_store: Any) -> Dict[Any, dict]: + state: Dict[Any, dict] = {} + for k, v in kv_store.get_range_startswith(self._prefix): + try: + state[bytes(k)] = json.loads(v) + except ValueError: + logger.warning('Skipping unparsable record %s', bytes(k)) + return state + + def _to_models(self, raw: Dict[Any, dict]) -> List[Any]: + return [self._model_cls().from_dict(d) for d in raw.values()] + + def _run(self, stop: threading.Event, wake: threading.Event) -> None: + from simplyblock_core.db_controller import DBController + kv_store = DBController().kv_store + set_watch = fdb.transactional(_watch_tx) + failures = 0 + while not stop.is_set(): + watch = None + try: + # Watch first, then read: see module docstring. + watch = set_watch(kv_store, self._counter_key) + raw = self._read_raw(kv_store) + failures = 0 + with self._lock: + changed = self._raw is _UNSET or raw != self._raw + if changed: + self._publish(raw, self._to_models(raw)) + wake.clear() + # Runs on the FDB network thread: must only set the event, + # never block or call back into fdb. + watch.on_ready(lambda _f: wake.set()) + wake.wait(timeout=WATCH_RECONCILE_SEC) + except Exception: + logger.exception('Watch hub for %s failed', self._name) + failures += 1 + if failures > len(_FDB_RETRY_BACKOFF_SEC): + self._fail_all() + return + stop.wait(timeout=_FDB_RETRY_BACKOFF_SEC[failures - 1]) + finally: + if watch is not None: + try: + watch.cancel() + except Exception: + # Best-effort cleanup: cancellation failures should not + # mask the main loop error/retry behavior. + logger.debug('Ignoring watch cancellation failure for %s', self._name, exc_info=True) + + +_hubs: Dict[type, WatchHub] = {} +_hubs_lock = threading.Lock() + + +def get_hub(model_cls: Type[Any]) -> WatchHub: + with _hubs_lock: + hub = _hubs.get(model_cls) + if hub is None: + hub = _hubs[model_cls] = WatchHub(model_cls) + return hub + + +def backend_available() -> bool: + """Whether the watch backend can serve streams (the DB is reachable).""" + from simplyblock_core.db_controller import DBController + return DBController().kv_store is not None + + +async def watch( + model_cls: Type[Any], + *, + select: Optional[Select] = None, + ancestors: Sequence[Tuple[Type[Any], str]] = (), +): + """Async stream of :class:`ChangeEvent` batches for one watched class. + + ``select`` filters the shared full model list to this subscription's scope + (reuse a DBController getter with ``source=`` so the filter is defined once); + the diff runs on its output, so entities entering/leaving the scoped set emit + ``created``/``deleted``. ``ancestors`` are ``(model_cls, id)`` pairs whose + disappearance closes the stream (a reconnect then hits the endpoint's regular + 404). The first yielded batch is the initial state (all ``created``). + + Raises :class:`WatchUnavailable` if the backend fails irrecoverably. + """ + loop = asyncio.get_running_loop() + notify = asyncio.Event() + main_sub = get_hub(model_cls).subscribe(loop, notify) + ancestor_subs = [ + (get_hub(cls).subscribe(loop, notify), cls, str(ancestor_id)) + for cls, ancestor_id in ancestors + ] + try: + prev: Dict[str, dict] = {} + while True: + await notify.wait() + notify.clear() + + if main_sub.failed or any(sub.failed for sub, _, _ in ancestor_subs): + raise WatchUnavailable() + + for ancestor_sub, _, ancestor_id in ancestor_subs: + ancestor_state = ancestor_sub.take() + if ancestor_state is _UNSET: + continue + if not any(m.get_id() == ancestor_id for m in ancestor_state): + return + + models = main_sub.take() + if models is _UNSET: + continue + # select() filtering and to_dict() diffing are CPU-bound; keep them + # off the event loop. Selects must not do blocking DB reads (they + # filter the supplied model list) — DTO building, which may, stays + # in the web layer's threadpool. + events, prev = await loop.run_in_executor( + None, _compute_batch, models, select, prev) + yield events + finally: + get_hub(model_cls).unsubscribe(main_sub) + for ancestor_sub, cls, _ in ancestor_subs: + get_hub(cls).unsubscribe(ancestor_sub) + + +def _compute_batch( + models: List[Any], select: Optional[Select], prev: Dict[str, dict], +) -> Tuple[List[ChangeEvent], Dict[str, dict]]: + full = {m.get_id(): m for m in models} + selected = select(models) if select is not None else models + new = {m.get_id(): m for m in selected} + return diff_by_id(prev, new, full) diff --git a/simplyblock_core/watches.py b/simplyblock_core/watches.py new file mode 100644 index 000000000..4e6f37820 --- /dev/null +++ b/simplyblock_core/watches.py @@ -0,0 +1,33 @@ +# coding=utf-8 +"""Change-counter keys for entity watches. + +Every write/remove of a watched model class atomically bumps a per-class +counter key (``watch_seq/``) in the same FoundationDB transaction +as the entity mutation. Watchers (the web API's SSE layer) set FDB watches on +these counter keys to learn that *something* of that class changed, then +re-read the entity range and diff. + +The ``watch_seq/`` namespace is disjoint from all entity range scans: +``BaseModel.read_from_db`` only scans ``//`` prefixes. + +Stdlib-only leaf module — importable from models without cycles. +""" + +import struct + +WATCH_SEQ_PREFIX = 'watch_seq/' + +# Operand for FDB's atomic ADD mutation: little-endian 64-bit integer 1. +ONE_LE64 = struct.pack(' bytes: + """FDB key of the per-class change counter, e.g. ``b'watch_seq/StorageNode'``.""" + return (WATCH_SEQ_PREFIX + model_cls.__name__).encode() + + +def unpack_counter(raw) -> int: + """Decode a counter value produced by atomic ADD; absent/``None`` -> 0.""" + if not raw: + return 0 + return struct.unpack(' Tuple[List[ServerSentEvent], Dict[str, str]]: + """Render the initial batch (all ``created``) as one ``snapshot`` event.""" + cache = {event.id: build_dto(event.model).model_dump_json() for event in events} + if single: + if not cache: + return [], cache + data = next(iter(cache.values())) + else: + data = '[' + ','.join(cache[event.id] for event in events) + ']' + return [ServerSentEvent(event='snapshot', data=data, retry=SSE_RETRY_MS)], cache + + +def _build_events( + events: List[ChangeEvent], cache: Dict[str, str], build_dto: DTOBuilder, single: bool, +) -> Tuple[List[ServerSentEvent], Dict[str, str], bool]: + """Render a diff batch as individual ``created``/``updated``/``deleted`` events.""" + out: List[ServerSentEvent] = [] + new_cache = dict(cache) + deleted = False + for event in events: + if event.kind == 'deleted': + deleted = True + if event.model is not None: + try: + data = build_dto(event.model).model_dump_json() + except Exception: + logger.exception('Failed to build DTO for deleted entity %s', event.id) + data = new_cache.get(event.id, '{}') + else: + data = '{}' + new_cache.pop(event.id, None) + out.append(ServerSentEvent(event='deleted', data=data)) + else: + data = build_dto(event.model).model_dump_json() + if event.kind == 'updated' and data == new_cache.get(event.id): + # Change invisible in the DTO (e.g. a JobSchedule lease + # heartbeat rewrite): suppress it. + continue + new_cache[event.id] = data + out.append(ServerSentEvent(event='updated' if single else event.kind, data=data)) + return out, new_cache, deleted + + +def sse_response(change_stream, build_dto: DTOBuilder, *, single: bool = False) -> EventSourceResponse: + """Frame a controller's ChangeEvent stream as an SSE response. + + ``change_stream`` is an async iterator of ``List[ChangeEvent]`` batches (the + first batch is the initial state). ``build_dto`` converts a model into its + wire DTO. ``single`` switches to detail semantics: a single-object snapshot + and closing the stream once the entity is deleted. + """ + if not backend_available(): + raise HTTPException(503, 'Database unavailable') + + async def event_stream() -> Any: + dto_cache: Dict[str, str] = {} + first = True + deadline = time.monotonic() + WATCH_MAX_LIFETIME_SEC + try: + while True: + timeout = deadline - time.monotonic() + if timeout <= 0: + return + try: + events = await asyncio.wait_for(change_stream.__anext__(), timeout) + except asyncio.TimeoutError: + return + except StopAsyncIteration: + return + except WatchUnavailable: + yield ServerSentEvent(event='error', data='{"detail": "backend unavailable"}') + return + + if first: + first = False + sse_events, dto_cache = await run_in_threadpool( + _build_snapshot, events, build_dto, single) + if single and not dto_cache: + # Deleted between dependency resolution and first + # publication; close so the reconnect 404s. + return + for event in sse_events: + yield event + else: + sse_events, dto_cache, deleted = await run_in_threadpool( + _build_events, events, dto_cache, build_dto, single) + for event in sse_events: + yield event + if single and deleted: + return + finally: + await change_stream.aclose() + + return EventSourceResponse( + event_stream(), + ping=PING_SEC, + headers={'Cache-Control': 'no-store', 'X-Accel-Buffering': 'no'}, + ) diff --git a/simplyblock_web/api/v2/cluster/__init__.py b/simplyblock_web/api/v2/cluster/__init__.py index 7cc78f0f7..1ee29a6e4 100644 --- a/simplyblock_web/api/v2/cluster/__init__.py +++ b/simplyblock_web/api/v2/cluster/__init__.py @@ -1,12 +1,14 @@ from threading import Thread -from typing import Annotated, List, Literal, Optional +from typing import Annotated, List, Literal, Optional, Union from uuid import UUID from fastapi import APIRouter, HTTPException, Request, Response from pydantic import BaseModel, Field, SecretStr from pydantic.networks import AnyUrl, UrlConstraints +from sse_starlette import EventSourceResponse from simplyblock_core.db_controller import DBController +from simplyblock_core.models.cluster import Cluster as ClusterModel from simplyblock_core.models.cluster import HashicorpVaultSettings as ModelVaultSettings from simplyblock_core import cluster_ops @@ -16,6 +18,7 @@ from .storage_node import api as storage_node_api from .task import api as task_api from .._dtos import ClusterDTO +from .._sse import WATCH_RESPONSES, WatchParam, sse_response from .. import util as util @@ -87,8 +90,15 @@ class ClusterParams(BaseModel): hashicorp_vault_settings: Optional[HashicorpVaultSettings] = None -@api.get('/', name='clusters:list') -def list() -> List[ClusterDTO]: +def _cluster_dto(cluster: ClusterModel) -> ClusterDTO: + ret = db.get_cluster_capacity(cluster, 1) + return ClusterDTO.from_model(cluster, ret[0] if ret else None) + + +@api.get('/', name='clusters:list', response_model=List[ClusterDTO], responses=WATCH_RESPONSES) +def list(watch: WatchParam = False) -> Union[List[ClusterDTO], EventSourceResponse]: + if watch: + return sse_response(cluster_ops.watch_clusters(), _cluster_dto) data = [] for cluster in db.get_clusters(): stat_obj = None @@ -127,8 +137,14 @@ def add(request: Request, parameters: ClusterParams, response_format: util.Creat instance_api = APIRouter(prefix='/{cluster_id}') -@instance_api.get('/', name='clusters:detail') -def get(cluster: Cluster) -> ClusterDTO: +@instance_api.get('/', name='clusters:detail', response_model=ClusterDTO, responses=WATCH_RESPONSES) +def get(cluster: Cluster, watch: WatchParam = False) -> Union[ClusterDTO, EventSourceResponse]: + if watch: + return sse_response( + cluster_ops.watch_cluster(cluster.get_id()), + _cluster_dto, + single=True, + ) stat_obj = None ret = db.get_cluster_capacity(cluster, 1) if ret: diff --git a/simplyblock_web/api/v2/cluster/storage_node/__init__.py b/simplyblock_web/api/v2/cluster/storage_node/__init__.py index 942242c99..9a209de50 100644 --- a/simplyblock_web/api/v2/cluster/storage_node/__init__.py +++ b/simplyblock_web/api/v2/cluster/storage_node/__init__.py @@ -1,16 +1,19 @@ from threading import Thread -from typing import List, Optional +from typing import List, Optional, Union from uuid import UUID from fastapi import APIRouter, HTTPException, Request, Response from pydantic import BaseModel, Field +from sse_starlette import EventSourceResponse from simplyblock_core.db_controller import DBController from simplyblock_core.controllers import tasks_controller +from simplyblock_core.models.storage_node import StorageNode as StorageNodeModel from simplyblock_core import storage_node_ops from ... import util as util from ..._dependencies import Cluster, StorageNode +from ..._sse import WATCH_RESPONSES, WatchParam, sse_response from .device import api as device_api from ..._dtos import StorageNodeDTO, TaskDTO @@ -19,8 +22,18 @@ db = DBController() -@api.get('/', name='clusters:storage-nodes:list') -def list(cluster: Cluster) -> List[StorageNodeDTO]: +def _storage_node_dto(storage_node: StorageNodeModel) -> StorageNodeDTO: + ret = db.get_node_capacity(storage_node, 1) + return StorageNodeDTO.from_model(storage_node, ret[0] if ret else None) + + +@api.get('/', name='clusters:storage-nodes:list', response_model=List[StorageNodeDTO], responses=WATCH_RESPONSES) +def list(cluster: Cluster, watch: WatchParam = False) -> Union[List[StorageNodeDTO], EventSourceResponse]: + if watch: + return sse_response( + storage_node_ops.watch_storage_nodes(cluster.get_id()), + _storage_node_dto, + ) data = [] for storage_node in db.get_storage_nodes_by_cluster_id(cluster.get_id()): node_stat_obj = None @@ -99,8 +112,14 @@ def add(request: Request, cluster: Cluster, parameters: StorageNodeParams, respo instance_api = APIRouter(prefix='/{storage_node_id}') -@instance_api.get('/', name='clusters:storage-nodes:detail') -def get(cluster: Cluster, storage_node: StorageNode): +@instance_api.get('/', name='clusters:storage-nodes:detail', response_model=StorageNodeDTO, responses=WATCH_RESPONSES) +def get(cluster: Cluster, storage_node: StorageNode, watch: WatchParam = False) -> Union[StorageNodeDTO, EventSourceResponse]: + if watch: + return sse_response( + storage_node_ops.watch_storage_node(cluster.get_id(), storage_node.get_id()), + _storage_node_dto, + single=True, + ) node_stat_obj = None ret = db.get_node_capacity(storage_node, 1) if ret: diff --git a/simplyblock_web/api/v2/cluster/storage_node/device.py b/simplyblock_web/api/v2/cluster/storage_node/device.py index 6fe7a60ae..08753f090 100644 --- a/simplyblock_web/api/v2/cluster/storage_node/device.py +++ b/simplyblock_web/api/v2/cluster/storage_node/device.py @@ -1,20 +1,36 @@ -from typing import List, Optional +from typing import Callable, List, Optional, Union from fastapi import APIRouter, Response +from sse_starlette import EventSourceResponse from simplyblock_core.db_controller import DBController from simplyblock_core.controllers import device_controller +from simplyblock_core.models.nvme_device import NVMeDevice from ..._dependencies import Cluster, StorageNode, Device from ..._dtos import DeviceDTO +from ..._sse import WATCH_RESPONSES, WatchParam, sse_response api = APIRouter() db = DBController() -@api.get('/', name='clusters:storage_nodes:devices:list') -def list(cluster: Cluster, storage_node: StorageNode) -> List[DeviceDTO]: +def _make_device_dto(storage_node_id: str) -> Callable[[NVMeDevice], DeviceDTO]: + def build(device: NVMeDevice) -> DeviceDTO: + ret = db.get_device_stats(device, 1) + return DeviceDTO.from_model(device, storage_node_id, ret[0] if ret else None) + return build + + +@api.get('/', name='clusters:storage_nodes:devices:list', response_model=List[DeviceDTO], responses=WATCH_RESPONSES) +def list(cluster: Cluster, storage_node: StorageNode, watch: WatchParam = False) -> Union[List[DeviceDTO], EventSourceResponse]: + if watch: + node_id = storage_node.get_id() + return sse_response( + device_controller.watch_devices(cluster.get_id(), node_id), + _make_device_dto(node_id), + ) data = [] for device in storage_node.nvme_devices: stat_obj = None @@ -28,8 +44,15 @@ def list(cluster: Cluster, storage_node: StorageNode) -> List[DeviceDTO]: instance_api = APIRouter(prefix='/{device_id}') -@instance_api.get('/', name='clusters:storage_nodes:devices:detail') -def get(cluster: Cluster, storage_node: StorageNode, device: Device) -> DeviceDTO: +@instance_api.get('/', name='clusters:storage_nodes:devices:detail', response_model=DeviceDTO, responses=WATCH_RESPONSES) +def get(cluster: Cluster, storage_node: StorageNode, device: Device, watch: WatchParam = False) -> Union[DeviceDTO, EventSourceResponse]: + if watch: + node_id = storage_node.get_id() + return sse_response( + device_controller.watch_device(cluster.get_id(), node_id, device.get_id()), + _make_device_dto(node_id), + single=True, + ) stat_obj = None ret = db.get_device_stats(device, 1) if ret: diff --git a/simplyblock_web/api/v2/cluster/storage_pool/__init__.py b/simplyblock_web/api/v2/cluster/storage_pool/__init__.py index f9627d420..28729b115 100644 --- a/simplyblock_web/api/v2/cluster/storage_pool/__init__.py +++ b/simplyblock_web/api/v2/cluster/storage_pool/__init__.py @@ -1,8 +1,9 @@ -from typing import Annotated, List, Optional +from typing import Annotated, List, Optional, Union from uuid import UUID from fastapi import APIRouter, HTTPException, Request, Response from pydantic import BaseModel, Field +from sse_starlette import EventSourceResponse from simplyblock_core.db_controller import DBController from simplyblock_core.controllers import pool_controller @@ -11,6 +12,7 @@ from ... import util as util from ..._dependencies import Cluster, StoragePool +from ..._sse import WATCH_RESPONSES, WatchParam, sse_response from .volume import api as volume_api from .snapshot import api as snapshot_api from ..._dtos import StoragePoolDTO @@ -20,8 +22,17 @@ db = DBController() -@api.get('/', name='clusters:storage-pools:list') -def list(cluster: Cluster) -> List[StoragePoolDTO]: +def _pool_dto(pool: PoolModel) -> StoragePoolDTO: + return StoragePoolDTO.from_model(pool, None) + + +@api.get('/', name='clusters:storage-pools:list', response_model=List[StoragePoolDTO], responses=WATCH_RESPONSES) +def list(cluster: Cluster, watch: WatchParam = False) -> Union[List[StoragePoolDTO], EventSourceResponse]: + if watch: + return sse_response( + pool_controller.watch_pools(cluster.get_id()), + _pool_dto, + ) return [StoragePoolDTO.from_model(pool, None) for pool in db.get_pools(cluster.get_id())] @@ -68,8 +79,14 @@ def add(request: Request, cluster: Cluster, parameters: StoragePoolParams, respo instance_api = APIRouter(prefix='/{pool_id}') -@instance_api.get('/', name='clusters:storage-pools:detail') -def get(cluster: Cluster, pool: StoragePool) -> StoragePoolDTO: +@instance_api.get('/', name='clusters:storage-pools:detail', response_model=StoragePoolDTO, responses=WATCH_RESPONSES) +def get(cluster: Cluster, pool: StoragePool, watch: WatchParam = False) -> Union[StoragePoolDTO, EventSourceResponse]: + if watch: + return sse_response( + pool_controller.watch_pool(cluster.get_id(), pool.get_id()), + _pool_dto, + single=True, + ) stat_obj = None return StoragePoolDTO.from_model(pool, stat_obj) diff --git a/simplyblock_web/api/v2/cluster/storage_pool/snapshot.py b/simplyblock_web/api/v2/cluster/storage_pool/snapshot.py index 96d091687..11dfc0ac4 100644 --- a/simplyblock_web/api/v2/cluster/storage_pool/snapshot.py +++ b/simplyblock_web/api/v2/cluster/storage_pool/snapshot.py @@ -1,20 +1,34 @@ -from typing import List +from typing import Callable, List, Union from fastapi import APIRouter, Response, Request +from sse_starlette import EventSourceResponse from simplyblock_core.db_controller import DBController from simplyblock_core.controllers import snapshot_controller +from simplyblock_core.models.snapshot import SnapShot as SnapshotModel from ..._dependencies import Cluster, StoragePool, Snapshot from ..._dtos import SnapshotDTO +from ..._sse import WATCH_RESPONSES, WatchParam, sse_response api = APIRouter() db = DBController() -@api.get('/', name='clusters:storage-pools:snapshots:list') -def list(request: Request, cluster: Cluster, pool: StoragePool) -> List[SnapshotDTO]: +def _make_snapshot_dto(request: Request, cluster_id: str, pool_id: str) -> Callable[[SnapshotModel], SnapshotDTO]: + def build(snapshot: SnapshotModel) -> SnapshotDTO: + return SnapshotDTO.from_model(snapshot, request, cluster_id=cluster_id, pool_id=pool_id) + return build + + +@api.get('/', name='clusters:storage-pools:snapshots:list', response_model=List[SnapshotDTO], responses=WATCH_RESPONSES) +def list(request: Request, cluster: Cluster, pool: StoragePool, watch: WatchParam = False) -> Union[List[SnapshotDTO], EventSourceResponse]: + if watch: + return sse_response( + snapshot_controller.watch_snapshots(cluster.get_id(), pool.get_id()), + _make_snapshot_dto(request, cluster.get_id(), pool.get_id()), + ) return [ SnapshotDTO.from_model(snapshot, request, cluster_id=cluster.get_id(), pool_id=pool.get_id()) for snapshot in db.get_snapshots_by_pool_id(pool.get_id()) @@ -24,8 +38,14 @@ def list(request: Request, cluster: Cluster, pool: StoragePool) -> List[Snapshot instance_api = APIRouter(prefix='/{snapshot_id}') -@instance_api.get('/', name='clusters:storage-pools:snapshots:detail') -def get(request: Request, cluster: Cluster, pool: StoragePool, snapshot: Snapshot) -> SnapshotDTO: +@instance_api.get('/', name='clusters:storage-pools:snapshots:detail', response_model=SnapshotDTO, responses=WATCH_RESPONSES) +def get(request: Request, cluster: Cluster, pool: StoragePool, snapshot: Snapshot, watch: WatchParam = False) -> Union[SnapshotDTO, EventSourceResponse]: + if watch: + return sse_response( + snapshot_controller.watch_snapshot(cluster.get_id(), pool.get_id(), snapshot.get_id()), + _make_snapshot_dto(request, cluster.get_id(), pool.get_id()), + single=True, + ) return SnapshotDTO.from_model(snapshot, request, cluster_id=cluster.get_id(), pool_id=pool.get_id()) diff --git a/simplyblock_web/api/v2/cluster/storage_pool/volume/__init__.py b/simplyblock_web/api/v2/cluster/storage_pool/volume/__init__.py index a2dbdedc0..e9f486e22 100644 --- a/simplyblock_web/api/v2/cluster/storage_pool/volume/__init__.py +++ b/simplyblock_web/api/v2/cluster/storage_pool/volume/__init__.py @@ -3,6 +3,7 @@ from fastapi import APIRouter, HTTPException, Request, Response from pydantic import BaseModel, Field, RootModel +from sse_starlette import EventSourceResponse from simplyblock_core.db_controller import DBController from simplyblock_core import utils as core_utils @@ -11,6 +12,7 @@ from ...._dependencies import Cluster, StoragePool, Volume from ...._dtos import BackupDTO, VolumeDTO, SnapshotDTO, TaskDTO +from ...._sse import WATCH_RESPONSES, WatchParam, sse_response from .... import util from .migration import api as migration_api @@ -19,8 +21,14 @@ db = DBController() -@api.get('/', name='clusters:storage-pools:volumes:list') -def list(request: Request, cluster: Cluster, pool: StoragePool) -> List[VolumeDTO]: +@api.get('/', name='clusters:storage-pools:volumes:list', response_model=List[VolumeDTO], responses=WATCH_RESPONSES) +def list(request: Request, cluster: Cluster, pool: StoragePool, watch: WatchParam = False) -> Union[List[VolumeDTO], EventSourceResponse]: + if watch: + cluster_id = cluster.get_id() + return sse_response( + lvol_controller.watch_volumes(cluster_id, pool.get_id()), + lambda lvol: VolumeDTO.from_model(lvol, request, cluster_id, None), + ) data = [] for lvol in db.get_lvols_by_pool_id(pool.get_id()): stat_obj = None @@ -140,8 +148,20 @@ def replicate_lvol_on_source_cluster(cluster: Cluster, pool: StoragePool, body: instance_api = APIRouter(prefix='/{volume_id}') -@instance_api.get('/', name='clusters:storage-pools:volumes:detail') -def get(request: Request, cluster: Cluster, pool: StoragePool, volume: Volume) -> VolumeDTO: +@instance_api.get('/', name='clusters:storage-pools:volumes:detail', response_model=VolumeDTO, responses=WATCH_RESPONSES) +def get(request: Request, cluster: Cluster, pool: StoragePool, volume: Volume, watch: WatchParam = False) -> Union[VolumeDTO, EventSourceResponse]: + if watch: + cluster_id = cluster.get_id() + + def _volume_dto(lvol: LVol) -> VolumeDTO: + rep_info = lvol_controller.get_replication_info(lvol.get_id()) + return VolumeDTO.from_model(lvol, request, cluster_id, None, rep_info) + + return sse_response( + lvol_controller.watch_volume(cluster_id, pool.get_id(), volume.get_id()), + _volume_dto, + single=True, + ) stat_obj = None rep_info = lvol_controller.get_replication_info(volume.get_id()) return VolumeDTO.from_model(volume, request, cluster.get_id(), stat_obj, rep_info) diff --git a/simplyblock_web/api/v2/cluster/task.py b/simplyblock_web/api/v2/cluster/task.py index 42ac831be..d81aec481 100644 --- a/simplyblock_web/api/v2/cluster/task.py +++ b/simplyblock_web/api/v2/cluster/task.py @@ -1,20 +1,32 @@ -from typing import List +from typing import List, Union from fastapi import APIRouter +from sse_starlette import EventSourceResponse from simplyblock_core.db_controller import DBController +from simplyblock_core.controllers import tasks_controller from simplyblock_core.models.job_schedule import JobSchedule from .._dependencies import Cluster, Task from .._dtos import TaskDTO +from .._sse import WATCH_RESPONSES, WatchParam, sse_response api = APIRouter() db = DBController() -@api.get('/', name='clusters:tasks:list') -def list(cluster: Cluster) -> List[TaskDTO]: +def _task_dto(task: JobSchedule) -> TaskDTO: + return TaskDTO.from_model(task) + + +@api.get('/', name='clusters:tasks:list', response_model=List[TaskDTO], responses=WATCH_RESPONSES) +def list(cluster: Cluster, watch: WatchParam = False) -> Union[List[TaskDTO], EventSourceResponse]: + if watch: + return sse_response( + tasks_controller.watch_tasks(cluster.get_id()), + _task_dto, + ) cluster_tasks = db.get_job_tasks(cluster.get_id(), limit=0) data = [] for t in cluster_tasks: @@ -27,8 +39,14 @@ def list(cluster: Cluster) -> List[TaskDTO]: instance_api = APIRouter(prefix='/{task_id}') -@instance_api.get('/', name='clusters:tasks:detail') -def get(cluster: Cluster, task: Task) -> TaskDTO: +@instance_api.get('/', name='clusters:tasks:detail', response_model=TaskDTO, responses=WATCH_RESPONSES) +def get(cluster: Cluster, task: Task, watch: WatchParam = False) -> Union[TaskDTO, EventSourceResponse]: + if watch: + return sse_response( + tasks_controller.watch_task(cluster.get_id(), task.uuid), + _task_dto, + single=True, + ) return TaskDTO.from_model(task) diff --git a/tests/integration/test_watch_counter_bumps.py b/tests/integration/test_watch_counter_bumps.py new file mode 100644 index 000000000..e36ebbf95 --- /dev/null +++ b/tests/integration/test_watch_counter_bumps.py @@ -0,0 +1,104 @@ +# coding=utf-8 +"""Every entity write path bumps the per-class change counter — real FDB.""" + +from uuid import uuid4 + +from simplyblock_core import watches +from simplyblock_core.db_controller import DBController +from simplyblock_core.models.pool import Pool +from simplyblock_core.models.stats import LVolStatObject +from simplyblock_core.models.storage_node import StorageNode + + +def _counter(db, model_cls): + raw = db.kv_store.get(watches.watch_counter_key(model_cls)) + return watches.unpack_counter(bytes(raw)) if raw is not None else 0 + + +def _make_pool(cluster_id=None): + return Pool({ + 'uuid': str(uuid4()), + 'cluster_id': cluster_id or str(uuid4()), + 'pool_name': 'watch-test-pool', + 'status': Pool.STATUS_ACTIVE, + }) + + +def test_write_to_db_bumps_counter_by_one(): + db = DBController() + pool = _make_pool() + before = _counter(db, Pool) + pool.write_to_db(db.kv_store) + assert _counter(db, Pool) == before + 1 + assert db.get_pool_by_id(pool.get_id()).get_id() == pool.get_id() + + +def test_remove_bumps_counter(): + db = DBController() + pool = _make_pool() + pool.write_to_db(db.kv_store) + before = _counter(db, Pool) + pool.remove(db.kv_store) + assert _counter(db, Pool) == before + 1 + + +def test_atomic_update_bumps_counter(): + db = DBController() + node = StorageNode({ + 'uuid': str(uuid4()), + 'cluster_id': str(uuid4()), + 'status': StorageNode.STATUS_ONLINE, + }) + node.write_to_db(db.kv_store) + before = _counter(db, StorageNode) + + def mutate(fresh): + fresh.status = StorageNode.STATUS_SUSPENDED + + updated = db.atomic_update(node, mutate) + assert updated.status == StorageNode.STATUS_SUSPENDED + assert _counter(db, StorageNode) == before + 1 + + +def test_try_set_node_restarting_bumps_counter(monkeypatch): + from simplyblock_core import distr_controller + monkeypatch.setattr(distr_controller, 'send_node_status_event', lambda *a, **kw: None) + + db = DBController() + cluster_id = str(uuid4()) + node = StorageNode({ + 'uuid': str(uuid4()), + 'cluster_id': cluster_id, + 'status': StorageNode.STATUS_ONLINE, + }) + node.write_to_db(db.kv_store) + before = _counter(db, StorageNode) + + acquired, reason = db.try_set_node_restarting(cluster_id, node.get_id()) + assert acquired, reason + assert _counter(db, StorageNode) == before + 1 + + +def test_counter_keys_invisible_to_entity_reads(): + db = DBController() + pool = _make_pool() + pool.write_to_db(db.kv_store) + # A counter key inside the entity prefix would blow up the JSON parse in + # read_from_db; parsing the whole range cleanly is the collision guard. + pools = Pool().read_from_db(db.kv_store) + assert pool.get_id() in {p.get_id() for p in pools} + entity_keys = [bytes(k) for k, _ in db.kv_store.get_range_startswith(b'object/Pool/')] + assert all(b'watch_seq' not in key for key in entity_keys) + + +def test_unwatched_write_creates_no_counter(): + db = DBController() + before = _counter(db, LVolStatObject) + stat = LVolStatObject({ + 'uuid': str(uuid4()), + 'cluster_id': str(uuid4()), + 'pool_id': str(uuid4()), + 'date': 1, + }) + stat.write_to_db(db.kv_store) + assert _counter(db, LVolStatObject) == before diff --git a/tests/integration/test_watch_hub.py b/tests/integration/test_watch_hub.py new file mode 100644 index 000000000..00323c320 --- /dev/null +++ b/tests/integration/test_watch_hub.py @@ -0,0 +1,100 @@ +# coding=utf-8 +"""WatchHub against real FDB: watch firing, reconcile poll, teardown.""" + +import asyncio +import json +import time +from uuid import uuid4 + +import pytest + +from simplyblock_core.db_controller import DBController +from simplyblock_core.models.pool import Pool +from simplyblock_core import watch +from simplyblock_core.watch import _UNSET, WatchHub + + +def _pool_dict(pool_id=None): + return { + 'uuid': pool_id or str(uuid4()), + 'cluster_id': str(uuid4()), + 'pool_name': 'hub-test-pool', + 'status': Pool.STATUS_ACTIVE, + } + + +async def _await_publication(sub, notify, timeout=10.0): + """Wait until the subscription has a fresh state and return it.""" + deadline = time.monotonic() + timeout + while True: + remaining = deadline - time.monotonic() + assert remaining > 0, 'no publication within timeout' + await asyncio.wait_for(notify.wait(), remaining) + notify.clear() + state = sub.take() + if state is not _UNSET: + return state + + +@pytest.mark.timeout(60) +def test_hub_publishes_writes_and_stops_on_unsubscribe(): + db = DBController() + + async def scenario(): + loop = asyncio.get_running_loop() + notify = asyncio.Event() + hub = WatchHub(Pool) + sub = hub.subscribe(loop, notify) + try: + # Initial publication (whatever the range currently holds). + await _await_publication(sub, notify) + + pool = Pool(_pool_dict()) + started = time.monotonic() + await asyncio.to_thread(pool.write_to_db, db.kv_store) + state = await _await_publication(sub, notify, timeout=5.0) + assert time.monotonic() - started < 5.0 + assert any(m.get_id() == pool.get_id() for m in state) + + await asyncio.to_thread(pool.remove, db.kv_store) + state = await _await_publication(sub, notify, timeout=5.0) + assert not any(m.get_id() == pool.get_id() for m in state) + finally: + hub.unsubscribe(sub) + + thread = hub._thread + deadline = time.monotonic() + 10 + while thread.is_alive() and time.monotonic() < deadline: + await asyncio.sleep(0.1) + assert not thread.is_alive() + + asyncio.run(scenario()) + + +@pytest.mark.timeout(60) +def test_reconcile_catches_writes_without_counter_bump(monkeypatch): + """Old-version writers don't bump counters; the reconcile poll must + still pick their writes up.""" + monkeypatch.setattr(watch, 'WATCH_RECONCILE_SEC', 0.5) + db = DBController() + + async def scenario(): + loop = asyncio.get_running_loop() + notify = asyncio.Event() + hub = WatchHub(Pool) + sub = hub.subscribe(loop, notify) + try: + await _await_publication(sub, notify) + + # Simulate a pre-upgrade writer: raw set, no counter bump. + legacy = _pool_dict() + key = f'object/Pool/{legacy["uuid"]}'.encode() + await asyncio.to_thread( + db.kv_store.set, key, json.dumps(legacy).encode()) + + state = await _await_publication(sub, notify, timeout=10.0) + assert any(m.get_id() == legacy['uuid'] for m in state) + finally: + hub.unsubscribe(sub) + + asyncio.run(scenario()) diff --git a/tests/integration/web/__init__.py b/tests/integration/web/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/integration/web/test_watch_sse_e2e.py b/tests/integration/web/test_watch_sse_e2e.py new file mode 100644 index 000000000..f85d8fdbd --- /dev/null +++ b/tests/integration/web/test_watch_sse_e2e.py @@ -0,0 +1,215 @@ +# coding=utf-8 +"""End-to-end SSE watch: real app in a real uvicorn server on a real FDB. + +Runs the actual ``simplyblock_web.app:app`` (including AccessLogMiddleware — +the BaseHTTPMiddleware whose interaction with streaming responses this pins) +in a background uvicorn thread and consumes the stream over a real socket +with ``requests``, so client-disconnect semantics are exercised for real. +""" + +import json +import threading +import time +from uuid import uuid4 + +import pytest +import requests +import uvicorn + +from simplyblock_core.db_controller import DBController +from simplyblock_core.models.cluster import Cluster +from simplyblock_core.models.lvol_model import LVol +from simplyblock_core.models.pool import Pool +from simplyblock_core.watch import get_hub + + +@pytest.fixture(scope='module') +def base_url(): + from simplyblock_web.app import app + from simplyblock_web.api.v2._auth import verify_api_token + + app.dependency_overrides[verify_api_token] = lambda: None + config = uvicorn.Config(app, host='127.0.0.1', port=0, log_level='warning') + server = uvicorn.Server(config) + thread = threading.Thread(target=server.run, daemon=True) + thread.start() + deadline = time.monotonic() + 15 + while not server.started: + assert time.monotonic() < deadline, 'uvicorn did not start' + time.sleep(0.05) + port = server.servers[0].sockets[0].getsockname()[1] + yield f'http://127.0.0.1:{port}' + server.should_exit = True + thread.join(timeout=10) + app.dependency_overrides.clear() + + +class SSEReader: + """Minimal SSE parser over a requests streaming response.""" + + def __init__(self, response): + # requests' iter_lines mangles \r\n-terminated SSE lines (spurious + # empty lines that are indistinguishable from event separators), so + # buffer raw content and split on \n ourselves. + self._content = response.iter_content(chunk_size=None, decode_unicode=True) + self._buffer = '' + + def _readline(self): + while '\n' not in self._buffer: + try: + self._buffer += next(self._content) + except StopIteration: + return None + line, self._buffer = self._buffer.split('\n', 1) + return line.rstrip('\r') + + def next_event(self): + """Return (event, data) for the next event, or (None, None) on EOF.""" + name, data = None, [] + while True: + line = self._readline() + if line is None: + return None, None + if line == '': + if name is not None or data: + return name, '\n'.join(data) + continue + if line.startswith(':'): + continue + field, _, value = line.partition(':') + value = value.removeprefix(' ') + if field == 'event': + name = value + elif field == 'data': + data.append(value) + + +def _seed_cluster(db): + cluster = Cluster({'uuid': str(uuid4()), 'status': Cluster.STATUS_ACTIVE}) + cluster.write_to_db(db.kv_store) + return cluster + + +def _seed_pool(db, cluster): + pool = Pool({ + 'uuid': str(uuid4()), + 'cluster_id': cluster.get_id(), + 'pool_name': f'pool-{uuid4().hex[:8]}', + 'status': Pool.STATUS_ACTIVE, + }) + pool.write_to_db(db.kv_store) + return pool + + +def _seed_lvol(db, pool, name): + lvol = LVol({ + 'uuid': str(uuid4()), + 'pool_uuid': pool.get_id(), + 'pool_name': pool.pool_name, + 'lvol_name': name, + 'node_id': str(uuid4()), + 'status': LVol.STATUS_ONLINE, + 'size': 1024, + }) + lvol.write_to_db(db.kv_store) + return lvol + + +def _volumes_watch_url(base_url, cluster, pool): + return (f'{base_url}/api/v2/clusters/{cluster.get_id()}' + f'/storage-pools/{pool.get_id()}/volumes/?watch=true') + + +@pytest.mark.timeout(120) +def test_volume_watch_stream_flow(base_url): + db = DBController() + cluster = _seed_cluster(db) + pool = _seed_pool(db, cluster) + other_pool = _seed_pool(db, cluster) + lvol = _seed_lvol(db, pool, 'watched-1') + + with requests.get(_volumes_watch_url(base_url, cluster, pool), + stream=True, timeout=(5, 30)) as response: + assert response.status_code == 200 + assert response.headers['content-type'].startswith('text/event-stream') + reader = SSEReader(response) + + name, data = reader.next_event() + assert name == 'snapshot' + snapshot = json.loads(data) + assert [v['id'] for v in snapshot] == [lvol.get_id()] + + # A volume in another pool must not produce an event; the next event + # observed has to be the update of the watched volume. + _seed_lvol(db, other_pool, 'other-pool-volume') + lvol.status = LVol.STATUS_OFFLINE + lvol.write_to_db(db.kv_store) + name, data = reader.next_event() + assert name == 'updated' + assert json.loads(data)['id'] == lvol.get_id() + assert json.loads(data)['status'] == LVol.STATUS_OFFLINE + + second = _seed_lvol(db, pool, 'watched-2') + name, data = reader.next_event() + assert name == 'created' + assert json.loads(data)['id'] == second.get_id() + + # Soft delete: the entity leaves the filtered set but its key is + # still retrievable — the event carries the final representation. + second.status = LVol.STATUS_DELETED + second.write_to_db(db.kv_store) + name, data = reader.next_event() + assert name == 'deleted' + payload = json.loads(data) + assert payload['id'] == second.get_id() + assert payload['status'] == LVol.STATUS_DELETED + + # Physical removal: no representation left to return. + lvol.remove(db.kv_store) + name, data = reader.next_event() + assert name == 'deleted' + assert json.loads(data) == {} + + +@pytest.mark.timeout(120) +def test_parent_pool_deletion_closes_stream(base_url): + db = DBController() + cluster = _seed_cluster(db) + pool = _seed_pool(db, cluster) + _seed_lvol(db, pool, 'orphan-to-be') + + with requests.get(_volumes_watch_url(base_url, cluster, pool), + stream=True, timeout=(5, 30)) as response: + reader = SSEReader(response) + name, _ = reader.next_event() + assert name == 'snapshot' + + pool.remove(db.kv_store) + name, data = reader.next_event() + assert (name, data) == (None, None) # stream closed + + # Reconnect gets the endpoint's regular 404. + response = requests.get(_volumes_watch_url(base_url, cluster, pool), timeout=5) + assert response.status_code == 404 + + +@pytest.mark.timeout(120) +def test_client_disconnect_unsubscribes(base_url): + db = DBController() + cluster = _seed_cluster(db) + pool = _seed_pool(db, cluster) + _seed_lvol(db, pool, 'disconnect-test') + + response = requests.get(_volumes_watch_url(base_url, cluster, pool), + stream=True, timeout=(5, 30)) + reader = SSEReader(response) + name, _ = reader.next_event() + assert name == 'snapshot' + assert get_hub(LVol).subscriber_count() > 0 + + response.close() + + deadline = time.monotonic() + 20 + while get_hub(LVol).subscriber_count() > 0 and time.monotonic() < deadline: + time.sleep(0.2) + assert get_hub(LVol).subscriber_count() == 0 diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index 5d0f08aa1..5ddd4b962 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -22,7 +22,13 @@ def _stub(name, **attrs): if 'fdb' not in sys.modules: class _FDBError(Exception): pass - _stub('fdb', open=lambda *a, **kw: None, FDBError=_FDBError) + # ``transactional`` mirrors the real decorator's calling convention: the + # wrapped function's first argument is the database/transaction. Unit + # tests pass a MagicMock kv_store, so running the body directly against + # it keeps existing ``kv_store.set(...)`` assertions working while + # watched-model writes land their counter ``add(...)`` on the same mock. + _stub('fdb', open=lambda *a, **kw: None, FDBError=_FDBError, + transactional=lambda f: f) _stub('fdb.tuple') diff --git a/tests/unit/test_watch_core.py b/tests/unit/test_watch_core.py new file mode 100644 index 000000000..963aba7e3 --- /dev/null +++ b/tests/unit/test_watch_core.py @@ -0,0 +1,197 @@ +# coding=utf-8 +"""Core watch primitive: diff_by_id and the async watch() generator.""" + +import asyncio + +import pytest + +from simplyblock_core import watch as watchmod +from simplyblock_core.watch import ( + WatchUnavailable, _Subscription, diff_by_id, watch, +) + + +class FakeModel: + def __init__(self, entity_id, **fields): + self._id = entity_id + self._fields = fields + + def get_id(self): + return self._id + + def to_dict(self): + return {'uuid': self._id, **self._fields} + + +def _m(entity_id, **fields): + return FakeModel(entity_id, **fields) + + +# ---- diff_by_id (pure function) ---- + +def test_diff_created_updated_deleted(): + prev = {'a': _m('a', status='x').to_dict(), 'b': _m('b', status='x').to_dict()} + b2, c = _m('b', status='y'), _m('c', status='x') + new = {'b': b2, 'c': c} + events, new_dicts = diff_by_id(prev, new, dict(new)) + by_kind = {e.kind: e for e in events} + assert set(by_kind) == {'created', 'updated', 'deleted'} + assert by_kind['created'].id == 'c' + assert by_kind['updated'].id == 'b' and by_kind['updated'].model is b2 + assert by_kind['deleted'].id == 'a' + assert by_kind['deleted'].model is None # 'a' absent from full: physically gone + assert set(new_dicts) == {'b', 'c'} + + +def test_diff_unchanged_emits_nothing(): + prev = {'a': _m('a', status='x').to_dict()} + events, _ = diff_by_id(prev, {'a': _m('a', status='x')}, {'a': _m('a', status='x')}) + assert events == [] + + +def test_diff_scope_leave_carries_current_state(): + """Left the scoped set but still present in full -> deleted carries the + current (unscoped) model, not a stale one.""" + prev = {'a': _m('a', status='online').to_dict()} + current = _m('a', status='deleted') + events, _ = diff_by_id(prev, {}, {'a': current}) + [event] = events + assert event.kind == 'deleted' and event.model is current + + +def test_diff_created_order_follows_new(): + events, _ = diff_by_id({}, {'a': _m('a'), 'b': _m('b')}, {'a': _m('a'), 'b': _m('b')}) + assert [e.id for e in events] == ['a', 'b'] + + +# ---- watch() generator with a fake hub ---- + +class FakeHub: + def __init__(self): + self.subs = [] + self.cache = None + + def subscribe(self, loop, notify): + sub = _Subscription(loop, notify) + self.subs.append(sub) + if self.cache is not None: + sub._deliver(self.cache) + return sub + + def unsubscribe(self, sub): + self.subs.remove(sub) + + def publish(self, models): + self.cache = models + for sub in list(self.subs): + sub._deliver(models) + + def fail(self): + for sub in list(self.subs): + sub._fail() + + +class Thing: + pass + + +class Parent: + pass + + +def _patch(monkeypatch, hubs): + monkeypatch.setattr(watchmod, 'get_hub', lambda cls: hubs[cls]) + + +async def _next(gen, timeout=2.0): + return await asyncio.wait_for(gen.__anext__(), timeout) + + +def test_watch_snapshot_update_delete(monkeypatch): + hub = FakeHub() + _patch(monkeypatch, {Thing: hub}) + hub.publish([_m('a', status='online')]) + + async def scenario(): + gen = watch(Thing) + [event] = await _next(gen) + assert event.kind == 'created' and event.id == 'a' + + hub.publish([_m('a', status='offline')]) + [event] = await _next(gen) + assert event.kind == 'updated' and event.model.to_dict()['status'] == 'offline' + + hub.publish([]) + [event] = await _next(gen) + assert event.kind == 'deleted' and event.id == 'a' + + await gen.aclose() + assert hub.subs == [] + + asyncio.run(scenario()) + + +def test_watch_coalesces_bursts(monkeypatch): + hub = FakeHub() + _patch(monkeypatch, {Thing: hub}) + hub.publish([_m('a', status='s0')]) + + async def scenario(): + gen = watch(Thing) + assert (await _next(gen))[0].kind == 'created' + + # Two publications before the generator runs: only the latest diffs. + hub.publish([_m('a', status='s1')]) + hub.publish([_m('a', status='s2')]) + [event] = await _next(gen) + assert event.kind == 'updated' and event.model.to_dict()['status'] == 's2' + + with pytest.raises(asyncio.TimeoutError): + await _next(gen, timeout=0.2) + + asyncio.run(scenario()) + + +def test_watch_scope_select(monkeypatch): + hub = FakeHub() + _patch(monkeypatch, {Thing: hub}) + hub.publish([_m('a', keep=True), _m('b', keep=False)]) + + async def scenario(): + gen = watch(Thing, select=lambda models: [m for m in models if m.to_dict()['keep']]) + events = await _next(gen) + assert [e.id for e in events] == ['a'] + + asyncio.run(scenario()) + + +def test_watch_ancestor_deletion_closes(monkeypatch): + hub, parent = FakeHub(), FakeHub() + _patch(monkeypatch, {Thing: hub, Parent: parent}) + hub.publish([_m('a')]) + parent.publish([_m('p')]) + + async def scenario(): + gen = watch(Thing, ancestors=[(Parent, 'p')]) + assert (await _next(gen))[0].id == 'a' + parent.publish([]) + with pytest.raises(StopAsyncIteration): + await _next(gen) + assert hub.subs == [] and parent.subs == [] + + asyncio.run(scenario()) + + +def test_watch_failure_raises(monkeypatch): + hub = FakeHub() + _patch(monkeypatch, {Thing: hub}) + hub.publish([_m('a')]) + + async def scenario(): + gen = watch(Thing) + await _next(gen) + hub.fail() + with pytest.raises(WatchUnavailable): + await _next(gen) + + asyncio.run(scenario()) diff --git a/tests/unit/test_watch_counters.py b/tests/unit/test_watch_counters.py new file mode 100644 index 000000000..0e13579e3 --- /dev/null +++ b/tests/unit/test_watch_counters.py @@ -0,0 +1,106 @@ +# coding=utf-8 +"""Change-counter keys and the transactional write+bump in BaseModel.""" + +import struct +from unittest.mock import MagicMock + +import pytest + +from simplyblock_core import watches +from simplyblock_core.models.base_model import BaseModel +from simplyblock_core.models.cluster import Cluster, ClusterAddNodeLock, PortReservation +from simplyblock_core.models.events import EventObj +from simplyblock_core.models.job_schedule import JobSchedule +from simplyblock_core.models.lvol_model import LVol, LVolMini +from simplyblock_core.models.pool import Pool +from simplyblock_core.models.snapshot import SnapShot, SnapShotMini +from simplyblock_core.models.stats import LVolStatObject, StatsObject +from simplyblock_core.models.storage_node import StorageNode + + +WATCHED_CLASSES = [Cluster, StorageNode, Pool, LVol, SnapShot, JobSchedule] +UNWATCHED_CLASSES = [ + EventObj, LVolMini, SnapShotMini, StatsObject, LVolStatObject, + ClusterAddNodeLock, PortReservation, +] + + +def test_counter_key_format(): + assert watches.watch_counter_key(StorageNode) == b'watch_seq/StorageNode' + assert watches.watch_counter_key(LVol) == b'watch_seq/LVol' + + +@pytest.mark.parametrize('model_cls', WATCHED_CLASSES) +def test_counter_keys_disjoint_from_entity_scans(model_cls): + """read_from_db scans '//' prefixes; counter keys + must never be picked up by any such scan.""" + counter_key = watches.watch_counter_key(model_cls) + for other_cls in WATCHED_CLASSES + UNWATCHED_CLASSES: + instance = other_cls() + prefix = f'{instance.object_type}/{instance.name}/'.encode() + assert not counter_key.startswith(prefix) + + +@pytest.mark.parametrize('model_cls', WATCHED_CLASSES) +def test_watched_flag_set(model_cls): + assert model_cls._WATCHED is True + + +@pytest.mark.parametrize('model_cls', UNWATCHED_CLASSES) +def test_unwatched_flag_not_set(model_cls): + assert model_cls._WATCHED is False + + +def test_watched_write_bumps_counter(): + kv = MagicMock() + pool = Pool({'uuid': 'pool-1'}) + pool.write_to_db(kv) + kv.set.assert_called_once() + key, _value = kv.set.call_args[0] + assert key == b'object/Pool/pool-1' + kv.add.assert_called_once_with(b'watch_seq/Pool', watches.ONE_LE64) + + +def test_unwatched_write_does_not_bump(): + kv = MagicMock() + event = EventObj({'uuid': 'ev-1'}) + event.write_to_db(kv) + kv.set.assert_called_once() + kv.add.assert_not_called() + + +def test_watched_remove_bumps_counter(): + kv = MagicMock() + pool = Pool({'uuid': 'pool-1'}) + pool.remove(kv) + kv.clear.assert_called_once_with(b'object/Pool/pool-1') + kv.add.assert_called_once_with(b'watch_seq/Pool', watches.ONE_LE64) + + +def test_unwatched_remove_does_not_bump(): + kv = MagicMock() + event = EventObj({'uuid': 'ev-1', 'cluster_uuid': 'c', 'date': 5}) + event.remove(kv) + kv.clear.assert_called_once() + kv.add.assert_not_called() + + +def test_compound_key_write_bumps_class_counter(): + """JobSchedule's compound get_id() must not leak into the counter key.""" + kv = MagicMock() + task = JobSchedule({'uuid': 'task-1', 'cluster_id': 'cl-1', 'date': 42}) + task.write_to_db(kv) + key, _value = kv.set.call_args[0] + assert key == b'object/JobSchedule/cl-1/42/task-1' + kv.add.assert_called_once_with(b'watch_seq/JobSchedule', watches.ONE_LE64) + + +def test_base_model_default_unwatched(): + assert BaseModel._WATCHED is False + + +def test_unpack_counter(): + assert watches.unpack_counter(None) == 0 + assert watches.unpack_counter(b'') == 0 + assert watches.unpack_counter(struct.pack(' _DTO: + return _DTO(id=model.get_id(), status=model.to_dict().get('status', '')) + + +def _created(entity_id, **f): + return ChangeEvent('created', entity_id, FakeModel(entity_id, **f)) + + +def _updated(entity_id, **f): + return ChangeEvent('updated', entity_id, FakeModel(entity_id, **f)) + + +def _deleted(entity_id, model=None): + return ChangeEvent('deleted', entity_id, model) + + +# ---- _build_snapshot ---- + +def test_snapshot_collection_is_json_array(): + events, cache = _build_snapshot([_created('a'), _created('b')], _dto, single=False) + [event] = events + assert event.event == 'snapshot' + assert event.data.startswith('[') and event.data.endswith(']') + assert '"a"' in event.data and '"b"' in event.data + assert set(cache) == {'a', 'b'} + + +def test_snapshot_single_is_bare_object(): + events, _ = _build_snapshot([_created('a')], _dto, single=True) + [event] = events + assert event.event == 'snapshot' + assert event.data.startswith('{') + + +def test_snapshot_single_empty_emits_nothing(): + events, cache = _build_snapshot([], _dto, single=True) + assert events == [] and cache == {} + + +# ---- _build_events ---- + +def test_events_created_updated_deleted(): + _, cache = _build_snapshot([_created('a', status='x')], _dto, single=False) + events, new_cache, deleted = _build_events( + [_updated('a', status='y'), _created('c', status='x'), _deleted('b')], + cache, _dto, single=False) + by_name = {e.event: e.data for e in events} + assert set(by_name) == {'updated', 'created', 'deleted'} + assert '"y"' in by_name['updated'] + assert '"c"' in by_name['created'] + assert by_name['deleted'] == '{}' # physically gone: no representation + assert deleted is True + assert set(new_cache) == {'a', 'c'} + + +def test_events_suppress_dto_invisible_update(): + """Raw model changed (lease heartbeat) but the DTO is identical -> no event.""" + task = { + 'uuid': str(uuid4()), 'cluster_id': str(uuid4()), 'date': 1, + 'function_name': 'node_restart', 'status': 'running', + 'updated_at': '2026-01-01T00:00:00', + } + build = lambda m: TaskDTO.from_model(m) # noqa: E731 + js1 = JobSchedule(task) + _, cache, _ = _build_events([ChangeEvent('created', js1.get_id(), js1)], {}, build, single=False) + js2 = JobSchedule({**task, 'updated_at': '2026-01-01T00:00:05'}) + events, _, _ = _build_events([ChangeEvent('updated', js2.get_id(), js2)], cache, build, single=False) + assert events == [] + + +def test_events_single_maps_created_to_updated(): + events, _, _ = _build_events([_created('a')], {}, _dto, single=True) + [event] = events + assert event.event == 'updated' + + +def test_events_deleted_carries_current_state(): + events, _, _ = _build_events( + [_deleted('a', FakeModel('a', status='deleted'))], {}, _dto, single=False) + [event] = events + assert event.event == 'deleted' + assert '"deleted"' in event.data + + +# ---- sse_response generator ---- + +class FakeStream: + def __init__(self, batches): + self._it = iter(batches) + self.closed = False + + def __aiter__(self): + return self + + async def __anext__(self): + try: + return next(self._it) + except StopIteration: + raise StopAsyncIteration + + async def aclose(self): + self.closed = True + + +class FailingStream: + def __init__(self): + self.n = 0 + + def __aiter__(self): + return self + + async def __anext__(self): + self.n += 1 + if self.n == 1: + return [_created('a')] + raise WatchUnavailable() + + async def aclose(self): + pass + + +async def _event(gen, timeout=2.0): + return await asyncio.wait_for(gen.__anext__(), timeout) + + +def test_stream_snapshot_update_delete(monkeypatch): + monkeypatch.setattr(_sse, 'backend_available', lambda: True) + + async def scenario(): + stream = FakeStream([[_created('a', status='online')], + [_updated('a', status='offline')], + [_deleted('a')]]) + gen = sse_response(stream, _dto).body_iterator + assert (await _event(gen)).event == 'snapshot' + event = await _event(gen) + assert event.event == 'updated' and 'offline' in event.data + assert (await _event(gen)).event == 'deleted' + with pytest.raises(StopAsyncIteration): + await _event(gen) + assert stream.closed + + asyncio.run(scenario()) + + +def test_detail_stream_closes_after_delete(monkeypatch): + monkeypatch.setattr(_sse, 'backend_available', lambda: True) + + async def scenario(): + stream = FakeStream([[_created('a')], [_deleted('a')]]) + gen = sse_response(stream, _dto, single=True).body_iterator + assert (await _event(gen)).event == 'snapshot' + assert (await _event(gen)).event == 'deleted' + with pytest.raises(StopAsyncIteration): + await _event(gen) + + asyncio.run(scenario()) + + +def test_stream_failure_emits_error_and_closes(monkeypatch): + monkeypatch.setattr(_sse, 'backend_available', lambda: True) + + async def scenario(): + gen = sse_response(FailingStream(), _dto).body_iterator + assert (await _event(gen)).event == 'snapshot' + event = await _event(gen) + assert event.event == 'error' and 'backend unavailable' in event.data + with pytest.raises(StopAsyncIteration): + await _event(gen) + + asyncio.run(scenario()) + + +def test_kv_store_unavailable_yields_503(monkeypatch): + from fastapi import HTTPException + monkeypatch.setattr(_sse, 'backend_available', lambda: False) + with pytest.raises(HTTPException) as exc_info: + sse_response(FakeStream([]), _dto) + assert exc_info.value.status_code == 503 + + +# ---- secrets on the stream (ClusterDTO carries `secret`) ---- + +def test_stream_payload_unwraps_secret_on_wire(caplog): + """Same behavior as the plain GET: JSON wire carries plaintext.""" + cluster = ClusterModel({'uuid': str(uuid4()), 'secret': 'hunter2', 'status': 'active'}) + build = lambda m: ClusterDTO.from_model(m, None) # noqa: E731 + with caplog.at_level(logging.DEBUG): + events, _ = _build_snapshot( + [ChangeEvent('created', cluster.get_id(), cluster)], build, single=False) + [event] = events + assert 'hunter2' in event.data + assert 'hunter2' not in caplog.text + + +def test_dto_python_mode_masks_secret(): + cluster = ClusterModel({'uuid': str(uuid4()), 'secret': 'hunter2', 'status': 'active'}) + dto = ClusterDTO.from_model(cluster, None) + assert 'hunter2' not in repr(dto) + assert str(dto.model_dump()['secret']) == '**********' diff --git a/tox.ini b/tox.ini index 9d03e70a6..48b55a310 100644 --- a/tox.ini +++ b/tox.ini @@ -1,5 +1,5 @@ [tox] -envlist = unit, integration, lint, types +envlist = lint, types, unit, integration skipsdist = true requires = tox-uv>=1