diff --git a/README.md b/README.md index ec6c797..6191b9f 100644 --- a/README.md +++ b/README.md @@ -131,6 +131,14 @@ Baseline storage settings are in `Settings`: - `Max values per entity field`: cap for distinct values per stream/entity/field. Lower this if `profile_values` grows too fast. +Retention and per-field value capping already run automatically every poll +cycle — nothing above needs a manual trigger. `Diagnostics -> Baseline +Maintenance` shows the result: current DB size, reclaimable bytes, rows +deleted last cycle, and specifically which `(stream, entity, field)` rows +got trimmed for exceeding the max-values-per-field cap, so you can tell +whether a field is worth excluding from its profile instead of just raising +the cap. Only disk-space reclamation (`VACUUM`) needs a manual, offline step: + For a large existing baseline, stop the service before compacting: ```bash diff --git a/ROADMAP.md b/ROADMAP.md index cb5d5a1..313a13a 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -105,7 +105,18 @@ Goal: make findings more accurate before adding more integrations. state (`BaselineStore.detector_readiness()`, mirroring the existing per-field `profile_readiness()`) and a plain-language "what triggers this" reference — all surfaced in a new Detector Health dashboard panel. -- [ ] Add baseline maintenance tooling in the dashboard for retention, high-cardinality fields, and database compaction status. +- [x] Add baseline maintenance tooling in the dashboard for retention, + high-cardinality fields, and database compaction status. Retention itself + already ran automatically every poll cycle; the gap was pure visibility — + the dashboard showed only a single "DB size" line, with no way to see + what got deleted, which fields are hitting the per-field value cap, or + whether disk space is reclaimable without SSHing in. New Baseline + Maintenance panel shows size/reclaimable bytes, retention settings, rows + deleted last cycle, and the specific `(stream, entity, field)` rows that + got trimmed for exceeding the max-values-per-field cap. No live-vacuum + button added deliberately — vacuum needs the service stopped first (the + existing How To guidance already says so), so the dashboard only surfaces + when it's worth doing and points at the offline CLI command. - [x] Add observed-field profile recommendations so streams can get suggested entity, time, categorical, numeric, and detector settings. - [x] Add optional Ollama profile advisor with validated JSON output for semantic field mapping. - [x] Let the dashboard's Ollama analyst note reason about a small, bounded diff --git a/src/fgai/baseline.py b/src/fgai/baseline.py index a8c2049..0317f8a 100644 --- a/src/fgai/baseline.py +++ b/src/fgai/baseline.py @@ -598,10 +598,11 @@ class BaselineStore: deleted["seen_events"] = connection.execute("delete from seen_events where first_seen > 0 and first_seen < ?", (bucket_cutoff,)).rowcount deleted["profile_seen_events"] = connection.execute("delete from profile_seen_events where first_seen > 0 and first_seen < ?", (bucket_cutoff,)).rowcount deleted["profile_values_stale_low_count"] = connection.execute("delete from profile_values where last_seen > 0 and last_seen < ? and seen_count <= 1", (value_cutoff,)).rowcount + high_cardinality_fields: list[dict[str, object]] = [] if max_values_per_field > 0: rows = connection.execute("select stream_id, entity, field, count(*) from profile_values group by stream_id, entity, field having count(*) > ?", (max_values_per_field,)).fetchall() trimmed = 0 - for stream_id, entity, field, _count in rows: + for stream_id, entity, field, count in rows: keep = { row[0] for row in connection.execute( @@ -617,13 +618,15 @@ class BaselineStore: f"delete from profile_values where stream_id=? and entity=? and field=? and value not in ({placeholders})", (stream_id, entity, field, *keep), ).rowcount + high_cardinality_fields.append({"stream_id": stream_id, "entity": entity, "field": field, "distinct_values": int(count), "trimmed_to": max_values_per_field}) deleted["profile_values_capped"] = trimmed + high_cardinality_fields.sort(key=lambda item: int(item["distinct_values"]), reverse=True) connection.execute("pragma optimize") if vacuum: with self._connect() as connection: connection.execute("vacuum") stats = self.stats(include_rows=include_rows) - return {"retention_days": retention_days, "value_retention_days": value_retention_days, "max_values_per_field": max_values_per_field, "vacuum": vacuum, "deleted": deleted, **stats} + return {"retention_days": retention_days, "value_retention_days": value_retention_days, "max_values_per_field": max_values_per_field, "vacuum": vacuum, "deleted": deleted, "high_cardinality_fields": high_cardinality_fields[:20], **stats} def stats(self, *, include_rows: bool = False) -> dict[str, object]: with self._connect() as connection: diff --git a/src/fgai/dashboard.py b/src/fgai/dashboard.py index 422314e..eba8e3b 100644 --- a/src/fgai/dashboard.py +++ b/src/fgai/dashboard.py @@ -1064,6 +1064,10 @@ async function refresh() { const quality = data.data_quality || {}; 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})); + const maintenance = (data.baseline || {}).maintenance || {}; + const maintenanceDeleted = maintenance.deleted || {}; + const maintenanceRowsDeleted = Object.values(maintenanceDeleted).reduce((sum, n) => sum + (n || 0), 0); + const maintenanceHighCardinality = maintenance.high_cardinality_fields || []; document.getElementById('diagnostics').innerHTML = collapsibleSection('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')) + collapsibleSection('shared-entity-baselines', 'Shared Entity Baselines', (data.shared_entity_baselines || []).length ? table(data.shared_entity_baselines || [], [{label:'Entity concept', key:'group'}, {label:'Shared across streams', render:r => esc((r.stream_names || r.streams || []).join(', '))}], 'shared-entity-baselines') : '

No entity concept is currently recognized across two or more enabled streams. Baselines stay per-stream until that happens.

') + @@ -1071,6 +1075,21 @@ async function refresh() { collapsibleSection('entities', 'Entities', table(context.source_profiles || [], [{label:'Entity', key:'entity', render:r => entityCell(r)}, {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')) + collapsibleSection('profile-readiness', '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'}, {label:'Action', render:r => r.stream_id ? `` : '-'}], 'profile-readiness')) + collapsibleSection('detector-health', 'Detector Health', table(data.detector_health || [], [{label:'Detector', key:'detector'}, {label:'Description', key:'description'}, {label:'Learning state', render:r => r.learning_state === 'no_data' ? 'no data yet' : r.learning_state === 'ready' ? `ready (${r.streams_ready}/${r.streams_configured} streams)` : `learning (${r.streams_ready}/${r.streams_configured} streams)`}, {label:'Precision (30d)', render:r => r.precision === null || r.precision === undefined ? 'n/a' : `${Math.round(r.precision * 100)}%`}, {label:'Confirmed', key:'confirmed'}, {label:'False positive', key:'false_positive'}, {label:'Expected', key:'expected'}], 'detector-health')) + + collapsibleSection('baseline-maintenance', 'Baseline Maintenance', table([{ + size: bytes(maintenance.size_bytes || 0), + reclaimable: bytes(maintenance.reclaimable_bytes || 0), + retention_days: maintenance.retention_days ?? '-', + value_retention_days: maintenance.value_retention_days ?? '-', + max_values_per_field: maintenance.max_values_per_field ?? '-', + rows_deleted: maintenanceRowsDeleted, + }], [ + {label:'DB size', key:'size'}, + {label:'Reclaimable (needs VACUUM)', key:'reclaimable'}, + {label:'Bucket retention (days)', key:'retention_days'}, + {label:'Value retention (days)', key:'value_retention_days'}, + {label:'Max values per field', key:'max_values_per_field'}, + {label:'Rows deleted last cycle', key:'rows_deleted'}, + ], 'baseline-maintenance-summary') + (maintenanceHighCardinality.length ? `

