From 50463aa4d64737b3b253319076ba0d701ff11ab8eadc961889a1c87208eb7900 Mon Sep 17 00:00:00 2001 From: larssand Date: Tue, 30 Jun 2026 10:15:06 +0200 Subject: [PATCH] Fixed the MCP error handling for large pulls. --- README.md | 4 ++++ src/fgai/dashboard.py | 6 +++--- src/fgai/graylog_source.py | 20 +++++++++++++++++--- src/fgai/monitor.py | 20 +++++++++++++------- tests/test_graylog_source.py | 14 ++++++++++++++ 5 files changed, 51 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 09bea09..f33aaa9 100644 --- a/README.md +++ b/README.md @@ -208,6 +208,10 @@ stream during that window. If a stream hits the cap, the dashboard marks the window as truncated because high EPS means SignalScope sampled only part of the Graylog result set. For very high-volume streams, prefer aggregate baselines and targeted drill-down queries over trying to pull every raw event through MCP. +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. ## Monitoring Export diff --git a/src/fgai/dashboard.py b/src/fgai/dashboard.py index 78cf61d..5e0527d 100644 --- a/src/fgai/dashboard.py +++ b/src/fgai/dashboard.py @@ -267,11 +267,11 @@ async function refresh() { `Baseline DB size: ${esc(bytes((data.baseline || {}).size_bytes || 0))}`, `MCP poll window: ${esc(mcp.range_seconds || configuration.graylog_range_seconds || 0)}s`, `MCP max events/stream: ${esc(mcp.max_events_per_stream || configuration.graylog_max_events_per_stream || 0)}`, - `MCP coverage: ${esc(mcp.coverage_status || 'unknown')}${mcp.truncated_streams ? ` (${esc(mcp.truncated_streams)} truncated)` : ''}`, + `MCP coverage: ${esc(mcp.coverage_status || 'unknown')}${mcp.partial_streams ? ` (${esc(mcp.partial_streams)} partial)` : ''}${mcp.truncated_streams ? ` (${esc(mcp.truncated_streams)} truncated)` : ''}`, mcp.coverage_warning ? `${esc(mcp.coverage_warning)}` : '' ].filter(Boolean).join('
'); document.getElementById('health').innerHTML = [ - metric('Enabled streams', enabledStreams.length), metric('Streams missing profile', streamsMissingProfile), metric('MCP events fetched', mcp.events_fetched || 0), metric('Truncated streams', mcp.truncated_streams || 0), metric('Correlated entities', correlations.length) + metric('Enabled streams', enabledStreams.length), metric('Streams missing profile', streamsMissingProfile), metric('MCP events fetched', mcp.events_fetched || 0), metric('Partial streams', mcp.partial_streams || 0), metric('Truncated streams', mcp.truncated_streams || 0), metric('Correlated entities', correlations.length) ].join(''); window.profileSuggestions = data.profile_suggestions || []; const advisor = ((data.capabilities || {}).profile_advisor || {}); @@ -400,7 +400,7 @@ async function refresh() { const profileNames = Object.fromEntries((data.stream_profiles || []).map(item => [item.stream_id, item.name || item.stream_id])); const profileReadiness = (data.profile_readiness || []).map(item => ({...item, profile_name: item.profile_name || profileNames[item.stream_id] || item.stream_id, stream_title: item.stream_name || item.stream_title || streamTitles[item.stream_id] || item.stream_id})); document.getElementById('diagnostics').innerHTML = - '

Stream Coverage

' + table(streamCoverage, [{label:'Stream', key:'stream_name'}, {label:'Enabled', key:'enabled', render:r => r.enabled ? 'yes' : 'no'}, {label:'Profile', render:r => esc(r.profile || 'missing')}, {label:'Entity Field', key:'entity_field'}, {label:'Tracked Fields', key:'tracked_fields'}, {label:'Ready Fields', key:'readiness'}, {label:'Events', key:'events_fetched'}, {label:'Latest Event', key:'latest_event_time'}, {label:'Health', key:'health'}], 'stream-coverage') + + '

