From 8be716b0b904a105268b170f2d97d918855ac28ef1e10c43828424f4026d67c7 Mon Sep 17 00:00:00 2001 From: larssand Date: Thu, 2 Jul 2026 11:01:27 +0200 Subject: [PATCH] fix --- src/fgai/dashboard.py | 24 ++++++++++++++++++------ src/fgai/monitor.py | 12 ++++++++++-- tests/test_monitor.py | 5 ++++- 3 files changed, 32 insertions(+), 9 deletions(-) diff --git a/src/fgai/dashboard.py b/src/fgai/dashboard.py index b21e073..09c328c 100644 --- a/src/fgai/dashboard.py +++ b/src/fgai/dashboard.py @@ -299,6 +299,11 @@ async function refresh() { const streamCoverage = data.stream_coverage || []; const enabledStreams = streamCoverage.filter(item => item.enabled); const streamsMissingProfile = enabledStreams.filter(item => !item.profile_ready).length; + const enabledWithNoRawEvents = enabledStreams.filter(item => Number(item.events_fetched || 0) === 0); + const displayedAggregateEvents = mcp.status === 'refreshing' ? Number(mcp.previous_aggregate_events || 0) : Number(mcp.aggregate_events || 0); + const displayedRawEvents = mcp.status === 'refreshing' ? Number(mcp.previous_raw_events_fetched || mcp.previous_events_fetched || 0) : Number(mcp.raw_events_fetched || mcp.events_fetched || 0); + const displayedFetchMode = mcp.fetch_mode || mcp.previous_fetch_mode || configuration.graylog_fetch_mode || 'raw'; + const displayedCoverage = mcp.coverage_status || mcp.previous_coverage_status || 'unknown'; const rawCorrelations = data.cross_source_correlations || []; const correlationsCached = rawCorrelations.length === 0 && uiCache.correlations.length > 0; const correlations = rawCorrelations.length ? rawCorrelations : uiCache.correlations; @@ -331,17 +336,24 @@ async function refresh() { `Baseline sources ready: ${esc((data.baseline || {}).sources_ready || 0)}`, `Baseline training days: ${esc((data.baseline || {}).training_days || 0)}`, `Baseline DB size: ${esc(bytes((data.baseline || {}).size_bytes || 0))}`, - `MCP fetch mode: ${esc(mcp.fetch_mode || configuration.graylog_fetch_mode || 'raw')}`, + `MCP status: ${esc(mcp.status || 'unknown')}`, + mcp.error ? `MCP error: ${esc(mcp.error)}` : '', + `Enabled streams: ${esc(enabledStreams.length)}`, + `MCP fetch mode: ${esc(displayedFetchMode)}${mcp.status === 'refreshing' ? ' (refreshing, showing previous counters)' : ''}`, `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 aggregate events: ${esc(mcp.aggregate_events || 0)}`, - `MCP raw sample events: ${esc(mcp.raw_events_fetched || mcp.events_fetched || 0)}`, - `MCP coverage: ${esc(mcp.coverage_status || 'unknown')}${mcp.partial_streams ? ` (${esc(mcp.partial_streams)} partial)` : ''}${mcp.truncated_streams ? ` (${esc(mcp.truncated_streams)} truncated)` : ''}${mcp.sample_limited_streams ? ` (${esc(mcp.sample_limited_streams)} sample capped)` : ''}`, + Number(mcp.max_events_per_stream || configuration.graylog_max_events_per_stream || 0) > 100000 ? `MCP max events/stream is very high. Use aggregate mode with a smaller raw sample, for example max 5000 and raw sample 1000-5000.` : '', + `MCP aggregate events: ${esc(displayedAggregateEvents)}`, + `MCP raw sample events: ${esc(displayedRawEvents)}`, + `MCP coverage: ${esc(displayedCoverage)}${mcp.partial_streams ? ` (${esc(mcp.partial_streams)} partial)` : ''}${mcp.truncated_streams ? ` (${esc(mcp.truncated_streams)} truncated)` : ''}${mcp.sample_limited_streams ? ` (${esc(mcp.sample_limited_streams)} sample capped)` : ''}`, + configuration.log_source === 'graylog_mcp' && mcp.status === 'no_streams_enabled' ? `No Graylog streams are enabled. Open Settings, Load streams, tick streams, then click Save streams.` : '', + configuration.log_source === 'graylog_mcp' && mcp.status !== 'refreshing' && enabledStreams.length > 0 && displayedRawEvents === 0 ? `No raw events fetched from enabled streams. Check poll window, Graylog query, stream permissions, and Diagnostics -> Stream Coverage.` : '', + enabledWithNoRawEvents.length ? `${esc(enabledWithNoRawEvents.length)} enabled stream(s) returned zero raw events in this poll.` : '', mcp.coverage_warning ? `${esc(mcp.coverage_warning)}` : '', advisor.status === 'error' ? `Profile advisor error: ${esc(advisor.error || 'unknown error')}` : '' ].filter(Boolean).join('
'); document.getElementById('health').innerHTML = [ - metric('Enabled streams', enabledStreams.length), metric('Streams missing profile', streamsMissingProfile), metric('MCP aggregate events', mcp.aggregate_events || 0), metric('MCP raw sample', mcp.raw_events_fetched || mcp.events_fetched || 0), metric('Partial streams', mcp.partial_streams || 0), metric('Sample capped', mcp.sample_limited_streams || 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 aggregate events', displayedAggregateEvents), metric('MCP raw sample', displayedRawEvents), metric('Partial streams', mcp.partial_streams || 0), metric('Sample capped', mcp.sample_limited_streams || 0), metric('Truncated streams', mcp.truncated_streams || 0), metric('Correlated entities', correlations.length) ].join(''); window.profileSuggestions = data.profile_suggestions || []; const showExistingProfiles = document.getElementById('showExistingProfiles')?.checked; @@ -476,7 +488,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:'Raw Events', key:'events_fetched'}, {label:'Aggregate Events', key:'aggregate_events'}, {label:'Aggregate', key:'aggregate_status'}, {label:'Aggregate Schema', key:'aggregate_schema_properties'}, {label:'Latest Event', key:'latest_event_time'}, {label:'Health', key:'health'}, {label:'Aggregate Error', render:r => esc(r.aggregate_error || '-')}, {label:'Raw Error', render:r => esc(r.raw_error || '-')}], '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:'Raw Events', key:'events_fetched'}, {label:'Aggregate Events', key:'aggregate_events'}, {label:'Aggregate', key:'aggregate_status'}, {label:'Aggregate Schema', key:'aggregate_schema_properties'}, {label:'Latest Event', key:'latest_event_time'}, {label:'Health', render:r => esc(`${r.health || ''}${r.health_detail ? ': ' + r.health_detail : ''}`)}, {label:'Aggregate Error', render:r => esc(r.aggregate_error || '-')}, {label:'Raw Error', render:r => esc(r.raw_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/monitor.py b/src/fgai/monitor.py index 1065f6b..cf9d9c2 100644 --- a/src/fgai/monitor.py +++ b/src/fgai/monitor.py @@ -76,6 +76,8 @@ def _stream_coverage(runtime_values: dict[str, object], stream_profiles: dict[st total_fields = len(readiness) status = status_by_id.get(stream_id, {}) enabled = next((bool(item.get("enabled")) for item in configured if str(item.get("id", "")) == stream_id), False) + events_fetched = int(status.get("events_fetched", 0) or 0) + health = "not_enabled" if not enabled else "partial_fetch" if status.get("partial") else "missing_profile" if not profile else "no_events" if events_fetched == 0 else "learning" if total_fields and ready_fields < total_fields else "ready" if total_fields else "profile_needs_fields" rows.append({ "stream_id": stream_id, "stream_name": _stream_name(stream_id, stream_titles, profile), @@ -87,7 +89,7 @@ def _stream_coverage(runtime_values: dict[str, object], stream_profiles: dict[st "ready_fields": ready_fields, "total_fields": total_fields, "readiness": f"{ready_fields}/{total_fields}" if total_fields else "0/0", - "events_fetched": int(status.get("events_fetched", 0) or 0), + "events_fetched": events_fetched, "aggregate_events": int(status.get("aggregate_events", 0) or 0), "aggregate_status": str(status.get("aggregate_status", "")), "aggregate_schema_properties": ", ".join(str(item) for item in status.get("aggregate_schema_properties", []) if item), @@ -97,7 +99,8 @@ def _stream_coverage(runtime_values: dict[str, object], stream_profiles: dict[st "raw_error": str(status.get("error", "")), "aggregate_error": str(status.get("aggregate_error", "")), "error": str(status.get("aggregate_error", "") or 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", + "health": health, + "health_detail": "No raw events returned for this stream in the current MCP poll window." if enabled and events_fetched == 0 else "", }) return rows @@ -465,6 +468,11 @@ def write_refreshing_status(output: str) -> None: "status": "refreshing", "previous_status": previous_mcp.get("status", "") if isinstance(previous_mcp, dict) else "", "previous_error": previous_mcp.get("error", "") if isinstance(previous_mcp, dict) else "", + "previous_events_fetched": previous_mcp.get("events_fetched", 0) if isinstance(previous_mcp, dict) else 0, + "previous_raw_events_fetched": previous_mcp.get("raw_events_fetched", 0) if isinstance(previous_mcp, dict) else 0, + "previous_aggregate_events": previous_mcp.get("aggregate_events", 0) if isinstance(previous_mcp, dict) else 0, + "previous_fetch_mode": previous_mcp.get("fetch_mode", "") if isinstance(previous_mcp, dict) else "", + "previous_coverage_status": previous_mcp.get("coverage_status", "") if isinstance(previous_mcp, dict) else "", } current["status_cache"] = {"served_from_cache": False, "reason": "refreshing"} write_status(current, output) diff --git a/tests/test_monitor.py b/tests/test_monitor.py index 5adb97c..fd44465 100644 --- a/tests/test_monitor.py +++ b/tests/test_monitor.py @@ -48,7 +48,7 @@ class MonitorTests(unittest.TestCase): "status_schema": 2, "stale": True, "stale_reason": "live_mcp_error", - "capabilities": {"graylog_mcp": {"status": "error", "error": "old dns error"}}, + "capabilities": {"graylog_mcp": {"status": "error", "error": "old dns error", "events_fetched": 99, "aggregate_events": 1234, "fetch_mode": "aggregate"}}, } ), encoding="utf-8", @@ -62,6 +62,9 @@ class MonitorTests(unittest.TestCase): self.assertEqual(status["capabilities"]["graylog_mcp"]["status"], "refreshing") self.assertEqual(status["capabilities"]["graylog_mcp"]["previous_status"], "error") self.assertEqual(status["capabilities"]["graylog_mcp"]["previous_error"], "old dns error") + self.assertEqual(status["capabilities"]["graylog_mcp"]["previous_events_fetched"], 99) + self.assertEqual(status["capabilities"]["graylog_mcp"]["previous_aggregate_events"], 1234) + self.assertEqual(status["capabilities"]["graylog_mcp"]["previous_fetch_mode"], "aggregate") self.assertFalse(status["status_cache"]["served_from_cache"]) def test_profile_readiness_includes_stream_and_profile_names(self):