Fields below hit the max-values-per-field cap and had their least-recently-seen values trimmed this cycle. Consider excluding a consistently-listed field from its profile, or raising Max values per entity field in Settings.

` + table(maintenanceHighCardinality, [{label:'Stream', key:'stream_id'}, {label:'Entity', key:'entity'}, {label:'Field', key:'field'}, {label:'Distinct values', key:'distinct_values'}, {label:'Trimmed to', key:'trimmed_to'}], 'baseline-high-cardinality') : '

No field is currently over the max-values-per-field cap.

') + (maintenance.reclaimable_bytes > 0 ? `

Reclaiming disk space needs an offline VACUUM: stop the service, then run signalscope baseline-maintenance --vacuum.

` : '')) + collapsibleSection('data-quality', 'Data Quality', table([quality], [{label:'Events', key:'events'}, {label:'Timestamp coverage', render:r => `${r.timestamp_coverage || 0}%`}, {label:'Source coverage', render:r => `${r.source_coverage || 0}%`}, {label:'Truncated streams', render:r => esc((r.truncated_streams || []).join(', ') || 'none')}], 'data-quality')) + collapsibleSection('feedback-rules', 'Feedback Rules', table( ['value', 'field', 'entity', 'stream'].map(scope => ({scope, count: feedbackRules.filter(item => (item.scope || 'value') === scope).length})), diff --git a/tests/test_baseline.py b/tests/test_baseline.py index a9f18e8..5ca9682 100644 --- a/tests/test_baseline.py +++ b/tests/test_baseline.py @@ -112,6 +112,35 @@ class BaselineTests(unittest.TestCase): self.assertIn("rows", result) self.assertIn("reclaimable_bytes", result) + def test_maintenance_reports_high_cardinality_fields_it_trims(self): + with tempfile.TemporaryDirectory() as directory: + store = BaselineStore(str(Path(directory) / "baseline.sqlite3")) + profiles = parse_profiles([{"stream_id": "dns", "entity_field": "srcip", "categorical_fields": ["query_domain"]}]) + events = [ + parse_log_line(f"fgai_stream_id=dns srcip=10.0.0.5 query_domain=value-{index}.example event={index}") + for index in range(5) + ] + store.ingest_profile_fields(events, profiles, observed_at=int(time.time()), max_values_per_field=0) + + result = store.maintenance(retention_days=7, value_retention_days=7, max_values_per_field=2) + + self.assertEqual(result["deleted"]["profile_values_capped"], 3) + row = next(item for item in result["high_cardinality_fields"] if item["field"] == "query_domain") + self.assertEqual(row["stream_id"], "dns") + self.assertEqual(row["entity"], "10.0.0.5") + self.assertEqual(row["distinct_values"], 5) + self.assertEqual(row["trimmed_to"], 2) + + def test_maintenance_reports_no_high_cardinality_fields_when_under_cap(self): + with tempfile.TemporaryDirectory() as directory: + store = BaselineStore(str(Path(directory) / "baseline.sqlite3")) + profiles = parse_profiles([{"stream_id": "dns", "entity_field": "srcip", "categorical_fields": ["query_domain"]}]) + store.ingest_profile_fields([parse_log_line("fgai_stream_id=dns srcip=10.0.0.5 query_domain=only.example")], profiles, observed_at=int(time.time())) + + result = store.maintenance(retention_days=7, value_retention_days=7, max_values_per_field=500) + + self.assertEqual(result["high_cardinality_fields"], []) + def test_profile_value_storage_skips_noisy_high_cardinality_fields(self): with tempfile.TemporaryDirectory() as directory: store = BaselineStore(str(Path(directory) / "baseline.sqlite3")) diff --git a/tests/test_dashboard.py b/tests/test_dashboard.py index 2e1eb73..858c07e 100644 --- a/tests/test_dashboard.py +++ b/tests/test_dashboard.py @@ -109,6 +109,12 @@ class DashboardTests(unittest.TestCase): self.assertIn("Detector Health", HTML) self.assertIn("detector_health", HTML) + def test_dashboard_shows_baseline_maintenance(self): + self.assertIn("baseline-maintenance", HTML) + self.assertIn("Baseline Maintenance", HTML) + self.assertIn("high_cardinality_fields", HTML) + self.assertIn("baseline-maintenance --vacuum", HTML) + def test_review_actions_carry_detector_for_precision_tracking(self): self.assertIn('data-detector="${esc(r.detector', HTML) self.assertIn("detector: button.dataset.detector", HTML)