Stream Coverage

' + table(streamCoverage, [{label:'Stream', key:'stream_name'}, {label:'Enabled', key:'enabled', render:r => r.enabled ? 'yes' : 'no'}, {label:'Profile', render:r => esc(r.profile || 'missing')}, {label:'Entity Field', key:'entity_field'}, {label:'Tracked Fields', key:'tracked_fields'}, {label:'Ready Fields', key:'readiness'}, {label:'Events', key:'events_fetched'}, {label:'Latest Event', key:'latest_event_time'}, {label:'Health', key:'health'}, {label:'Error', render:r => esc(r.error || '-')}], 'stream-coverage') + '

Cross-Source Correlations

' + table(correlations, [{label:'Entity', key:'entity', render:r => esc(`${r.entity || r.source_ip} (${r.entity_type || 'ip'})`)}, {label:'Streams', render:r => esc((r.streams || []).join(', '))}, {label:'Events', key:'events'}, {label:'Security Events', key:'security_events'}], 'correlations') + '

Entities

' + table(context.source_profiles || [], [{label:'Entity', key:'entity'}, {label:'Events', key:'events'}, {label:'UTM', key:'utm_events'}, {label:'Deny', key:'deny_or_threat_actions'}, {label:'Destinations', key:'distinct_destinations'}, {label:'Actions', render:r => esc((r.top_actions || []).join(', '))}], 'entities') + '

Profile Baseline Readiness

