From 286b532aa7b12b4910f872d0330fc74e9887785f Mon Sep 17 00:00:00 2001 From: flamboh Date: Tue, 24 Mar 2026 09:49:27 -0700 Subject: [PATCH 1/6] fix(pipeline): align stale day detection with local midnight --- tests/python/test_discovery.py | 29 +++++++++++++++++++++++++++++ tools/netflow-db/discovery.py | 7 ++++++- 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/tests/python/test_discovery.py b/tests/python/test_discovery.py index 6050aa3..01fdc8c 100644 --- a/tests/python/test_discovery.py +++ b/tests/python/test_discovery.py @@ -111,3 +111,32 @@ def test_scan_filesystem_skips_unparseable_and_pre_start_files( rows = list(discovery.scan_filesystem()) assert rows == [] + + +def test_get_stale_days_uses_local_day_boundaries() -> None: + common, discovery = load_modules() + conn = sqlite3.connect(':memory:') + common.init_processed_files_table(conn) + + same_local_day = [ + ( + '/captures/r1/2025/03/05/nfcapd.202503050045', + 'r1', + common.timestamp_to_unix(datetime(2025, 3, 5, 0, 45)), + 1, + ), + ( + '/captures/r1/2025/03/05/nfcapd.202503052355', + 'r1', + common.timestamp_to_unix(datetime(2025, 3, 5, 23, 55)), + None, + ), + ] + conn.executemany( + 'INSERT INTO processed_files (file_path, router, timestamp, flow_stats_status) VALUES (?, ?, ?, ?)', + same_local_day, + ) + + assert discovery.get_stale_days(conn, 'flow_stats') == { + ('r1', common.timestamp_to_unix(datetime(2025, 3, 5, 0, 0))) + } diff --git a/tools/netflow-db/discovery.py b/tools/netflow-db/discovery.py index 5629418..3b6a7c8 100644 --- a/tools/netflow-db/discovery.py +++ b/tools/netflow-db/discovery.py @@ -503,7 +503,12 @@ def get_stale_days( WITH day_status AS ( SELECT router, - (timestamp / 86400) * 86400 AS day_start, + CAST( + strftime( + '%s', + datetime(timestamp, 'unixepoch', 'localtime', 'start of day') + ) AS INTEGER + ) AS day_start, MAX(CASE WHEN {status_column} = 1 THEN 1 ELSE 0 END) AS has_processed, MAX(CASE WHEN {status_column} IS NULL THEN 1 ELSE 0 END) AS has_pending FROM processed_files From 0cffde31bf9d4b15727c218bcbc14a0f25ac6f5a Mon Sep 17 00:00:00 2001 From: flamboh Date: Tue, 24 Mar 2026 10:04:43 -0700 Subject: [PATCH 2/6] fix(web): fallback sqlite runtime and align bucket epochs --- apps/web/src/lib/server/datasets.ts | 46 +++++++++++- .../src/routes/api/netflow/stats/+server.ts | 2 +- apps/web/tests/lib/server/datasets.test.ts | 75 ++++++++++++++----- .../tests/routes/api-netflow-stats.test.ts | 4 +- 4 files changed, 104 insertions(+), 23 deletions(-) diff --git a/apps/web/src/lib/server/datasets.ts b/apps/web/src/lib/server/datasets.ts index 18a672d..bc7ad56 100644 --- a/apps/web/src/lib/server/datasets.ts +++ b/apps/web/src/lib/server/datasets.ts @@ -1,6 +1,8 @@ import fs from 'fs'; import path from 'path'; import Database from 'better-sqlite3'; +import { DatabaseSync } from 'node:sqlite'; +import type { SQLInputValue } from 'node:sqlite'; import type { DatasetSummary } from '$lib/types/types'; import { getDatasetsConfigPath, getRepoRoot } from '$lib/server/paths'; @@ -17,9 +19,20 @@ export interface DatasetConfig { const repoRoot = getRepoRoot(); const defaultRegistryPath = getDatasetsConfigPath(); + +type PreparedStatement = { + get(...params: unknown[]): unknown; + all(...params: unknown[]): unknown[]; +}; + +export interface ReadonlyDatasetDb { + prepare(sql: string): PreparedStatement; + close(): void; +} + const datasetDbCache = new Map< string, - { db: Database.Database; dbPath: string; mtimeMs: number } + { db: ReadonlyDatasetDb; dbPath: string; mtimeMs: number } >(); const datasetDefaultStartCache = new Map< string, @@ -197,7 +210,23 @@ export function getDatasetDbPath(datasetId: string): string { return getDatasetConfig(datasetId).db_path; } -export function getDatasetDb(datasetId: string): Database.Database { +function openNodeSqliteDatabase(dbPath: string): ReadonlyDatasetDb { + const db = new DatabaseSync(dbPath, { open: true, readOnly: true }); + return { + prepare(sql: string): PreparedStatement { + const stmt = db.prepare(sql); + return { + get: (...params: unknown[]) => stmt.get(...(params as SQLInputValue[])), + all: (...params: unknown[]) => stmt.all(...(params as SQLInputValue[])) + }; + }, + close() { + db.close(); + } + }; +} + +export function getDatasetDb(datasetId: string): ReadonlyDatasetDb { const dbPath = getDatasetDbPath(datasetId); if (!fs.existsSync(dbPath)) { throw new Error(`Dataset database not found for '${datasetId}' at ${dbPath}`); @@ -217,7 +246,18 @@ export function getDatasetDb(datasetId: string): Database.Database { } } - const db = new Database(dbPath, { readonly: true }); + let db: ReadonlyDatasetDb; + try { + db = new Database(dbPath, { readonly: true }); + } catch (error) { + if (error instanceof Error && 'code' in error && error.code === 'ERR_DLOPEN_FAILED') { + console.warn(`better-sqlite3 failed to load for '${datasetId}', falling back to node:sqlite`); + db = openNodeSqliteDatabase(dbPath); + } else { + throw error; + } + } + datasetDbCache.set(datasetId, { db, dbPath, mtimeMs: stat.mtimeMs }); return db; } diff --git a/apps/web/src/routes/api/netflow/stats/+server.ts b/apps/web/src/routes/api/netflow/stats/+server.ts index 1eeb005..5db7c5e 100644 --- a/apps/web/src/routes/api/netflow/stats/+server.ts +++ b/apps/web/src/routes/api/netflow/stats/+server.ts @@ -17,7 +17,7 @@ const BUCKET_SIZES: Record = { */ function getBucketStartQuery(groupBy: string): string { const bucketSize = BUCKET_SIZES[groupBy] ?? BUCKET_SIZES.date; - return `(CAST(strftime('%s', datetime(timestamp, 'unixepoch', 'localtime')) AS integer) / ${bucketSize}) * ${bucketSize}`; + return `(CAST(strftime('%s', datetime(timestamp, 'unixepoch', 'localtime', 'start of day', 'utc', printf('+%d seconds', ((CAST(strftime('%s', datetime(timestamp, 'unixepoch', 'localtime')) AS integer) - CAST(strftime('%s', datetime(timestamp, 'unixepoch', 'localtime', 'start of day')) AS integer)) / ${bucketSize}) * ${bucketSize}))) AS integer))`; } export const GET: RequestHandler = async ({ url }) => { diff --git a/apps/web/tests/lib/server/datasets.test.ts b/apps/web/tests/lib/server/datasets.test.ts index e490594..a015465 100644 --- a/apps/web/tests/lib/server/datasets.test.ts +++ b/apps/web/tests/lib/server/datasets.test.ts @@ -4,26 +4,25 @@ import path from 'path'; import { spawnSync } from 'child_process'; import { afterEach, describe, expect, it, vi } from 'vitest'; -vi.mock('better-sqlite3', () => ({ - default: class MockDatabase { - constructor(private readonly dbPath: string) {} - - prepare(query: string) { - return { - get: () => { - const result = spawnSync('sqlite3', [this.dbPath, query], { encoding: 'utf-8' }); - if (result.status !== 0) { - throw new Error(result.stderr || 'sqlite3 query failed'); - } - - const minTimestamp = Number(result.stdout.trim()); - return { minTimestamp: Number.isFinite(minTimestamp) ? minTimestamp : null }; +const betterSqlite3Factory = vi.fn((dbPath: string) => ({ + prepare(query: string) { + return { + get: () => { + const result = spawnSync('sqlite3', [dbPath, query], { encoding: 'utf-8' }); + if (result.status !== 0) { + throw new Error(result.stderr || 'sqlite3 query failed'); } - }; - } - close() {} - } + const minTimestamp = Number(result.stdout.trim()); + return { minTimestamp: Number.isFinite(minTimestamp) ? minTimestamp : null }; + } + }; + }, + close() {} +})); + +vi.mock('better-sqlite3', () => ({ + default: vi.fn().mockImplementation((dbPath: string) => betterSqlite3Factory(dbPath)) })); async function loadDatasetsModule() { @@ -34,6 +33,7 @@ async function loadDatasetsModule() { describe('dataset server helpers', () => { afterEach(() => { vi.unstubAllEnvs(); + betterSqlite3Factory.mockClear(); }); it('lists dataset summaries from registry + sqlite min timestamp', async () => { @@ -115,4 +115,43 @@ describe('dataset server helpers', () => { expect(datasets.listDatasetSources('alpha')).toEqual(['r1', 'r2']); expect(() => datasets.getDatasetConfig('missing')).toThrow(/Unknown dataset 'missing'/); }); + + it('falls back to node:sqlite when better-sqlite3 fails to load', async () => { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'datasets-test-')); + const dbPath = path.join(tempDir, 'netflow.sqlite'); + const registryPath = path.join(tempDir, 'datasets.json'); + + const seedResult = spawnSync( + 'sqlite3', + [ + dbPath, + 'CREATE TABLE netflow_stats (timestamp INTEGER NOT NULL); INSERT INTO netflow_stats (timestamp) VALUES (1740823200);' + ], + { encoding: 'utf-8' } + ); + expect(seedResult.status).toBe(0); + + fs.writeFileSync( + registryPath, + JSON.stringify([ + { + dataset_id: 'alpha', + label: 'Alpha Label', + root_path: tempDir, + db_path: dbPath + } + ]) + ); + + betterSqlite3Factory.mockImplementationOnce(() => { + const error = new Error('Module did not self-register') as Error & { code: string }; + error.code = 'ERR_DLOPEN_FAILED'; + throw error; + }); + + vi.stubEnv('DATASETS_CONFIG_PATH', registryPath); + + const datasets = await loadDatasetsModule(); + expect(datasets.getDatasetDefaultStartDate('alpha')).toBe('2025-03-01'); + }); }); diff --git a/apps/web/tests/routes/api-netflow-stats.test.ts b/apps/web/tests/routes/api-netflow-stats.test.ts index a52af00..820c3f8 100644 --- a/apps/web/tests/routes/api-netflow-stats.test.ts +++ b/apps/web/tests/routes/api-netflow-stats.test.ts @@ -40,9 +40,10 @@ describe('/api/netflow/stats GET', () => { bytesOther: 14 } ]); + const prepare = vi.fn().mockReturnValue({ all }); vi.mocked(getRequestedDataset).mockReturnValue('alpha'); vi.mocked(getDatasetDb).mockReturnValue({ - prepare: vi.fn().mockReturnValue({ all }) + prepare } as never); const response = await GET({ @@ -75,6 +76,7 @@ describe('/api/netflow/stats GET', () => { ] }); expect(all).toHaveBeenCalledWith('r1', 'r2', '1', '2'); + expect(prepare).toHaveBeenCalledWith(expect.stringContaining("'start of day', 'utc'")); }); it('returns 500 when the database query fails', async () => { From f45ff21106910deef5c67cb58e8b5abfea73f4ab Mon Sep 17 00:00:00 2001 From: flamboh Date: Tue, 24 Mar 2026 10:23:02 -0700 Subject: [PATCH 3/6] fix(pipeline): dedupe mirrored netflow paths --- tests/python/test_discovery.py | 37 +++++ tests/python/test_flow_db.py | 37 +++++ tools/netflow-db/common.py | 11 +- tools/netflow-db/discovery.py | 91 ++++++++--- tools/netflow-db/flow_db.py | 13 ++ .../repair_mirrored_root_duplicates.py | 149 ++++++++++++++++++ 6 files changed, 318 insertions(+), 20 deletions(-) create mode 100644 tools/netflow-db/repair_mirrored_root_duplicates.py diff --git a/tests/python/test_discovery.py b/tests/python/test_discovery.py index 01fdc8c..7cfae4d 100644 --- a/tests/python/test_discovery.py +++ b/tests/python/test_discovery.py @@ -140,3 +140,40 @@ def test_get_stale_days_uses_local_day_boundaries() -> None: assert discovery.get_stale_days(conn, 'flow_stats') == { ('r1', common.timestamp_to_unix(datetime(2025, 3, 5, 0, 0))) } + + +def test_sync_processed_files_table_updates_mirrored_path_without_duplication( + monkeypatch: pytest.MonkeyPatch, +) -> None: + common, discovery = load_modules() + conn = sqlite3.connect(':memory:') + common.init_processed_files_table(conn) + + ts = common.timestamp_to_unix(datetime(2025, 3, 2, 0, 0)) + conn.execute( + 'INSERT INTO processed_files (file_path, router, timestamp, file_exists) VALUES (?, ?, ?, ?)', + ('/old-root/r1/2025/03/02/nfcapd.202503020000', 'r1', ts, 1), + ) + + monkeypatch.setattr(discovery, 'AVAILABLE_ROUTERS', ['r1']) + monkeypatch.setattr(discovery, 'DATA_START_DATE', datetime(2025, 3, 1, 0, 0)) + monkeypatch.setattr( + discovery, + 'scan_filesystem', + lambda discovery_window_days=0: iter( + [('/new-root/r1/2025/03/02/nfcapd.202503020000', 'r1', datetime(2025, 3, 2, 0, 0))] + ), + ) + + stats = discovery.sync_processed_files_table( + conn, + include_gaps=False, + reprocess_window_days=0, + discovery_window_days=0, + ) + + row = conn.execute( + 'SELECT file_path, router, timestamp, file_exists FROM processed_files' + ).fetchone() + assert stats == {'discovered': 1, 'new_files': 0, 'gaps': 0} + assert row == ('/new-root/r1/2025/03/02/nfcapd.202503020000', 'r1', ts, 1) diff --git a/tests/python/test_flow_db.py b/tests/python/test_flow_db.py index 7b2bf8f..d2b2d6b 100644 --- a/tests/python/test_flow_db.py +++ b/tests/python/test_flow_db.py @@ -56,3 +56,40 @@ def test_batch_insert_results_inserts_successful_rows() -> None: ).fetchone() assert inserted == 1 assert row == ('/tmp/a', 'r1', 3, 4, 5, 0) + + +def test_batch_insert_results_replaces_mirrored_path_duplicate() -> None: + _, flow_db = load_modules() + conn = sqlite3.connect(':memory:') + flow_db.init_netflow_stats_table(conn) + conn.execute( + """ + INSERT INTO netflow_stats ( + file_path, router, timestamp, + flows, flows_tcp, flows_udp, flows_icmp, flows_other, + packets, packets_tcp, packets_udp, packets_icmp, packets_other, + bytes, bytes_tcp, bytes_udp, bytes_icmp, bytes_other, + first_timestamp, last_timestamp, msec_first, msec_last, sequence_failures + ) VALUES (?, ?, ?, ?, 0, 0, 0, 0, ?, 0, 0, 0, 0, ?, 0, 0, 0, 0, 0, 0, 0, 0, 0) + """, + ('/old-root/a', 'r1', 123, 3, 4, 5), + ) + + inserted = flow_db.batch_insert_results( + conn, + [ + { + 'file_path': '/new-root/a', + 'router': 'r1', + 'timestamp': 123, + 'success': True, + 'data': {'flows': 7, 'packets': 8, 'bytes': 9}, + } + ], + ) + + rows = conn.execute( + 'SELECT file_path, router, timestamp, flows, packets, bytes FROM netflow_stats' + ).fetchall() + assert inserted == 1 + assert rows == [('/new-root/a', 'r1', 123, 7, 8, 9)] diff --git a/tools/netflow-db/common.py b/tools/netflow-db/common.py index 27daf40..fb0c7be 100644 --- a/tools/netflow-db/common.py +++ b/tools/netflow-db/common.py @@ -441,7 +441,16 @@ def init_processed_files_table(conn: sqlite3.Connection) -> None: CREATE INDEX IF NOT EXISTS idx_processed_files_router_timestamp ON processed_files(router, timestamp) """) - + + try: + cursor.execute(""" + CREATE UNIQUE INDEX IF NOT EXISTS idx_processed_files_router_timestamp_unique + ON processed_files(router, timestamp) + """) + except sqlite3.IntegrityError: + print("Warning: processed_files has duplicate router/timestamp rows; " + "skipping unique index creation until repaired") + cursor.execute(""" CREATE INDEX IF NOT EXISTS idx_processed_files_pending ON processed_files(processed_at) diff --git a/tools/netflow-db/discovery.py b/tools/netflow-db/discovery.py index 3b6a7c8..f1a54c9 100644 --- a/tools/netflow-db/discovery.py +++ b/tools/netflow-db/discovery.py @@ -215,16 +215,60 @@ def sync_processed_files_table( print("Scanning filesystem for NetFlow files...") for file_path, router, timestamp in scan_filesystem(discovery_window_days): stats['discovered'] += 1 - - # Insert or ignore if already exists - cursor.execute(""" - INSERT OR IGNORE INTO processed_files - (file_path, router, timestamp, file_exists, discovered_at) - VALUES (?, ?, ?, 1, CURRENT_TIMESTAMP) - """, (file_path, router, timestamp_to_unix(timestamp))) - - if cursor.rowcount > 0: - stats['new_files'] += 1 + + timestamp_unix = timestamp_to_unix(timestamp) + existing = cursor.execute(""" + SELECT file_path, file_exists + FROM processed_files + WHERE router = ? AND timestamp = ? + LIMIT 1 + """, (router, timestamp_unix)).fetchone() + + if existing is None: + cursor.execute(""" + INSERT OR IGNORE INTO processed_files + (file_path, router, timestamp, file_exists, discovered_at) + VALUES (?, ?, ?, 1, CURRENT_TIMESTAMP) + """, (file_path, router, timestamp_unix)) + if cursor.rowcount > 0: + stats['new_files'] += 1 + else: + existing_path, existing_file_exists = existing + if existing_path != file_path or existing_file_exists == 0: + cursor.execute(""" + UPDATE processed_files + SET file_path = ?, + file_exists = 1, + discovered_at = CASE + WHEN file_exists = 0 THEN CURRENT_TIMESTAMP + ELSE discovered_at + END, + processed_at = CASE + WHEN file_exists = 0 THEN NULL + ELSE processed_at + END, + flow_stats_status = CASE + WHEN file_exists = 0 THEN NULL + ELSE flow_stats_status + END, + ip_stats_status = CASE + WHEN file_exists = 0 THEN NULL + ELSE ip_stats_status + END, + protocol_stats_status = CASE + WHEN file_exists = 0 THEN NULL + ELSE protocol_stats_status + END, + spectrum_stats_status = CASE + WHEN file_exists = 0 THEN NULL + ELSE spectrum_stats_status + END, + structure_stats_status = CASE + WHEN file_exists = 0 THEN NULL + ELSE structure_stats_status + END + WHERE router = ? AND timestamp = ? + """, (file_path, router, timestamp_unix)) # Commit periodically if stats['discovered'] % 1000 == 0: @@ -267,15 +311,24 @@ def sync_processed_files_table( for gap_timestamp in gaps: # Construct the expected file path for this gap gap_path = construct_file_path(router, gap_timestamp) - - cursor.execute(""" - INSERT OR IGNORE INTO processed_files - (file_path, router, timestamp, file_exists, discovered_at) - VALUES (?, ?, ?, 0, CURRENT_TIMESTAMP) - """, (gap_path, router, timestamp_to_unix(gap_timestamp))) - - if cursor.rowcount > 0: - stats['gaps'] += 1 + gap_timestamp_unix = timestamp_to_unix(gap_timestamp) + + existing = cursor.execute(""" + SELECT 1 + FROM processed_files + WHERE router = ? AND timestamp = ? + LIMIT 1 + """, (router, gap_timestamp_unix)).fetchone() + + if existing is None: + cursor.execute(""" + INSERT OR IGNORE INTO processed_files + (file_path, router, timestamp, file_exists, discovered_at) + VALUES (?, ?, ?, 0, CURRENT_TIMESTAMP) + """, (gap_path, router, gap_timestamp_unix)) + + if cursor.rowcount > 0: + stats['gaps'] += 1 print(f" Router {router}: {len(gaps)} gaps identified, {stats['gaps']} new gap entries") diff --git a/tools/netflow-db/flow_db.py b/tools/netflow-db/flow_db.py index 4bb403e..8bfbaa2 100644 --- a/tools/netflow-db/flow_db.py +++ b/tools/netflow-db/flow_db.py @@ -73,6 +73,15 @@ def init_netflow_stats_table(conn: sqlite3.Connection) -> None: CREATE INDEX IF NOT EXISTS idx_router_timestamp ON netflow_stats (router, timestamp) """) + try: + cursor.execute(""" + CREATE UNIQUE INDEX IF NOT EXISTS idx_netflow_router_timestamp_unique + ON netflow_stats (router, timestamp) + """) + except sqlite3.IntegrityError: + print("[flow_stats] Warning: netflow_stats has duplicate router/timestamp rows; " + "skipping unique index creation until repaired") + cursor.execute(""" CREATE INDEX IF NOT EXISTS idx_file_path ON netflow_stats (file_path) """) @@ -170,6 +179,10 @@ def batch_insert_results(conn: sqlite3.Connection, results: list[dict]) -> int: data = result['data'] try: + cursor.execute(""" + DELETE FROM netflow_stats + WHERE router = ? AND timestamp = ? AND file_path != ? + """, (result['router'], result['timestamp'], result['file_path'])) cursor.execute(""" INSERT OR REPLACE INTO netflow_stats ( file_path, router, timestamp, diff --git a/tools/netflow-db/repair_mirrored_root_duplicates.py b/tools/netflow-db/repair_mirrored_root_duplicates.py new file mode 100644 index 0000000..76fc350 --- /dev/null +++ b/tools/netflow-db/repair_mirrored_root_duplicates.py @@ -0,0 +1,149 @@ +#!/usr/bin/env python3 +""" +Repair mirrored-root duplicates in NetFlow SQLite databases. + +Keeps a preferred root for each logical capture slot (router + timestamp), +deletes duplicate rows from netflow_stats and processed_files for the other root, +and optionally adds unique indexes on (router, timestamp). + +Example: + python tools/netflow-db/repair_mirrored_root_duplicates.py \ + --db data/uoregon/netflow.sqlite \ + --preferred-prefix /research/obo/netflow_datasets/uoregon/ \ + --remove-prefix /research/tango_cis/uonet-in/ \ + --start-date 2026-03-02 +""" + +import argparse +import sqlite3 +from pathlib import Path + + +def build_prefix_pattern(prefix: str) -> str: + return prefix.rstrip('/') + '/%' + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--db', required=True, help='Path to SQLite database') + parser.add_argument('--preferred-prefix', required=True, help='Path prefix to keep') + parser.add_argument('--remove-prefix', required=True, help='Path prefix to remove') + parser.add_argument('--start-date', required=True, help='Inclusive local date YYYY-MM-DD') + parser.add_argument('--end-date', default=None, help='Inclusive local date YYYY-MM-DD') + parser.add_argument('--dry-run', action='store_true', help='Report changes without writing') + args = parser.parse_args() + + db_path = Path(args.db).expanduser().resolve() + preferred_pattern = build_prefix_pattern(args.preferred_prefix) + remove_pattern = build_prefix_pattern(args.remove_prefix) + end_date = args.end_date or '9999-12-31' + + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + + duplicate_slots = conn.execute( + """ + SELECT router, timestamp + FROM processed_files + WHERE date(datetime(timestamp, 'unixepoch', 'localtime')) BETWEEN ? AND ? + GROUP BY router, timestamp + HAVING + SUM(CASE WHEN file_path LIKE ? THEN 1 ELSE 0 END) > 0 + AND SUM(CASE WHEN file_path LIKE ? THEN 1 ELSE 0 END) > 0 + """, + (args.start_date, end_date, preferred_pattern, remove_pattern), + ).fetchall() + + duplicate_count = len(duplicate_slots) + print(f"Logical duplicate slots matched: {duplicate_count}") + if duplicate_count == 0: + return + + delete_netflow = conn.execute( + """ + SELECT COUNT(*) + FROM netflow_stats + WHERE date(datetime(timestamp, 'unixepoch', 'localtime')) BETWEEN ? AND ? + AND file_path LIKE ? + AND EXISTS ( + SELECT 1 + FROM netflow_stats preferred + WHERE preferred.router = netflow_stats.router + AND preferred.timestamp = netflow_stats.timestamp + AND preferred.file_path LIKE ? + ) + """, + (args.start_date, end_date, remove_pattern, preferred_pattern), + ).fetchone()[0] + delete_processed = conn.execute( + """ + SELECT COUNT(*) + FROM processed_files + WHERE date(datetime(timestamp, 'unixepoch', 'localtime')) BETWEEN ? AND ? + AND file_path LIKE ? + AND EXISTS ( + SELECT 1 + FROM processed_files preferred + WHERE preferred.router = processed_files.router + AND preferred.timestamp = processed_files.timestamp + AND preferred.file_path LIKE ? + ) + """, + (args.start_date, end_date, remove_pattern, preferred_pattern), + ).fetchone()[0] + + print(f"netflow_stats rows to delete: {delete_netflow}") + print(f"processed_files rows to delete: {delete_processed}") + + if args.dry_run: + return + + with conn: + conn.execute( + """ + DELETE FROM netflow_stats + WHERE date(datetime(timestamp, 'unixepoch', 'localtime')) BETWEEN ? AND ? + AND file_path LIKE ? + AND EXISTS ( + SELECT 1 + FROM netflow_stats preferred + WHERE preferred.router = netflow_stats.router + AND preferred.timestamp = netflow_stats.timestamp + AND preferred.file_path LIKE ? + ) + """, + (args.start_date, end_date, remove_pattern, preferred_pattern), + ) + conn.execute( + """ + DELETE FROM processed_files + WHERE date(datetime(timestamp, 'unixepoch', 'localtime')) BETWEEN ? AND ? + AND file_path LIKE ? + AND EXISTS ( + SELECT 1 + FROM processed_files preferred + WHERE preferred.router = processed_files.router + AND preferred.timestamp = processed_files.timestamp + AND preferred.file_path LIKE ? + ) + """, + (args.start_date, end_date, remove_pattern, preferred_pattern), + ) + conn.execute( + """ + CREATE UNIQUE INDEX IF NOT EXISTS idx_processed_files_router_timestamp_unique + ON processed_files(router, timestamp) + """ + ) + conn.execute( + """ + CREATE UNIQUE INDEX IF NOT EXISTS idx_netflow_router_timestamp_unique + ON netflow_stats(router, timestamp) + """ + ) + + print("Repair complete") + + +if __name__ == '__main__': + main() From 3e811afd9dd60a81c584c0a4c854e109dc047494 Mon Sep 17 00:00:00 2001 From: flamboh Date: Tue, 24 Mar 2026 10:29:31 -0700 Subject: [PATCH 4/6] update sqlite loading --- apps/web/src/lib/server/datasets.ts | 6 +++-- apps/web/tests/lib/server/datasets.test.ts | 30 ++++++++++++++++++++++ term-update.md | 3 ++- 3 files changed, 36 insertions(+), 3 deletions(-) diff --git a/apps/web/src/lib/server/datasets.ts b/apps/web/src/lib/server/datasets.ts index bc7ad56..2e13921 100644 --- a/apps/web/src/lib/server/datasets.ts +++ b/apps/web/src/lib/server/datasets.ts @@ -1,8 +1,7 @@ import fs from 'fs'; import path from 'path'; +import { createRequire } from 'node:module'; import Database from 'better-sqlite3'; -import { DatabaseSync } from 'node:sqlite'; -import type { SQLInputValue } from 'node:sqlite'; import type { DatasetSummary } from '$lib/types/types'; import { getDatasetsConfigPath, getRepoRoot } from '$lib/server/paths'; @@ -19,6 +18,7 @@ export interface DatasetConfig { const repoRoot = getRepoRoot(); const defaultRegistryPath = getDatasetsConfigPath(); +const require = createRequire(import.meta.url); type PreparedStatement = { get(...params: unknown[]): unknown; @@ -211,6 +211,8 @@ export function getDatasetDbPath(datasetId: string): string { } function openNodeSqliteDatabase(dbPath: string): ReadonlyDatasetDb { + const { DatabaseSync } = require('node:sqlite') as typeof import('node:sqlite'); + type SQLInputValue = import('node:sqlite').SQLInputValue; const db = new DatabaseSync(dbPath, { open: true, readOnly: true }); return { prepare(sql: string): PreparedStatement { diff --git a/apps/web/tests/lib/server/datasets.test.ts b/apps/web/tests/lib/server/datasets.test.ts index a015465..88fe7bb 100644 --- a/apps/web/tests/lib/server/datasets.test.ts +++ b/apps/web/tests/lib/server/datasets.test.ts @@ -4,6 +4,27 @@ import path from 'path'; import { spawnSync } from 'child_process'; import { afterEach, describe, expect, it, vi } from 'vitest'; +class MockDatabaseSync { + constructor(private readonly dbPath: string) {} + + prepare(query: string) { + return { + get: () => { + const result = spawnSync('sqlite3', [this.dbPath, query], { encoding: 'utf-8' }); + if (result.status !== 0) { + throw new Error(result.stderr || 'sqlite3 query failed'); + } + + const minTimestamp = Number(result.stdout.trim()); + return { minTimestamp: Number.isFinite(minTimestamp) ? minTimestamp : null }; + }, + all: () => [] + }; + } + + close() {} +} + const betterSqlite3Factory = vi.fn((dbPath: string) => ({ prepare(query: string) { return { @@ -25,6 +46,15 @@ vi.mock('better-sqlite3', () => ({ default: vi.fn().mockImplementation((dbPath: string) => betterSqlite3Factory(dbPath)) })); +vi.mock('node:module', async () => ({ + createRequire: () => (specifier: string) => { + if (specifier === 'node:sqlite') { + return { DatabaseSync: MockDatabaseSync }; + } + throw new Error(`Unexpected require: ${specifier}`); + } +})); + async function loadDatasetsModule() { vi.resetModules(); return import('../../../src/lib/server/datasets'); diff --git a/term-update.md b/term-update.md index 2a73bc6..5516984 100644 --- a/term-update.md +++ b/term-update.md @@ -1,4 +1,5 @@ # Term Update: NetFlow Analysis & Predictor + **Period: December 15, 2025 – March 23, 2026** --- @@ -7,7 +8,7 @@ Over this term I made substantial progress on the NetFlow analysis project, evol **Data pipeline:** The ingestion and aggregation pipeline now handles multiple independent datasets and routers, with configurable reprocessing windows and improved reliability around stale data detection. Processing performance and observability were both improved. -**Visualization platform:** The web-based dashboard received significant work — it now supports multi-dataset and multi-router views, faster page loads, and a more interactive analysis experience including synchronized drilldowns, per-chart filters, and better time-range handling. +**Visualization platform:** The web dashboard now supports multi-dataset and multi-router views, has faster page loads, and is a more interactive analysis experience including synchronized drilldowns, per-chart filters, and better time-range handling. **New project — NetFlow Predictor:** I initialized a companion repository to develop predictive models over the collected traffic data. The project is in early planning stages, with the data infrastructure from the analysis pipeline serving as the foundation. From 1eed41c47b26db12b025125d98c507cda793a4e8 Mon Sep 17 00:00:00 2001 From: flamboh Date: Tue, 24 Mar 2026 11:22:45 -0700 Subject: [PATCH 5/6] fix(pipeline): preserve repair deletions on index failure --- .../repair_mirrored_root_duplicates.py | 149 ------------------ 1 file changed, 149 deletions(-) delete mode 100644 tools/netflow-db/repair_mirrored_root_duplicates.py diff --git a/tools/netflow-db/repair_mirrored_root_duplicates.py b/tools/netflow-db/repair_mirrored_root_duplicates.py deleted file mode 100644 index 76fc350..0000000 --- a/tools/netflow-db/repair_mirrored_root_duplicates.py +++ /dev/null @@ -1,149 +0,0 @@ -#!/usr/bin/env python3 -""" -Repair mirrored-root duplicates in NetFlow SQLite databases. - -Keeps a preferred root for each logical capture slot (router + timestamp), -deletes duplicate rows from netflow_stats and processed_files for the other root, -and optionally adds unique indexes on (router, timestamp). - -Example: - python tools/netflow-db/repair_mirrored_root_duplicates.py \ - --db data/uoregon/netflow.sqlite \ - --preferred-prefix /research/obo/netflow_datasets/uoregon/ \ - --remove-prefix /research/tango_cis/uonet-in/ \ - --start-date 2026-03-02 -""" - -import argparse -import sqlite3 -from pathlib import Path - - -def build_prefix_pattern(prefix: str) -> str: - return prefix.rstrip('/') + '/%' - - -def main() -> None: - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument('--db', required=True, help='Path to SQLite database') - parser.add_argument('--preferred-prefix', required=True, help='Path prefix to keep') - parser.add_argument('--remove-prefix', required=True, help='Path prefix to remove') - parser.add_argument('--start-date', required=True, help='Inclusive local date YYYY-MM-DD') - parser.add_argument('--end-date', default=None, help='Inclusive local date YYYY-MM-DD') - parser.add_argument('--dry-run', action='store_true', help='Report changes without writing') - args = parser.parse_args() - - db_path = Path(args.db).expanduser().resolve() - preferred_pattern = build_prefix_pattern(args.preferred_prefix) - remove_pattern = build_prefix_pattern(args.remove_prefix) - end_date = args.end_date or '9999-12-31' - - conn = sqlite3.connect(db_path) - conn.row_factory = sqlite3.Row - - duplicate_slots = conn.execute( - """ - SELECT router, timestamp - FROM processed_files - WHERE date(datetime(timestamp, 'unixepoch', 'localtime')) BETWEEN ? AND ? - GROUP BY router, timestamp - HAVING - SUM(CASE WHEN file_path LIKE ? THEN 1 ELSE 0 END) > 0 - AND SUM(CASE WHEN file_path LIKE ? THEN 1 ELSE 0 END) > 0 - """, - (args.start_date, end_date, preferred_pattern, remove_pattern), - ).fetchall() - - duplicate_count = len(duplicate_slots) - print(f"Logical duplicate slots matched: {duplicate_count}") - if duplicate_count == 0: - return - - delete_netflow = conn.execute( - """ - SELECT COUNT(*) - FROM netflow_stats - WHERE date(datetime(timestamp, 'unixepoch', 'localtime')) BETWEEN ? AND ? - AND file_path LIKE ? - AND EXISTS ( - SELECT 1 - FROM netflow_stats preferred - WHERE preferred.router = netflow_stats.router - AND preferred.timestamp = netflow_stats.timestamp - AND preferred.file_path LIKE ? - ) - """, - (args.start_date, end_date, remove_pattern, preferred_pattern), - ).fetchone()[0] - delete_processed = conn.execute( - """ - SELECT COUNT(*) - FROM processed_files - WHERE date(datetime(timestamp, 'unixepoch', 'localtime')) BETWEEN ? AND ? - AND file_path LIKE ? - AND EXISTS ( - SELECT 1 - FROM processed_files preferred - WHERE preferred.router = processed_files.router - AND preferred.timestamp = processed_files.timestamp - AND preferred.file_path LIKE ? - ) - """, - (args.start_date, end_date, remove_pattern, preferred_pattern), - ).fetchone()[0] - - print(f"netflow_stats rows to delete: {delete_netflow}") - print(f"processed_files rows to delete: {delete_processed}") - - if args.dry_run: - return - - with conn: - conn.execute( - """ - DELETE FROM netflow_stats - WHERE date(datetime(timestamp, 'unixepoch', 'localtime')) BETWEEN ? AND ? - AND file_path LIKE ? - AND EXISTS ( - SELECT 1 - FROM netflow_stats preferred - WHERE preferred.router = netflow_stats.router - AND preferred.timestamp = netflow_stats.timestamp - AND preferred.file_path LIKE ? - ) - """, - (args.start_date, end_date, remove_pattern, preferred_pattern), - ) - conn.execute( - """ - DELETE FROM processed_files - WHERE date(datetime(timestamp, 'unixepoch', 'localtime')) BETWEEN ? AND ? - AND file_path LIKE ? - AND EXISTS ( - SELECT 1 - FROM processed_files preferred - WHERE preferred.router = processed_files.router - AND preferred.timestamp = processed_files.timestamp - AND preferred.file_path LIKE ? - ) - """, - (args.start_date, end_date, remove_pattern, preferred_pattern), - ) - conn.execute( - """ - CREATE UNIQUE INDEX IF NOT EXISTS idx_processed_files_router_timestamp_unique - ON processed_files(router, timestamp) - """ - ) - conn.execute( - """ - CREATE UNIQUE INDEX IF NOT EXISTS idx_netflow_router_timestamp_unique - ON netflow_stats(router, timestamp) - """ - ) - - print("Repair complete") - - -if __name__ == '__main__': - main() From f8bf29c47944dfd86c86bb2917628eea252252d8 Mon Sep 17 00:00:00 2001 From: flamboh Date: Tue, 24 Mar 2026 13:55:14 -0700 Subject: [PATCH 6/6] delete docs --- term-update-accomplishments-2025-12-15.md | 39 ----------------------- term-update.md | 15 --------- 2 files changed, 54 deletions(-) delete mode 100644 term-update-accomplishments-2025-12-15.md delete mode 100644 term-update.md diff --git a/term-update-accomplishments-2025-12-15.md b/term-update-accomplishments-2025-12-15.md deleted file mode 100644 index cf2d2c8..0000000 --- a/term-update-accomplishments-2025-12-15.md +++ /dev/null @@ -1,39 +0,0 @@ -# Term Update Accomplishments - -Coverage: work reflected in git history for `netflow-analysis` and `../netflow-predictor` since 2025-12-15. - -## netflow-analysis - -### Major accomplishments - -- Expanded the project from a basic NetFlow dashboard into a broader analysis platform with support for richer traffic characterizations, including structure- and spectrum-based views. -- Strengthened the backend data pipeline and database workflow so processing is more reliable, better organized, and more scalable for ongoing data ingestion. -- Added dataset-aware workflow support, making it easier to manage multiple datasets and expose them cleanly through the web application. -- Improved the system’s resilience around data discovery, reprocessing, migration, and stale-data handling, which should make routine maintenance and updates more dependable. -- Made substantial progress on frontend usability and performance, including faster detail views, better chart interactions, more intuitive controls, and smoother exploratory analysis workflows. -- Improved caching and request handling in the web app to reduce redundant work and support more responsive analysis over larger data windows. -- Continued cleanup and documentation work to make the codebase easier to maintain and extend. - -### Rough timeline - -- Dec 2025: initial structure/spectrum analysis support and spectrum drilldown. -- Jan 2026: pipeline/database refactors, migration updates, batching/parallelization redesign, stale-day/schema fixes, and timezone/chart correctness work. -- Feb 2026: router migration support, stronger recovery behavior, SQLite write-path refactor, and major dashboard interaction improvements. -- Mar 2026: configurable reprocessing/discovery windows, faster file analysis pages, dataset-centric dashboards/configuration, caching layers, and broader UI polish/performance work. - -## netflow-predictor - -### Major accomplishments - -- Set up the initial predictor project and development environment. -- Defined the first-pass research direction for forecasting future network behavior from processed NetFlow data and MAAD-derived analyses. -- Wrote down an initial modeling and evaluation plan so the prediction work now has a concrete starting point. - -### Rough timeline - -- Feb 2026: repository initialization. -- Mar 2026: uv-based project setup and initial research/problem-definition documents. - -## Overall term-level takeaway - -Across the term, the main body of work was in `netflow-analysis`, where the project matured into a more capable and reliable platform for ingesting, organizing, and exploring network-flow data. In parallel, `netflow-predictor` was started as the next stage of the research effort, with the initial environment and planning work in place for future forecasting experiments. diff --git a/term-update.md b/term-update.md deleted file mode 100644 index 5516984..0000000 --- a/term-update.md +++ /dev/null @@ -1,15 +0,0 @@ -# Term Update: NetFlow Analysis & Predictor - -**Period: December 15, 2025 – March 23, 2026** - ---- - -Over this term I made substantial progress on the NetFlow analysis project, evolving it from a working prototype into a more complete and reliable research platform. - -**Data pipeline:** The ingestion and aggregation pipeline now handles multiple independent datasets and routers, with configurable reprocessing windows and improved reliability around stale data detection. Processing performance and observability were both improved. - -**Visualization platform:** The web dashboard now supports multi-dataset and multi-router views, has faster page loads, and is a more interactive analysis experience including synchronized drilldowns, per-chart filters, and better time-range handling. - -**New project — NetFlow Predictor:** I initialized a companion repository to develop predictive models over the collected traffic data. The project is in early planning stages, with the data infrastructure from the analysis pipeline serving as the foundation. - -Overall, the system is in a much stronger state for supporting ongoing analysis and, going forward, for experimenting with traffic prediction.