"""A live, fixed-in-place terminal dashboard for monitoring a large fleet against a target deployment state — the kind of thing you'd run on a screen in an ops room while rolling out or watching a large deployment with many nodes and many app versions in the field. You define one or more target rules in a JSON config, each selecting a set of nodes (by Barbara ID, tag, or group — or several at once) and the state those nodes are expected to be in: { "refresh_seconds": 10, "targets": [ { "barbara_ids": ["line-3-plc-01", "line-3-plc-02"], "state": { "online": true, "barbara_core_updated": true, "apps": [{"name": "Alert Manager", "version": "1.2.4"}] } }, {"tags": ["production"], "state": {"online": true}}, {"group": "Line 3", "state": {"barbara_core_updated": true}} ] } A node matched by more than one rule gets the union of their checks (a later rule's ``online``/``barbara_core_updated`` overrides an earlier one for that node; ``apps`` lists accumulate). Every rule needs at least one selector (``barbara_ids``/``tags``/``group``) and at least one state check. Every ``refresh_seconds`` (default 10), the dashboard redraws in place (clears and repaints the screen, so it never scrolls) showing, per node, which checks currently pass or fail. A rolling alert log underneath the table persists across redraws and gains a line every time a check flips from passing to failing (or back). State checked, per node, straight from the API: - **online**: ``alive`` on the node's own document. - **barbara_core_updated**: whether ``deviceVersion.lastVersion. updateAvailable`` is false. - **apps**: for each expected ``{"name", "version"}``, whether a workload with that application name is deployed on the node at that exact version — read straight from the same node document as the two checks above (it already embeds every deployed app's name and installed version), so watching any number of apps on a node adds no extra calls. Which nodes match which rule (by tag/group) is resolved once at startup, not every cycle — a node added to a monitored group/tag afterwards won't be picked up without restarting. Combined with the point above, this keeps the steady-state API call cost per cycle at exactly one call per monitored node, regardless of how many checks or apps are being watched on it or how large the fleet is. Usage: python deployment_dashboard.py deployment_dashboard.sample.json Credentials are read from environment variables (BBR_API_CLIENT_ID / BBR_API_CLIENT_SECRET / BBR_API_USERNAME / BBR_API_PASSWORD), optionally loaded from a .env file first — see load_dotenv() below and README.md. """ from __future__ import annotations import argparse import json import os import re import sys import time from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any, Dict, List, Optional from barbara import BarbaraApiError, BarbaraAuthError, BarbaraClient from barbara.models import Node INTERNAL_ID_RE = re.compile(r"^[0-9a-f]{24}$") # -- ANSI styling (no extra dependency) -------------------------------------- RESET = "\033[0m" BOLD = "\033[1m" DIM = "\033[2m" RED = "\033[31m" GREEN = "\033[32m" YELLOW = "\033[33m" CYAN = "\033[36m" CLEAR_AND_HOME = "\033[2J\033[H" def paint(text: str, *styles: str) -> str: return "".join(styles) + text + RESET def load_dotenv(path: str) -> None: if not os.path.isfile(path): return with open(path, encoding="utf-8") as fh: for line in fh: line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, _, value = line.partition("=") os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'")) def resolve_node(client: BarbaraClient, identifier: str) -> Node: if INTERNAL_ID_RE.match(identifier): return client.nodes.get(identifier) return client.nodes.resolve(identifier) def _oid(value: Any) -> Any: """A node's ``spaces`` entries come back as plain id strings from ``list()``/``resolve()`` but as embedded objects (with the id under ``_id``) from ``get()`` — normalize to the raw id either way. """ return value.get("_id") if isinstance(value, dict) else value def resolve_group_node_ids(client: BarbaraClient, identifier: str) -> List[str]: if INTERNAL_ID_RE.match(identifier): return client.groups.get(identifier).node_ids for group in client.groups.list(): if group.name == identifier: return group.node_ids raise SystemExit(f"No group found matching '{identifier}'") # -- Target state ------------------------------------------------------------- @dataclass class TargetState: online: Optional[bool] = None barbara_core_updated: Optional[bool] = None apps: Dict[str, str] = field(default_factory=dict) # app name -> expected version def merge(self, other: TargetState) -> TargetState: return TargetState( online=other.online if other.online is not None else self.online, barbara_core_updated=( other.barbara_core_updated if other.barbara_core_updated is not None else self.barbara_core_updated ), apps={**self.apps, **other.apps}, ) def _parse_state(raw: Dict[str, Any]) -> TargetState: apps = {app["name"]: app["version"] for app in raw.get("apps") or []} state = TargetState(online=raw.get("online"), barbara_core_updated=raw.get("barbara_core_updated"), apps=apps) if state.online is None and state.barbara_core_updated is None and not state.apps: raise SystemExit("Each target's \"state\" needs at least one of online/barbara_core_updated/apps.") return state def resolve_target_states(client: BarbaraClient, config: Dict[str, Any]) -> Dict[str, TargetState]: """Resolves every rule's selector once, and merges per-node target states for nodes matched by more than one rule. """ states: Dict[str, TargetState] = {} for rule in config.get("targets") or []: selectors = [k for k in ("barbara_ids", "tags", "group") if rule.get(k)] if not selectors: raise SystemExit("Each target needs at least one of barbara_ids/tags/group.") state = _parse_state(rule.get("state") or {}) node_ids: List[str] = [] for barbara_id in rule.get("barbara_ids") or []: node_ids.append(resolve_node(client, barbara_id).id) if rule.get("tags"): node_ids.extend(n.id for n in client.nodes.list_by_tags(rule["tags"])) if rule.get("group"): node_ids.extend(resolve_group_node_ids(client, rule["group"])) for node_id in node_ids: states[node_id] = states[node_id].merge(state) if node_id in states else state if not states: raise SystemExit("No nodes matched by any target rule.") return states # -- Evaluation ----------------------------------------------------------- def _workload_app_name(workload_raw: Dict[str, Any]) -> Optional[str]: current = workload_raw.get("current") or {} for key in ("marketApplicationId", "applicationId", "modelApplicationId"): ref = current.get(key) if isinstance(ref, dict): return ref.get("name") return None def _installed_apps(node: Node) -> Dict[str, Optional[str]]: """Maps every app name deployed on the node to its installed version — read directly from the space objects ``client.nodes.get()`` already embeds (name, current version, everything), so checking any number of apps on a node costs nothing beyond the one call already made for the online/Barbara Core checks below — no per-workload call needed. """ installed: Dict[str, Optional[str]] = {} for space in node.raw.get("spaces") or []: if not isinstance(space, dict): continue name = _workload_app_name(space) if name: installed[name] = (space.get("current") or {}).get("name") return installed @dataclass class CheckResult: ok: bool detail: str def evaluate_node(client: BarbaraClient, node_id: str, state: TargetState) -> Dict[str, CheckResult]: """One API call total, regardless of how many checks or apps this node has — ``client.nodes.get()`` already carries everything each check needs (``alive``, ``deviceVersion``, and every deployed app's name/version). """ results: Dict[str, CheckResult] = {} node = client.nodes.get(node_id) if state.online is not None: alive = bool(node.raw.get("alive")) results["online"] = CheckResult(alive == state.online, "online" if alive else "offline") if state.barbara_core_updated is not None: last = (node.raw.get("deviceVersion") or {}).get("lastVersion") or {} update_available = bool(last.get("updateAvailable")) up_to_date = not update_available results["barbara_core_updated"] = CheckResult( up_to_date == state.barbara_core_updated, "up to date" if up_to_date else "update available", ) if state.apps: installed = _installed_apps(node) for app_name, expected_version in state.apps.items(): check_name = f"app:{app_name}" if app_name not in installed: results[check_name] = CheckResult(False, "not deployed") continue installed_version = installed[app_name] if installed_version == expected_version: results[check_name] = CheckResult(True, installed_version or "") else: results[check_name] = CheckResult( False, f"version {installed_version!r} (expected {expected_version!r})" ) return results # -- Dashboard rendering ---------------------------------------------------- def render( nodes: Dict[str, Node], results: Dict[str, Dict[str, CheckResult]], alerts: List[str], refresh_seconds: int, ) -> None: now = datetime.now(timezone.utc).strftime("%H:%M:%S UTC") lines = [ paint(" Barbara Deployment Dashboard ", BOLD, CYAN), paint(f" refresh: {refresh_seconds}s last update: {now} Ctrl+C to stop", DIM), "", ] for node_id, checks in results.items(): node = nodes[node_id] node_ok = all(c.ok for c in checks.values()) status = paint("OK", BOLD, GREEN) if node_ok else paint("FAIL", BOLD, RED) lines.append(f"{paint(node.node_name, BOLD):30s} [{status}]") for check_name, result in checks.items(): mark = paint("✓", GREEN) if result.ok else paint("✗", RED) lines.append(f" {mark} {check_name:20s} {result.detail}") lines.append("") lines.append(paint(" Alerts ", BOLD, YELLOW)) if alerts: lines.extend(f" {a}" for a in alerts[-15:]) else: lines.append(paint(" (none yet)", DIM)) sys.stdout.write(CLEAR_AND_HOME + "\n".join(lines) + "\n") sys.stdout.flush() def run_dashboard(client: BarbaraClient, config: Dict[str, Any]) -> None: refresh_seconds = int(config.get("refresh_seconds", 10)) target_states = resolve_target_states(client, config) nodes = {node_id: client.nodes.get(node_id) for node_id in target_states} previous: Dict[str, Dict[str, bool]] = {} alerts: List[str] = [] try: while True: results = { node_id: evaluate_node(client, node_id, target_states[node_id]) for node_id in target_states } timestamp = datetime.now(timezone.utc).strftime("%H:%M:%S") for node_id, checks in results.items(): node_name = nodes[node_id].node_name prior = previous.get(node_id, {}) for check_name, result in checks.items(): was_ok = prior.get(check_name) if was_ok is True and not result.ok: alerts.append( paint(f"[{timestamp}] ALERT", BOLD, RED) + f" {node_name}: {check_name} started failing — {result.detail}" ) elif was_ok is False and result.ok: alerts.append( paint(f"[{timestamp}] RECOVERED", BOLD, GREEN) + f" {node_name}: {check_name}" ) previous[node_id] = {name: r.ok for name, r in checks.items()} render(nodes, results, alerts, refresh_seconds) time.sleep(refresh_seconds) except KeyboardInterrupt: print("\nStopped.") def main() -> None: parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("config", help="Path to a target-state JSON config (see module docstring)") parser.add_argument( "--env-file", default=".env", help="Path to a .env file with Barbara API Credentials" ) args = parser.parse_args() load_dotenv(args.env_file) with open(args.config, encoding="utf-8") as fh: config = json.load(fh) with BarbaraClient.from_env() as client: try: client.api_version() except BarbaraAuthError as exc: raise SystemExit(f"Login failed — check your Barbara API Credentials: {exc}") try: run_dashboard(client, config) except BarbaraApiError as exc: raise SystemExit(f"API error: {exc}") if __name__ == "__main__": main()