Public/fgAI
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix poll mcp

Browse Source
This commit is contained in:
larssand
2026-07-02 21:03:37 +02:00
parent 825bd0247a
commit 7d1bdd35c9
6 changed files with 58 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
src/fgai/monitor.py
View File

@@ -92,7 +92,7 @@ def _stream_coverage(runtime_values: dict[str, object], stream_profiles: dict[st
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"
health = "not_enabled" if not enabled else "poll_budget_skipped" if status.get("error") == "skipped_poll_budget" 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),
@@ -115,7 +115,7 @@ def _stream_coverage(runtime_values: dict[str, object], stream_profiles: dict[st
"aggregate_error": str(status.get("aggregate_error", "")),
"error": str(status.get("aggregate_error", "") or status.get("error", "")),
"health": health,
"health_detail": "No raw events returned for this stream in the current MCP poll window." if enabled and events_fetched == 0 else "",
"health_detail": str(status.get("health_detail", "")) or ("No raw events returned for this stream in the current MCP poll window." if enabled and events_fetched == 0 else ""),
})
return rows
@@ -165,15 +165,36 @@ def build_status(
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))
mcp_call_timeout = max(1, int(runtime_values.get("graylog_mcp_call_timeout_seconds", 8) or 8))
mcp_poll_timeout = max(60, int(runtime_values.get("graylog_mcp_poll_timeout_seconds", 120) or 120))
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
client = GraylogMcpClient(url, token, verify_tls=verify_tls)
poll_started_monotonic = time.monotonic()
poll_deadline = poll_started_monotonic + mcp_poll_timeout
client = GraylogMcpClient(url, token, timeout=mcp_call_timeout, verify_tls=verify_tls)
probe_status = client.probe()
discovery_store = FieldDiscoveryStore(history_path) if history_path else None
catalog_fields_total = 0
for stream_config in stream_configs:
stream_id = str(stream_config["id"])
stream_name = str(stream_config.get("title", "") or stream_titles.get(stream_id) or stream_id)
if time.monotonic() >= poll_deadline:
stream_statuses.append({
"stream_id": stream_id,
"stream_name": stream_name,
"source": "graylog_mcp",
"events_fetched": 0,
"aggregate_events": 0,
"pages": 0,
"partial": True,
"error": "skipped_poll_budget",
"truncated": False,
"latest_event_time": "",
"raw_sample_limit": min(raw_sample_events, 10_000) if use_aggregate else max_events_per_stream,
"health_detail": "Skipped because the MCP poll time budget was reached before this stream.",
})
continue
profile = stream_profiles.get(stream_id)
profile_fields = (
str(getattr(profile, "entity_field", "")),
@@ -184,7 +205,6 @@ def build_status(
*tuple(str(getattr(relation, "left", "")) for relation in getattr(profile, "relationship_fields", ())),
*tuple(str(getattr(relation, "right", "")) for relation in getattr(profile, "relationship_fields", ())),
) if profile else ()
stream_name = str(stream_config.get("title", "") or stream_titles.get(stream_id) or stream_id)
if discovery_store:
try:
catalog_fields_total += discovery_store.ingest_catalog(stream_id, stream_name, _graylog_fields_from_result(client.call_tool("list_fields", {"streams": [stream_id]})))
@@ -201,8 +221,11 @@ def build_status(
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")]
skipped_streams = [item for item in stream_statuses if item.get("error") == "skipped_poll_budget"]
aggregate_errors = [item for item in stream_statuses if item.get("aggregate_status") == "error"]
warnings = []
if skipped_streams:
warnings.append(f"{len(skipped_streams)} stream(s) skipped because the MCP poll time budget was reached.")
if aggregate_errors:
warnings.append(f"{len(aggregate_errors)} stream(s) returned aggregate MCP errors.")
if partial_streams:
@@ -216,11 +239,15 @@ def build_status(
"raw_events_fetched": len(events),
"aggregate_events": aggregate_events_total,
"poll_completed_at": int(time.time()),
"poll_duration_seconds": round(time.monotonic() - poll_started_monotonic, 2),
"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,
"call_timeout_seconds": mcp_call_timeout,
"poll_timeout_seconds": mcp_poll_timeout,
"partial_streams": len(partial_streams),
"skipped_streams": len(skipped_streams),
"sample_limited_streams": len(sample_limited_streams),
"truncated_streams": len(truncated_streams),
"aggregate_error_streams": len(aggregate_errors),