from __future__ import annotations import contextlib import json import signal import threading import time from pathlib import Path from .anomaly import anomaly_summary, detect_source_anomalies from .baseline import BaselineStore 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 FieldDiscoveryStore, HistoryStore, StatusSnapshotStore from .incidents import IncidentStore, build_incidents from .data_quality import assess_data_quality from .llm import ollama_dashboard_assessment, ollama_profile_advice from .logs import local_in_failures, read_events, summarize_events, top_field_values from .mitigation import parse_allowlist, suggest_block_candidates from .policies import audit_policies, read_policies from .profile_suggestions import apply_profile_advice, suggest_stream_profiles from .recommendations import build_recommendations from .sequences import detect_sequences from .threat_intel import ThreatIntelClient, enrich_ips, is_public_ip from .triage import build_triage_queue from .stream_profiles import parse_profiles @contextlib.contextmanager def _cycle_timeout(seconds: int): if seconds <= 0 or threading.current_thread() is not threading.main_thread() or not hasattr(signal, "SIGALRM"): yield return previous_handler = signal.getsignal(signal.SIGALRM) previous_timer = signal.getitimer(signal.ITIMER_REAL) started = time.monotonic() def _raise_timeout(_signum, _frame): raise TimeoutError(f"status_cycle_timeout_{seconds}s") signal.signal(signal.SIGALRM, _raise_timeout) signal.setitimer(signal.ITIMER_REAL, seconds) try: yield finally: elapsed = time.monotonic() - started remaining = max(0.0, previous_timer[0] - elapsed) if previous_timer[0] else 0.0 signal.setitimer(signal.ITIMER_REAL, remaining, previous_timer[1]) signal.signal(signal.SIGALRM, previous_handler) def _stream_titles(config: dict[str, object]) -> dict[str, str]: return { str(item.get("id", "")): str(item.get("title", "") or item.get("id", "")) for item in config.get("graylog_streams", []) if isinstance(item, dict) and item.get("id") } def _public_runtime_config(config: dict[str, object]) -> dict[str, object]: public = { key: value for key, value in config.items() if key not in {"graylog_mcp_token", "abuseipdb_api_key", "virustotal_api_key"} } public["graylog_mcp_token_configured"] = bool(config.get("graylog_mcp_token")) public["abuseipdb_api_key_configured"] = bool(config.get("abuseipdb_api_key")) public["virustotal_api_key_configured"] = bool(config.get("virustotal_api_key")) public["enabled_streams"] = sum( 1 for item in config.get("graylog_streams", []) if isinstance(item, dict) and item.get("enabled") ) return public def _stream_name(stream_id: str, stream_titles: dict[str, str], profile: object | None = None) -> str: return stream_titles.get(stream_id) or getattr(profile, "name", "") or stream_id def _profile_name(stream_id: str, stream_titles: dict[str, str], profile: object | None = None) -> str: name = str(getattr(profile, "name", "") or "").strip() if not name or name == stream_id: return f"{_stream_name(stream_id, stream_titles, profile)} profile" return name def _range_seconds(value: object) -> int: try: return max(60, int(value)) except (TypeError, ValueError): return 300 def _graylog_fields_from_result(result: dict[str, object]) -> list[dict[str, object]]: content = result.get("result", {}).get("content", []) if isinstance(result.get("result"), dict) else [] text = next((item.get("text", "") for item in content if isinstance(item, dict)), "") if not text: return [] try: payload = json.loads(text) except json.JSONDecodeError: return [] fields = payload.get("fields", payload) if isinstance(payload, dict) else payload if isinstance(fields, dict) and isinstance(fields.get("fields"), list): fields = fields["fields"] return [item for item in fields if isinstance(item, dict)] if isinstance(fields, list) else [] def _stream_coverage(runtime_values: dict[str, object], stream_profiles: dict[str, object], stream_status: dict[str, object], profile_readiness: list[dict[str, object]], stream_titles: dict[str, str]) -> list[dict[str, object]]: configured = [ item for item in runtime_values.get("graylog_streams", []) if isinstance(item, dict) and item.get("id") ] status_by_id = { str(item.get("stream_id", "")): item for item in stream_status.get("streams", []) if isinstance(item, dict) and item.get("stream_id") } readiness_by_stream: dict[str, list[dict[str, object]]] = {} for item in profile_readiness: readiness_by_stream.setdefault(str(item.get("stream_id", "")), []).append(item) ids = list(dict.fromkeys([str(item.get("id", "")) for item in configured] + list(stream_profiles) + list(status_by_id))) rows = [] for stream_id in ids: profile = stream_profiles.get(stream_id) readiness = readiness_by_stream.get(stream_id, []) ready_fields = sum(1 for item in readiness if item.get("ready")) 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 "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), "enabled": enabled, "profile": _profile_name(stream_id, stream_titles, profile) if profile else "", "profile_ready": bool(profile), "entity_field": ", ".join(getattr(profile, "entity_fields", ()) or (str(getattr(profile, "entity_field", "")),)) if profile else "", "tracked_fields": len(getattr(profile, "categorical_fields", ())) + len(getattr(profile, "numeric_fields", ())) + len(getattr(profile, "relationship_fields", ())) if profile else 0, "ready_fields": ready_fields, "total_fields": total_fields, "readiness": f"{ready_fields}/{total_fields}" if total_fields else "0/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), "latest_event_time": str(status.get("latest_event_time", "")), "truncated": bool(status.get("truncated")), "partial": bool(status.get("partial")), "raw_error": str(status.get("error", "")), "aggregate_error": str(status.get("aggregate_error", "")), "error": str(status.get("aggregate_error", "") or status.get("error", "")), "health": health, "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 def build_status( log_path: str, *, policy_path: str | None = None, min_block_events: int = 3, min_block_score: int = 7, anomaly_limit: int = 20, baseline_path: str | None = None, 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()) runtime_values = config_store.read() if config_exists and config_store else {} runtime_config = _public_runtime_config(runtime_values) if config_store else {} stream_profiles = parse_profiles(runtime_values.get("graylog_stream_profiles", [])) stream_titles = _stream_titles(runtime_values) events = read_events(log_path) if Path(log_path).exists() else [] mcp_status: dict[str, object] = {"status": "not_configured"} if runtime_values.get("log_source") == "graylog_mcp": url, token = str(runtime_values.get("graylog_mcp_url", "")), str(runtime_values.get("graylog_mcp_token", "")) verify_tls = bool(runtime_values.get("graylog_tls_verify", True)) if not url or not token: mcp_status = {"status": "missing_configuration"} events = [] else: try: configured_streams = runtime_values.get("graylog_streams", []) stream_configs = [item for item in configured_streams if isinstance(item, dict) and item.get("enabled") and item.get("id")] stream_ids = [str(item.get("id")) for item in stream_configs] if not stream_ids: legacy_stream = str(runtime_values.get("graylog_stream", "") or "") if legacy_stream: stream_configs = [{"id": legacy_stream, "title": "Graylog"}] else: mcp_status = {"status": "no_streams_enabled", "streams": [], "events_fetched": 0, "coverage_status": "no_streams_enabled"} events = [] raise StopIteration stream_statuses = [] 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)) 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 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 def budget_exceeded() -> bool: return time.monotonic() >= poll_deadline def skipped_status(stream_id: str, stream_name: str) -> dict[str, object]: return { "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.", } 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 budget_exceeded(): stream_statuses.append(skipped_status(stream_id, stream_name)) continue profile = stream_profiles.get(stream_id) profile_fields = ( str(getattr(profile, "entity_field", "")), *tuple(str(field) for field in getattr(profile, "entity_fields", ())), str(getattr(profile, "timestamp_field", "")), *tuple(str(field) for field in getattr(profile, "categorical_fields", ())), *tuple(str(field) for field in getattr(profile, "numeric_fields", ())), *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 () if discovery_store and not budget_exceeded(): try: catalog_fields_total += discovery_store.ingest_catalog(stream_id, stream_name, _graylog_fields_from_result(client.call_tool("list_fields", {"streams": [stream_id]}))) except RuntimeError: pass aggregate_status: dict[str, object] = {} if use_aggregate and not budget_exceeded(): aggregate_status = GraylogAggregateSource(client, stream_id, str(runtime_values.get("graylog_query", "*"))).fetch_count(range_seconds=range_seconds, probe_status=probe_status, deadline_monotonic=poll_deadline) aggregate_events_total += int(aggregate_status.get("aggregate_events", 0) or 0) raw_limit = min(raw_sample_events, 10_000) if use_aggregate else max_events_per_stream if budget_exceeded(): stream_statuses.append({**skipped_status(stream_id, stream_name), **aggregate_status}) continue stream_events, stream_status = GraylogStreamSource(client, 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, probe_status=probe_status, deadline_monotonic=poll_deadline) events.extend(stream_events) 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")] 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: 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": "partial" if partial_streams or aggregate_errors else "connected", "streams": stream_statuses, "events_fetched": len(events), "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), "catalog_fields": catalog_fields_total, "coverage_status": "partial" if partial_streams else "truncated" if truncated_streams else "complete_window", "coverage_warning": " ".join(warnings), } except StopIteration: pass except RuntimeError as exc: mcp_status = {"status": "error", "error": str(exc)} events = [] baseline = BaselineStore(baseline_path) if baseline_path else None profiles = baseline.profiles({event.src_ip for event in events if event.src_ip}) if baseline else {} baseline_training_days = int(runtime_values.get("baseline_training_days", 7) or 7) field_deviations = baseline.profile_deviations(events, stream_profiles, min_training_days=baseline_training_days) if baseline else {} sequence_findings = detect_sequences(events) for entity, findings in sequence_findings.items(): field_deviations.setdefault(entity, []).extend(findings) for deviations in field_deviations.values(): for deviation in deviations: stream_id = str(deviation.get("stream_id", "")) profile = stream_profiles.get(stream_id) name = _stream_name(stream_id, stream_titles, profile) deviation["stream_name"] = name deviation["stream_title"] = name deviation["profile_name"] = _profile_name(stream_id, stream_titles, profile) deviation["sample_events"] = [ {"stream": name, **sample} if isinstance(sample, dict) and not sample.get("stream") else sample for sample in deviation.get("sample_events", []) ] feedback = FeedbackStore().entries() for entity, deviations in field_deviations.items(): for deviation in deviations: match = next(( item for item in feedback if item.get("entity") == entity and item.get("stream_id") == deviation.get("stream_id") and item.get("field") == deviation.get("field") and (not item.get("value") or item.get("value") == deviation.get("value", "")) ), None) if match: deviation["feedback"] = match["status"] if match["status"] in {"false_positive", "expected"}: deviation["score"] = 0 anomalies = detect_source_anomalies(events, limit=anomaly_limit, baselines=profiles, field_deviations=field_deviations) baseline_events = baseline.ingest(events) if baseline else 0 profile_baseline_fields = baseline.ingest_profile_fields(events, stream_profiles) if baseline else 0 baseline_maintenance = ( baseline.maintenance( retention_days=int(runtime_values.get("baseline_retention_days", 14) or 14), value_retention_days=int(runtime_values.get("baseline_value_retention_days", 7) or 7), max_values_per_field=int(runtime_values.get("baseline_max_values_per_field", 2000) or 2000), ) if baseline else {} ) profile_readiness = baseline.profile_readiness(stream_profiles, min_training_days=baseline_training_days) if baseline else [] profile_readiness = [ { **item, "profile_name": _profile_name(str(item.get("stream_id", "")), stream_titles, stream_profiles.get(str(item.get("stream_id", "")))), "stream_name": _stream_name(str(item.get("stream_id", "")), stream_titles, stream_profiles.get(str(item.get("stream_id", "")))), } for item in profile_readiness ] stream_coverage = _stream_coverage(runtime_values, stream_profiles, mcp_status, profile_readiness, stream_titles) discovery_cache_events = [] discovered_profile_fields = 0 if history_path: discovery_store = FieldDiscoveryStore(history_path) discovered_profile_fields = discovery_store.ingest(events) discovery_cache_events = discovery_store.synthetic_events() profile_suggestions = suggest_stream_profiles([*discovery_cache_events, *events], existing_profiles=stream_profiles) profile_advisor_status = {"enabled": bool(runtime_values.get("profile_advisor_enabled")), "status": "disabled"} if runtime_values.get("profile_advisor_enabled") and profile_suggestions: try: advice = ollama_profile_advice( profile_suggestions, model=str(runtime_values.get("profile_advisor_model", "") or "qwen3:8b"), timeout=int(runtime_values.get("profile_advisor_timeout", 120) or 120), ) profile_suggestions = apply_profile_advice(profile_suggestions, advice) profile_advisor_status = {"enabled": True, "status": "ok" if advice else "empty", "profiles_returned": len(advice), "model": str(runtime_values.get("profile_advisor_model", "") or "qwen3:8b")} except Exception as exc: profile_advisor_status = {"enabled": True, "status": "error", "error": str(exc), "model": str(runtime_values.get("profile_advisor_model", "") or "qwen3:8b")} for suggestion in profile_suggestions: suggestion.setdefault("profile_advisor", {"status": "heuristic", "error": str(exc)}) intel_ips = sorted( { ip for event in events for ip in (event.src_ip, event.dst_ip) if is_public_ip(ip) } ) threat_enabled = bool(runtime_values.get("threat_intel_enabled")) if runtime_values else None reputation = enrich_ips(intel_ips, limit=25, enabled=threat_enabled, config=runtime_values) threat_intel_status = ThreatIntelClient( enabled=threat_enabled, provider=str(runtime_values.get("threat_intel_provider", "auto")), abuseipdb_key=str(runtime_values.get("abuseipdb_api_key", "") or "") or None, virustotal_key=str(runtime_values.get("virustotal_api_key", "") or "") or None, daily_limit=int(runtime_values.get("threat_intel_daily_limit", 100) or 100), ttl_seconds=int(runtime_values.get("threat_intel_ttl_seconds", 604800) or 604800), error_ttl_seconds=int(runtime_values.get("threat_intel_error_ttl_seconds", 3600) or 3600), abuseipdb_max_age_days=int(runtime_values.get("abuseipdb_max_age_days", 90) or 90), ).status() recommendations = build_recommendations(events, anomalies, reputation) correlations = correlate_source_ips(events) incidents = IncidentStore(incident_path or "state/signalscope-incidents.json").apply(build_incidents(anomalies, field_deviations, correlations)) triage_queue = build_triage_queue(incidents, field_deviations, correlations, recommendations) block_candidates = suggest_block_candidates( events, min_events=min_block_events, min_score=min_block_score, allowlist=parse_allowlist(), ) policy_findings: list[dict[str, str | None]] = [] policy_error: str | None = None if policy_path and Path(policy_path).exists(): try: policy_findings = [finding.__dict__ for finding in audit_policies(read_policies(policy_path))] 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 = { "status_schema": 2, "stale": False, "generated_at": int(time.time()), "log_path": log_path, "policy_path": policy_path, "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, "discovery_fields_recorded": discovered_profile_fields, "discovery_cache_events": len(discovery_cache_events), "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}, "configuration": runtime_config, "stream_profiles": [{"stream_id": item.stream_id, "name": _profile_name(item.stream_id, stream_titles, item), "stream_name": _stream_name(item.stream_id, stream_titles, item), "entity_field": item.entity_field, "entity_fields": list(item.entity_fields), "timestamp_field": item.timestamp_field, "categorical_fields": list(item.categorical_fields), "numeric_fields": list(item.numeric_fields), "detectors": item.detectors, "field_weights": item.field_weights, "relationship_fields": [{"left": relation.left, "right": relation.right, "name": relation.name} for relation in item.relationship_fields]} for item in stream_profiles.values()], "stream_coverage": stream_coverage, "profile_suggestions": profile_suggestions, "profile_readiness": profile_readiness, "diagnostics": { "top_source_ips": top_field_values(events, "srcip", limit=10), "top_destination_ips": top_field_values(events, "dstip", limit=10), "top_policy_ids": top_field_values(events, "policyid", limit=10), "top_destination_ports": top_field_values(events, "dstport", limit=10), "top_source_ports": top_field_values(events, "srcport", limit=10), "top_services": top_field_values(events, "service", limit=10), "top_actions": top_field_values(events, "action", limit=10), "top_subtypes": top_field_values(events, "subtype", limit=10), "local_in_failures": local_in_failures(events, limit=10), }, "event_context": build_event_context(events), "field_deviations": field_deviations, "triage_queue": triage_queue, "sequence_findings": sequence_findings, "feedback": feedback, "cross_source_correlations": correlations, "incidents": incidents, "data_quality": assess_data_quality(events, mcp_status), "status_cache": {"served_from_cache": False, "reason": ""}, "anomalies": [ { "subject": finding.subject, "score": finding.score, "severity": finding.severity, "confidence": finding.confidence, "reasons": finding.reasons, "evidence": finding.evidence, } for finding in anomalies ], "recommendations": [ { "subject": item.subject, "score": item.score, "severity": item.severity, "title": item.title, "recommendation": item.recommendation, "reasons": item.reasons, "related_policy_ids": item.related_policy_ids, "related_services": item.related_services, } for item in recommendations ], "reputation": reputation, "block_candidates": [ {"src_ip": candidate.src_ip, "score": candidate.score, "reasons": candidate.reasons} for candidate in block_candidates ], "policy_findings": policy_findings, "policy_error": policy_error, } if history_path: 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 for row in cached.get("stream_coverage", []) if isinstance(cached.get("stream_coverage"), list) else []: if isinstance(row, dict): row.setdefault("raw_error", "") row.setdefault("aggregate_error", "") row.setdefault("error", row.get("aggregate_error") or row.get("raw_error") or "") if row.get("aggregate_status") == "error" and not row.get("aggregate_error") and row.get("error"): row["aggregate_error"] = str(row.get("error", "")) if row.get("partial") and not row.get("raw_error") and row.get("error"): row["raw_error"] = str(row.get("error", "")) cached["status_schema"] = 2 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"] = { "enabled": True, "status": "ok", "generated_at": int(time.time()), "text": ollama_dashboard_assessment(status, model=model, timeout=timeout), } except Exception as exc: status["llm_assessment"] = { "enabled": True, "status": "error", "generated_at": int(time.time()), "error": str(exc), "text": previous or "", } def write_status(status: dict[str, object], output: str) -> None: output_path = Path(output) output_path.parent.mkdir(parents=True, exist_ok=True) tmp_path = output_path.with_suffix(f"{output_path.suffix}.tmp") tmp_path.write_text(json.dumps(status, indent=2, sort_keys=True), encoding="utf-8") tmp_path.replace(output_path) def write_refreshing_status(output: str, *, cache_path: str | None = None, call_timeout_seconds: int = 0, poll_timeout_seconds: int = 0, runtime_values: dict[str, object] | None = None) -> None: output_path = Path(output) now = int(time.time()) try: current = json.loads(output_path.read_text(encoding="utf-8")) if output_path.exists() else {} except (json.JSONDecodeError, OSError): current = {} if not isinstance(current, dict): current = {} used_cache = False if cache_path: cached = StatusSnapshotStore(cache_path).load("last_good") if cached: current = cached used_cache = True previous_mcp = current.get("capabilities", {}).get("graylog_mcp", {}) if isinstance(current.get("capabilities"), dict) else {} previous_completed_at = previous_mcp.get("poll_completed_at", current.get("generated_at", 0)) if isinstance(previous_mcp, dict) else current.get("generated_at", 0) current.setdefault("status_schema", 2) current["generated_at"] = now current["stale"] = False current["stale_reason"] = "" if runtime_values is not None: current["configuration"] = _public_runtime_config(runtime_values) current["stream_coverage"] = _stream_coverage(runtime_values, {}, {"streams": []}, [], _stream_titles(runtime_values)) capabilities = current.setdefault("capabilities", {}) if isinstance(capabilities, dict): previous_events = previous_mcp.get("events_fetched", 0) if isinstance(previous_mcp, dict) else 0 previous_raw_events = 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 "" capabilities["graylog_mcp"] = { "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_events, "previous_raw_events_fetched": previous_raw_events, "previous_aggregate_events": previous_aggregate_events, "previous_fetch_mode": previous_fetch_mode, "previous_coverage_status": previous_coverage_status, "events_fetched": previous_events, "raw_events_fetched": previous_raw_events, "aggregate_events": previous_aggregate_events, "poll_started_at": now, "previous_poll_completed_at": previous_completed_at, "fetch_mode": previous_fetch_mode, "coverage_status": previous_coverage_status or "refreshing", "call_timeout_seconds": call_timeout_seconds or previous_mcp.get("call_timeout_seconds", 0), "poll_timeout_seconds": poll_timeout_seconds or previous_mcp.get("poll_timeout_seconds", 0), } current["status_cache"] = {"served_from_cache": used_cache, "reason": "refreshing"} write_status(current, output) def monitor_loop( log_path: str, output: str, *, policy_path: str | None = None, interval: int = 10, anomaly_limit: int = 20, llm: bool = False, llm_interval: int = 300, llm_model: str | None = None, llm_timeout: int | None = None, 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}", flush=True) print(f"Writing status to {output}", flush=True) last_llm_at = 0 last_llm_text: str | None = None while True: runtime = ConfigStore(config_path).read() if config_path and Path(config_path).exists() else {} effective_llm = bool(runtime.get("llm_enabled")) if runtime else llm effective_model = str(runtime.get("llm_model") or llm_model or "") mcp_call_timeout = max(1, int(runtime.get("graylog_mcp_call_timeout_seconds", 8) or 8)) mcp_poll_timeout = max(60, int(runtime.get("graylog_mcp_poll_timeout_seconds", 120) or 120)) if runtime.get("log_source") == "graylog_mcp": write_refreshing_status(output, cache_path=status_cache_path, call_timeout_seconds=mcp_call_timeout, poll_timeout_seconds=mcp_poll_timeout, runtime_values=runtime) try: status_timeout = mcp_poll_timeout + max(30, mcp_call_timeout * 2) with _cycle_timeout(status_timeout if runtime.get("log_source") == "graylog_mcp" else 0): 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, ) except Exception as exc: error_status = {"status": "error", "error": f"monitor_error: {exc}", "call_timeout_seconds": mcp_call_timeout, "poll_timeout_seconds": mcp_poll_timeout} status = cached_status_with_error(status_cache_path, error_status) if status_cache_path else None if not status: status = { "status_schema": 2, "generated_at": int(time.time()), "stale": True, "stale_reason": "monitor_error", "capabilities": {"graylog_mcp": error_status}, "summary": {"total": 0}, "status_cache": {"served_from_cache": False, "reason": "monitor_error"}, } 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: add_llm_assessment(status, previous=last_llm_text, model=effective_model or None, timeout=llm_timeout) assessment = status.get("llm_assessment", {}) if isinstance(assessment, dict): last_llm_text = str(assessment.get("text", "") or last_llm_text or "") last_llm_at = now else: status["llm_assessment"] = { "enabled": True, "status": "cached", "generated_at": last_llm_at, "text": last_llm_text or "", } else: status["llm_assessment"] = {"enabled": False, "status": "disabled", "text": ""} if status_cache_path and isinstance(mcp, dict) and mcp.get("status") not in {"error", "refreshing"}: StatusSnapshotStore(status_cache_path).save("last_good", status) write_status(status, output) time.sleep(interval)