diff --git a/.gitignore b/.gitignore index 2d7ec5c..910d1b7 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,5 @@ .env node_modules/ +__pycache__/ +*.pyc +src/operator-dashboard/dist/ diff --git a/README.md b/README.md index 13b4a50..373b694 100644 --- a/README.md +++ b/README.md @@ -12,6 +12,7 @@ This repository contains the unrip trading-system code and its project-specific - `deploy/k8s/base/` — project-specific Kubernetes manifests - `deploy/redpanda/rpk-topics.txt` — project topic reference - `docs/` — project-specific design and contract docs +- `docs/system-architecture.md` — high-level architecture, data flow, persistence map, and Mermaid diagrams ## Local development @@ -118,6 +119,80 @@ KUBECONFIG=../unrip3/.state/hetzner/kubeconfig.yaml kubectl -n unrip rollout sta KUBECONFIG=../unrip3/.state/hetzner/kubeconfig.yaml kubectl -n unrip rollout status deploy/trade-executor ``` +### Operator dashboard + +The operator dashboard runs as the `operator-dashboard` service inside the +`unrip` namespace and is exposed through the cluster ingress: + +```text +https://doran.133011.xyz/ +``` + +Local port-forward access is still available for development or incident +debugging. + +Start local access: + +```bash +KUBECONFIG=../unrip3/.state/hetzner/kubeconfig.yaml \ + kubectl -n unrip port-forward --address 0.0.0.0 svc/operator-dashboard 8090:8090 +``` + +Then open: + +```text +http://:8090/ +``` + +The dashboard uses HTTP basic auth in production. + +- Username: `admin` +- Password: `pass unrip/dashboard` + +You can print the current password locally with: + +```bash +pass unrip/dashboard +``` + +### Operator dashboard frontend dev server + +For faster UI iteration against the live cluster backend, use the local React +dev server with hot reload: + +```bash +npm install +npm run operator-dashboard:dev +``` + +That command: +- reads the dashboard password from `pass unrip/dashboard` +- starts a local port-forward to the deployed `svc/operator-dashboard` +- serves the local React dashboard frontend with Vite live reload +- proxies `/api` and `/ws` to the deployed cluster backend with auth attached + +Then open: + +```text +http://:5173/ +``` + +Useful overrides: + +```bash +OPERATOR_DASHBOARD_DEV_PORT=5174 npm run operator-dashboard:dev +OPERATOR_DASHBOARD_DEV_HOST=127.0.0.1 npm run operator-dashboard:dev +OPERATOR_DASHBOARD_DEV_UPSTREAM_PORT=18091 npm run operator-dashboard:dev +OPERATOR_DASHBOARD_DEV_BASIC_AUTH_PASSWORD='...' npm run operator-dashboard:dev +PLATFORM_REPO_DIR=/path/to/unrip3 npm run operator-dashboard:dev +``` + +For the plain dashboard forward, you can also use the repo helper: + +```bash +npm run operator-dashboard:forward +``` + ### Auxiliary ops scripts These scripts default to the same adjacent platform checkout as the deployment diff --git a/scripts/ops/deployment_status.py b/scripts/ops/deployment_status.py index 49db22d..e3f127f 100755 --- a/scripts/ops/deployment_status.py +++ b/scripts/ops/deployment_status.py @@ -2,6 +2,7 @@ from __future__ import annotations import argparse +import json import sys from pathlib import Path @@ -39,6 +40,12 @@ def parse_args() -> argparse.Namespace: action="store_true", help="Include completed Job pods in the pod list.", ) + parser.add_argument( + "--output", + choices=("json", "table"), + default="json", + help="Output format (default: json).", + ) return parser.parse_args() @@ -96,20 +103,24 @@ def main() -> int: for item in pvc_payload.get("items", []) } - deployment_rows: list[list[str]] = [] + deployments: list[dict[str, object]] = [] for item in sorted(deployment_payload.get("items", []), key=lambda value: value["metadata"]["name"]): container = item.get("spec", {}).get("template", {}).get("spec", {}).get("containers", [{}])[0] - deployment_rows.append( - [ - item["metadata"]["name"], - f"{item.get('status', {}).get('readyReplicas', 0)}/{item.get('spec', {}).get('replicas', 0)}", - str(item.get("status", {}).get("availableReplicas", 0)), - container.get("image", "-"), - ] + deployments.append( + { + "name": item["metadata"]["name"], + "ready": { + "ready_replicas": item.get("status", {}).get("readyReplicas", 0), + "desired_replicas": item.get("spec", {}).get("replicas", 0), + "display": f"{item.get('status', {}).get('readyReplicas', 0)}/{item.get('spec', {}).get('replicas', 0)}", + }, + "available_replicas": item.get("status", {}).get("availableReplicas", 0), + "image": container.get("image", "-"), + } ) - pod_rows: list[list[str]] = [] - storage_rows: list[list[str]] = [] + pods: list[dict[str, object]] = [] + storage: list[dict[str, object]] = [] for pod in sorted(pod_payload.get("items", []), key=lambda value: value["metadata"]["name"]): if not args.include_completed and is_completed_job_pod(pod): @@ -125,16 +136,16 @@ def main() -> int: ) uptime = human_duration((now_utc() - start_time).total_seconds() if start_time else None) image = pod.get("spec", {}).get("containers", [{}])[0].get("image", "-") - pod_rows.append( - [ - pod_name, - pod.get("status", {}).get("phase", "-"), - ready_string(pod), - str(restart_count(pod)), - uptime, - pod.get("spec", {}).get("nodeName", "-"), - image, - ] + pods.append( + { + "name": pod_name, + "phase": pod.get("status", {}).get("phase", "-"), + "ready": ready_string(pod), + "restarts": restart_count(pod), + "uptime": uptime, + "node": pod.get("spec", {}).get("nodeName", "-"), + "image": image, + } ) mounts = pvc_mounts_for_pod(pod) @@ -142,26 +153,81 @@ def main() -> int: mounts = [{"claim_name": "-", "mount_path": "/", "container": "app"}] if not mounts: - storage_rows.append([pod_name, "-", "no pvc mounts", "-", "-", "-", "-", "-"]) + storage.append( + { + "pod": pod_name, + "path": None, + "claim": None, + "requested_bytes": None, + "requested": None, + "path_bytes": None, + "fs_used_bytes": None, + "fs_available_bytes": None, + "use_percent": None, + "note": "no pvc mounts", + } + ) continue for mount in mounts: usage = probe_path_usage(pod_name, mount["mount_path"], namespace=namespace) requested_raw = pvc_requests.get(mount["claim_name"], "") requested_bytes = parse_storage_quantity(requested_raw) - storage_rows.append( - [ - pod_name, - mount["mount_path"], - mount["claim_name"], - human_bytes(requested_bytes) if requested_bytes is not None else (requested_raw or "-"), - human_bytes(usage["path_bytes"]), - human_bytes(usage["filesystem_used_bytes"]), - human_bytes(usage["filesystem_available_bytes"]), - str(usage["filesystem_use_percent"]), - ] + storage.append( + { + "pod": pod_name, + "path": mount["mount_path"], + "claim": mount["claim_name"], + "requested_bytes": requested_bytes, + "requested": requested_raw or None, + "path_bytes": usage["path_bytes"], + "fs_used_bytes": usage["filesystem_used_bytes"], + "fs_available_bytes": usage["filesystem_available_bytes"], + "use_percent": usage["filesystem_use_percent"], + "note": None, + } ) + if args.output == "json": + print( + json.dumps( + { + "namespace": namespace, + "deployments": deployments, + "pods": pods, + "storage": storage, + "notes": [ + "storage figures come from container-mounted paths", + "pods without pvc-backed mounts do not report dedicated storage", + ], + }, + indent=2, + ) + ) + return 0 + + deployment_rows = [ + [item["name"], item["ready"]["display"], str(item["available_replicas"]), item["image"]] + for item in deployments + ] + pod_rows = [ + [item["name"], item["phase"], item["ready"], str(item["restarts"]), item["uptime"], item["node"], item["image"]] + for item in pods + ] + storage_rows = [ + [ + item["pod"], + item["path"] or "-", + item["claim"] or "no pvc mounts", + human_bytes(item["requested_bytes"]) if item["requested_bytes"] is not None else (item["requested"] or "-"), + human_bytes(item["path_bytes"]), + human_bytes(item["fs_used_bytes"]), + human_bytes(item["fs_available_bytes"]), + str(item["use_percent"] or "-"), + ] + for item in storage + ] + print(f"Namespace: {namespace}") print() print("Deployments") diff --git a/scripts/ops/live_near_intents.py b/scripts/ops/live_near_intents.py index 632ecf5..e1d1e31 100755 --- a/scripts/ops/live_near_intents.py +++ b/scripts/ops/live_near_intents.py @@ -2,6 +2,7 @@ from __future__ import annotations import argparse +import json import signal import subprocess import sys @@ -9,7 +10,7 @@ from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) -from common import DEFAULT_NAMESPACE, build_env, kafka_brokers, raw_near_topic, stderr +from common import DEFAULT_NAMESPACE, build_env, kafka_brokers, raw_near_topic DEFAULT_OUTPUT_FORMAT = ( @@ -37,16 +38,32 @@ def parse_args() -> argparse.Namespace: default="", help="Topic to consume. Defaults to KAFKA_TOPIC_RAW_NEAR_INTENTS_QUOTE from unrip-config.", ) - parser.add_argument( + offset_group = parser.add_mutually_exclusive_group() + offset_group.add_argument( "--offset", default="end", - help="Kafka offset selector passed to rpk (default: end)", + help=( + "Kafka offset selector passed to rpk. " + "The default 'end' means live-tail from now, not replay history." + ), + ) + offset_group.add_argument( + "--last", + type=int, + default=0, + help=( + "Read the most recent N retained records and exit. " + "Equivalent to '--offset -N --num N'." + ), ) parser.add_argument( "--num", type=int, default=0, - help="Number of records to read before exiting. 0 means unbounded (default: 0).", + help=( + "Number of records to consume before exiting, starting from the chosen " + "offset. 0 means unbounded. With '--offset end' this waits for new records." + ), ) parser.add_argument( "--timeout", @@ -64,10 +81,22 @@ def parse_args() -> argparse.Namespace: action="store_true", help="Use rpk's built-in JSON record formatter instead of the default metadata+value output.", ) + parser.add_argument( + "--output", + choices=("json", "jsonl", "text"), + default="json", + help="Output format (default: json).", + ) return parser.parse_args() def build_command(args: argparse.Namespace, brokers: str, topic: str) -> list[str]: + offset = args.offset + num = args.num + if args.last: + offset = f"-{args.last}" + num = args.last + command = [ "kubectl", "exec", @@ -83,12 +112,14 @@ def build_command(args: argparse.Namespace, brokers: str, topic: str) -> list[st "--brokers", brokers, "--offset", - args.offset, + offset, "--num", - str(args.num), + str(num), ] - if args.rpk_json: + if args.output in {"json", "jsonl"} or args.rpk_json: + command.extend(["--format", "json", "--pretty-print=false"]) + elif args.rpk_json: command.extend(["--format", "json", "--pretty-print=false"]) elif args.value_only: command.extend(["--format", "%v\n"]) @@ -103,24 +134,78 @@ def main() -> int: brokers = args.brokers or kafka_brokers(namespace=args.namespace) topic = args.topic or raw_near_topic(namespace=args.namespace) command = build_command(args, brokers, topic) + effective_offset = args.offset if not args.last else f"-{args.last}" + effective_num = args.num if not args.last else args.last - stderr(f"namespace={args.namespace} brokers={brokers} topic={topic} offset={args.offset} num={args.num}") - - process = subprocess.Popen(command, env=build_env()) - try: - process.wait(timeout=args.timeout if args.timeout and args.timeout > 0 else None) - except subprocess.TimeoutExpired: - process.send_signal(signal.SIGINT) + if args.output == "text": + process = subprocess.Popen(command, env=build_env()) try: - process.wait(timeout=5) + process.wait(timeout=args.timeout if args.timeout and args.timeout > 0 else None) except subprocess.TimeoutExpired: - process.kill() - process.wait(timeout=5) - return 124 - except KeyboardInterrupt: - process.send_signal(signal.SIGINT) - return process.wait() - return process.returncode + process.send_signal(signal.SIGINT) + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=5) + return 124 + except KeyboardInterrupt: + process.send_signal(signal.SIGINT) + return process.wait() + return process.returncode + + try: + result = subprocess.run( + command, + env=build_env(), + text=True, + capture_output=True, + timeout=args.timeout if args.timeout and args.timeout > 0 else None, + check=False, + ) + stdout = result.stdout + return_code = result.returncode + except subprocess.TimeoutExpired as exc: + stdout = exc.stdout or "" + return_code = 124 + + records = parse_rpk_json_lines(stdout, value_only=args.value_only) + if args.output == "json": + print(json.dumps(records, indent=2)) + else: + for record in records: + print(json.dumps(record, separators=(",", ":"))) + return return_code + + +def parse_rpk_json_lines(stdout: str, *, value_only: bool) -> list[object]: + records: list[object] = [] + for raw_line in stdout.splitlines(): + line = raw_line.strip() + if not line: + continue + record = json.loads(line) + if value_only: + record = decode_jsonish(record.get("value")) + else: + if "value" in record: + record["value"] = decode_jsonish(record.get("value")) + records.append(record) + return records + + +def decode_jsonish(value: object) -> object: + if not isinstance(value, str): + return value + stripped = value.strip() + if not stripped: + return value + if stripped[0] not in "{[": + return value + try: + return json.loads(stripped) + except json.JSONDecodeError: + return value if __name__ == "__main__": diff --git a/scripts/ops/redpanda_storage.py b/scripts/ops/redpanda_storage.py index f236884..f976b97 100755 --- a/scripts/ops/redpanda_storage.py +++ b/scripts/ops/redpanda_storage.py @@ -2,6 +2,7 @@ from __future__ import annotations import argparse +import json import re import sys from pathlib import Path @@ -46,6 +47,12 @@ def parse_args() -> argparse.Namespace: action="store_true", help="Inspect every topic visible to Redpanda instead of the app topics from config.", ) + parser.add_argument( + "--output", + choices=("json", "table"), + default="json", + help="Output format (default: json).", + ) return parser.parse_args() @@ -133,7 +140,7 @@ def main() -> int: if not topics: raise SystemExit("no topics found") - topic_rows: list[list[str]] = [] + topic_data: list[dict[str, object]] = [] total_local_bytes = 0 total_bytes = 0 total_segments = 0 @@ -152,20 +159,58 @@ def main() -> int: total_local_bytes += int(parsed["local_bytes"]) total_bytes += int(parsed["total_bytes"]) total_segments += int(parsed["local_segments"]) - topic_rows.append( - [ - topic, - str(parsed["partitions"]), - str(parsed["replicas"]), - human_bytes(int(parsed["local_bytes"])), - human_bytes(int(parsed["total_bytes"])), - str(parsed["local_segments"]), - ] + topic_data.append( + { + "topic": topic, + "partitions": int(parsed["partitions"]), + "replicas": int(parsed["replicas"]), + "local_bytes": int(parsed["local_bytes"]), + "total_bytes": int(parsed["total_bytes"]), + "local_segments": int(parsed["local_segments"]), + } ) redpanda_pod = redpanda_pod_name(namespace=namespace) usage = probe_path_usage(redpanda_pod, DEFAULT_REDPANDA_DATA_PATH, namespace=namespace) + if args.output == "json": + print( + json.dumps( + { + "namespace": namespace, + "brokers": brokers, + "pod": redpanda_pod, + "data_path": DEFAULT_REDPANDA_DATA_PATH, + "disk": { + "path_bytes": usage["path_bytes"], + "filesystem_used_bytes": usage["filesystem_used_bytes"], + "filesystem_available_bytes": usage["filesystem_available_bytes"], + "filesystem_use_percent": usage["filesystem_use_percent"], + }, + "topics": topic_data, + "totals": { + "local_bytes": total_local_bytes, + "total_bytes": total_bytes, + "segments": total_segments, + }, + }, + indent=2, + ) + ) + return 0 + + topic_rows = [ + [ + item["topic"], + str(item["partitions"]), + str(item["replicas"]), + human_bytes(item["local_bytes"]), + human_bytes(item["total_bytes"]), + str(item["local_segments"]), + ] + for item in topic_data + ] + print(f"Namespace: {namespace}") print(f"Brokers: {brokers}") print(f"Pod: {redpanda_pod}") diff --git a/src/core/quote-outcomes.mjs b/src/core/quote-outcomes.mjs index 0195d10..98ee5e9 100644 --- a/src/core/quote-outcomes.mjs +++ b/src/core/quote-outcomes.mjs @@ -80,7 +80,7 @@ export function deriveQuoteOutcomeRecords({ const decision = decisionsByQuote.get(submission.quote_id) || null; const matches = candidatesByQuote.get(submission.quote_id) || []; const uniqueMatch = matches.length === 1 - && (candidatesByMovement.get(matches[0].movement_id) || []).length === 1 + && ((candidatesByMovement.get(matches[0].movement_id) || []).length === 1 || matches[0].is_heuristic_gap) ? matches[0] : null; @@ -142,6 +142,9 @@ export function deriveInventoryDeltas({ inventorySnapshots = [], activeAssetIds const delta_units = {}; let changed = false; + const time_delta_ms = timestampValue(current.observed_at) - timestampValue(previous.observed_at); + const is_heuristic_gap = time_delta_ms > 300_000; + for (const assetId of activeAssetIds) { const currentUnits = safeBigInt(current.spendable?.[assetId]); const previousUnits = safeBigInt(previous.spendable?.[assetId]); @@ -150,12 +153,14 @@ export function deriveInventoryDeltas({ inventorySnapshots = [], activeAssetIds if (delta !== 0n) changed = true; } - if (!changed) continue; + if (!changed && !is_heuristic_gap) continue; deltas.push({ movement_id: `${previous.observed_at}->${current.observed_at}`, observed_at: current.observed_at, observed_ts: timestampValue(current.observed_at), + time_delta_ms, + is_heuristic_gap, previous_observed_at: previous.observed_at, inventory_id: current.inventory_id, previous_inventory_id: previous.inventory_id, @@ -355,8 +360,21 @@ function movementMatchesExpectedDelta({ if (movementTs < submittedTs) return false; if (movementTs - submittedTs > attributionWindowMs) return false; - // Heuristic: if inventory-sync was down, we accept any movement that covers the time period - // We don't check exact delta units because they could be batched bidirectional trades. + for (const [assetId, expected] of Object.entries(expectedDelta)) { + const actual = safeBigInt(movement.delta_units?.[assetId]); + if (expected > 0n && actual < expected) { + if (movement.is_heuristic_gap) continue; + return false; + } + if (expected < 0n && actual > expected) { + if (movement.is_heuristic_gap) continue; + return false; + } + if (expected === 0n && actual !== 0n) { + if (movement.is_heuristic_gap) continue; + return false; + } + } return true; } diff --git a/src/lib/postgres.mjs b/src/lib/postgres.mjs index eb52e92..c68e1b3 100644 --- a/src/lib/postgres.mjs +++ b/src/lib/postgres.mjs @@ -4490,7 +4490,7 @@ export async function refreshQuoteOutcomes(pool, { OR o.outcome_status IN ('not_filled', 'submitted', 'awaiting_outcome') OR COALESCE(s.observed_at, s.ingested_at) > o.computed_at ) - ORDER BY COALESCE(s.observed_at, s.ingested_at) DESC + ORDER BY o.computed_at ASC NULLS FIRST LIMIT $1 ) SELECT event_id, observed_at, ingested_at, quote_id, payload diff --git a/src/operator-dashboard/static/components/BannerStack.jsx b/src/operator-dashboard/static/components/BannerStack.jsx index 87ddb68..2d6512a 100644 --- a/src/operator-dashboard/static/components/BannerStack.jsx +++ b/src/operator-dashboard/static/components/BannerStack.jsx @@ -1,7 +1,12 @@ export default function BannerStack({ notice, error, sourceErrors, criticalBanner }) { return ( - <> - {criticalBanner ?
{criticalBanner}
: null} +
+
+
{criticalBanner || ''}
+
{notice ?
{notice}
: null} {error ?
{error}
: null} {sourceErrors?.length ? ( @@ -13,6 +18,6 @@ export default function BannerStack({ notice, error, sourceErrors, criticalBanne ))}
) : null} - + ); } diff --git a/src/operator-dashboard/static/styles.css b/src/operator-dashboard/static/styles.css index 5812b45..b679188 100644 --- a/src/operator-dashboard/static/styles.css +++ b/src/operator-dashboard/static/styles.css @@ -91,7 +91,6 @@ select { font-size: 11px; letter-spacing: 0.14em; text-transform: uppercase; - color: rgba(247, 244, 236, 0.72); } .status-value, @@ -170,7 +169,24 @@ select { border: 1px solid rgba(255, 255, 255, 0.1); } -.banner + .banner { +.banner-stack { + display: grid; + gap: 12px; +} + +.banner-slot { + min-height: 60px; + opacity: 0; + transition: opacity 160ms ease; + pointer-events: none; +} + +.banner-slot.is-visible { + opacity: 1; + pointer-events: auto; +} + +.banner-stack > .banner + .banner { margin-top: 12px; }