' + table(profileReadiness, [{label:'Profile', key:'profile_name'}, {label:'Stream', key:'stream_title'}, {label:'Field', key:'field'}, {label:'Buckets', key:'buckets'}, {label:'Age days', key:'age_days'}, {label:'Training days', key:'training_days'}, {label:'Ready', key:'ready', render:r => r.ready ? 'ready' : 'learning'}], 'profile-readiness') + diff --git a/src/fgai/graylog_source.py b/src/fgai/graylog_source.py index be03498..0692167 100644 --- a/src/fgai/graylog_source.py +++ b/src/fgai/graylog_source.py @@ -58,13 +58,19 @@ class GraylogStreamSource: events: list[LogEvent] = [] page_size = 1_000 pages = 0 + partial_error = "" while len(events) < max_events: request_limit = min(page_size, max_events - len(events)) - result = self.client.call_tool("search_messages", {**arguments, "limit": request_limit, "offset": len(events)}) + try: + result = self.client.call_tool("search_messages", {**arguments, "limit": request_limit, "offset": len(events)}) + except RuntimeError as exc: + partial_error = str(exc) + break content = result.get("result", {}).get("content", []) if isinstance(result.get("result"), dict) else [] if isinstance(result.get("result"), dict) and result["result"].get("isError"): detail = next((str(item.get("text")) for item in content if isinstance(item, dict) and item.get("type") == "text"), "Graylog search failed") - raise RuntimeError(f"graylog_search_error: {detail}") + partial_error = f"graylog_search_error: {detail}" + break records: list[dict[str, object]] = [] for item in content if isinstance(content, list) else []: if isinstance(item, dict) and item.get("type") == "text": @@ -77,7 +83,15 @@ class GraylogStreamSource: if len(records) < request_limit: break latest = max((event.fields.get("eventtime", "") for event in events), default="") - status.update({"source": "graylog_mcp", "events_fetched": len(events), "pages": pages, "truncated": len(events) >= max_events, "latest_event_time": latest}) + status.update({ + "source": "graylog_mcp", + "events_fetched": len(events), + "pages": pages, + "partial": bool(partial_error), + "error": partial_error, + "truncated": len(events) >= max_events, + "latest_event_time": latest, + }) return events, status def _event(self, record: dict[str, object]) -> LogEvent: diff --git a/src/fgai/monitor.py b/src/fgai/monitor.py index 50595d5..70cbaa6 100644 --- a/src/fgai/monitor.py +++ b/src/fgai/monitor.py @@ -89,7 +89,9 @@ def _stream_coverage(runtime_values: dict[str, object], stream_profiles: dict[st "events_fetched": int(status.get("events_fetched", 0) or 0), "latest_event_time": str(status.get("latest_event_time", "")), "truncated": bool(status.get("truncated")), - "health": "not_enabled" if not enabled else "missing_profile" if not profile else "no_events" if int(status.get("events_fetched", 0) or 0) == 0 else "learning" if total_fields and ready_fields < total_fields else "ready" if total_fields else "profile_needs_fields", + "partial": bool(status.get("partial")), + "error": str(status.get("error", "")), + "health": "not_enabled" if not enabled else "partial_fetch" if status.get("partial") else "missing_profile" if not profile else "no_events" if int(status.get("events_fetched", 0) or 0) == 0 else "learning" if total_fields and ready_fields < total_fields else "ready" if total_fields else "profile_needs_fields", }) return rows @@ -145,18 +147,22 @@ def build_status( events.extend(stream_events) stream_statuses.append({"stream_id": stream_id, "stream_name": stream_name, **stream_status}) truncated_streams = [item for item in stream_statuses if item.get("truncated")] + partial_streams = [item for item in stream_statuses if item.get("partial")] + warnings = [] + if partial_streams: + warnings.append(f"{len(partial_streams)} stream(s) returned a partial MCP fetch; Graylog likely timed out or rejected a large paged query.") + if truncated_streams: + warnings.append(f"{len(truncated_streams)} stream(s) hit max_events_per_stream; high EPS means the analysis window is only partially sampled.") mcp_status = { - "status": "connected", + "status": "partial" if partial_streams else "connected", "streams": stream_statuses, "events_fetched": len(events), "range_seconds": range_seconds, "max_events_per_stream": max_events_per_stream, + "partial_streams": len(partial_streams), "truncated_streams": len(truncated_streams), - "coverage_status": "truncated" if truncated_streams else "complete_window", - "coverage_warning": ( - f"{len(truncated_streams)} stream(s) hit max_events_per_stream; high EPS means the analysis window is only partially sampled." - if truncated_streams else "" - ), + "coverage_status": "partial" if partial_streams else "truncated" if truncated_streams else "complete_window", + "coverage_warning": " ".join(warnings), } except RuntimeError as exc: mcp_status = {"status": "error", "error": str(exc)} diff --git a/tests/test_graylog_source.py b/tests/test_graylog_source.py index 788759d..732fd54 100644 --- a/tests/test_graylog_source.py +++ b/tests/test_graylog_source.py @@ -15,6 +15,14 @@ class _Client: return {"result": {"content": [{"type": "text", "text": '{"schema":[{"field":"client"},{"field":"server"},{"field":"result"}],"datarows":[["10.0.0.1","8.8.8.8","deny"],["10.0.0.2","8.8.8.8","accept"]]}' }]}} +class _ErrorClient: + def probe(self): + return {"status": "connected"} + + def call_tool(self, _name, _arguments): + return {"result": {"isError": True, "content": [{"type": "text", "text": "Tool call failed: timeout"}]}} + + class GraylogSourceTests(unittest.TestCase): def test_applies_custom_mapping_to_generic_stream_message(self): client = _Client() @@ -42,6 +50,12 @@ class GraylogSourceTests(unittest.TestCase): self.assertEqual(client.arguments["limit"], 2) self.assertTrue(status["truncated"]) + def test_returns_partial_status_instead_of_raising_on_search_error(self): + events, status = GraylogStreamSource(_ErrorClient(), "vpn").fetch() + self.assertEqual(events, []) + self.assertTrue(status["partial"]) + self.assertIn("graylog_search_error", status["error"]) + def test_requests_selected_profile_fields(self): client = _Client() GraylogStreamSource(client, "windows", profile_fields=("TargetUserName", "EventID")).fetch()