diff --git a/README.md b/README.md index f33aaa9..0a1989c 100644 --- a/README.md +++ b/README.md @@ -212,6 +212,11 @@ Large values such as 100000 can require hundreds of paged MCP searches across enabled streams. If Graylog times out or rejects the query, SignalScope keeps the events already fetched, marks the stream as a partial fetch, and shows the MCP error in Diagnostics instead of failing the whole dashboard update. +The monitor also stores the last successful dashboard status in +`state/signalscope-status-cache.sqlite3`. If a later live MCP fetch fails before +usable data is available, the dashboard keeps showing the last good findings, +graphs, incidents, and correlations with a stale-data warning instead of going +blank. ## Monitoring Export diff --git a/src/fgai/cli.py b/src/fgai/cli.py index e136368..6ef8f7c 100644 --- a/src/fgai/cli.py +++ b/src/fgai/cli.py @@ -213,6 +213,7 @@ def run_monitor(args: argparse.Namespace) -> int: baseline_path=args.baseline_db, config_path=args.config_file, history_path=args.history_db, + status_cache_path=args.status_cache_db, ) return 0 @@ -421,6 +422,7 @@ def build_parser() -> argparse.ArgumentParser: monitor.add_argument("--baseline-db", default="state/fgai-baseline.sqlite3", help="SQLite database for historical behavior baselines") monitor.add_argument("--config-file", default="state/fgai-config.json", help="Runtime configuration written by dashboard") monitor.add_argument("--history-db", default="state/signalscope-history.sqlite3", help="Aggregate history database for dashboard trends") + monitor.add_argument("--status-cache-db", default="state/signalscope-status-cache.sqlite3", help="SQLite cache for last successful dashboard status") monitor.add_argument("--llm", action="store_true", help="Generate cached Ollama analyst note for dashboard") monitor.add_argument("--llm-interval", type=int, default=300, help="Seconds between Ollama dashboard assessments") monitor.add_argument("--model", default=None, help="Ollama model name") diff --git a/src/fgai/dashboard.py b/src/fgai/dashboard.py index 5e0527d..f090ae8 100644 --- a/src/fgai/dashboard.py +++ b/src/fgai/dashboard.py @@ -233,6 +233,7 @@ async function refresh() { const threat = (data.capabilities || {}).threat_intel || {}; const mcp = (data.capabilities || {}).graylog_mcp || {}; const configuration = data.configuration || {}; + const cache = data.status_cache || {}; const streamCoverage = data.stream_coverage || []; const enabledStreams = streamCoverage.filter(item => item.enabled); const streamsMissingProfile = enabledStreams.filter(item => !item.profile_ready).length; @@ -258,6 +259,8 @@ async function refresh() { capability('Graylog MCP', mcp.status === 'connected' ? 'on' : 'warn', configuration.log_source === 'graylog_mcp' ? (mcp.status || 'checking') : 'not selected') ].join(''); document.getElementById('liveStatus').innerHTML = [ + data.stale ? `Showing cached dashboard data because live MCP fetch failed.` : '', + cache.stored_at ? `Cached snapshot: ${esc(new Date(cache.stored_at * 1000).toLocaleString())}` : '', `Log file: ${esc(data.log_path || '')}`, `Policy file: ${esc(data.policy_path || 'none')}`, `Critical anomalies: ${esc((a.critical || 0))}`, diff --git a/src/fgai/history.py b/src/fgai/history.py index e12a688..5470fa8 100644 --- a/src/fgai/history.py +++ b/src/fgai/history.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json import sqlite3 import time from pathlib import Path @@ -26,3 +27,34 @@ class HistoryStore: with sqlite3.connect(self.path) as connection: rows = connection.execute("select at, events, anomalies, critical, high, baseline_ready, mcp_events from snapshots order by at desc limit ?", (limit,)).fetchall() return [dict(zip(("at", "events", "anomalies", "critical", "high", "baseline_ready", "mcp_events"), row)) for row in reversed(rows)] + + +class StatusSnapshotStore: + def __init__(self, path: str) -> None: + self.path = Path(path) + self.path.parent.mkdir(parents=True, exist_ok=True) + with sqlite3.connect(self.path) as connection: + connection.execute("create table if not exists status_snapshots (name text primary key, at integer not null, payload text not null)") + + def save(self, name: str, status: dict[str, object]) -> None: + with sqlite3.connect(self.path) as connection: + connection.execute( + "insert or replace into status_snapshots values (?, ?, ?)", + (name, int(time.time()), json.dumps(status, sort_keys=True)), + ) + + def load(self, name: str) -> dict[str, object] | None: + with sqlite3.connect(self.path) as connection: + row = connection.execute("select at, payload from status_snapshots where name = ?", (name,)).fetchone() + if not row: + return None + try: + payload = json.loads(str(row[1])) + except json.JSONDecodeError: + return None + if not isinstance(payload, dict): + return None + payload.setdefault("status_cache", {}) + if isinstance(payload["status_cache"], dict): + payload["status_cache"]["stored_at"] = int(row[0]) + return payload diff --git a/src/fgai/monitor.py b/src/fgai/monitor.py index 70cbaa6..fecef17 100644 --- a/src/fgai/monitor.py +++ b/src/fgai/monitor.py @@ -12,7 +12,7 @@ from .event_context import build_event_context from .feedback import FeedbackStore from .graylog_mcp import GraylogMcpClient from .graylog_source import GraylogStreamSource -from .history import HistoryStore +from .history import HistoryStore, StatusSnapshotStore from .incidents import IncidentStore, build_incidents from .data_quality import assess_data_quality from .llm import ollama_dashboard_assessment, ollama_profile_advice @@ -107,6 +107,7 @@ def build_status( config_path: str | None = None, history_path: str | None = None, incident_path: str | None = None, + status_cache_path: str | None = None, ) -> dict[str, object]: config_store = ConfigStore(config_path) if config_path else None config_exists = bool(config_store and config_store.path.exists()) @@ -342,9 +343,27 @@ def build_status( history = HistoryStore(history_path) history.record(status) status["history"] = history.recent() + if status_cache_path and not (isinstance(mcp_status, dict) and mcp_status.get("status") == "error"): + StatusSnapshotStore(status_cache_path).save("last_good", status) return status +def cached_status_with_error(cache_path: str, error_status: dict[str, object]) -> dict[str, object] | None: + cached = StatusSnapshotStore(cache_path).load("last_good") + if not cached: + return None + cached["generated_at"] = int(time.time()) + cached["stale"] = True + cached["stale_reason"] = "live_mcp_error" + capabilities = cached.setdefault("capabilities", {}) + if isinstance(capabilities, dict): + capabilities["graylog_mcp"] = error_status + cached.setdefault("status_cache", {}) + if isinstance(cached["status_cache"], dict): + cached["status_cache"].update({"served_from_cache": True, "reason": "live_mcp_error"}) + return cached + + def add_llm_assessment(status: dict[str, object], *, previous: str | None = None, model: str | None = None, timeout: int | None = None) -> None: try: status["llm_assessment"] = { @@ -385,6 +404,7 @@ def monitor_loop( baseline_path: str | None = None, config_path: str | None = None, history_path: str | None = None, + status_cache_path: str | None = None, ) -> None: print(f"Monitoring {log_path}") print(f"Writing status to {output}") @@ -397,7 +417,13 @@ def monitor_loop( status = build_status( log_path, policy_path=policy_path, anomaly_limit=anomaly_limit, baseline_path=baseline_path, config_path=config_path, history_path=history_path, + status_cache_path=status_cache_path, ) + mcp = status.get("capabilities", {}).get("graylog_mcp", {}) if isinstance(status.get("capabilities"), dict) else {} + if status_cache_path and isinstance(mcp, dict) and mcp.get("status") == "error": + cached = cached_status_with_error(status_cache_path, mcp) + if cached: + status = cached if effective_llm: now = int(time.time()) if now - last_llm_at >= llm_interval: diff --git a/tests/test_monitor.py b/tests/test_monitor.py index 56e1cb8..f9d7e5e 100644 --- a/tests/test_monitor.py +++ b/tests/test_monitor.py @@ -4,7 +4,8 @@ import unittest from pathlib import Path from unittest.mock import patch -from fgai.monitor import add_llm_assessment, build_status, write_status +from fgai.history import StatusSnapshotStore +from fgai.monitor import add_llm_assessment, build_status, cached_status_with_error, write_status class MonitorTests(unittest.TestCase): @@ -109,6 +110,26 @@ class MonitorTests(unittest.TestCase): self.assertEqual(advisor["status"], "error") self.assertIn("timeout", advisor["error"]) + def test_cached_status_with_error_keeps_last_good_dashboard_data(self): + with tempfile.TemporaryDirectory() as tmp: + cache_path = str(Path(tmp) / "status-cache.sqlite3") + StatusSnapshotStore(cache_path).save( + "last_good", + { + "summary": {"total": 42}, + "capabilities": {"graylog_mcp": {"status": "connected"}}, + "cross_source_correlations": [{"entity": "10.0.0.1"}], + }, + ) + + status = cached_status_with_error(cache_path, {"status": "error", "error": "mcp down"}) + + self.assertIsNotNone(status) + self.assertEqual(status["summary"]["total"], 42) + self.assertTrue(status["stale"]) + self.assertEqual(status["capabilities"]["graylog_mcp"]["status"], "error") + self.assertTrue(status["status_cache"]["served_from_cache"]) + if __name__ == "__main__": unittest.main()