add agregated search
This commit is contained in:
larssand
14
README.md
14
README.md
@@ -218,6 +218,20 @@ 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.
|
||||
|
||||
Graylog fetch mode controls how high-volume streams are read:
|
||||
|
||||
- `raw`: fetch raw events up to `graylog_max_events_per_stream`.
|
||||
- `aggregate`: use Graylog MCP `aggregate_messages` for total event volume, then
|
||||
fetch only `graylog_raw_sample_events` raw events per stream for findings and
|
||||
drill-down context.
|
||||
- `auto`: use aggregate mode automatically when `graylog_max_events_per_stream`
|
||||
is larger than `graylog_raw_sample_events`.
|
||||
|
||||
For high EPS environments, keep `graylog_range_seconds` at 300, set
|
||||
`graylog_fetch_mode` to `auto` or `aggregate`, and use a modest raw sample such
|
||||
as 5000. The dashboard then shows aggregate event volume without forcing every
|
||||
raw log line through MCP each poll.
|
||||
|
||||
## Monitoring Export
|
||||
|
||||
The dashboard also exposes Prometheus text metrics at:
|
||||
|
||||
@@ -12,8 +12,10 @@ DEFAULT_CONFIG: dict[str, object] = {
|
||||
"graylog_streams": [],
|
||||
"graylog_stream_profiles": [],
|
||||
"graylog_query": "*",
|
||||
"graylog_fetch_mode": "auto",
|
||||
"graylog_range_seconds": 300,
|
||||
"graylog_max_events_per_stream": 5000,
|
||||
"graylog_raw_sample_events": 5000,
|
||||
"baseline_retention_days": 14,
|
||||
"baseline_value_retention_days": 7,
|
||||
"baseline_max_values_per_field": 2000,
|
||||
@@ -66,7 +68,9 @@ class ConfigStore:
|
||||
current[key] = bool(value)
|
||||
elif key == "log_source" and value in {"local_syslog", "graylog_mcp"}:
|
||||
current[key] = value
|
||||
elif key in {"graylog_range_seconds", "graylog_max_events_per_stream", "baseline_retention_days", "baseline_value_retention_days", "baseline_max_values_per_field", "baseline_training_days", "profile_advisor_timeout", "threat_intel_daily_limit", "threat_intel_ttl_seconds", "threat_intel_error_ttl_seconds", "abuseipdb_max_age_days"}:
|
||||
elif key == "graylog_fetch_mode" and value in {"auto", "raw", "aggregate"}:
|
||||
current[key] = value
|
||||
elif key in {"graylog_range_seconds", "graylog_max_events_per_stream", "graylog_raw_sample_events", "baseline_retention_days", "baseline_value_retention_days", "baseline_max_values_per_field", "baseline_training_days", "profile_advisor_timeout", "threat_intel_daily_limit", "threat_intel_ttl_seconds", "threat_intel_error_ttl_seconds", "abuseipdb_max_age_days"}:
|
||||
try:
|
||||
minimum = 60 if key == "graylog_range_seconds" else 1
|
||||
current[key] = max(minimum, int(value))
|
||||
|
||||
@@ -116,7 +116,7 @@ HTML = """<!doctype html>
|
||||
<div data-view="overview" class="active"><section class="split"><div class="panel"><h2>Events and Anomalies</h2><canvas id="trendChart" class="chart"></canvas></div><div class="panel"><h2>Baseline and Stream Health</h2><div id="health"></div></div></section><section class="split"><div class="panel"><h2>Correlation Map</h2><canvas id="correlationGraph" class="graph"></canvas><div id="correlationGraphInfo" class="muted"></div></div><div class="panel"><h2>AI Assessment</h2><div id="llmAssessment" class="muted">LLM assessment disabled.</div></div></section><section class="panel"><h2>Investigation Incidents</h2><div id="incidents"></div></section><section class="split"><div class="panel"><h2>Anomalies</h2><div id="anomalies"></div></div><div class="panel"><h2>Recommendations</h2><div id="recommendations"></div></div></section></div>
|
||||
<div data-view="findings"><section class="panel"><h2>Triage Queue</h2><div id="triageQueue"></div></section><section class="panel"><h2>Field Baseline Deviations</h2><div class="toolbar"><label><input id="showReviewedFindings" type="checkbox"> show reviewed</label><label><input id="showLowFindings" type="checkbox"> show low score</label><span id="findingSummary"></span></div><div id="feedbackNotice" class="muted" role="status"></div><div id="fieldDeviations"></div></section><section class="panel"><h2>Related Activity Across Sources</h2><div id="relatedActivity"></div></section><section class="split"><div class="panel"><h2>Block Candidates</h2><div id="blocks"></div></div><div class="panel"><h2>Threat Intelligence</h2><div id="reputation"></div></div></section><section class="panel"><h2>Policy Findings</h2><div id="policies"></div></section></div>
|
||||
<div data-view="diagnostics"><section class="panel"><h2>Diagnostics</h2><div id="diagnostics"></div></section></div>
|
||||
<div data-view="settings"><section class="panel"><h2>Recommended Stream Profiles</h2><div id="profileSuggestions" class="muted">Waiting for observed stream data.</div></section><section class="panel"><h2>Installed Ollama Models</h2><div id="ollamaModels" class="muted">Loading local Ollama models.</div></section><section class="panel"><h2>Runtime Configuration</h2><form id="settingsForm"><div class="grid"><label>Log source<br><select name="log_source"><option value="local_syslog">Local syslog</option><option value="graylog_mcp">Graylog MCP</option></select></label><label>Graylog MCP URL<br><input name="graylog_mcp_url" type="url" placeholder="https://graylog.example/api/mcp"></label><label>Enabled streams<br><button type="button" id="loadStreams">Load streams</button><div id="streamPicker" class="muted">Load streams after URL and token are saved.</div></label><label>Profile editor<br><button type="button" id="loadFields">Edit first checked stream profile</button><div id="profileEditorStatus" class="muted">No profile selected for editing.</div><div id="fieldPicker" class="muted">Click Edit profile on one stream.</div></label><label>Profile name<br><input name="profile_name" placeholder="Example: Windows login behavior"></label><label>Detector thresholds (JSON)<br><textarea name="profile_detectors" placeholder='{"auth_failure":{"enabled":true,"minimum":5,"z_threshold":3}}'></textarea></label><label>Field weights (JSON)<br><textarea name="profile_field_weights" placeholder='{"url":1.5,"auth_failure_burst":2,"query_domain":{"rare_value":1.8}}'></textarea></label><label>Graylog query<br><input name="graylog_query" placeholder="*"></label><label>Graylog MCP poll window seconds<br><input name="graylog_range_seconds" type="number" min="60" step="60" placeholder="300"></label><label>Graylog max events per stream<br><input name="graylog_max_events_per_stream" type="number" min="1" step="1000" placeholder="5000"></label><label>Baseline training days<br><input name="baseline_training_days" type="number" min="1" step="1" placeholder="7"></label><label>Baseline bucket retention days<br><input name="baseline_retention_days" type="number" min="1" step="1" placeholder="14"></label><label>Baseline value retention days<br><input name="baseline_value_retention_days" type="number" min="1" step="1" placeholder="7"></label><label>Max values per entity field<br><input name="baseline_max_values_per_field" type="number" min="1" step="100" placeholder="2000"></label><label>Threat intel provider<br><select name="threat_intel_provider"><option value="auto">Auto</option><option value="abuseipdb">AbuseIPDB</option><option value="virustotal">VirusTotal</option></select></label><label>AbuseIPDB API key<br><input name="abuseipdb_api_key" type="password" placeholder="Leave blank to keep current key"></label><label>VirusTotal API key<br><input name="virustotal_api_key" type="password" placeholder="Leave blank to keep current key"></label><label>Threat intel daily limit<br><input name="threat_intel_daily_limit" type="number" min="1" step="1" placeholder="100"></label><label>Threat intel cache TTL seconds<br><input name="threat_intel_ttl_seconds" type="number" min="60" step="60" placeholder="604800"></label><label>Threat intel error TTL seconds<br><input name="threat_intel_error_ttl_seconds" type="number" min="60" step="60" placeholder="3600"></label><label>AbuseIPDB max age days<br><input name="abuseipdb_max_age_days" type="number" min="1" step="1" placeholder="90"></label><label>Graylog field mapping (JSON)<br><textarea name="graylog_field_mapping" placeholder='{"srcip":"client_ip","dstip":"server_ip","action":"event_action"}'></textarea></label><label>Graylog MCP token<br><input name="graylog_mcp_token" type="password" placeholder="Leave blank to keep current token"></label><label>Ollama model<br><input name="llm_model" placeholder="llama3.1"></label><label>Profile advisor model<br><input name="profile_advisor_model" placeholder="qwen3:8b"></label><label>Profile advisor timeout seconds<br><input name="profile_advisor_timeout" type="number" min="1" step="1" placeholder="120"></label><label><input name="llm_enabled" type="checkbox"> Enable Ollama analysis</label><label><input name="profile_advisor_enabled" type="checkbox"> Enable Ollama profile advisor</label><label><input name="threat_intel_enabled" type="checkbox"> Enable threat intelligence</label></div><p><button type="submit">Save configuration</button> <span id="settingsResult" class="muted"></span></p></form></section></div>
|
||||
<div data-view="settings"><section class="panel"><h2>Recommended Stream Profiles</h2><div id="profileSuggestions" class="muted">Waiting for observed stream data.</div></section><section class="panel"><h2>Installed Ollama Models</h2><div id="ollamaModels" class="muted">Loading local Ollama models.</div></section><section class="panel"><h2>Runtime Configuration</h2><form id="settingsForm"><div class="grid"><label>Log source<br><select name="log_source"><option value="local_syslog">Local syslog</option><option value="graylog_mcp">Graylog MCP</option></select></label><label>Graylog MCP URL<br><input name="graylog_mcp_url" type="url" placeholder="https://graylog.example/api/mcp"></label><label>Enabled streams<br><button type="button" id="loadStreams">Load streams</button><div id="streamPicker" class="muted">Load streams after URL and token are saved.</div></label><label>Profile editor<br><button type="button" id="loadFields">Edit first checked stream profile</button><div id="profileEditorStatus" class="muted">No profile selected for editing.</div><div id="fieldPicker" class="muted">Click Edit profile on one stream.</div></label><label>Profile name<br><input name="profile_name" placeholder="Example: Windows login behavior"></label><label>Detector thresholds (JSON)<br><textarea name="profile_detectors" placeholder='{"auth_failure":{"enabled":true,"minimum":5,"z_threshold":3}}'></textarea></label><label>Field weights (JSON)<br><textarea name="profile_field_weights" placeholder='{"url":1.5,"auth_failure_burst":2,"query_domain":{"rare_value":1.8}}'></textarea></label><label>Graylog query<br><input name="graylog_query" placeholder="*"></label><label>Graylog fetch mode<br><select name="graylog_fetch_mode"><option value="auto">Auto aggregate</option><option value="aggregate">Aggregate + sample</option><option value="raw">Raw only</option></select></label><label>Graylog MCP poll window seconds<br><input name="graylog_range_seconds" type="number" min="60" step="60" placeholder="300"></label><label>Graylog max events per stream<br><input name="graylog_max_events_per_stream" type="number" min="1" step="1000" placeholder="5000"></label><label>Raw sample events per stream<br><input name="graylog_raw_sample_events" type="number" min="1" step="1000" placeholder="5000"></label><label>Baseline training days<br><input name="baseline_training_days" type="number" min="1" step="1" placeholder="7"></label><label>Baseline bucket retention days<br><input name="baseline_retention_days" type="number" min="1" step="1" placeholder="14"></label><label>Baseline value retention days<br><input name="baseline_value_retention_days" type="number" min="1" step="1" placeholder="7"></label><label>Max values per entity field<br><input name="baseline_max_values_per_field" type="number" min="1" step="100" placeholder="2000"></label><label>Threat intel provider<br><select name="threat_intel_provider"><option value="auto">Auto</option><option value="abuseipdb">AbuseIPDB</option><option value="virustotal">VirusTotal</option></select></label><label>AbuseIPDB API key<br><input name="abuseipdb_api_key" type="password" placeholder="Leave blank to keep current key"></label><label>VirusTotal API key<br><input name="virustotal_api_key" type="password" placeholder="Leave blank to keep current key"></label><label>Threat intel daily limit<br><input name="threat_intel_daily_limit" type="number" min="1" step="1" placeholder="100"></label><label>Threat intel cache TTL seconds<br><input name="threat_intel_ttl_seconds" type="number" min="60" step="60" placeholder="604800"></label><label>Threat intel error TTL seconds<br><input name="threat_intel_error_ttl_seconds" type="number" min="60" step="60" placeholder="3600"></label><label>AbuseIPDB max age days<br><input name="abuseipdb_max_age_days" type="number" min="1" step="1" placeholder="90"></label><label>Graylog field mapping (JSON)<br><textarea name="graylog_field_mapping" placeholder='{"srcip":"client_ip","dstip":"server_ip","action":"event_action"}'></textarea></label><label>Graylog MCP token<br><input name="graylog_mcp_token" type="password" placeholder="Leave blank to keep current token"></label><label>Ollama model<br><input name="llm_model" placeholder="llama3.1"></label><label>Profile advisor model<br><input name="profile_advisor_model" placeholder="qwen3:8b"></label><label>Profile advisor timeout seconds<br><input name="profile_advisor_timeout" type="number" min="1" step="1" placeholder="120"></label><label><input name="llm_enabled" type="checkbox"> Enable Ollama analysis</label><label><input name="profile_advisor_enabled" type="checkbox"> Enable Ollama profile advisor</label><label><input name="threat_intel_enabled" type="checkbox"> Enable threat intelligence</label></div><p><button type="submit">Save configuration</button> <span id="settingsResult" class="muted"></span></p></form></section></div>
|
||||
</main>
|
||||
<script>
|
||||
function esc(value) {
|
||||
@@ -268,13 +268,16 @@ 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 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.partial_streams ? ` (${esc(mcp.partial_streams)} partial)` : ''}${mcp.truncated_streams ? ` (${esc(mcp.truncated_streams)} truncated)` : ''}`,
|
||||
`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)` : ''}`,
|
||||
mcp.coverage_warning ? `<span class="sev-high">${esc(mcp.coverage_warning)}</span>` : ''
|
||||
].filter(Boolean).join('<br>');
|
||||
document.getElementById('health').innerHTML = [
|
||||
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)
|
||||
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)
|
||||
].join('');
|
||||
window.profileSuggestions = data.profile_suggestions || [];
|
||||
const advisor = ((data.capabilities || {}).profile_advisor || {});
|
||||
@@ -403,7 +406,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 =
|
||||
'<h3>Stream Coverage</h3>' + 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') +
|
||||
'<h3>Stream Coverage</h3>' + 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:'Latest Event', key:'latest_event_time'}, {label:'Health', key:'health'}, {label:'Error', render:r => esc(r.error || '-')}], 'stream-coverage') +
|
||||
'<h3>Cross-Source Correlations</h3>' + 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') +
|
||||
'<h3>Entities</h3>' + 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') +
|
||||
'<h3>Profile Baseline Readiness</h3>' + 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') +
|
||||
@@ -474,8 +477,10 @@ async function applySuggestedProfile(streamId) {
|
||||
log_source: config.log_source || 'graylog_mcp',
|
||||
graylog_mcp_url: config.graylog_mcp_url || '',
|
||||
graylog_query: config.graylog_query || '*',
|
||||
graylog_fetch_mode: config.graylog_fetch_mode || 'auto',
|
||||
graylog_range_seconds: config.graylog_range_seconds || 300,
|
||||
graylog_max_events_per_stream: config.graylog_max_events_per_stream || 5000,
|
||||
graylog_raw_sample_events: config.graylog_raw_sample_events || 5000,
|
||||
graylog_field_mapping: config.graylog_field_mapping || '',
|
||||
baseline_training_days: config.baseline_training_days || 7,
|
||||
baseline_retention_days: config.baseline_retention_days || 14,
|
||||
|
||||
114
src/fgai/graylog_aggregate.py
Normal file
114
src/fgai/graylog_aggregate.py
Normal file
@@ -0,0 +1,114 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from collections.abc import Iterable
|
||||
|
||||
from .graylog_mcp import GraylogMcpClient
|
||||
|
||||
|
||||
def _records(value: object) -> Iterable[dict[str, object]]:
|
||||
if isinstance(value, dict):
|
||||
schema = value.get("schema")
|
||||
datarows = value.get("datarows")
|
||||
if isinstance(schema, list) and isinstance(datarows, list):
|
||||
fields = [str(column.get("field", column.get("name", ""))) for column in schema if isinstance(column, dict)]
|
||||
for row in datarows:
|
||||
if isinstance(row, list):
|
||||
yield {field: row[index] for index, field in enumerate(fields) if field and index < len(row)}
|
||||
return
|
||||
for key in ("rows", "data", "results", "messages"):
|
||||
if isinstance(value.get(key), list):
|
||||
yield from (item for item in value[key] if isinstance(item, dict))
|
||||
return
|
||||
if value:
|
||||
yield value
|
||||
|
||||
|
||||
def _number(value: object) -> int:
|
||||
try:
|
||||
return int(float(str(value)))
|
||||
except (TypeError, ValueError):
|
||||
return 0
|
||||
|
||||
|
||||
def _count_from_records(records: list[dict[str, object]]) -> int:
|
||||
if not records:
|
||||
return 0
|
||||
keys = ("count", "event_count", "events", "total", "COUNT()", "count()")
|
||||
for record in records:
|
||||
for key, value in record.items():
|
||||
normalized_key = str(key).lower()
|
||||
if normalized_key in {item.lower() for item in keys} or "count" in normalized_key:
|
||||
count = _number(value)
|
||||
if count:
|
||||
return count
|
||||
if len(records) == 1:
|
||||
numeric_values = [_number(value) for value in records[0].values()]
|
||||
return max(numeric_values or [0])
|
||||
return len(records)
|
||||
|
||||
|
||||
class GraylogAggregateSource:
|
||||
def __init__(self, client: GraylogMcpClient, stream: str, query: str = "*") -> None:
|
||||
self.client = client
|
||||
self.stream = stream
|
||||
self.query = query or "*"
|
||||
|
||||
def fetch_count(self, *, range_seconds: int = 300) -> dict[str, object]:
|
||||
status = self.client.probe()
|
||||
variants: list[dict[str, object]] = [
|
||||
{
|
||||
"query": self.query,
|
||||
"streams": [self.stream] if self.stream else [],
|
||||
"range_seconds": max(1, int(range_seconds)),
|
||||
"group_by": [],
|
||||
"metrics": [{"function": "count"}],
|
||||
},
|
||||
{
|
||||
"query": self.query,
|
||||
"streams": [self.stream] if self.stream else [],
|
||||
"range_seconds": max(1, int(range_seconds)),
|
||||
"groups": [],
|
||||
"series": [{"function": "count"}],
|
||||
},
|
||||
{
|
||||
"query": self.query,
|
||||
"streams": [self.stream] if self.stream else [],
|
||||
"range_seconds": max(1, int(range_seconds)),
|
||||
},
|
||||
]
|
||||
errors: list[str] = []
|
||||
for arguments in variants:
|
||||
try:
|
||||
result = self.client.call_tool("aggregate_messages", arguments)
|
||||
except RuntimeError as exc:
|
||||
errors.append(str(exc))
|
||||
continue
|
||||
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 aggregate failed")
|
||||
errors.append(detail)
|
||||
continue
|
||||
records: list[dict[str, object]] = []
|
||||
for item in content if isinstance(content, list) else []:
|
||||
if isinstance(item, dict) and item.get("type") == "text":
|
||||
try:
|
||||
records.extend(_records(json.loads(str(item.get("text", "")))))
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
return {
|
||||
**status,
|
||||
"source": "graylog_mcp_aggregate",
|
||||
"aggregate_status": "ok",
|
||||
"aggregate_events": _count_from_records(records),
|
||||
"aggregate_records": len(records),
|
||||
"aggregate_arguments": arguments,
|
||||
}
|
||||
return {
|
||||
**status,
|
||||
"source": "graylog_mcp_aggregate",
|
||||
"aggregate_status": "error",
|
||||
"aggregate_events": 0,
|
||||
"aggregate_records": 0,
|
||||
"aggregate_error": "; ".join(errors[-3:]) or "aggregate_messages failed",
|
||||
}
|
||||
@@ -10,6 +10,7 @@ from .config import ConfigStore
|
||||
from .correlation import correlate_source_ips
|
||||
from .event_context import build_event_context
|
||||
from .feedback import FeedbackStore
|
||||
from .graylog_aggregate import GraylogAggregateSource
|
||||
from .graylog_mcp import GraylogMcpClient
|
||||
from .graylog_source import GraylogStreamSource
|
||||
from .history import HistoryStore, StatusSnapshotStore
|
||||
@@ -87,6 +88,8 @@ def _stream_coverage(runtime_values: dict[str, object], stream_profiles: dict[st
|
||||
"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),
|
||||
"aggregate_events": int(status.get("aggregate_events", 0) or 0),
|
||||
"aggregate_status": str(status.get("aggregate_status", "")),
|
||||
"latest_event_time": str(status.get("latest_event_time", "")),
|
||||
"truncated": bool(status.get("truncated")),
|
||||
"partial": bool(status.get("partial")),
|
||||
@@ -133,6 +136,10 @@ def build_status(
|
||||
events = []
|
||||
range_seconds = _range_seconds(runtime_values.get("graylog_range_seconds", 300))
|
||||
max_events_per_stream = max(1, int(runtime_values.get("graylog_max_events_per_stream", 5000) or 5000))
|
||||
raw_sample_events = max(1, int(runtime_values.get("graylog_raw_sample_events", 5000) or 5000))
|
||||
fetch_mode = str(runtime_values.get("graylog_fetch_mode", "auto") or "auto")
|
||||
use_aggregate = fetch_mode == "aggregate" or (fetch_mode == "auto" and max_events_per_stream > raw_sample_events)
|
||||
aggregate_events_total = 0
|
||||
for stream_config in stream_configs:
|
||||
stream_id = str(stream_config["id"])
|
||||
profile = stream_profiles.get(stream_id)
|
||||
@@ -144,12 +151,21 @@ def build_status(
|
||||
*tuple(str(field) for field in getattr(profile, "numeric_fields", ())),
|
||||
) if profile else ()
|
||||
stream_name = str(stream_config.get("title", "") or stream_titles.get(stream_id) or stream_id)
|
||||
stream_events, stream_status = GraylogStreamSource(GraylogMcpClient(url, token), stream_id, str(runtime_values.get("graylog_query", "*")), str(runtime_values.get("graylog_field_mapping", "")), stream_name, profile_fields).fetch(max_events=max_events_per_stream, range_seconds=range_seconds)
|
||||
aggregate_status: dict[str, object] = {}
|
||||
if use_aggregate:
|
||||
aggregate_status = GraylogAggregateSource(GraylogMcpClient(url, token), stream_id, str(runtime_values.get("graylog_query", "*"))).fetch_count(range_seconds=range_seconds)
|
||||
aggregate_events_total += int(aggregate_status.get("aggregate_events", 0) or 0)
|
||||
raw_limit = raw_sample_events if use_aggregate else max_events_per_stream
|
||||
stream_events, stream_status = GraylogStreamSource(GraylogMcpClient(url, token), stream_id, str(runtime_values.get("graylog_query", "*")), str(runtime_values.get("graylog_field_mapping", "")), stream_name, profile_fields).fetch(max_events=raw_limit, range_seconds=range_seconds)
|
||||
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")]
|
||||
stream_statuses.append({"stream_id": stream_id, "stream_name": stream_name, **aggregate_status, **stream_status, "raw_sample_limit": raw_limit})
|
||||
sample_limited_streams = [item for item in stream_statuses if item.get("truncated") and use_aggregate]
|
||||
truncated_streams = [item for item in stream_statuses if item.get("truncated") and not use_aggregate]
|
||||
partial_streams = [item for item in stream_statuses if item.get("partial")]
|
||||
aggregate_errors = [item for item in stream_statuses if item.get("aggregate_status") == "error"]
|
||||
warnings = []
|
||||
if aggregate_errors:
|
||||
warnings.append(f"{len(aggregate_errors)} stream(s) returned aggregate MCP errors.")
|
||||
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:
|
||||
@@ -158,10 +174,16 @@ def build_status(
|
||||
"status": "partial" if partial_streams else "connected",
|
||||
"streams": stream_statuses,
|
||||
"events_fetched": len(events),
|
||||
"raw_events_fetched": len(events),
|
||||
"aggregate_events": aggregate_events_total,
|
||||
"fetch_mode": "aggregate" if use_aggregate else "raw",
|
||||
"range_seconds": range_seconds,
|
||||
"max_events_per_stream": max_events_per_stream,
|
||||
"raw_sample_events": raw_sample_events,
|
||||
"partial_streams": len(partial_streams),
|
||||
"sample_limited_streams": len(sample_limited_streams),
|
||||
"truncated_streams": len(truncated_streams),
|
||||
"aggregate_error_streams": len(aggregate_errors),
|
||||
"coverage_status": "partial" if partial_streams else "truncated" if truncated_streams else "complete_window",
|
||||
"coverage_warning": " ".join(warnings),
|
||||
}
|
||||
@@ -275,11 +297,16 @@ def build_status(
|
||||
except Exception as exc:
|
||||
policy_error = str(exc)
|
||||
|
||||
summary = summarize_events(events)
|
||||
if isinstance(mcp_status, dict) and int(mcp_status.get("aggregate_events", 0) or 0) > summary.get("total", 0):
|
||||
summary["total"] = int(mcp_status.get("aggregate_events", 0) or 0)
|
||||
summary["raw_sample_total"] = len(events)
|
||||
summary["aggregate_backed"] = True
|
||||
status = {
|
||||
"generated_at": int(time.time()),
|
||||
"log_path": log_path,
|
||||
"policy_path": policy_path,
|
||||
"summary": summarize_events(events),
|
||||
"summary": summary,
|
||||
"anomaly_summary": anomaly_summary(anomalies),
|
||||
"baseline": {"enabled": bool(baseline), "sources_ready": len(profiles), "training_days": baseline_training_days, "new_events_recorded": baseline_events, "profile_fields_recorded": profile_baseline_fields, "maintenance": baseline_maintenance, "size_bytes": baseline_maintenance.get("size_bytes", 0) if isinstance(baseline_maintenance, dict) else 0},
|
||||
"capabilities": {"threat_intel": threat_intel_status, "graylog_mcp": mcp_status, "profile_advisor": profile_advisor_status},
|
||||
|
||||
@@ -31,6 +31,13 @@ class ConfigTests(unittest.TestCase):
|
||||
public = store.update({"graylog_max_events_per_stream": "25000"})
|
||||
self.assertEqual(public["graylog_max_events_per_stream"], 25000)
|
||||
|
||||
def test_graylog_fetch_mode_and_raw_sample_events_are_saved(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
store = ConfigStore(str(Path(directory) / "config.json"))
|
||||
public = store.update({"graylog_fetch_mode": "aggregate", "graylog_raw_sample_events": "2500"})
|
||||
self.assertEqual(public["graylog_fetch_mode"], "aggregate")
|
||||
self.assertEqual(public["graylog_raw_sample_events"], 2500)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
45
tests/test_graylog_aggregate.py
Normal file
45
tests/test_graylog_aggregate.py
Normal file
@@ -0,0 +1,45 @@
|
||||
import unittest
|
||||
|
||||
from fgai.graylog_aggregate import GraylogAggregateSource
|
||||
|
||||
|
||||
class _AggregateClient:
|
||||
def __init__(self, responses):
|
||||
self.responses = list(responses)
|
||||
self.arguments = []
|
||||
|
||||
def probe(self):
|
||||
return {"status": "connected"}
|
||||
|
||||
def call_tool(self, _name, arguments):
|
||||
self.arguments.append(arguments)
|
||||
return self.responses.pop(0)
|
||||
|
||||
|
||||
class GraylogAggregateTests(unittest.TestCase):
|
||||
def test_reads_count_from_graylog_schema_rows(self):
|
||||
client = _AggregateClient([
|
||||
{"result": {"content": [{"type": "text", "text": '{"schema":[{"name":"metric: count()"}],"datarows":[[12345]]}'}]}}
|
||||
])
|
||||
|
||||
status = GraylogAggregateSource(client, "firewall").fetch_count(range_seconds=300)
|
||||
|
||||
self.assertEqual(status["aggregate_status"], "ok")
|
||||
self.assertEqual(status["aggregate_events"], 12345)
|
||||
self.assertEqual(client.arguments[0]["streams"], ["firewall"])
|
||||
|
||||
def test_tries_fallback_argument_shape_after_tool_error(self):
|
||||
client = _AggregateClient([
|
||||
{"result": {"isError": True, "content": [{"type": "text", "text": "bad metrics"}]}},
|
||||
{"result": {"content": [{"type": "text", "text": '{"events": 42}'}]}},
|
||||
])
|
||||
|
||||
status = GraylogAggregateSource(client, "firewall").fetch_count()
|
||||
|
||||
self.assertEqual(status["aggregate_status"], "ok")
|
||||
self.assertEqual(status["aggregate_events"], 42)
|
||||
self.assertEqual(len(client.arguments), 2)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user