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 46ec21931..cc0d0846d 100755 --- a/simplyblock_core/cluster_ops.py +++ b/simplyblock_core/cluster_ops.py @@ -37,6 +37,18 @@ db_controller = DBController() + +async def watch_clusters(): + """Stream changes across all clusters.""" + async for batch in db_controller.watch(Cluster): + yield batch + + +async def watch_cluster(cluster_id): + """Stream changes for a single cluster.""" + async for batch in db_controller.watch(Cluster, entity_id=cluster_id): + yield batch + 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 bbf836daf..fba87de48 100644 --- a/simplyblock_core/controllers/device_controller.py +++ b/simplyblock_core/controllers/device_controller.py @@ -7,6 +7,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 @@ -21,6 +22,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 [] + + +async def watch_devices(cluster_id, node_id): + """Stream device changes for one storage node (a device change is a node write).""" + db = DBController() + async for batch in db.watch( + StorageNode, scope=(cluster_id,), entity_id=node_id, + select=lambda nodes: _explode_devices(nodes, node_id), + ancestors=[(Cluster, (), cluster_id), (StorageNode, (cluster_id,), node_id)]): + yield batch + + +async def watch_device(cluster_id, node_id, device_id): + """Stream changes for a single device.""" + db = DBController() + async for batch in db.watch( + StorageNode, scope=(cluster_id,), entity_id=node_id, + select=lambda nodes: [ + device for device in _explode_devices(nodes, node_id) + if device.get_id() == device_id + ], + ancestors=[(Cluster, (), cluster_id), (StorageNode, (cluster_id,), node_id)]): + yield batch + + 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 7c05614fb..415f0ba14 100755 --- a/simplyblock_core/controllers/lvol_controller.py +++ b/simplyblock_core/controllers/lvol_controller.py @@ -28,6 +28,28 @@ logger = utils.get_logger(__name__) +async def watch_volumes(cluster_id, pool_id): + """Stream volume changes for one pool (same scope as get_lvols_by_pool_id).""" + db = DBController() + async for batch in db.watch( + LVol, scope=(pool_id,), + select=lambda models: db.get_lvols_by_pool_id(pool_id, source=models), + ancestors=[(Cluster, (), cluster_id), (Pool, (cluster_id,), pool_id)]): + yield batch + + +async def watch_volume(cluster_id, pool_id, volume_id): + """Stream changes for a single volume.""" + db = DBController() + async for batch in db.watch( + LVol, scope=(pool_id,), entity_id=volume_id, + # Reuse the pool getter so deleted volumes leave the set (soft-delete + # -> deleted event + close), matching the list endpoint's filter. + select=lambda models: db.get_lvols_by_pool_id(pool_id, source=models), + ancestors=[(Cluster, (), cluster_id), (Pool, (cluster_id,), pool_id)]): + yield batch + + 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..3f02cc30e 100644 --- a/simplyblock_core/controllers/pool_controller.py +++ b/simplyblock_core/controllers/pool_controller.py @@ -13,12 +13,32 @@ 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() +async def watch_pools(cluster_id): + """Stream pool changes for one cluster (same scope as get_pools).""" + db = DBController() + async for batch in db.watch( + Pool, scope=(cluster_id,), + select=lambda models: db.get_pools(cluster_id, source=models), + ancestors=[(Cluster, (), cluster_id)]): + yield batch + + +async def watch_pool(cluster_id, pool_id): + """Stream changes for a single pool.""" + db = DBController() + async for batch in db.watch( + Pool, scope=(cluster_id,), entity_id=pool_id, + ancestors=[(Cluster, (), cluster_id)]): + yield batch + + 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 50248d279..2fa8a0809 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,25 @@ logger = lg.getLogger() + +async def watch_snapshots(cluster_id, pool_id): + """Stream snapshot changes for one pool (same scope as get_snapshots_by_pool_id).""" + db = DBController() + async for batch in db.watch( + SnapShot, scope=(pool_id,), + select=lambda models: db.get_snapshots_by_pool_id(pool_id, source=models), + ancestors=[(Cluster, (), cluster_id), (Pool, (cluster_id,), pool_id)]): + yield batch + + +async def watch_snapshot(cluster_id, pool_id, snapshot_id): + """Stream changes for a single snapshot.""" + db = DBController() + async for batch in db.watch( + SnapShot, scope=(pool_id,), entity_id=snapshot_id, + ancestors=[(Cluster, (), cluster_id), (Pool, (cluster_id,), pool_id)]): + yield batch + db_controller = DBController() diff --git a/simplyblock_core/controllers/tasks_controller.py b/simplyblock_core/controllers/tasks_controller.py index 2e6d1d268..780fea586 100644 --- a/simplyblock_core/controllers/tasks_controller.py +++ b/simplyblock_core/controllers/tasks_controller.py @@ -14,6 +14,32 @@ logger = logging.getLogger() db = db_controller.DBController() + +async def watch_tasks(cluster_id): + """Stream task changes for one cluster (excludes device-migration tasks, + matching the task list endpoint).""" + async for batch in db.watch( + JobSchedule, scope=(cluster_id,), + 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)]): + yield batch + + +async def watch_task(cluster_id, task_id): + """Stream changes for a single task. + + JobSchedule's compound object key has no uuid-only version key to watch, so + this watches the cluster rollup and filters to the task uuid. + """ + async for batch in db.watch( + JobSchedule, scope=(cluster_id,), + select=lambda models: [task for task in models if task.uuid == task_id], + ancestors=[(Cluster, (), cluster_id)]): + yield batch + # 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 f4ae87734..fd2ca61b0 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,23 @@ def __init__(self): except Exception: logger.exception("FDB initialization failed") + def watch(self, model_cls, *, scope=(), entity_id=None, select=None, ancestors=()): + """Async stream of watch.ChangeEvent batches for a watched scope. + + 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, scope=scope, entity_id=entity_id, 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 +111,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 +129,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 +161,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 +176,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 +289,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 +307,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 +337,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 +686,10 @@ 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): + scope = obj.watch_scope() + tr.add(watches.watch_index_rollup_key(model_cls, scope), watches.ONE_LE64) + tr.add(watches.watch_index_version_key(model_cls, scope, obj.get_id()), watches.ONE_LE64) return obj def atomic_update(self, obj, mutate_fn): @@ -920,6 +935,9 @@ def _try_set_node_restarting_tx(self, tr, cluster_id, node_id, allow_concurrent_ prefix = target.get_db_id() data = json.dumps(target.get_clean_dict(unwrap_secrets=True)) tr[prefix.encode()] = data.encode() + scope = target.watch_scope() + tr.add(watches.watch_index_rollup_key(StorageNode, scope), watches.ONE_LE64) + tr.add(watches.watch_index_version_key(StorageNode, scope, target.get_id()), watches.ONE_LE64) return True, None diff --git a/simplyblock_core/models/base_model.py b/simplyblock_core/models/base_model.py index 4550a2c32..a890cabb1 100644 --- a/simplyblock_core/models/base_model.py +++ b/simplyblock_core/models/base_model.py @@ -9,11 +9,19 @@ from pydantic import SecretBytes, SecretStr +from simplyblock_core import watches + class BaseModel(object): _STATUS_CODE_MAP: dict = {} + # When True, write_to_db()/remove() atomically maintain the watch_index/ + # version index (rollup + per-entity version keyed by watch_scope()) 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 = "" @@ -239,14 +247,43 @@ def get_last(self, kv_store): return objects[0] return None + def watch_scope(self): + """Parent-id path locating this entity in the watch index. + + Watched subclasses override to return their ancestor ids (e.g. + ``(self.pool_uuid,)``); the default empty tuple places the entity + directly under its class (a root, e.g. Cluster). + """ + return () + + @staticmethod + def _write_tx(tr, key, value, rollup_key, version_key): + tr.set(key, value) + tr.add(rollup_key, watches.ONE_LE64) + tr.add(version_key, watches.ONE_LE64) + + @staticmethod + def _remove_tx(tr, key, rollup_key, version_key): + tr.clear(key) + tr.add(rollup_key, watches.ONE_LE64) + tr.clear(version_key) + 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 + scope = self.watch_scope() + fdb.transactional(BaseModel._write_tx)( + kv_store, key, value, + watches.watch_index_rollup_key(type(self), scope), + watches.watch_index_version_key(type(self), scope, self.get_id())) + else: + kv_store.set(key, value) return True except Exception: from simplyblock_core import utils @@ -254,8 +291,15 @@ 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 + scope = self.watch_scope() + return fdb.transactional(BaseModel._remove_tx)( + kv_store, key, + watches.watch_index_rollup_key(type(self), scope), + watches.watch_index_version_key(type(self), scope, self.get_id())) def keys(self): return self.get_attrs_map().keys() diff --git a/simplyblock_core/models/cluster.py b/simplyblock_core/models/cluster.py index 72017e4c6..59620da06 100644 --- a/simplyblock_core/models/cluster.py +++ b/simplyblock_core/models/cluster.py @@ -17,6 +17,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 88476b450..5312ec444 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' @@ -55,6 +57,9 @@ class JobSchedule(BaseModel): # goes stale (see constants.TASK_LEASE_TTL_SEC). See tasks_controller.claim_task. owner: str = "" + def watch_scope(self): + return (self.cluster_id,) + def write_to_db(self, kv_store=None): self.updated_at = str(datetime.datetime.now(datetime.timezone.utc)) super().write_to_db(kv_store) diff --git a/simplyblock_core/models/lvol_model.py b/simplyblock_core/models/lvol_model.py index ea658eb75..182cb2461 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' @@ -79,6 +81,9 @@ class LVol(BaseModel): # replication. 0 disables interval snapshots (only user snaps replicate). replication_interval_min: int = 0 + def watch_scope(self): + return (self.pool_uuid,) + def has_qos(self): return (self.rw_ios_per_sec > 0 or self.rw_mbytes_per_sec > 0 or self.r_mbytes_per_sec > 0 or self.w_mbytes_per_sec > 0) diff --git a/simplyblock_core/models/pool.py b/simplyblock_core/models/pool.py index cf3f19f50..960d8e35c 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" @@ -41,6 +43,9 @@ class Pool(BaseModel): dhchap: bool = False dhchap_key: SecretStr = SecretStr("") dhchap_ctrlr_key: SecretStr = SecretStr("") + + def watch_scope(self): + return (self.cluster_id,) allowed_hosts: List[str] = [] diff --git a/simplyblock_core/models/snapshot.py b/simplyblock_core/models/snapshot.py index 46559e8d0..896d8bdb8 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' @@ -48,6 +50,9 @@ class SnapShot(BaseModel): # This value can be used to identify the same snapshot on other nodes data_uuid: str = "" + def watch_scope(self): + return (self.pool_uuid,) + def write_to_db(self, kv_store=None): super().write_to_db(kv_store) snap_mini = SnapShotMini().from_snapshot(self) diff --git a/simplyblock_core/models/storage_node.py b/simplyblock_core/models/storage_node.py index 7294c7027..c6dc3a8d8 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" @@ -176,6 +178,9 @@ def get_hublvol_port(self, lvs_name=None): return self.hublvol.nvmf_port return 0 + def watch_scope(self): + return (self.cluster_id,) + def client(self, **kwargs): """Return API client to this node """ diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index f717e6406..9e473c48e 100755 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -46,6 +46,26 @@ logger = utils.get_logger(__name__) +async def watch_storage_nodes(cluster_id): + """Stream storage-node changes for one cluster (same scope as + get_storage_nodes_by_cluster_id).""" + db = DBController() + async for batch in db.watch( + StorageNode, scope=(cluster_id,), + select=lambda models: db.get_storage_nodes_by_cluster_id(cluster_id, source=models), + ancestors=[(Cluster, (), cluster_id)]): + yield batch + + +async def watch_storage_node(cluster_id, node_id): + """Stream changes for a single storage node.""" + db = DBController() + async for batch in db.watch( + StorageNode, scope=(cluster_id,), entity_id=node_id, + ancestors=[(Cluster, (), cluster_id)]): + yield batch + + 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..374896636 --- /dev/null +++ b/simplyblock_core/watch.py @@ -0,0 +1,366 @@ +# coding=utf-8 +"""Generic entity-watch primitive for streaming state changes (version index). + +Change signal: every write/remove of a watched model class atomically maintains a +hierarchical version index in the same FDB transaction as the mutation (see +:mod:`simplyblock_core.watches`): a per-scope ``rollup`` counter and a per-entity +``version`` counter, keyed by the model's ``watch_scope()`` path. + +A subscription watches a single FDB key — the scope's rollup key (a *list*) or an +entity's version key (a *detail*). When it fires, one shared :class:`ScopeWatch` +per watch key reads that scope's version subtree, diffs versions against the +previous read to learn exactly which entities changed, point-reads only those, and +fans the resulting scope entity list out to every subscriber. Unrelated scopes do +not wake the watch, and only changed entities are re-read. + +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``). The +``ChangeEvent`` interface is what lets this version-index backend replace the +earlier counter+full-re-read backend without touching controllers or the API. +""" + +import asyncio +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 / insurance against a missed watch: re-scan the scope +# index even without a watch fire (the version-map diff makes a no-op re-scan cheap +# — it publishes nothing when nothing changed). +WATCH_RECONCILE_SEC = 30.0 + +_FDB_RETRY_BACKOFF_SEC = (1.0, 2.0, 4.0) + +_UNSET: Any = object() + +# Filters the shared scope entity list down to a subscription's presentation set. +# 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 watch 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: # watch thread + self._loop.call_soon_threadsafe(self._deliver, state) + + def fail(self) -> None: # watch 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 ScopeWatch: + """Single FDB watch on one key + version-diff read for one watched scope. + + Delivers ``List[model]`` — the current entity set of the scope — to + subscribers, exactly as the old class-wide hub delivered the full class list, + so ``watch()``/``diff_by_id``/``select`` are unchanged. A list watch targets + the scope's rollup key and reads its version subtree; a detail watch + (``entity_id`` given) targets that entity's version key and reads it alone. + Change is detected by diffing the version map, so a reconcile re-scan that + finds nothing changed publishes nothing; only changed entities are re-read. + """ + + def __init__(self, model_cls: Type[Any], scope: Sequence[Any], entity_id: Optional[str]) -> None: + self._model_cls = model_cls + self._scope = tuple(scope) + self._entity_id = entity_id + if entity_id is not None: + self._detail = True + self._watch_key = watches.watch_index_version_key(model_cls, self._scope, entity_id) + else: + self._detail = False + self._watch_key = watches.watch_index_rollup_key(model_cls, self._scope) + self._version_prefix = watches.watch_index_version_prefix(model_cls, self._scope) + self._name = f'{model_cls.__name__}/{"/".join(map(str, self._scope))}/{entity_id or "*"}' + self._lock = threading.Lock() + self._subs: set = set() + self._cache: Any = _UNSET # last delivered List[model] + self._stop = threading.Event() + self._wake = threading.Event() + self._thread: Optional[threading.Thread] = None + # Reader state (owned by the watch thread): + self._prev_versions: Dict[str, int] = {} + self._entities: Dict[str, Any] = {} # leaf (get_id) -> model + self._started = False + + 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._prev_versions = {} + self._entities = {} + self._started = False + 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, entities: List[Any]) -> None: + with self._lock: + self._cache = entities + subs = list(self._subs) + for sub in subs: + sub.deliver(entities) + + def _fail_all(self) -> None: + with self._lock: + subs = list(self._subs) + for sub in subs: + sub.fail() + + def _read(self, kv_store: Any) -> Tuple[List[Any], bool]: + """Return (current scope entity list, changed?) via a version-map diff.""" + cur: Dict[str, int] = {} + if self._detail: + for _k, v in kv_store.get_range_startswith(self._watch_key): + cur[str(self._entity_id)] = watches.unpack_counter(v) + else: + plen = len(self._version_prefix) + for k, v in kv_store.get_range_startswith(self._version_prefix): + leaf = bytes(k)[plen:].decode() + cur[leaf] = watches.unpack_counter(v) + + changed = [leaf for leaf, ver in cur.items() if self._prev_versions.get(leaf) != ver] + removed = [leaf for leaf in self._prev_versions if leaf not in cur] + if self._started and not changed and not removed: + return list(self._entities.values()), False + + for leaf in changed: + # Point-read only the changed entity; leaf is its get_id() (object-key + # suffix), so read_from_db(id=leaf) resolves the exact object key. + rows = self._model_cls().read_from_db(kv_store, id=leaf) + if rows: + self._entities[leaf] = rows[0] + else: + self._entities.pop(leaf, None) # raced away between index and read + for leaf in removed: + self._entities.pop(leaf, None) + self._prev_versions = cur + self._started = True + return list(self._entities.values()), True + + 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: any write not visible to the read has a + # commit version at or after the watch's read version, so it fires. + watch = set_watch(kv_store, self._watch_key) + entities, changed = self._read(kv_store) + failures = 0 + if changed or self._cache is _UNSET: + self._publish(entities) + wake.clear() + # Runs on the FDB network thread: must only set the event. + watch.on_ready(lambda _f: wake.set()) + wake.wait(timeout=WATCH_RECONCILE_SEC) + except Exception: + logger.exception('Watch 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: + logger.debug('Ignoring watch cancellation failure for %s', self._name, exc_info=True) + + +_scope_watches: Dict[bytes, ScopeWatch] = {} +_scope_watches_lock = threading.Lock() + + +def get_scope_watch(model_cls: Type[Any], scope: Sequence[Any] = (), entity_id: Optional[str] = None) -> ScopeWatch: + """Shared :class:`ScopeWatch` for one watch key (rollup for a list, version + key for a detail). Subscribers of the same scope share one FDB watch + read.""" + scope = tuple(scope) + if entity_id is not None: + key = watches.watch_index_version_key(model_cls, scope, entity_id) + else: + key = watches.watch_index_rollup_key(model_cls, scope) + with _scope_watches_lock: + sw = _scope_watches.get(key) + if sw is None: + sw = _scope_watches[key] = ScopeWatch(model_cls, scope, entity_id) + return sw + + +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], + *, + scope: Sequence[Any] = (), + entity_id: Optional[str] = None, + select: Optional[Select] = None, + ancestors: Sequence[Tuple[Type[Any], Sequence[Any], str]] = (), +): + """Async stream of :class:`ChangeEvent` batches for one watched scope. + + ``scope`` is the parent-id path (e.g. ``(pool_id,)`` for a pool's volumes, + ``()`` for clusters); ``entity_id`` narrows to a single entity's version key + (detail). ``select`` filters the shared scope entity list to this + subscription's presentation set (reuse a DBController getter with ``source=`` + so the filter is defined once); the diff runs on its output. ``ancestors`` are + ``(ParentClass, parent_scope, parent_id)`` triples 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_scope_watch(model_cls, scope, entity_id).subscribe(loop, notify) + ancestor_specs = [(pcls, tuple(pscope), str(pid)) for pcls, pscope, pid in ancestors] + ancestor_subs = [ + (get_scope_watch(pcls, pscope, pid).subscribe(loop, notify), pcls, pscope, pid) + for pcls, pscope, pid in ancestor_specs + ] + 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, _, _, _ in ancestor_subs: + ancestor_state = ancestor_sub.take() + if ancestor_state is _UNSET: + continue + if not ancestor_state: # parent's detail watch delivered []: gone + return + + models = main_sub.take() + if models is _UNSET: + continue + events, prev = await loop.run_in_executor( + None, _compute_batch, models, select, prev) + yield events + finally: + get_scope_watch(model_cls, scope, entity_id).unsubscribe(main_sub) + for ancestor_sub, pcls, pscope, pid in ancestor_subs: + get_scope_watch(pcls, pscope, pid).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..a5be8df31 --- /dev/null +++ b/simplyblock_core/watches.py @@ -0,0 +1,72 @@ +# coding=utf-8 +"""Version-index keys for entity watches. + +Every write/remove of a watched model class atomically maintains a hierarchical +version index in FoundationDB, in the same transaction as the entity mutation: + + watch_index// rollup counter (atomic ADD) + watch_index/// per-entity version (ADD; cleared on remove) + +```` is the model's parent-id path (see ``BaseModel.watch_scope``). +Watchers (the web API's SSE layer) set a single FDB watch on the rollup key for +their scope (a list) or on an entity's version key (a detail); when it fires they +read the scope's version subtree, diff versions to learn exactly which entities +changed, and re-read only those. + +The ``watch_index/`` 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_INDEX_PREFIX = 'watch_index/' + +# Operand for FDB's atomic ADD mutation: little-endian 64-bit integer 1. +ONE_LE64 = struct.pack(' str: + """``watch_index/[//...]`` for scope tuple ``scope``.""" + parts = [WATCH_INDEX_PREFIX + model_cls.__name__] + parts.extend(str(s) for s in scope) + return '/'.join(parts) + + +def watch_index_rollup_key(model_cls: type, scope=()) -> bytes: + """Rollup key for a scope, e.g. ``b'watch_index/LVol/'``. + + Bumped on any write/remove of an entity in the scope; the watch target for a + list stream. For a root class with empty scope this is ``watch_index/``. + """ + return _scope_path(model_cls, scope).encode() + + +def watch_index_version_key(model_cls: type, scope, leaf: str) -> bytes: + """Per-entity version key, e.g. ``b'watch_index/LVol//'``. + + ``leaf`` is the entity's ``get_id()`` — its object-key suffix under + ``object//`` (a plain uuid for flat classes, the compound + ``//`` for JobSchedule) — so a watcher can reconstruct + the object key and point-read the entity via ``read_from_db(id=leaf)``. + Bumped on each write of the entity and cleared on remove; the watch target + for a detail stream (flat-keyed classes). + """ + return (_scope_path(model_cls, scope) + '/' + str(leaf)).encode() + + +def watch_index_version_prefix(model_cls: type, scope) -> bytes: + """Prefix for scanning a scope's version keys (rollup key + ``/``). + + Disjoint from the rollup key itself (which has no trailing ``/``), so a + ``get_range_startswith`` on this prefix returns only per-entity versions. + """ + return (_scope_path(model_cls, scope) + '/').encode() + + +def unpack_counter(raw) -> int: + """Decode a counter/version 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 2ae5e2e6d..bc88ae4c7 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 @@ -88,8 +91,15 @@ class ClusterParams(BaseModel): enable_failure_domain: bool = False -@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 @@ -128,8 +138,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 d4446a908..a846a8f40 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 +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 @@ -103,8 +116,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 e60865f6a..186c362f6 100644 --- a/simplyblock_web/api/v2/cluster/storage_node/device.py +++ b/simplyblock_web/api/v2/cluster/storage_node/device.py @@ -1,19 +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, DeviceHealthInfoDTO +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 @@ -27,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..ea57eb562 --- /dev/null +++ b/tests/integration/test_watch_counter_bumps.py @@ -0,0 +1,116 @@ +# coding=utf-8 +"""Every entity write path maintains the version index — 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 _rollup(db, model_cls, scope): + raw = db.kv_store.get(watches.watch_index_rollup_key(model_cls, scope)) + return watches.unpack_counter(bytes(raw)) if raw is not None else 0 + + +def _version_present(db, model_cls, scope, leaf): + return db.kv_store.get(watches.watch_index_version_key(model_cls, scope, leaf)) is not None + + +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_rollup_and_version(): + db = DBController() + pool = _make_pool() + scope = pool.watch_scope() + before = _rollup(db, Pool, scope) + pool.write_to_db(db.kv_store) + assert _rollup(db, Pool, scope) == before + 1 + assert _version_present(db, Pool, scope, pool.get_id()) + assert db.get_pool_by_id(pool.get_id()).get_id() == pool.get_id() + + +def test_remove_bumps_rollup_and_clears_version(): + db = DBController() + pool = _make_pool() + pool.write_to_db(db.kv_store) + scope = pool.watch_scope() + before = _rollup(db, Pool, scope) + pool.remove(db.kv_store) + assert _rollup(db, Pool, scope) == before + 1 + assert not _version_present(db, Pool, scope, pool.get_id()) + + +def test_atomic_update_bumps_index(): + db = DBController() + node = StorageNode({ + 'uuid': str(uuid4()), + 'cluster_id': str(uuid4()), + 'status': StorageNode.STATUS_ONLINE, + }) + node.write_to_db(db.kv_store) + scope = node.watch_scope() + before = _rollup(db, StorageNode, scope) + + def mutate(fresh): + fresh.status = StorageNode.STATUS_SUSPENDED + + updated = db.atomic_update(node, mutate) + assert updated.status == StorageNode.STATUS_SUSPENDED + assert _rollup(db, StorageNode, scope) == before + 1 + assert _version_present(db, StorageNode, scope, node.get_id()) + + +def test_try_set_node_restarting_bumps_index(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) + scope = node.watch_scope() + before = _rollup(db, StorageNode, scope) + + acquired, reason = db.try_set_node_restarting(cluster_id, node.get_id()) + assert acquired, reason + assert _rollup(db, StorageNode, scope) == before + 1 + + +def test_index_keys_invisible_to_entity_reads(): + db = DBController() + pool = _make_pool() + pool.write_to_db(db.kv_store) + # An index 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_index' not in key for key in entity_keys) + + +def test_unwatched_write_creates_no_index(): + db = DBController() + scope = () # LVolStatObject is unwatched; default scope + before = _rollup(db, LVolStatObject, scope) + stat = LVolStatObject({ + 'uuid': str(uuid4()), + 'cluster_id': str(uuid4()), + 'pool_id': str(uuid4()), + 'date': 1, + }) + stat.write_to_db(db.kv_store) + assert _rollup(db, LVolStatObject, scope) == before diff --git a/tests/integration/test_watch_hub.py b/tests/integration/test_watch_hub.py new file mode 100644 index 000000000..eb0dde38b --- /dev/null +++ b/tests/integration/test_watch_hub.py @@ -0,0 +1,97 @@ +# coding=utf-8 +"""ScopeWatch against real FDB: scoped delivery, scoped wakeup, teardown.""" + +import asyncio +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.watch import _UNSET, ScopeWatch + + +def _pool_dict(cluster_id): + return { + 'uuid': str(uuid4()), + 'cluster_id': cluster_id, + '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_scopewatch_delivers_scoped_writes_and_stops_on_unsubscribe(): + db = DBController() + cluster_id = str(uuid4()) + + async def scenario(): + loop = asyncio.get_running_loop() + notify = asyncio.Event() + sw = ScopeWatch(Pool, (cluster_id,), None) + sub = sw.subscribe(loop, notify) + try: + # Initial publication for this (empty) scope. + assert await _await_publication(sub, notify) == [] + + pool = Pool(_pool_dict(cluster_id)) + 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: + sw.unsubscribe(sub) + + thread = sw._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_scopewatch_ignores_other_scope(): + """A write in a different cluster's scope must not wake this watch.""" + db = DBController() + cluster_a = str(uuid4()) + cluster_b = str(uuid4()) + + async def scenario(): + loop = asyncio.get_running_loop() + notify = asyncio.Event() + sw = ScopeWatch(Pool, (cluster_a,), None) + sub = sw.subscribe(loop, notify) + try: + assert await _await_publication(sub, notify) == [] + + other = Pool(_pool_dict(cluster_b)) + await asyncio.to_thread(other.write_to_db, db.kv_store) + + # cluster_a's rollup key never bumped -> no publication (< reconcile). + with pytest.raises(asyncio.TimeoutError): + await _await_publication(sub, notify, timeout=3.0) + finally: + sw.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..ef2fe65ae --- /dev/null +++ b/tests/integration/web/test_watch_sse_e2e.py @@ -0,0 +1,216 @@ +# 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_scope_watch + + +@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' + watch = get_scope_watch(LVol, (pool.get_id(),)) + assert watch.subscriber_count() > 0 + + response.close() + + deadline = time.monotonic() + 20 + while watch.subscriber_count() > 0 and time.monotonic() < deadline: + time.sleep(0.2) + assert watch.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..2f25a7c87 --- /dev/null +++ b/tests/unit/test_watch_core.py @@ -0,0 +1,275 @@ +# coding=utf-8 +"""Core watch primitive: diff_by_id, ScopeWatch index-diff, and watch().""" + +import asyncio +import json +import struct + +import pytest + +from simplyblock_core import watch as watchmod +from simplyblock_core import watches +from simplyblock_core.models.pool import Pool +from simplyblock_core.watch import ( + ScopeWatch, 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(): + 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'] + + +# ---- ScopeWatch._read (version-map diff + point-read), no FDB ---- + +class FakeKV: + def __init__(self): + self.data = {} + + def get_range_startswith(self, prefix, limit=0, reverse=False): + prefix = bytes(prefix) + items = [(k, v) for k, v in self.data.items() if k.startswith(prefix)] + items.sort(key=lambda kv: kv[0], reverse=reverse) + return items[:limit] if limit else items + + +def _write_obj(kv, model): + kv.data[model.get_db_id().encode()] = json.dumps(model.to_dict(unwrap_secrets=True)).encode() + + +def _set_version(kv, cls, scope, leaf, n): + kv.data[watches.watch_index_version_key(cls, scope, leaf)] = struct.pack(' no re-publish + + pool.status = Pool.STATUS_INACTIVE + _write_obj(kv, pool) + _set_version(kv, Pool, ('cl-1',), 'p1', 2) # a write bumps the version + entities, changed = sw._read(kv) + assert changed and entities[0].status == Pool.STATUS_INACTIVE # re-read the changed entity + + _clear_version(kv, Pool, ('cl-1',), 'p1') # remove clears the version key + entities, changed = sw._read(kv) + assert changed and entities == [] + + +def test_scopewatch_scoped_to_its_subtree(): + kv = FakeKV() + other = Pool({'uuid': 'p2', 'cluster_id': 'cl-2', 'pool_name': 'y'}) + _write_obj(kv, other) + _set_version(kv, Pool, ('cl-2',), 'p2', 1) # a pool in a different cluster + sw = ScopeWatch(Pool, ('cl-1',), None) + entities, _ = sw._read(kv) + assert entities == [] # out-of-scope entity is invisible + + +def test_scopewatch_detail_read(): + kv = FakeKV() + pool = Pool({'uuid': 'p1', 'cluster_id': 'cl-1', 'pool_name': 'x'}) + sw = ScopeWatch(Pool, ('cl-1',), 'p1') + + assert sw._read(kv) == ([], True) # not present yet + _write_obj(kv, pool) + _set_version(kv, Pool, ('cl-1',), 'p1', 1) + entities, changed = sw._read(kv) + assert changed and [p.get_id() for p in entities] == ['p1'] + _clear_version(kv, Pool, ('cl-1',), 'p1') + entities, changed = sw._read(kv) + assert changed and entities == [] + + +# ---- watch() generator with a fake scope watch ---- + +class FakeScopeWatch: + 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, watches_map): + monkeypatch.setattr( + watchmod, 'get_scope_watch', + lambda cls, scope=(), entity_id=None: watches_map[(cls, tuple(scope), entity_id)]) + + +async def _next(gen, timeout=2.0): + return await asyncio.wait_for(gen.__anext__(), timeout) + + +def test_watch_snapshot_update_delete(monkeypatch): + sw = FakeScopeWatch() + _patch(monkeypatch, {(Thing, (), None): sw}) + sw.publish([_m('a', status='online')]) + + async def scenario(): + gen = watch(Thing) + [event] = await _next(gen) + assert event.kind == 'created' and event.id == 'a' + + sw.publish([_m('a', status='offline')]) + [event] = await _next(gen) + assert event.kind == 'updated' and event.model.to_dict()['status'] == 'offline' + + sw.publish([]) + [event] = await _next(gen) + assert event.kind == 'deleted' and event.id == 'a' + + await gen.aclose() + assert sw.subs == [] + + asyncio.run(scenario()) + + +def test_watch_coalesces_bursts(monkeypatch): + sw = FakeScopeWatch() + _patch(monkeypatch, {(Thing, (), None): sw}) + sw.publish([_m('a', status='s0')]) + + async def scenario(): + gen = watch(Thing) + assert (await _next(gen))[0].kind == 'created' + + sw.publish([_m('a', status='s1')]) + sw.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): + sw = FakeScopeWatch() + _patch(monkeypatch, {(Thing, (), None): sw}) + sw.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): + sw, parent = FakeScopeWatch(), FakeScopeWatch() + _patch(monkeypatch, {(Thing, (), None): sw, (Parent, (), 'p'): parent}) + sw.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([]) # parent gone + with pytest.raises(StopAsyncIteration): + await _next(gen) + assert sw.subs == [] and parent.subs == [] + + asyncio.run(scenario()) + + +def test_watch_failure_raises(monkeypatch): + sw = FakeScopeWatch() + _patch(monkeypatch, {(Thing, (), None): sw}) + sw.publish([_m('a')]) + + async def scenario(): + gen = watch(Thing) + await _next(gen) + sw.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..def2858d6 --- /dev/null +++ b/tests/unit/test_watch_counters.py @@ -0,0 +1,139 @@ +# coding=utf-8 +"""Version-index keys and the transactional write+index maintenance 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_index_key_format(): + assert watches.watch_index_rollup_key(Cluster, ()) == b'watch_index/Cluster' + assert watches.watch_index_rollup_key(Pool, ('cl-1',)) == b'watch_index/Pool/cl-1' + assert watches.watch_index_version_key(LVol, ('p1',), 'v1') == b'watch_index/LVol/p1/v1' + assert watches.watch_index_version_prefix(LVol, ('p1',)) == b'watch_index/LVol/p1/' + + +def test_version_prefix_excludes_rollup_key(): + """Scanning the version prefix must not pick up the rollup key itself.""" + rollup = watches.watch_index_rollup_key(LVol, ('p1',)) + prefix = watches.watch_index_version_prefix(LVol, ('p1',)) + assert not rollup.startswith(prefix) + assert prefix.startswith(rollup) + + +def test_scope_derivation_per_class(): + assert Cluster({'uuid': 'c'}).watch_scope() == () + assert Pool({'uuid': 'p', 'cluster_id': 'c'}).watch_scope() == ('c',) + assert StorageNode({'uuid': 'n', 'cluster_id': 'c'}).watch_scope() == ('c',) + assert LVol({'uuid': 'l', 'pool_uuid': 'p'}).watch_scope() == ('p',) + assert SnapShot({'uuid': 's', 'pool_uuid': 'p'}).watch_scope() == ('p',) + assert JobSchedule({'uuid': 't', 'cluster_id': 'c', 'date': 1}).watch_scope() == ('c',) + + +@pytest.mark.parametrize('model_cls', WATCHED_CLASSES) +def test_index_keys_disjoint_from_entity_scans(model_cls): + """read_from_db scans '//' prefixes; index keys must + never be picked up by any such scan.""" + keys = [ + watches.watch_index_rollup_key(model_cls, ('x',)), + watches.watch_index_version_key(model_cls, ('x',), 'y'), + ] + for other_cls in WATCHED_CLASSES + UNWATCHED_CLASSES: + instance = other_cls() + prefix = f'{instance.object_type}/{instance.name}/'.encode() + for key in keys: + assert not 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_rollup_and_version(): + kv = MagicMock() + pool = Pool({'uuid': 'pool-1', 'cluster_id': 'cl-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' + assert kv.add.call_count == 2 + kv.add.assert_any_call(b'watch_index/Pool/cl-1', watches.ONE_LE64) + kv.add.assert_any_call(b'watch_index/Pool/cl-1/pool-1', watches.ONE_LE64) + + +def test_unwatched_write_does_not_index(): + 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_rollup_and_clears_version(): + kv = MagicMock() + pool = Pool({'uuid': 'pool-1', 'cluster_id': 'cl-1'}) + pool.remove(kv) + assert kv.clear.call_count == 2 + kv.clear.assert_any_call(b'object/Pool/pool-1') + kv.clear.assert_any_call(b'watch_index/Pool/cl-1/pool-1') + kv.add.assert_called_once_with(b'watch_index/Pool/cl-1', watches.ONE_LE64) + + +def test_unwatched_remove_does_not_index(): + 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_indexes_by_get_id(): + """JobSchedule's compound get_id() is the version leaf so a watcher can + reconstruct the object key and point-read the task.""" + 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_any_call(b'watch_index/JobSchedule/cl-1', watches.ONE_LE64) + kv.add.assert_any_call(b'watch_index/JobSchedule/cl-1/cl-1/42/task-1', watches.ONE_LE64) + + +def test_base_model_default_unwatched(): + assert BaseModel._WATCHED is False + + +def test_base_model_default_scope_empty(): + assert BaseModel().watch_scope() == () + + +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']) == '**********'