diff --git a/README.md b/README.md index b4cbab6..d555b12 100644 --- a/README.md +++ b/README.md @@ -230,7 +230,9 @@ Graylog fetch mode controls how high-volume streams are read: 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. +raw log line through MCP each poll. In aggregate mode the raw sample is capped at +10000 events per stream to stay within Graylog's default result-window limit; +aggregate counts are used for volume above that. ## Monitoring Export diff --git a/src/fgai/graylog_aggregate.py b/src/fgai/graylog_aggregate.py index d617eff..fba4806 100644 --- a/src/fgai/graylog_aggregate.py +++ b/src/fgai/graylog_aggregate.py @@ -81,7 +81,7 @@ class GraylogAggregateSource: } schema_variants: list[dict[str, object]] = [] metric_keys = [key for key in ("metrics", "series") if key in properties] - group_keys = [key for key in ("group_by", "groups", "fields") if key in properties] + group_keys = [key for key in ("groupings", "group_by", "groups", "fields") if key in properties] if properties and metric_keys: for metric_key in metric_keys: for metric_value in (["count()"], ["count"], [{"function": "count"}]): @@ -92,6 +92,27 @@ class GraylogAggregateSource: if properties: schema_variants.append(_filter_supported(base, properties)) fallback_variants: list[dict[str, object]] = [ + { + "query": self.query, + "streams": [self.stream] if self.stream else [], + "range_seconds": max(1, int(range_seconds)), + "groupings": [], + "metrics": ["count()"], + }, + { + "query": self.query, + "streams": [self.stream] if self.stream else [], + "range_seconds": max(1, int(range_seconds)), + "groupings": [], + "metrics": ["count"], + }, + { + "query": self.query, + "streams": [self.stream] if self.stream else [], + "range_seconds": max(1, int(range_seconds)), + "groupings": [], + "metrics": [{"function": "count"}], + }, { "query": self.query, "streams": [self.stream] if self.stream else [], diff --git a/src/fgai/monitor.py b/src/fgai/monitor.py index 20a660f..2f70a4e 100644 --- a/src/fgai/monitor.py +++ b/src/fgai/monitor.py @@ -156,7 +156,7 @@ def build_status( 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 + raw_limit = min(raw_sample_events, 10_000) 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, **aggregate_status, **stream_status, "raw_sample_limit": raw_limit}) diff --git a/tests/test_graylog_aggregate.py b/tests/test_graylog_aggregate.py index 1b15508..aab99d4 100644 --- a/tests/test_graylog_aggregate.py +++ b/tests/test_graylog_aggregate.py @@ -46,7 +46,7 @@ class GraylogAggregateTests(unittest.TestCase): self.assertEqual(len(client.arguments), 2) def test_uses_tool_schema_to_avoid_unsupported_fields(self): - schema = {"properties": {"query": {}, "streams": {}, "range_seconds": {}, "series": {}}} + schema = {"properties": {"query": {}, "streams": {}, "range_seconds": {}, "metrics": {}, "groupings": {}}} client = _AggregateClient([ {"result": {"content": [{"type": "text", "text": '{"schema":[{"name":"count"}],"datarows":[[7]]}'}]}} ], schema=schema) @@ -54,7 +54,8 @@ class GraylogAggregateTests(unittest.TestCase): status = GraylogAggregateSource(client, "firewall").fetch_count() self.assertEqual(status["aggregate_events"], 7) - self.assertIn("series", client.arguments[0]) + self.assertIn("metrics", client.arguments[0]) + self.assertIn("groupings", client.arguments[0]) self.assertNotIn("group_by", client.arguments[0])