# Files changed · PR \#128 · flambo/atlantis [View on GitCafe](https://git.cafe/flambo/atlantis/pull/128/diffs) Repository: [flambo/atlantis](https://git.cafe/flambo/atlantis) Visibility: public [feat\(netflow-db\): add concentration controls for MAAD address sets](https://git.cafe/flambo/atlantis/pull/128.md) Version: 0 Head commit: 4346f62ecf63ed37d4bdc8b2ccf466c0b7170a35 Base commit: 11072e9906bdc333a830fe9465c0c5f1c0580b87 Comparison base: 11072e9906bdc333a830fe9465c0c5f1c0580b87 Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## apps/web/drizzle/20261003025340_concentration_stats/migration.sql Status: added; +60; −0 ```diff @@ -1,0 +1,60 @@ +CREATE TABLE `address_concentration_stats` ( + `source_id` text NOT NULL, + `granularity` text NOT NULL, + `bucket_start` integer NOT NULL, + `bucket_end` integer NOT NULL, + `ip_version` integer NOT NULL, + `src_locality` text NOT NULL, + `dst_locality` text NOT NULL, + `address_side` text NOT NULL, + `measure` text NOT NULL, + `weight_total` real NOT NULL, + `entry_count` integer NOT NULL, + `hhi` real, + `top1_share` real, + `top10_share` real, + `top100_share` real, + `entropy_p8` real, + `entropy_p16` real, + `entropy_p24` real, + `entropy_p32` real, + `entropy_p48` real, + `entropy_p64` real, + `entropy_p128` real, + CONSTRAINT "address_concentration_stats_granularity_check" CHECK("granularity" IN ('5m', '10m', '30m', '1h', '1d')), + CONSTRAINT "address_concentration_stats_bucket_check" CHECK("bucket_end" > "bucket_start"), + CONSTRAINT "address_concentration_stats_ip_version_check" CHECK("ip_version" IN (4, 6)), + CONSTRAINT "address_concentration_stats_src_locality_check" CHECK("src_locality" IN ('all', 'internal', 'external')), + CONSTRAINT "address_concentration_stats_dst_locality_check" CHECK("dst_locality" IN ('all', 'internal', 'external')), + CONSTRAINT "address_concentration_stats_address_side_check" CHECK("address_side" IN ('source', 'destination')), + CONSTRAINT "address_concentration_stats_measure_check" CHECK("measure" IN ('addresses', 'packets', 'bytes')), + CONSTRAINT "address_concentration_stats_weight_total_check" CHECK("weight_total" >= 0 AND "weight_total" <= 1.7976931348623157e308), + CONSTRAINT "address_concentration_stats_entry_count_check" CHECK("entry_count" >= 0), + CONSTRAINT "address_concentration_stats_empty_weight_check" CHECK(("entry_count" = 0) = ("weight_total" = 0)), + CONSTRAINT "address_concentration_stats_minimum_count_check" CHECK(("hhi" IS NOT NULL) = ("entry_count" >= 2)), + CONSTRAINT "address_concentration_stats_hhi_check" CHECK("hhi" > 0 AND "hhi" <= 1), + CONSTRAINT "address_concentration_stats_top1_share_check" CHECK("top1_share" >= 0 AND "top1_share" <= 1), + CONSTRAINT "address_concentration_stats_top10_share_check" CHECK("top10_share" >= 0 AND "top10_share" <= 1), + CONSTRAINT "address_concentration_stats_top100_share_check" CHECK("top100_share" >= 0 AND "top100_share" <= 1), + CONSTRAINT "address_concentration_stats_entropy_p8_check" CHECK("entropy_p8" >= 0 AND "entropy_p8" <= 1.7976931348623157e308), + CONSTRAINT "address_concentration_stats_entropy_p16_check" CHECK("entropy_p16" >= 0 AND "entropy_p16" <= 1.7976931348623157e308), + CONSTRAINT "address_concentration_stats_entropy_p24_check" CHECK("entropy_p24" >= 0 AND "entropy_p24" <= 1.7976931348623157e308), + CONSTRAINT "address_concentration_stats_entropy_p32_check" CHECK("entropy_p32" >= 0 AND "entropy_p32" <= 1.7976931348623157e308), + CONSTRAINT "address_concentration_stats_entropy_p48_check" CHECK("entropy_p48" >= 0 AND "entropy_p48" <= 1.7976931348623157e308), + CONSTRAINT "address_concentration_stats_entropy_p64_check" CHECK("entropy_p64" >= 0 AND "entropy_p64" <= 1.7976931348623157e308), + CONSTRAINT "address_concentration_stats_entropy_p128_check" CHECK("entropy_p128" >= 0 AND "entropy_p128" <= 1.7976931348623157e308), + CONSTRAINT "address_concentration_stats_top1_share_presence_check" CHECK(("top1_share" IS NULL) = ("hhi" IS NULL)), + CONSTRAINT "address_concentration_stats_top10_share_presence_check" CHECK(("top10_share" IS NULL) = ("hhi" IS NULL)), + CONSTRAINT "address_concentration_stats_top100_share_presence_check" CHECK(("top100_share" IS NULL) = ("hhi" IS NULL)), + CONSTRAINT "address_concentration_stats_entropy_p32_presence_check" CHECK(("entropy_p32" IS NULL) = ("hhi" IS NULL)), + CONSTRAINT "address_concentration_stats_entropy_p8_presence_check" CHECK(("entropy_p8" IS NOT NULL) = ("ip_version" = 4 AND "hhi" IS NOT NULL)), + CONSTRAINT "address_concentration_stats_entropy_p16_presence_check" CHECK(("entropy_p16" IS NOT NULL) = ("ip_version" = 4 AND "hhi" IS NOT NULL)), + CONSTRAINT "address_concentration_stats_entropy_p24_presence_check" CHECK(("entropy_p24" IS NOT NULL) = ("ip_version" = 4 AND "hhi" IS NOT NULL)), + CONSTRAINT "address_concentration_stats_entropy_p48_presence_check" CHECK(("entropy_p48" IS NOT NULL) = ("ip_version" = 6 AND "hhi" IS NOT NULL)), + CONSTRAINT "address_concentration_stats_entropy_p64_presence_check" CHECK(("entropy_p64" IS NOT NULL) = ("ip_version" = 6 AND "hhi" IS NOT NULL)), + CONSTRAINT "address_concentration_stats_entropy_p128_presence_check" CHECK(("entropy_p128" IS NOT NULL) = ("ip_version" = 6 AND "hhi" IS NOT NULL)), + CONSTRAINT "address_concentration_stats_top_order_check" CHECK("top10_share" >= "top1_share" AND "top100_share" >= "top10_share") +); +--> statement-breakpoint +CREATE UNIQUE INDEX `idx_address_concentration_stats_key` ON `address_concentration_stats` (`source_id`,`granularity`,`src_locality`,`dst_locality`,`ip_version`,`measure`,`bucket_start`,`address_side`);--> statement-breakpoint +CREATE INDEX `idx_address_concentration_stats_bucket` ON `address_concentration_stats` (`granularity`,`bucket_start`); ``` ## apps/web/drizzle/20261003025340_concentration_stats/snapshot.json Status: added; +2255; −0 ```diff @@ -1,0 +1,2255 @@ +{ + "version": "7", + "dialect": "sqlite", + "id": "7976b462-f23a-430d-bbec-5966fccfb1a2", + "prevIds": ["09dd8c45-29e1-4934-bd9c-e92174d89c92"], + "ddl": [ + { + "name": "address_concentration_stats", + "entityType": "tables" + }, + { + "name": "address_count_stats", + "entityType": "tables" + }, + { + "name": "address_maad_stats", + "entityType": "tables" + }, + { + "name": "bucket_coverage", + "entityType": "tables" + }, + { + "name": "datasets", + "entityType": "tables" + }, + { + "name": "maad_q_grid", + "entityType": "tables" + }, + { + "name": "port_count_stats", + "entityType": "tables" + }, + { + "name": "processed_inputs", + "entityType": "tables" + }, + { + "name": "protocol_stats", + "entityType": "tables" + }, + { + "name": "source_members", + "entityType": "tables" + }, + { + "name": "traffic_stats", + "entityType": "tables" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "source_id", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "granularity", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bucket_start", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bucket_end", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "ip_version", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "src_locality", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "dst_locality", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "address_side", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "measure", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "real", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "weight_total", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "entry_count", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "real", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "hhi", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "real", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "top1_share", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "real", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "top10_share", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "real", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "top100_share", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "real", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "entropy_p8", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "real", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "entropy_p16", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "real", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "entropy_p24", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "real", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "entropy_p32", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "real", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "entropy_p48", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "real", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "entropy_p64", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "real", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "entropy_p128", + "entityType": "columns", + "table": "address_concentration_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "source_id", + "entityType": "columns", + "table": "address_count_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "granularity", + "entityType": "columns", + "table": "address_count_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bucket_start", + "entityType": "columns", + "table": "address_count_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bucket_end", + "entityType": "columns", + "table": "address_count_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "ip_version", + "entityType": "columns", + "table": "address_count_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "src_locality", + "entityType": "columns", + "table": "address_count_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "dst_locality", + "entityType": "columns", + "table": "address_count_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "address_side", + "entityType": "columns", + "table": "address_count_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "unique_address_count", + "entityType": "columns", + "table": "address_count_stats" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": "CURRENT_TIMESTAMP", + "generated": null, + "name": "processed_at", + "entityType": "columns", + "table": "address_count_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "source_id", + "entityType": "columns", + "table": "address_maad_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "granularity", + "entityType": "columns", + "table": "address_maad_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bucket_start", + "entityType": "columns", + "table": "address_maad_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bucket_end", + "entityType": "columns", + "table": "address_maad_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "ip_version", + "entityType": "columns", + "table": "address_maad_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "src_locality", + "entityType": "columns", + "table": "address_maad_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "dst_locality", + "entityType": "columns", + "table": "address_maad_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "address_side", + "entityType": "columns", + "table": "address_maad_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "measure", + "entityType": "columns", + "table": "address_maad_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "total_addrs", + "entityType": "columns", + "table": "address_maad_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": "0", + "generated": null, + "name": "zero_weight_addrs", + "entityType": "columns", + "table": "address_maad_stats" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "min_prefix_length", + "entityType": "columns", + "table": "address_maad_stats" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "max_prefix_length", + "entityType": "columns", + "table": "address_maad_stats" + }, + { + "type": "real", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "d0", + "entityType": "columns", + "table": "address_maad_stats" + }, + { + "type": "real", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "d1", + "entityType": "columns", + "table": "address_maad_stats" + }, + { + "type": "real", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "d2", + "entityType": "columns", + "table": "address_maad_stats" + }, + { + "type": "blob", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "tau", + "entityType": "columns", + "table": "address_maad_stats" + }, + { + "type": "blob", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "tau_sd", + "entityType": "columns", + "table": "address_maad_stats" + }, + { + "type": "blob", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "spectrum", + "entityType": "columns", + "table": "address_maad_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "source_id", + "entityType": "columns", + "table": "bucket_coverage" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "granularity", + "entityType": "columns", + "table": "bucket_coverage" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bucket_start", + "entityType": "columns", + "table": "bucket_coverage" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bucket_end", + "entityType": "columns", + "table": "bucket_coverage" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "coverage_state", + "entityType": "columns", + "table": "bucket_coverage" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "observed_units", + "entityType": "columns", + "table": "bucket_coverage" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "expected_units", + "entityType": "columns", + "table": "bucket_coverage" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "rejected_units", + "entityType": "columns", + "table": "bucket_coverage" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "datasets" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "label", + "entityType": "columns", + "table": "datasets" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "default_start_date", + "entityType": "columns", + "table": "datasets" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": "'static'", + "generated": null, + "name": "source_mode", + "entityType": "columns", + "table": "datasets" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": "'static'", + "generated": null, + "name": "discovery_mode", + "entityType": "columns", + "table": "datasets" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": "0", + "generated": null, + "name": "sort_order", + "entityType": "columns", + "table": "datasets" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": "false", + "generated": null, + "name": "has_locality", + "entityType": "columns", + "table": "datasets" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": "true", + "generated": null, + "name": "maad_internal_side", + "entityType": "columns", + "table": "datasets" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "ip_version", + "entityType": "columns", + "table": "maad_q_grid" + }, + { + "type": "real", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "q_min", + "entityType": "columns", + "table": "maad_q_grid" + }, + { + "type": "real", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "q_step", + "entityType": "columns", + "table": "maad_q_grid" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "q_count", + "entityType": "columns", + "table": "maad_q_grid" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "source_id", + "entityType": "columns", + "table": "port_count_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "granularity", + "entityType": "columns", + "table": "port_count_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bucket_start", + "entityType": "columns", + "table": "port_count_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bucket_end", + "entityType": "columns", + "table": "port_count_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "ip_version", + "entityType": "columns", + "table": "port_count_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "src_locality", + "entityType": "columns", + "table": "port_count_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "dst_locality", + "entityType": "columns", + "table": "port_count_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "port_side", + "entityType": "columns", + "table": "port_count_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "port_range", + "entityType": "columns", + "table": "port_count_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "unique_port_count", + "entityType": "columns", + "table": "port_count_stats" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": "CURRENT_TIMESTAMP", + "generated": null, + "name": "processed_at", + "entityType": "columns", + "table": "port_count_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "input_kind", + "entityType": "columns", + "table": "processed_inputs" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "input_locator", + "entityType": "columns", + "table": "processed_inputs" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "source_id", + "entityType": "columns", + "table": "processed_inputs" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bucket_start", + "entityType": "columns", + "table": "processed_inputs" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bucket_end", + "entityType": "columns", + "table": "processed_inputs" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": "'pending'", + "generated": null, + "name": "status", + "entityType": "columns", + "table": "processed_inputs" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "error_message", + "entityType": "columns", + "table": "processed_inputs" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": "CURRENT_TIMESTAMP", + "generated": null, + "name": "discovered_at", + "entityType": "columns", + "table": "processed_inputs" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "processed_at", + "entityType": "columns", + "table": "processed_inputs" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "source_id", + "entityType": "columns", + "table": "protocol_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "granularity", + "entityType": "columns", + "table": "protocol_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bucket_start", + "entityType": "columns", + "table": "protocol_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bucket_end", + "entityType": "columns", + "table": "protocol_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "ip_version", + "entityType": "columns", + "table": "protocol_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "src_locality", + "entityType": "columns", + "table": "protocol_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "dst_locality", + "entityType": "columns", + "table": "protocol_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "unique_protocols_count", + "entityType": "columns", + "table": "protocol_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "protocols_list", + "entityType": "columns", + "table": "protocol_stats" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": "CURRENT_TIMESTAMP", + "generated": null, + "name": "processed_at", + "entityType": "columns", + "table": "protocol_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "dataset_id", + "entityType": "columns", + "table": "source_members" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "source_id", + "entityType": "columns", + "table": "source_members" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "member_id", + "entityType": "columns", + "table": "source_members" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "source_id", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "granularity", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bucket_start", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bucket_end", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "ip_version", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "src_locality", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "dst_locality", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "flows", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "flows_tcp", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "flows_udp", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "flows_icmp", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "flows_other", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "packets", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "packets_tcp", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "packets_udp", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "packets_icmp", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "packets_other", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bytes", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bytes_tcp", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bytes_udp", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bytes_icmp", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "bytes_other", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "duration_sum_ms", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "duration_count", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "real", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "average_duration_ms", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "min_ttl_sum", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "min_ttl_count", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "real", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "average_min_ttl", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "max_ttl_sum", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "max_ttl_count", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "real", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "average_max_ttl", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": "CURRENT_TIMESTAMP", + "generated": null, + "name": "processed_at", + "entityType": "columns", + "table": "traffic_stats" + }, + { + "columns": [ + "source_id", + "granularity", + "bucket_start", + "ip_version", + "src_locality", + "dst_locality", + "address_side" + ], + "nameExplicit": false, + "name": "address_count_stats_source_id_granularity_bucket_start_ip_version_src_locality_dst_locality_address_side_pk", + "entityType": "pks", + "table": "address_count_stats" + }, + { + "columns": ["source_id", "granularity", "bucket_start"], + "nameExplicit": false, + "name": "bucket_coverage_source_id_granularity_bucket_start_pk", + "entityType": "pks", + "table": "bucket_coverage" + }, + { + "columns": [ + "source_id", + "granularity", + "bucket_start", + "ip_version", + "src_locality", + "dst_locality", + "port_side", + "port_range" + ], + "nameExplicit": false, + "name": "port_count_stats_source_id_granularity_bucket_start_ip_version_src_locality_dst_locality_port_side_port_range_pk", + "entityType": "pks", + "table": "port_count_stats" + }, + { + "columns": ["input_kind", "input_locator", "source_id", "bucket_start"], + "nameExplicit": false, + "name": "processed_inputs_input_kind_input_locator_source_id_bucket_start_pk", + "entityType": "pks", + "table": "processed_inputs" + }, + { + "columns": [ + "source_id", + "granularity", + "bucket_start", + "ip_version", + "src_locality", + "dst_locality" + ], + "nameExplicit": false, + "name": "protocol_stats_source_id_granularity_bucket_start_ip_version_src_locality_dst_locality_pk", + "entityType": "pks", + "table": "protocol_stats" + }, + { + "columns": ["dataset_id", "source_id", "member_id"], + "nameExplicit": false, + "name": "source_members_dataset_id_source_id_member_id_pk", + "entityType": "pks", + "table": "source_members" + }, + { + "columns": [ + "source_id", + "granularity", + "bucket_start", + "ip_version", + "src_locality", + "dst_locality" + ], + "nameExplicit": false, + "name": "traffic_stats_source_id_granularity_bucket_start_ip_version_src_locality_dst_locality_pk", + "entityType": "pks", + "table": "traffic_stats" + }, + { + "columns": ["id"], + "nameExplicit": false, + "name": "datasets_pk", + "table": "datasets", + "entityType": "pks" + }, + { + "columns": ["ip_version"], + "nameExplicit": false, + "name": "maad_q_grid_pk", + "table": "maad_q_grid", + "entityType": "pks" + }, + { + "columns": [ + { + "value": "source_id", + "isExpression": false + }, + { + "value": "granularity", + "isExpression": false + }, + { + "value": "src_locality", + "isExpression": false + }, + { + "value": "dst_locality", + "isExpression": false + }, + { + "value": "ip_version", + "isExpression": false + }, + { + "value": "measure", + "isExpression": false + }, + { + "value": "bucket_start", + "isExpression": false + }, + { + "value": "address_side", + "isExpression": false + } + ], + "isUnique": true, + "where": null, + "origin": "manual", + "name": "idx_address_concentration_stats_key", + "entityType": "indexes", + "table": "address_concentration_stats" + }, + { + "columns": [ + { + "value": "granularity", + "isExpression": false + }, + { + "value": "bucket_start", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "idx_address_concentration_stats_bucket", + "entityType": "indexes", + "table": "address_concentration_stats" + }, + { + "columns": [ + { + "value": "granularity", + "isExpression": false + }, + { + "value": "bucket_start", + "isExpression": false + }, + { + "value": "source_id", + "isExpression": false + }, + { + "value": "ip_version", + "isExpression": false + }, + { + "value": "src_locality", + "isExpression": false + }, + { + "value": "dst_locality", + "isExpression": false + }, + { + "value": "address_side", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "idx_address_count_stats_query", + "entityType": "indexes", + "table": "address_count_stats" + }, + { + "columns": [ + { + "value": "source_id", + "isExpression": false + }, + { + "value": "granularity", + "isExpression": false + }, + { + "value": "src_locality", + "isExpression": false + }, + { + "value": "dst_locality", + "isExpression": false + }, + { + "value": "bucket_start", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "idx_address_count_stats_timeseries", + "entityType": "indexes", + "table": "address_count_stats" + }, + { + "columns": [ + { + "value": "source_id", + "isExpression": false + }, + { + "value": "granularity", + "isExpression": false + }, + { + "value": "src_locality", + "isExpression": false + }, + { + "value": "dst_locality", + "isExpression": false + }, + { + "value": "ip_version", + "isExpression": false + }, + { + "value": "measure", + "isExpression": false + }, + { + "value": "bucket_start", + "isExpression": false + }, + { + "value": "address_side", + "isExpression": false + } + ], + "isUnique": true, + "where": null, + "origin": "manual", + "name": "idx_address_maad_stats_key", + "entityType": "indexes", + "table": "address_maad_stats" + }, + { + "columns": [ + { + "value": "granularity", + "isExpression": false + }, + { + "value": "bucket_start", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "idx_address_maad_stats_bucket", + "entityType": "indexes", + "table": "address_maad_stats" + }, + { + "columns": [ + { + "value": "granularity", + "isExpression": false + }, + { + "value": "bucket_start", + "isExpression": false + }, + { + "value": "source_id", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "idx_bucket_coverage_query", + "entityType": "indexes", + "table": "bucket_coverage" + }, + { + "columns": [ + { + "value": "source_id", + "isExpression": false + }, + { + "value": "granularity", + "isExpression": false + }, + { + "value": "src_locality", + "isExpression": false + }, + { + "value": "dst_locality", + "isExpression": false + }, + { + "value": "bucket_start", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "idx_port_count_stats_timeseries", + "entityType": "indexes", + "table": "port_count_stats" + }, + { + "columns": [ + { + "value": "source_id", + "isExpression": false + }, + { + "value": "bucket_start", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "idx_processed_inputs_source_bucket", + "entityType": "indexes", + "table": "processed_inputs" + }, + { + "columns": [ + { + "value": "source_id", + "isExpression": false + }, + { + "value": "granularity", + "isExpression": false + }, + { + "value": "src_locality", + "isExpression": false + }, + { + "value": "dst_locality", + "isExpression": false + }, + { + "value": "bucket_start", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "idx_protocol_stats_timeseries", + "entityType": "indexes", + "table": "protocol_stats" + }, + { + "columns": [ + { + "value": "granularity", + "isExpression": false + }, + { + "value": "bucket_start", + "isExpression": false + }, + { + "value": "source_id", + "isExpression": false + }, + { + "value": "ip_version", + "isExpression": false + }, + { + "value": "src_locality", + "isExpression": false + }, + { + "value": "dst_locality", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "idx_traffic_stats_query", + "entityType": "indexes", + "table": "traffic_stats" + }, + { + "columns": [ + { + "value": "source_id", + "isExpression": false + }, + { + "value": "granularity", + "isExpression": false + }, + { + "value": "src_locality", + "isExpression": false + }, + { + "value": "dst_locality", + "isExpression": false + }, + { + "value": "bucket_start", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "idx_traffic_stats_timeseries", + "entityType": "indexes", + "table": "traffic_stats" + }, + { + "value": "\"granularity\" IN ('5m', '10m', '30m', '1h', '1d')", + "name": "address_concentration_stats_granularity_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"bucket_end\" > \"bucket_start\"", + "name": "address_concentration_stats_bucket_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"ip_version\" IN (4, 6)", + "name": "address_concentration_stats_ip_version_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"src_locality\" IN ('all', 'internal', 'external')", + "name": "address_concentration_stats_src_locality_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"dst_locality\" IN ('all', 'internal', 'external')", + "name": "address_concentration_stats_dst_locality_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"address_side\" IN ('source', 'destination')", + "name": "address_concentration_stats_address_side_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"measure\" IN ('addresses', 'packets', 'bytes')", + "name": "address_concentration_stats_measure_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"weight_total\" >= 0 AND \"weight_total\" <= 1.7976931348623157e308", + "name": "address_concentration_stats_weight_total_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"entry_count\" >= 0", + "name": "address_concentration_stats_entry_count_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "(\"entry_count\" = 0) = (\"weight_total\" = 0)", + "name": "address_concentration_stats_empty_weight_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "(\"hhi\" IS NOT NULL) = (\"entry_count\" >= 2)", + "name": "address_concentration_stats_minimum_count_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"hhi\" > 0 AND \"hhi\" <= 1", + "name": "address_concentration_stats_hhi_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"top1_share\" >= 0 AND \"top1_share\" <= 1", + "name": "address_concentration_stats_top1_share_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"top10_share\" >= 0 AND \"top10_share\" <= 1", + "name": "address_concentration_stats_top10_share_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"top100_share\" >= 0 AND \"top100_share\" <= 1", + "name": "address_concentration_stats_top100_share_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"entropy_p8\" >= 0 AND \"entropy_p8\" <= 1.7976931348623157e308", + "name": "address_concentration_stats_entropy_p8_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"entropy_p16\" >= 0 AND \"entropy_p16\" <= 1.7976931348623157e308", + "name": "address_concentration_stats_entropy_p16_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"entropy_p24\" >= 0 AND \"entropy_p24\" <= 1.7976931348623157e308", + "name": "address_concentration_stats_entropy_p24_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"entropy_p32\" >= 0 AND \"entropy_p32\" <= 1.7976931348623157e308", + "name": "address_concentration_stats_entropy_p32_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"entropy_p48\" >= 0 AND \"entropy_p48\" <= 1.7976931348623157e308", + "name": "address_concentration_stats_entropy_p48_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"entropy_p64\" >= 0 AND \"entropy_p64\" <= 1.7976931348623157e308", + "name": "address_concentration_stats_entropy_p64_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"entropy_p128\" >= 0 AND \"entropy_p128\" <= 1.7976931348623157e308", + "name": "address_concentration_stats_entropy_p128_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "(\"top1_share\" IS NULL) = (\"hhi\" IS NULL)", + "name": "address_concentration_stats_top1_share_presence_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "(\"top10_share\" IS NULL) = (\"hhi\" IS NULL)", + "name": "address_concentration_stats_top10_share_presence_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "(\"top100_share\" IS NULL) = (\"hhi\" IS NULL)", + "name": "address_concentration_stats_top100_share_presence_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "(\"entropy_p32\" IS NULL) = (\"hhi\" IS NULL)", + "name": "address_concentration_stats_entropy_p32_presence_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "(\"entropy_p8\" IS NOT NULL) = (\"ip_version\" = 4 AND \"hhi\" IS NOT NULL)", + "name": "address_concentration_stats_entropy_p8_presence_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "(\"entropy_p16\" IS NOT NULL) = (\"ip_version\" = 4 AND \"hhi\" IS NOT NULL)", + "name": "address_concentration_stats_entropy_p16_presence_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "(\"entropy_p24\" IS NOT NULL) = (\"ip_version\" = 4 AND \"hhi\" IS NOT NULL)", + "name": "address_concentration_stats_entropy_p24_presence_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "(\"entropy_p48\" IS NOT NULL) = (\"ip_version\" = 6 AND \"hhi\" IS NOT NULL)", + "name": "address_concentration_stats_entropy_p48_presence_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "(\"entropy_p64\" IS NOT NULL) = (\"ip_version\" = 6 AND \"hhi\" IS NOT NULL)", + "name": "address_concentration_stats_entropy_p64_presence_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "(\"entropy_p128\" IS NOT NULL) = (\"ip_version\" = 6 AND \"hhi\" IS NOT NULL)", + "name": "address_concentration_stats_entropy_p128_presence_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"top10_share\" >= \"top1_share\" AND \"top100_share\" >= \"top10_share\"", + "name": "address_concentration_stats_top_order_check", + "entityType": "checks", + "table": "address_concentration_stats" + }, + { + "value": "\"ip_version\" IN (4, 6)", + "name": "address_count_stats_ip_version_check", + "entityType": "checks", + "table": "address_count_stats" + }, + { + "value": "\"src_locality\" IN ('all', 'internal', 'external')", + "name": "address_count_stats_src_locality_check", + "entityType": "checks", + "table": "address_count_stats" + }, + { + "value": "\"dst_locality\" IN ('all', 'internal', 'external')", + "name": "address_count_stats_dst_locality_check", + "entityType": "checks", + "table": "address_count_stats" + }, + { + "value": "\"ip_version\" IN (4, 6)", + "name": "address_maad_stats_ip_version_check", + "entityType": "checks", + "table": "address_maad_stats" + }, + { + "value": "\"src_locality\" IN ('all', 'internal', 'external')", + "name": "address_maad_stats_src_locality_check", + "entityType": "checks", + "table": "address_maad_stats" + }, + { + "value": "\"dst_locality\" IN ('all', 'internal', 'external')", + "name": "address_maad_stats_dst_locality_check", + "entityType": "checks", + "table": "address_maad_stats" + }, + { + "value": "\"measure\" IN ('addresses', 'packets', 'bytes')", + "name": "address_maad_stats_measure_check", + "entityType": "checks", + "table": "address_maad_stats" + }, + { + "value": "\"bucket_end\" > \"bucket_start\"", + "name": "address_maad_stats_bucket_check", + "entityType": "checks", + "table": "address_maad_stats" + }, + { + "value": "\"total_addrs\" >= 0", + "name": "address_maad_stats_total_addrs_check", + "entityType": "checks", + "table": "address_maad_stats" + }, + { + "value": "\"zero_weight_addrs\" >= 0 AND (\"measure\" <> 'addresses' OR \"zero_weight_addrs\" = 0)", + "name": "address_maad_stats_zero_weight_addrs_check", + "entityType": "checks", + "table": "address_maad_stats" + }, + { + "value": "\"max_prefix_length\" >= \"min_prefix_length\"", + "name": "address_maad_stats_prefix_length_check", + "entityType": "checks", + "table": "address_maad_stats" + }, + { + "value": "typeof(\"tau\") IN ('null', 'blob') AND length(\"tau\") > 0 AND length(\"tau\") % 4 = 0", + "name": "address_maad_stats_tau_check", + "entityType": "checks", + "table": "address_maad_stats" + }, + { + "value": "typeof(\"tau_sd\") = typeof(\"tau\") AND length(\"tau_sd\") IS length(\"tau\")", + "name": "address_maad_stats_tau_sd_check", + "entityType": "checks", + "table": "address_maad_stats" + }, + { + "value": "typeof(\"spectrum\") IN ('null', 'blob') AND length(\"spectrum\") % 8 = 0", + "name": "address_maad_stats_spectrum_check", + "entityType": "checks", + "table": "address_maad_stats" + }, + { + "value": "\"measure\" = 'addresses' OR \"spectrum\" IS NULL", + "name": "address_maad_stats_measure_spectrum_check", + "entityType": "checks", + "table": "address_maad_stats" + }, + { + "value": "(\"tau\" IS NULL) = (\"d0\" IS NULL)", + "name": "address_maad_stats_tau_d0_check", + "entityType": "checks", + "table": "address_maad_stats" + }, + { + "value": "\"bucket_end\" > \"bucket_start\"", + "name": "bucket_coverage_interval_check", + "entityType": "checks", + "table": "bucket_coverage" + }, + { + "value": "\"expected_units\" > 0", + "name": "bucket_coverage_expected_check", + "entityType": "checks", + "table": "bucket_coverage" + }, + { + "value": "\"observed_units\" >= 0 AND \"observed_units\" <= \"expected_units\"", + "name": "bucket_coverage_observed_check", + "entityType": "checks", + "table": "bucket_coverage" + }, + { + "value": "\"rejected_units\" >= 0 AND \"rejected_units\" <= \"expected_units\"", + "name": "bucket_coverage_rejected_check", + "entityType": "checks", + "table": "bucket_coverage" + }, + { + "value": "(\"coverage_state\" = 'complete' AND \"observed_units\" = \"expected_units\" AND \"rejected_units\" = 0) OR (\"coverage_state\" = 'unknown' AND \"observed_units\" = 0 AND \"rejected_units\" = 0) OR (\"coverage_state\" = 'partial' AND NOT (\"observed_units\" = \"expected_units\" AND \"rejected_units\" = 0) AND NOT (\"observed_units\" = 0 AND \"rejected_units\" = 0))", + "name": "bucket_coverage_state_check", + "entityType": "checks", + "table": "bucket_coverage" + }, + { + "value": "\"ip_version\" IN (4, 6)", + "name": "maad_q_grid_ip_version_check", + "entityType": "checks", + "table": "maad_q_grid" + }, + { + "value": "\"q_step\" > 0", + "name": "maad_q_grid_q_step_check", + "entityType": "checks", + "table": "maad_q_grid" + }, + { + "value": "\"q_count\" > 0", + "name": "maad_q_grid_q_count_check", + "entityType": "checks", + "table": "maad_q_grid" + }, + { + "value": "\"ip_version\" IN (4, 6)", + "name": "port_count_stats_ip_version_check", + "entityType": "checks", + "table": "port_count_stats" + }, + { + "value": "\"src_locality\" IN ('all', 'internal', 'external')", + "name": "port_count_stats_src_locality_check", + "entityType": "checks", + "table": "port_count_stats" + }, + { + "value": "\"dst_locality\" IN ('all', 'internal', 'external')", + "name": "port_count_stats_dst_locality_check", + "entityType": "checks", + "table": "port_count_stats" + }, + { + "value": "\"ip_version\" IN (4, 6)", + "name": "protocol_stats_ip_version_check", + "entityType": "checks", + "table": "protocol_stats" + }, + { + "value": "\"src_locality\" IN ('all', 'internal', 'external')", + "name": "protocol_stats_src_locality_check", + "entityType": "checks", + "table": "protocol_stats" + }, + { + "value": "\"dst_locality\" IN ('all', 'internal', 'external')", + "name": "protocol_stats_dst_locality_check", + "entityType": "checks", + "table": "protocol_stats" + }, + { + "value": "\"ip_version\" IN (4, 6)", + "name": "traffic_stats_ip_version_check", + "entityType": "checks", + "table": "traffic_stats" + }, + { + "value": "\"src_locality\" IN ('all', 'internal', 'external')", + "name": "traffic_stats_src_locality_check", + "entityType": "checks", + "table": "traffic_stats" + }, + { + "value": "\"dst_locality\" IN ('all', 'internal', 'external')", + "name": "traffic_stats_dst_locality_check", + "entityType": "checks", + "table": "traffic_stats" + } + ], + "renames": [] +} ``` ## apps/web/src/lib/server/db/local-schema.ts Status: modified; +45; −0 ```diff @@ -136,6 +136,51 @@ CHECK ((tau IS NULL) = (d0 IS NULL)) ); + CREATE TABLE IF NOT EXISTS address_concentration_stats ( + source_id TEXT NOT NULL, + granularity TEXT NOT NULL CHECK (granularity IN ('5m', '10m', '30m', '1h', '1d')), + bucket_start INTEGER NOT NULL, + bucket_end INTEGER NOT NULL CHECK (bucket_end > bucket_start), + ip_version INTEGER NOT NULL CHECK (ip_version IN (4, 6)), + src_locality TEXT NOT NULL CHECK (src_locality IN ('all', 'internal', 'external')), + dst_locality TEXT NOT NULL CHECK (dst_locality IN ('all', 'internal', 'external')), + address_side TEXT NOT NULL CHECK (address_side IN ('source', 'destination')), + measure TEXT NOT NULL CHECK (measure IN ('addresses', 'packets', 'bytes')), + weight_total REAL NOT NULL CHECK (weight_total >= 0 AND weight_total <= 1.7976931348623157e308), + entry_count INTEGER NOT NULL CHECK (entry_count >= 0), + hhi REAL CHECK (hhi > 0 AND hhi <= 1), + top1_share REAL CHECK (top1_share >= 0 AND top1_share <= 1), + top10_share REAL CHECK (top10_share >= 0 AND top10_share <= 1), + top100_share REAL CHECK (top100_share >= 0 AND top100_share <= 1), + entropy_p8 REAL CHECK (entropy_p8 >= 0 AND entropy_p8 <= 1.7976931348623157e308), + entropy_p16 REAL CHECK (entropy_p16 >= 0 AND entropy_p16 <= 1.7976931348623157e308), + entropy_p24 REAL CHECK (entropy_p24 >= 0 AND entropy_p24 <= 1.7976931348623157e308), + entropy_p32 REAL CHECK (entropy_p32 >= 0 AND entropy_p32 <= 1.7976931348623157e308), + entropy_p48 REAL CHECK (entropy_p48 >= 0 AND entropy_p48 <= 1.7976931348623157e308), + entropy_p64 REAL CHECK (entropy_p64 >= 0 AND entropy_p64 <= 1.7976931348623157e308), + entropy_p128 REAL CHECK (entropy_p128 >= 0 AND entropy_p128 <= 1.7976931348623157e308), + CHECK ((entry_count = 0) = (weight_total = 0)), + CHECK ((hhi IS NOT NULL) = (entry_count >= 2)), + CHECK ((top1_share IS NULL) = (hhi IS NULL)), + CHECK ((top10_share IS NULL) = (hhi IS NULL)), + CHECK ((top100_share IS NULL) = (hhi IS NULL)), + CHECK ((entropy_p32 IS NULL) = (hhi IS NULL)), + CHECK ((entropy_p8 IS NOT NULL) = (ip_version = 4 AND hhi IS NOT NULL)), + CHECK ((entropy_p16 IS NOT NULL) = (ip_version = 4 AND hhi IS NOT NULL)), + CHECK ((entropy_p24 IS NOT NULL) = (ip_version = 4 AND hhi IS NOT NULL)), + CHECK ((entropy_p48 IS NOT NULL) = (ip_version = 6 AND hhi IS NOT NULL)), + CHECK ((entropy_p64 IS NOT NULL) = (ip_version = 6 AND hhi IS NOT NULL)), + CHECK ((entropy_p128 IS NOT NULL) = (ip_version = 6 AND hhi IS NOT NULL)), + CHECK (top10_share >= top1_share AND top100_share >= top10_share) + ); + CREATE UNIQUE INDEX IF NOT EXISTS idx_address_concentration_stats_key + ON address_concentration_stats ( + source_id, granularity, src_locality, dst_locality, + ip_version, measure, bucket_start, address_side + ); + CREATE INDEX IF NOT EXISTS idx_address_concentration_stats_bucket + ON address_concentration_stats (granularity, bucket_start); + CREATE TABLE IF NOT EXISTS maad_q_grid ( ip_version INTEGER PRIMARY KEY CHECK (ip_version IN (4, 6)), q_min REAL NOT NULL, ``` ## apps/web/src/lib/server/db/schema.ts Status: modified; +164; −0 ```diff @@ -401,6 +401,170 @@ ] ); +export const addressConcentrationStats = sqliteTable( + 'address_concentration_stats', + { + sourceId: text('source_id').notNull(), + granularity: text('granularity', { enum: ['5m', '10m', '30m', '1h', '1d'] }).notNull(), + bucketStart: integer('bucket_start').notNull(), + bucketEnd: integer('bucket_end').notNull(), + ipVersion: integer('ip_version').notNull(), + srcLocality: text('src_locality', { enum: ['all', 'internal', 'external'] }).notNull(), + dstLocality: text('dst_locality', { enum: ['all', 'internal', 'external'] }).notNull(), + addressSide: text('address_side', { enum: ['source', 'destination'] }).notNull(), + measure: text('measure', { enum: ['addresses', 'packets', 'bytes'] }).notNull(), + weightTotal: real('weight_total').notNull(), + entryCount: integer('entry_count').notNull(), + hhi: real('hhi'), + top1Share: real('top1_share'), + top10Share: real('top10_share'), + top100Share: real('top100_share'), + entropyP8: real('entropy_p8'), + entropyP16: real('entropy_p16'), + entropyP24: real('entropy_p24'), + entropyP32: real('entropy_p32'), + entropyP48: real('entropy_p48'), + entropyP64: real('entropy_p64'), + entropyP128: real('entropy_p128') + }, + (table) => [ + uniqueIndex('idx_address_concentration_stats_key').on( + table.sourceId, + table.granularity, + table.srcLocality, + table.dstLocality, + table.ipVersion, + table.measure, + table.bucketStart, + table.addressSide + ), + index('idx_address_concentration_stats_bucket').on(table.granularity, table.bucketStart), + check( + 'address_concentration_stats_granularity_check', + sql`${table.granularity} IN ('5m', '10m', '30m', '1h', '1d')` + ), + check( + 'address_concentration_stats_bucket_check', + sql`${table.bucketEnd} > ${table.bucketStart}` + ), + check('address_concentration_stats_ip_version_check', sql`${table.ipVersion} IN (4, 6)`), + check( + 'address_concentration_stats_src_locality_check', + sql`${table.srcLocality} IN ('all', 'internal', 'external')` + ), + check( + 'address_concentration_stats_dst_locality_check', + sql`${table.dstLocality} IN ('all', 'internal', 'external')` + ), + check( + 'address_concentration_stats_address_side_check', + sql`${table.addressSide} IN ('source', 'destination')` + ), + check( + 'address_concentration_stats_measure_check', + sql`${table.measure} IN ('addresses', 'packets', 'bytes')` + ), + check( + 'address_concentration_stats_weight_total_check', + sql`${table.weightTotal} >= 0 AND ${table.weightTotal} <= 1.7976931348623157e308` + ), + check('address_concentration_stats_entry_count_check', sql`${table.entryCount} >= 0`), + check( + 'address_concentration_stats_empty_weight_check', + sql`(${table.entryCount} = 0) = (${table.weightTotal} = 0)` + ), + check( + 'address_concentration_stats_minimum_count_check', + sql`(${table.hhi} IS NOT NULL) = (${table.entryCount} >= 2)` + ), + check('address_concentration_stats_hhi_check', sql`${table.hhi} > 0 AND ${table.hhi} <= 1`), + check( + 'address_concentration_stats_top1_share_check', + sql`${table.top1Share} >= 0 AND ${table.top1Share} <= 1` + ), + check( + 'address_concentration_stats_top10_share_check', + sql`${table.top10Share} >= 0 AND ${table.top10Share} <= 1` + ), + check( + 'address_concentration_stats_top100_share_check', + sql`${table.top100Share} >= 0 AND ${table.top100Share} <= 1` + ), + check( + 'address_concentration_stats_entropy_p8_check', + sql`${table.entropyP8} >= 0 AND ${table.entropyP8} <= 1.7976931348623157e308` + ), + check( + 'address_concentration_stats_entropy_p16_check', + sql`${table.entropyP16} >= 0 AND ${table.entropyP16} <= 1.7976931348623157e308` + ), + check( + 'address_concentration_stats_entropy_p24_check', + sql`${table.entropyP24} >= 0 AND ${table.entropyP24} <= 1.7976931348623157e308` + ), + check( + 'address_concentration_stats_entropy_p32_check', + sql`${table.entropyP32} >= 0 AND ${table.entropyP32} <= 1.7976931348623157e308` + ), + check( + 'address_concentration_stats_entropy_p48_check', + sql`${table.entropyP48} >= 0 AND ${table.entropyP48} <= 1.7976931348623157e308` + ), + check( + 'address_concentration_stats_entropy_p64_check', + sql`${table.entropyP64} >= 0 AND ${table.entropyP64} <= 1.7976931348623157e308` + ), + check( + 'address_concentration_stats_entropy_p128_check', + sql`${table.entropyP128} >= 0 AND ${table.entropyP128} <= 1.7976931348623157e308` + ), + check( + 'address_concentration_stats_top1_share_presence_check', + sql`(${table.top1Share} IS NULL) = (${table.hhi} IS NULL)` + ), + check( + 'address_concentration_stats_top10_share_presence_check', + sql`(${table.top10Share} IS NULL) = (${table.hhi} IS NULL)` + ), + check( + 'address_concentration_stats_top100_share_presence_check', + sql`(${table.top100Share} IS NULL) = (${table.hhi} IS NULL)` + ), + check( + 'address_concentration_stats_entropy_p32_presence_check', + sql`(${table.entropyP32} IS NULL) = (${table.hhi} IS NULL)` + ), + check( + 'address_concentration_stats_entropy_p8_presence_check', + sql`(${table.entropyP8} IS NOT NULL) = (${table.ipVersion} = 4 AND ${table.hhi} IS NOT NULL)` + ), + check( + 'address_concentration_stats_entropy_p16_presence_check', + sql`(${table.entropyP16} IS NOT NULL) = (${table.ipVersion} = 4 AND ${table.hhi} IS NOT NULL)` + ), + check( + 'address_concentration_stats_entropy_p24_presence_check', + sql`(${table.entropyP24} IS NOT NULL) = (${table.ipVersion} = 4 AND ${table.hhi} IS NOT NULL)` + ), + check( + 'address_concentration_stats_entropy_p48_presence_check', + sql`(${table.entropyP48} IS NOT NULL) = (${table.ipVersion} = 6 AND ${table.hhi} IS NOT NULL)` + ), + check( + 'address_concentration_stats_entropy_p64_presence_check', + sql`(${table.entropyP64} IS NOT NULL) = (${table.ipVersion} = 6 AND ${table.hhi} IS NOT NULL)` + ), + check( + 'address_concentration_stats_entropy_p128_presence_check', + sql`(${table.entropyP128} IS NOT NULL) = (${table.ipVersion} = 6 AND ${table.hhi} IS NOT NULL)` + ), + check( + 'address_concentration_stats_top_order_check', + sql`${table.top10Share} >= ${table.top1Share} AND ${table.top100Share} >= ${table.top10Share}` + ) + ] +); + export const maadQGrid = sqliteTable( 'maad_q_grid', { ``` ## apps/web/tests/lib/server/migrations.test.ts Status: modified; +51; −0 ```diff @@ -308,6 +308,57 @@ } }); + it.each([ + [ + 'migrated', + (database: Database.Database) => + migrationFiles().forEach((file) => applyMigration(database, file)) + ], + ['local', (database: Database.Database) => database.exec(localSchemaSql)] + ])('stores concentration with the same constraints in the %s schema', (_schema, createSchema) => { + const database = new Database(':memory:'); + try { + createSchema(database); + const insert = database.prepare(` + INSERT INTO address_concentration_stats ( + source_id, granularity, bucket_start, bucket_end, ip_version, + src_locality, dst_locality, address_side, measure, weight_total, entry_count, + hhi, top1_share, top10_share, top100_share, + entropy_p8, entropy_p16, entropy_p24, entropy_p32, entropy_p48, entropy_p64, entropy_p128 + ) VALUES ('edge', '5m', 0, 300, ?, 'all', 'all', 'source', 'addresses', 2, 2, + 0.5, 0.5, 1, 1, ?, ?, ?, 1, ?, ?, ?) + `); + insert.run(4, 0, 0, 0, null, null, null); + insert.run(6, null, null, null, 0, 0, 1); + expect( + database.prepare('SELECT COUNT(*) AS count FROM address_concentration_stats').get() + ).toEqual({ count: 2 }); + for (const change of [ + 'hhi = 1.1', + 'hhi = NULL', + 'weight_total = 0', + 'top10_share = 0.1', + 'entropy_p32 = NULL', + 'entry_count = 1' + ]) { + expect(() => database.exec(`UPDATE address_concentration_stats SET ${change}`)).toThrow( + /CHECK constraint failed/ + ); + } + expect(() => + database.exec('UPDATE address_concentration_stats SET entropy_p8 = 0 WHERE ip_version = 6') + ).toThrow(/CHECK constraint failed/); + expect(() => + database.exec( + 'UPDATE address_concentration_stats SET entropy_p128 = 0 WHERE ip_version = 4' + ) + ).toThrow(/CHECK constraint failed/); + expect(() => insert.run(4, 0, 0, 0, null, null, null)).toThrow(/UNIQUE constraint failed/); + } finally { + database.close(); + } + }); + it('upgrades the deployed coverage schema without retaining unused bucket-first indexes', () => { const database = new Database(':memory:'); const migrations = migrationFiles(); ``` ## docs/code/pipeline-contract.md Status: modified; +29; −0 ```diff @@ -11,6 +11,7 @@ - The canonical endpoint-locality rules - The pipeline timezone - The native decoder contract +- The concentration contract version, measure set, and fixed prefix lengths - The MAAD enabled state, contract version, configuration, and measure set, plus `"internal_side": false` when the product skips internal-side address sets @@ -151,6 +152,34 @@ coverage, and publishing logs the dropped flow count per bucket. A CSV row without a packet count has an unknown count rather than a reported zero, so it still contributes. +## Concentration statistics + +`address_concentration_stats` stores conventional concentration controls for every address set and measure that MAAD computes. +It uses the same bucket, locality scope, address side, and IP version keys as `address_maad_stats`. +Disabling MAAD disables concentration statistics. The `maad_internal_side` setting applies to both tables. + +Each measure uses the same per-address weights as MAAD and excludes zero weights. +`entry_count` counts the retained addresses. `weight_total` stores their summed weight, including below the minimum count of two. +Below that minimum, all statistical columns are `NULL`. +An empty MAAD curve caused by prefix pruning does not suppress concentration values above that minimum. + +- `hhi` is the sum of squared weight shares. For `addresses`, it equals `1 / entry_count`. +- `top1_share`, `top10_share`, and `top100_share` sum the largest address weights and divide by `weight_total`. +- Prefix entropy is Shannon entropy in bits after summing weights by fixed-length prefix. + IPv4 uses `/8`, `/16`, `/24`, and `/32`. IPv6 uses `/32`, `/48`, `/64`, and `/128`. + Columns for the other IP version are `NULL`. Both families use `entropy_p32`. + +The pipeline computes concentration from MAAD's sorted address entries without another flow scan or full weight sort. +Rollups recompute from merged per-address child maps. They never average child statistics. +All statistic columns and `weight_total` are `REAL` values without f32 rounding. `entry_count` is an integer. + +The schema and result configuration record this contract. Existing products require a fresh database path. +`verify --require-maad-data` requires matching concentration rows for every computed MAAD set and measure. +`verify` checks retained counts against MAAD and rejects skipped internal-side rows. +`compare` applies `--maad-absolute-tolerance` to concentration values and `weight_total`, and compares `entry_count` exactly. +`merge-shards` validates concentration day ownership and internal-side skipping before copying rows. +Analysis-window exports include the new table. + ## Coordinated subset runs Repeat `--dataset` for two or more registry entries to build coordinated subset products. Each entry ``` ## tools/netflow-db/src/compare.rs Status: modified; +35; −1 ```diff @@ -104,6 +104,36 @@ CandidateOnlyPolicy::MissingReferenceBucket, ), TableSpec::new( + "address_concentration_stats", + &[ + "granularity", + "bucket_start", + "source_id", + "ip_version", + "src_locality", + "dst_locality", + "address_side", + "measure", + "bucket_end", + ], + &[], + &[ + "hhi", + "top1_share", + "top10_share", + "top100_share", + "entropy_p8", + "entropy_p16", + "entropy_p24", + "entropy_p32", + "entropy_p48", + "entropy_p64", + "entropy_p128", + "weight_total", + ], + CandidateOnlyPolicy::MissingReferenceBucket, + ), + TableSpec::new( "processed_inputs", &[ "input_kind", @@ -191,7 +221,11 @@ let mut compatible = true; let mut tables = BTreeMap::new(); for spec in TABLES { + let skip_internal_side = candidate_skips_internal_side + && matches!( - let skip_internal_side = candidate_skips_internal_side && spec.name == "address_maad_stats"; + spec.name, + "address_maad_stats" | "address_concentration_stats" + ); let comparison = compare_table(&candidate, &reference, spec, skip_internal_side, options)?; compatible &= comparison.reference_only_rows == 0 && comparison.unexpected_candidate_only_rows == 0 ``` ## tools/netflow-db/src/concentration.rs Status: added; +186; −0 ```diff @@ -1,0 +1,186 @@ +use crate::maad::{MIN_MAAD_ADDRESSES, PrefixBits}; + +#[derive(Clone, Debug, PartialEq)] +pub struct ConcentrationResult { + pub entry_count: usize, + pub weight_total: f64, + pub values: Option, +} + +#[derive(Clone, Debug, PartialEq)] +pub struct ConcentrationValues { + pub hhi: f64, + pub top1_share: f64, + pub top10_share: f64, + pub top100_share: f64, + pub entropy: [f64; 4], +} + +pub(crate) fn compute_sorted( + entries: impl Iterator + Clone, + uniform: bool, +) -> ConcentrationResult { + let mut entry_count = 0; + let mut weight_total = 0.0; + for (_, weight) in entries.clone().filter(|(_, weight)| *weight > 0.0) { + entry_count += 1; + weight_total += weight; + } + let values = (entry_count >= MIN_MAAD_ADDRESSES).then(|| { + let (hhi, top) = if uniform { + ( + 1.0 / weight_total, + [1, 10, 100].map(|k| k.min(entry_count) as f64 / weight_total), + ) + } else { + let mut weights = Vec::with_capacity(entry_count); + let mut hhi = 0.0; + for (_, weight) in entries.clone().filter(|(_, weight)| *weight > 0.0) { + let share = weight / weight_total; + hhi += share * share; + weights.push(weight); + } + if weights.len() > 100 { + weights.select_nth_unstable_by(99, |a, b| b.total_cmp(a)); + weights.truncate(100); + } + weights.sort_unstable_by(|a, b| b.total_cmp(a)); + ( + hhi.min(1.0), + [1, 10, 100].map(|k| { + if k >= entry_count { + 1.0 + } else { + (weights.iter().take(k).sum::() / weight_total).min(1.0) + } + }), + ) + }; + let prefix_lengths = if B::WIDTH == 32 { + [8, 16, 24, 32] + } else { + [32, 48, 64, 128] + }; + let entropy = prefix_lengths.map(|length| { + let mut current = None; + let mut mass = 0.0; + let mut entropy = 0.0; + let contribution = |mass: f64| { + let share = mass / weight_total; + if share > 0.0 { + -share * share.log2() + } else { + 0.0 + } + }; + for (address, weight) in entries.clone().filter(|(_, weight)| *weight > 0.0) { + let prefix = address.prefix(length); + if current.is_some_and(|current| current != prefix) { + entropy += contribution(mass); + mass = 0.0; + } + current = Some(prefix); + mass += weight; + } + entropy + contribution(mass) + }); + ConcentrationValues { + hhi, + top1_share: top[0], + top10_share: top[1], + top100_share: top[2], + entropy, + } + }); + ConcentrationResult { + entry_count, + weight_total, + values, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::{Ipv4Addr, Ipv6Addr}; + + fn close(actual: f64, expected: f64) { + assert!((actual - expected).abs() < 1e-12, "{actual} != {expected}"); + } + + #[test] + fn ipv4_small_set_matches_hand_computed_values() { + let entries = [ + (u32::from(Ipv4Addr::new(10, 1, 1, 1)), 1.0), + (u32::from(Ipv4Addr::new(10, 1, 1, 2)), 1.0), + (u32::from(Ipv4Addr::new(10, 1, 2, 1)), 2.0), + (u32::from(Ipv4Addr::new(11, 1, 1, 1)), 4.0), + ]; + let result = compute_sorted(entries.into_iter(), false); + assert_eq!(result.entry_count, 4); + close(result.weight_total, 8.0); + let values = result.values.unwrap(); + close(values.hhi, 22.0 / 64.0); + close(values.top1_share, 0.5); + close(values.top10_share, 1.0); + close(values.top100_share, 1.0); + for (actual, expected) in values.entropy.into_iter().zip([1.0, 1.0, 1.5, 1.75]) { + close(actual, expected); + } + let uniform = compute_sorted(entries.into_iter().map(|(a, _)| (a, 1.0)), true) + .values + .unwrap(); + close(uniform.hhi, 0.25); + close(uniform.top1_share, 0.25); + for (actual, expected) in + uniform + .entropy + .into_iter() + .zip([0.8112781244591328, 0.8112781244591328, 1.5, 2.0]) + { + close(actual, expected); + } + } + + #[test] + fn ipv6_fixed_prefixes_include_full_address_entropy() { + let entries = [ + ("2001:db8:1:1::1", 1.0), + ("2001:db8:1:1::2", 1.0), + ("2001:db8:1:2::1", 2.0), + ("2001:db9:1:1::1", 4.0), + ] + .map(|(a, w)| (u128::from(a.parse::().unwrap()), w)); + let result = compute_sorted(entries.into_iter(), false); + let values = result.values.unwrap(); + close(values.hhi, 22.0 / 64.0); + for (actual, expected) in values.entropy.into_iter().zip([1.0, 1.0, 1.5, 1.75]) { + close(actual, expected); + } + } + + #[test] + fn zero_weights_are_excluded_and_small_sets_keep_audit_fields() { + let values = compute_sorted([(0u32, 0.0), (1, 2.0), (2, 2.0)].into_iter(), false); + assert_eq!(values.entry_count, 2); + close(values.weight_total, 4.0); + close(values.values.unwrap().hhi, 0.5); + for entries in [vec![], vec![(0u32, 0.0)], vec![(0u32, 0.0), (1, 9.0)]] { + let result = compute_sorted(entries.iter().copied(), false); + assert!(result.values.is_none()); + assert_eq!(result.entry_count, usize::from(result.weight_total > 0.0)); + } + } + + #[test] + fn partial_selection_finds_top_k_without_sorting_all_weights() { + let entries = (1u32..=200).map(|a| (a, a as f64)).collect::>(); + let values = compute_sorted(entries.iter().copied(), false) + .values + .unwrap(); + close(values.top1_share, 200.0 / 20100.0); + close(values.top10_share, 1955.0 / 20100.0); + close(values.top100_share, 15050.0 / 20100.0); + close(values.hhi, 2686700.0 / (20100.0 * 20100.0)); + } +} ``` ## tools/netflow-db/src/export.rs Status: modified; +25; −0 ```diff @@ -363,6 +363,31 @@ ]); return columns; } + if table == "address_concentration_stats" { + let mut columns = COMMON_COLUMNS + .iter() + .copied() + .filter(|column| *column != "processed_at") + .collect::>(); + columns.extend([ + "address_side", + "measure", + "weight_total", + "entry_count", + "hhi", + "top1_share", + "top10_share", + "top100_share", + "entropy_p8", + "entropy_p16", + "entropy_p24", + "entropy_p32", + "entropy_p48", + "entropy_p64", + "entropy_p128", + ]); + return columns; + } if table == "bucket_coverage" { return vec![ "source_id", ``` ## tools/netflow-db/src/lib.rs Status: modified; +2; −1 ```diff @@ -3,9 +3,10 @@ #![forbid(unsafe_code)] /// The persisted pipeline contract version produced by this implementation. -pub const PIPELINE_CONTRACT_VERSION: u32 = 7; +pub const PIPELINE_CONTRACT_VERSION: u32 = 8; pub mod compare; +pub mod concentration; pub mod config; pub mod coverage; pub mod domain; ``` ## tools/netflow-db/src/maad.rs Status: modified; +93; −11 ```diff @@ -5,7 +5,7 @@ use std::io::Write; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; -const MIN_MAAD_ADDRESSES: usize = 2; +pub(crate) const MIN_MAAD_ADDRESSES: usize = 2; const SCHEMA_VERSION: u32 = 3; const DEFAULT_FULL_THRESHOLD: f64 = 0.05; const DEFAULT_Q_STEP: f64 = 1.0 / 8.0; @@ -291,11 +291,97 @@ )) } +pub(crate) struct MeasuresWithConcentration { + pub addresses: MaadResult, + pub weighted: [MaadResult; K], + pub concentration_addresses: crate::concentration::ConcentrationResult, + pub concentration_weighted: [crate::concentration::ConcentrationResult; K], +} + -/// Validate, sort and sum weighted entries, then walk their prefix levels once. +pub(crate) fn compute_measures_with_concentration( + entries: impl IntoIterator, +) -> Result, MaadError> { + let config = A::default_config(); + let prepared = prepare_measures(entries, &config)?; + let concentration_addresses = crate::concentration::compute_sorted( + prepared.entries.iter().map(|&(address, _)| (address, 1.0)), + true, + ); + let concentration_weighted = std::array::from_fn(|k| { + crate::concentration::compute_sorted( + prepared + .entries + .iter() + .map(|&(address, weights)| (address, weights[k])), + false, + ) + }); + let (analysis, q_values) = prepared.analyze(&config); + Ok(MeasuresWithConcentration { + addresses: analysis.address_result(&q_values, config.q_step), + weighted: std::array::from_fn(|k| analysis.weighted_result(k, &q_values)), + concentration_addresses, + concentration_weighted, + }) +} + +pub(crate) fn compute_weighted_with_concentration( + entries: impl IntoIterator, +) -> Result<(MaadResult, crate::concentration::ConcentrationResult), MaadError> { + let config = A::default_config(); + let prepared = prepare_measures( + entries + .into_iter() + .map(|(address, weight)| (address, [weight])), + &config, + )?; + let concentration = crate::concentration::compute_sorted( + prepared + .entries + .iter() + .map(|&(address, weights)| (address, weights[0])), + false, + ); + let (analysis, q_values) = prepared.analyze(&config); + Ok((analysis.weighted_result(0, &q_values), concentration)) +} + +struct PreparedMeasures { + entries: Vec<(B, [f64; K])>, + totals: [f64; K], + q_values: Vec, +} + +impl PreparedMeasures { + fn analyze(self, config: &MaadConfig) -> (Analysis, Vec) { + if self.entries.len() < MIN_MAAD_ADDRESSES { + return (Analysis::empty(self.entries.len(), K), self.q_values); + } + let addresses: Vec<_> = self.entries.iter().map(|&(address, _)| address).collect(); + let columns: Vec> = (0..K) + .map(|k| self.entries.iter().map(|(_, weights)| weights[k]).collect()) + .collect(); + drop(self.entries); + ( + analyze(&addresses, &columns, &self.totals, config), + self.q_values, + ) + } +} + fn analyze_measures( entries: impl IntoIterator, config: &MaadConfig, ) -> Result<(Analysis, Vec), MaadError> { + let prepared = prepare_measures(entries, config)?; + Ok(prepared.analyze(config)) +} + +/// Validate, sort and sum weighted entries. +fn prepare_measures( + entries: impl IntoIterator, + config: &MaadConfig, +) -> Result, MaadError> { let q_values = validate_config(config, A::Bits::WIDTH)?; let mut entries = entries .into_iter() @@ -327,15 +413,11 @@ if let Some(&total) = totals.iter().find(|total| !total.is_finite()) { return Err(MaadError::NonFiniteTotalWeight { total }); } - if entries.len() < MIN_MAAD_ADDRESSES { + Ok(PreparedMeasures { + entries, + totals, + q_values, - return Ok((Analysis::empty(entries.len(), K), q_values)); + }) - } - let addresses: Vec<_> = entries.iter().map(|&(address, _)| address).collect(); - let columns: Vec> = (0..K) - .map(|k| entries.iter().map(|(_, weights)| weights[k]).collect()) - .collect(); - drop(entries); - Ok((analyze(&addresses, &columns, &totals, config), q_values)) } /// The uniform q grid on which a family's default configuration evaluates the structure function. ``` ## tools/netflow-db/src/merge.rs Status: modified; +66; −14 ```diff @@ -19,12 +19,13 @@ }, }; -const DAY_OWNED_TABLES: [&str; 8] = [ +const DAY_OWNED_TABLES: [&str; 9] = [ "traffic_stats", "protocol_stats", "address_count_stats", "port_count_stats", "address_maad_stats", + "address_concentration_stats", "bucket_coverage", "input_evidence", "processed_inputs", @@ -340,18 +341,18 @@ } drop(rows); drop(statement); - if !stored_maad_internal_side(&connection)? - && connection.query_row( - &format!( - "SELECT EXISTS(SELECT 1 FROM address_maad_stats WHERE {INTERNAL_SIDE_MAAD_SQL})" - ), - [], - |row| row.get::<_, bool>(0), - )? - { - return Err(refused(format!( - "{label} skips internal-side MAAD but stores internal-side MAAD rows" - ))); + if !stored_maad_internal_side(&connection)? { + for table in ["address_maad_stats", "address_concentration_stats"] { + if connection.query_row( + &format!("SELECT EXISTS(SELECT 1 FROM {table} WHERE {INTERNAL_SIDE_MAAD_SQL})"), + [], + |row| row.get::<_, bool>(0), + )? { + return Err(refused(format!( + "{label} skips internal-side MAAD but stores internal-side rows in {table}" + ))); + } + } } let source_members = connection .prepare("SELECT dataset_id, source_id, member_id FROM source_members")? @@ -1094,7 +1095,11 @@ fixture.merge(&merged, &[&first, &second]).unwrap(); let connection = Connection::open(&merged).unwrap(); + for table in [ - for table in ["maad_q_grid", "address_maad_stats"] { + "maad_q_grid", + "address_maad_stats", + "address_concentration_stats", + ] { let rows: i64 = connection .query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |row| { row.get(0) @@ -1186,6 +1191,53 @@ assert!(!merged.exists()); } + #[test] + fn concentration_rows_outside_completed_days_are_refused() { + let fixture = Fixture::new(&["2025-06-01", "2025-06-02"]); + let first = fixture.path("first.sqlite"); + let second = fixture.path("second.sqlite"); + let output = fixture.path("merged.sqlite"); + fixture.run(&first, "2025-06-01", "2025-06-01", true); + fixture.run(&second, "2025-06-02", "2025-06-02", true); + Connection::open(&second).unwrap().execute( + "UPDATE address_concentration_stats SET bucket_start = bucket_start + 86400, bucket_end = bucket_end + 86400", + [], + ).unwrap(); + assert_refused( + fixture.merge(&output, &[&first, &second]), + "address_concentration_stats rows", + ); + assert!(!output.exists()); + } + + #[test] + fn skipping_shards_reject_internal_side_concentration_rows() { + let fixture = Fixture::new(&["2025-06-01"]); + let shard = fixture.path("shard.sqlite"); + fixture.run(&shard, "2025-06-01", "2025-06-01", true); + let connection = Connection::open(&shard).unwrap(); + let config: String = connection + .query_row("SELECT config_json FROM pipeline_product", [], |row| { + row.get(0) + }) + .unwrap(); + let mut config: serde_json::Value = serde_json::from_str(&config).unwrap(); + config["maad"]["internal_side"] = false.into(); + connection + .execute( + "UPDATE pipeline_product SET config_json = ?1", + [config.to_string()], + ) + .unwrap(); + connection.execute("UPDATE address_concentration_stats SET src_locality = 'internal' WHERE address_side = 'source' AND src_locality = 'all'", []).unwrap(); + let output = fixture.path("merged.sqlite"); + assert_refused( + fixture.merge(&output, &[&shard]), + "address_concentration_stats", + ); + assert!(!output.exists()); + } + fn assert_refused(result: Result, fragment: &str) { let error = result.unwrap_err(); assert!( ``` ## tools/netflow-db/src/pipeline.rs Status: modified; +10; −1 ```diff @@ -1532,7 +1532,7 @@ maad["internal_side"] = json!(false); } let result_config = json!({ - "version": 6, + "version": 7, "timezone": pipeline.timezone, "locality": pipeline .selection @@ -1548,6 +1548,15 @@ "executable": nfdump_executable, }, "maad": maad, + "concentration": { + "enabled": pipeline.maad.enabled(), + "contract_version": 1, + "measures": crate::domain::MaadMeasure::ALL.map(crate::domain::MaadMeasure::as_str), + "ipv4_prefix_lengths": [8, 16, 24, 32], + "ipv6_prefix_lengths": [32, 48, 64, 128], + "minimum_addresses": crate::maad::MIN_MAAD_ADDRESSES, + "zero_weight_exclusion": true, + }, }); let identity = ProductIdentity::create( &schema, ``` ## tools/netflow-db/src/publish.rs Status: modified; +190; −31 ```diff @@ -18,9 +18,10 @@ }, maad, storage::{ + AddressConcentrationStatsRow, AddressCountStatsRow, AddressMaadStatsRow, BucketCoverageRow, + MaadCurve, MaadQGridRow, PortCountStatsRow, ProtocolStatsRow, StatsBucketKey, + StatsDimensions, StorageError, TrafficStatsRow, delete_stats_bucket_keys, + insert_address_concentration_stats_rows, insert_address_count_stats_rows, - AddressCountStatsRow, AddressMaadStatsRow, BucketCoverageRow, MaadCurve, MaadQGridRow, - PortCountStatsRow, ProtocolStatsRow, StatsBucketKey, StatsDimensions, StorageError, - TrafficStatsRow, delete_stats_bucket_keys, insert_address_count_stats_rows, insert_address_maad_stats_rows, insert_bucket_coverage_rows, insert_maad_q_grid_rows, insert_port_count_stats_rows, insert_protocol_stats_rows, insert_traffic_stats_rows, }, @@ -192,7 +193,7 @@ .map(|addresses| count(addresses.addresses.len())) .sum::(); let maad_started = Instant::now(); - let (address_maad, zero_weight_addresses) = maad_rows(&address_sets)?; + let (address_maad, concentration, zero_weight_addresses) = maad_rows(&address_sets)?; profile.maad_elapsed += maad_started.elapsed(); profile.maad_zero_weight_addresses += zero_weight_addresses; if zero_weight_addresses > 0 { @@ -218,6 +219,7 @@ .sum::(); let insert_started = Instant::now(); insert_address_maad_stats_rows(connection, &address_maad)?; + insert_address_concentration_stats_rows(connection, &concentration)?; profile.address_maad_insert_elapsed += insert_started.elapsed(); } } @@ -335,9 +337,16 @@ /// because their summed weight was zero. fn maad_rows( address_sets: &[AddressSetRow<'_>], +) -> Result< + ( + Vec, -) -> Result<(Vec, u64), PublishError> { + Vec, + u64, + ), + PublishError, +> { if address_sets.is_empty() { + return Ok((Vec::new(), Vec::new(), 0)); - return Ok((Vec::new(), 0)); } let pool = maad_pool()?; @@ -347,11 +356,17 @@ .map(scope_rows) .collect::, _>>() })?; - let zero_weight_addresses = measured.iter().map(|(_, excluded)| count(*excluded)).sum(); - Ok(( + let zero_weight_addresses = measured + .iter() - measured.into_iter().flat_map(|(rows, _)| rows).collect(), - zero_weight_addresses, - )) + .map(|(_, _, excluded)| count(*excluded)) + .sum(); + let mut maad = Vec::new(); + let mut concentration = Vec::new(); + for (rows, controls, _) in measured { + maad.extend(rows); + concentration.extend(controls); + } + Ok((maad, concentration, zero_weight_addresses)) } /// One scope's MAAD results: distinct addresses, then packets and bytes with @@ -359,11 +374,19 @@ struct ScopeResults { addresses: maad::MaadResult, weighted: [(maad::MaadResult, usize); 2], + concentration: [crate::concentration::ConcentrationResult; 3], } fn scope_rows( addresses: &AddressSetRow<'_>, +) -> Result< + ( + Vec, -) -> Result<(Vec, usize), PublishError> { + Vec, + usize, + ), + PublishError, +> { let results = match addresses.scope.ip_version { IpVersion::V4 => scope_results(addresses.addresses.iter().filter_map( |(address, traffic)| match address { @@ -397,7 +420,17 @@ rows.push(row(measure, result, *zero_weight_addrs)); zero_weight_addresses += zero_weight_addrs; } + let concentration = MaadMeasure::ALL + .into_iter() + .zip(results.concentration) + .map(|(measure, result)| AddressConcentrationStatsRow { + dimensions: dimensions.clone(), + address_side: addresses.address_side.as_str().to_owned(), + measure: measure.as_str().to_owned(), + result, + }) - Ok((rows, zero_weight_addresses)) + .collect(); + Ok((rows, concentration, zero_weight_addresses)) } /// Store one measure's result with its curve rounded to f32. Only the addresses measure @@ -449,28 +482,42 @@ .filter(|(_, traffic)| traffic.bytes == 0) .count(); if zero_packets == 0 && zero_bytes == 0 { - let (addresses, [packets, bytes]) = - maad::compute_measures(entries.iter().map(|&(address, traffic)| { - (address, [traffic.packets as f64, traffic.bytes as f64]) - }))?; + let measured = + maad::compute_measures_with_concentration(entries.iter().map( + |&(address, traffic)| (address, [traffic.packets as f64, traffic.bytes as f64]), + ))?; + let [packets, bytes] = measured.weighted; + let [concentration_packets, concentration_bytes] = measured.concentration_weighted; return Ok(ScopeResults { - addresses, + addresses: measured.addresses, weighted: [(packets, 0), (bytes, 0)], + concentration: [ + measured.concentration_addresses, + concentration_packets, + concentration_bytes, + ], }); } let weighted = |weight: fn(AddressTraffic) -> u64| { - maad::compute_weighted(entries.iter().filter_map(|&(address, traffic)| { - match weight(traffic) { + maad::compute_weighted_with_concentration(entries.iter().filter_map( + |&(address, traffic)| match weight(traffic) { 0 => None, value => Some((address, value as f64)), + }, - } + )) - })) }; + let addresses = maad::compute_measures_with_concentration( + entries.iter().map(|&(address, _)| (address, [])), + )?; + let (packet_result, packet_concentration) = weighted(|traffic| traffic.packets)?; + let (byte_result, byte_concentration) = weighted(|traffic| traffic.bytes)?; Ok(ScopeResults { - addresses: maad::compute(entries.iter().map(|&(address, _)| address)), - weighted: [ + addresses: addresses.addresses, + weighted: [(packet_result, zero_packets), (byte_result, zero_bytes)], + concentration: [ + addresses.concentration_addresses, + packet_concentration, + byte_concentration, - (weighted(|traffic| traffic.packets)?, zero_packets), - (weighted(|traffic| traffic.bytes)?, zero_bytes), ], }) } @@ -674,12 +721,111 @@ "4:internal->external:destination", ] ); + let concentration_scopes = connection.prepare( + "SELECT DISTINCT ip_version || ':' || src_locality || '->' || dst_locality || ':' || address_side + FROM address_concentration_stats WHERE ip_version = 4 ORDER BY 1", + ).unwrap().query_map([], |row| row.get::<_, String>(0)).unwrap() + .collect::, _>>().unwrap(); + assert_eq!(concentration_scopes, scopes); + assert_eq!( + connection + .query_row( + "SELECT COUNT(*) FROM address_concentration_stats", + [], + |row| row.get::<_, i64>(0) + ) + .unwrap(), + 36 + ); assert_eq!(profile.address_count_rows, 20); assert_eq!(profile.maad_address_sets, 12); assert_eq!(profile.address_maad_rows, 36); } #[test] + fn concentration_rollups_recompute_after_merging_repeated_addresses() { + let address = |last| IpAddr::V4(Ipv4Addr::new(10, 0, 0, last)); + let first = [ + (address(1), AddressTraffic::new(1, 10)), + (address(2), AddressTraffic::new(3, 30)), + ] + .into_iter() + .collect::(); + let second = [ + (address(1), AddressTraffic::new(5, 50)), + (address(3), AddressTraffic::new(3, 30)), + ] + .into_iter() + .collect::(); + let child = |start, totals: AddressTotals| { + let mut bucket = StatisticalBucket::dense(BucketKey::new( + "r1", + Granularity::FiveMinutes, + start, + start + 300, + )); + bucket + .add(ScopedAddressesFact::new( + Scope::new(IpVersion::V4, Locality::All, Locality::All), + AddressSide::Source, + totals.iter(), + )) + .unwrap(); + bucket.finish() + }; + let mut rollup = + StatisticalBucket::new(BucketKey::new("r1", Granularity::TenMinutes, 0, 600)); + rollup.include(&child(0, first.clone())).unwrap(); + rollup.include(&child(300, second.clone())).unwrap(); + let connection = Connection::open_in_memory().unwrap(); + init_stats_tables(&connection).unwrap(); + write_buckets(&connection, &[rollup.finish()], MaadScopes::All).unwrap(); + let mut merged = first; + merged.merge(&second); + let recomputed = scope_results(merged.iter().map(|(a, t)| { + let IpAddr::V4(a) = a else { unreachable!() }; + (a, t) + })) + .unwrap(); + let (hhi, top, total, count): (f64, f64, f64, i64) = connection + .query_row( + "SELECT hhi, top1_share, weight_total, entry_count FROM address_concentration_stats + WHERE ip_version = 4 AND src_locality = 'all' AND dst_locality = 'all' + AND address_side = 'source' AND measure = 'packets'", + [], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), + ) + .unwrap(); + let control = &recomputed.concentration[1]; + let values = connection + .query_row( + "SELECT hhi, top1_share, top10_share, top100_share, + entropy_p8, entropy_p16, entropy_p24, entropy_p32 + FROM address_concentration_stats WHERE ip_version = 4 + AND src_locality = 'all' AND dst_locality = 'all' + AND address_side = 'source' AND measure = 'packets'", + [], + |row| { + Ok(crate::concentration::ConcentrationValues { + hhi: row.get(0)?, + top1_share: row.get(1)?, + top10_share: row.get(2)?, + top100_share: row.get(3)?, + entropy: [row.get(4)?, row.get(5)?, row.get(6)?, row.get(7)?], + }) + }, + ) + .unwrap(); + assert_eq!(control.values.as_ref(), Some(&values)); + assert_eq!(values.entropy, [0.0, 0.0, 0.0, 1.5]); + assert_eq!((total, count), (12.0, 3)); + assert_eq!(hhi, control.values.as_ref().unwrap().hhi); + assert!((hhi - 0.375).abs() < 1e-12); + assert_eq!(top, 0.5); + assert!((hhi - (0.625 + 0.53125) / 2.0).abs() > 0.1); + } + + #[test] fn maad_workers_cannot_resize_an_initialized_pool() { assert!(maad_pool().unwrap().current_num_threads() >= 1); let error = set_maad_workers(NonZeroUsize::new(2).unwrap()).unwrap_err(); @@ -704,7 +850,7 @@ connection } - fn product_rows(connection: &Connection) -> (Vec, Vec) { + fn product_rows(connection: &Connection) -> (Vec, Vec, Vec) { fn query(connection: &Connection, sql: &str) -> Vec { connection .prepare(sql) @@ -724,6 +870,10 @@ connection, "SELECT printf('%s|%s|%s|%s|%s|%d|%d|%s|%s|%s|%s|%s|%s|%s|%s', ip_version, src_locality, dst_locality, address_side, measure, total_addrs, zero_weight_addrs, min_prefix_length, max_prefix_length, d0, d1, d2, hex(tau), hex(tau_sd), hex(spectrum)) FROM address_maad_stats ORDER BY ip_version, src_locality, dst_locality, address_side, measure", ), + query( + connection, + "SELECT printf('%s|%s|%s|%s|%s|%d|%.17g|%.17g|%.17g|%.17g|%.17g|%.17g|%.17g|%.17g|%.17g', ip_version, src_locality, dst_locality, address_side, measure, entry_count, weight_total, hhi, top1_share, top10_share, top100_share, entropy_p8, entropy_p16, entropy_p24, entropy_p32) FROM address_concentration_stats ORDER BY ip_version, src_locality, dst_locality, address_side, measure", + ), ) } @@ -781,10 +931,11 @@ }, ]; + let (first, first_controls, first_excluded) = maad_rows(&rows).unwrap(); - let (first, first_excluded) = maad_rows(&rows).unwrap(); + let (second, second_controls, second_excluded) = maad_rows(&rows).unwrap(); - let (second, second_excluded) = maad_rows(&rows).unwrap(); assert_eq!(first, second); + assert_eq!(first_controls, second_controls); assert_eq!((first_excluded, second_excluded), (0, 0)); let measures = |src_locality: &str, side: &str| { ["addresses", "packets", "bytes"] @@ -916,9 +1067,10 @@ addresses: &addresses, }]; - let (rows, _) = maad_rows(&rows).unwrap(); + let (rows, controls, _) = maad_rows(&rows).unwrap(); assert_eq!(rows.len(), 3); + assert!(controls.iter().all(|row| row.result.entry_count == 256)); assert!(rows.iter().all(|row| row.dimensions.ip_version == 6)); for row in &rows { assert_eq!(row.total_addrs, 256, "{}", row.measure); @@ -950,11 +1102,18 @@ addresses: &addresses, }]; - let (rows, excluded) = maad_rows(&rows).unwrap(); + let (rows, controls, excluded) = maad_rows(&rows).unwrap(); let row = |measure: &str| rows.iter().find(|row| row.measure == measure).unwrap(); let counts = |measure: &str| (row(measure).total_addrs, row(measure).zero_weight_addrs); assert_eq!(excluded, 64 + 10); + assert_eq!( + controls + .iter() + .map(|row| row.result.entry_count) + .collect::>(), + [256, 192, 246] + ); assert_eq!(counts("addresses"), (256, 0)); assert_eq!(counts("packets"), (192, 64)); assert_eq!(counts("bytes"), (246, 10)); ``` ## tools/netflow-db/src/storage.rs Status: modified; +175; −22 ```diff @@ -26,24 +26,26 @@ pub const BUSY_TIMEOUT_MS: u64 = 60_000; /// Tables that form the portable pipeline product. -pub const STATS_TABLE_NAMES: [&str; 6] = [ +pub const STATS_TABLE_NAMES: [&str; 7] = [ "traffic_stats", "protocol_stats", "address_count_stats", "port_count_stats", "address_maad_stats", + "address_concentration_stats", "bucket_coverage", ]; /// Versioned product schema bound into every pipeline database identity. pub fn product_schema() -> serde_json::Value { serde_json::json!({ + "version": 8, - "version": 7, "tables": [ {"name": "traffic_stats", "version": 4}, {"name": "protocol_stats", "version": 3}, {"name": "address_count_stats", "version": 3}, {"name": "port_count_stats", "version": 3}, {"name": "address_maad_stats", "version": 1}, + {"name": "address_concentration_stats", "version": 1}, {"name": "maad_q_grid", "version": 1}, {"name": "bucket_coverage", "version": 2} ] @@ -1535,6 +1537,51 @@ CREATE INDEX IF NOT EXISTS idx_address_maad_stats_bucket ON address_maad_stats (granularity, bucket_start); + CREATE TABLE IF NOT EXISTS address_concentration_stats ( + source_id TEXT NOT NULL, + granularity TEXT NOT NULL CHECK (granularity IN ('5m', '10m', '30m', '1h', '1d')), + bucket_start INTEGER NOT NULL, + bucket_end INTEGER NOT NULL CHECK (bucket_end > bucket_start), + ip_version INTEGER NOT NULL CHECK (ip_version IN (4, 6)), + src_locality TEXT NOT NULL CHECK (src_locality IN ('all', 'internal', 'external')), + dst_locality TEXT NOT NULL CHECK (dst_locality IN ('all', 'internal', 'external')), + address_side TEXT NOT NULL CHECK (address_side IN ('source', 'destination')), + measure TEXT NOT NULL CHECK (measure IN ('addresses', 'packets', 'bytes')), + weight_total REAL NOT NULL CHECK (weight_total >= 0 AND weight_total <= 1.7976931348623157e308), + entry_count INTEGER NOT NULL CHECK (entry_count >= 0), + hhi REAL CHECK (hhi > 0 AND hhi <= 1), + top1_share REAL CHECK (top1_share >= 0 AND top1_share <= 1), + top10_share REAL CHECK (top10_share >= 0 AND top10_share <= 1), + top100_share REAL CHECK (top100_share >= 0 AND top100_share <= 1), + entropy_p8 REAL CHECK (entropy_p8 >= 0 AND entropy_p8 <= 1.7976931348623157e308), + entropy_p16 REAL CHECK (entropy_p16 >= 0 AND entropy_p16 <= 1.7976931348623157e308), + entropy_p24 REAL CHECK (entropy_p24 >= 0 AND entropy_p24 <= 1.7976931348623157e308), + entropy_p32 REAL CHECK (entropy_p32 >= 0 AND entropy_p32 <= 1.7976931348623157e308), + entropy_p48 REAL CHECK (entropy_p48 >= 0 AND entropy_p48 <= 1.7976931348623157e308), + entropy_p64 REAL CHECK (entropy_p64 >= 0 AND entropy_p64 <= 1.7976931348623157e308), + entropy_p128 REAL CHECK (entropy_p128 >= 0 AND entropy_p128 <= 1.7976931348623157e308), + CHECK ((entry_count = 0) = (weight_total = 0)), + CHECK ((hhi IS NOT NULL) = (entry_count >= 2)), + CHECK ((top1_share IS NULL) = (hhi IS NULL)), + CHECK ((top10_share IS NULL) = (hhi IS NULL)), + CHECK ((top100_share IS NULL) = (hhi IS NULL)), + CHECK ((entropy_p32 IS NULL) = (hhi IS NULL)), + CHECK ((entropy_p8 IS NOT NULL) = (ip_version = 4 AND hhi IS NOT NULL)), + CHECK ((entropy_p16 IS NOT NULL) = (ip_version = 4 AND hhi IS NOT NULL)), + CHECK ((entropy_p24 IS NOT NULL) = (ip_version = 4 AND hhi IS NOT NULL)), + CHECK ((entropy_p48 IS NOT NULL) = (ip_version = 6 AND hhi IS NOT NULL)), + CHECK ((entropy_p64 IS NOT NULL) = (ip_version = 6 AND hhi IS NOT NULL)), + CHECK ((entropy_p128 IS NOT NULL) = (ip_version = 6 AND hhi IS NOT NULL)), + CHECK (top10_share >= top1_share AND top100_share >= top10_share) + ); + CREATE UNIQUE INDEX IF NOT EXISTS idx_address_concentration_stats_key + ON address_concentration_stats ( + source_id, granularity, src_locality, dst_locality, + ip_version, measure, bucket_start, address_side + ); + CREATE INDEX IF NOT EXISTS idx_address_concentration_stats_bucket + ON address_concentration_stats (granularity, bucket_start); + CREATE TABLE IF NOT EXISTS maad_q_grid ( ip_version INTEGER PRIMARY KEY CHECK (ip_version IN (4, 6)), q_min REAL NOT NULL, @@ -1947,6 +1994,14 @@ pub spectrum: Option>, } +#[derive(Clone, Debug, PartialEq)] +pub struct AddressConcentrationStatsRow { + pub dimensions: StatsDimensions, + pub address_side: String, + pub measure: String, + pub result: crate::concentration::ConcentrationResult, +} + /// The dimensions and structure function of a non-empty MAAD result. #[derive(Clone, Debug, PartialEq)] pub struct MaadCurve { @@ -2153,6 +2208,55 @@ Ok(()) } +pub fn insert_address_concentration_stats_rows( + connection: &Connection, + rows: &[AddressConcentrationStatsRow], +) -> Result<(), StorageError> { + let mut statement = connection.prepare_cached( + "INSERT OR REPLACE INTO address_concentration_stats ( + source_id, granularity, bucket_start, bucket_end, ip_version, + src_locality, dst_locality, address_side, measure, weight_total, entry_count, + hhi, top1_share, top10_share, top100_share, + entropy_p8, entropy_p16, entropy_p24, entropy_p32, entropy_p48, entropy_p64, entropy_p128 + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, + ?16, ?17, ?18, ?19, ?20, ?21, ?22)", + )?; + for row in rows { + let d = &row.dimensions; + let values = row.result.values.as_ref(); + let entropy = |ip, index| { + values + .filter(|_| d.ip_version == ip) + .map(|v| v.entropy[index]) + }; + statement.execute(params![ + d.source_id, + d.granularity, + d.bucket_start, + d.bucket_end, + d.ip_version, + d.src_locality, + d.dst_locality, + row.address_side, + row.measure, + row.result.weight_total, + i64::try_from(row.result.entry_count).unwrap_or(i64::MAX), + values.map(|v| v.hhi), + values.map(|v| v.top1_share), + values.map(|v| v.top10_share), + values.map(|v| v.top100_share), + entropy(4, 0), + entropy(4, 1), + entropy(4, 2), + values.map(|v| v.entropy[if d.ip_version == 4 { 3 } else { 0 }]), + entropy(6, 1), + entropy(6, 2), + entropy(6, 3) + ])?; + } + Ok(()) +} + pub fn insert_maad_q_grid_rows( connection: &Connection, rows: &[MaadQGridRow], @@ -2174,6 +2278,7 @@ pub address_count_rows: Vec, pub port_count_rows: Vec, pub address_maad_rows: Vec, + pub address_concentration_rows: Vec, } #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -2183,15 +2288,17 @@ AddressCount, PortCount, AddressMaad, + AddressConcentration, } impl StatsTable { + pub const ALL: [Self; 6] = [ - pub const ALL: [Self; 5] = [ Self::Traffic, Self::Protocol, Self::AddressCount, Self::PortCount, Self::AddressMaad, + Self::AddressConcentration, ]; pub const fn table_name(self) -> &'static str { @@ -2201,6 +2308,7 @@ Self::AddressCount => "address_count_stats", Self::PortCount => "port_count_stats", Self::AddressMaad => "address_maad_stats", + Self::AddressConcentration => "address_concentration_stats", } } } @@ -2224,6 +2332,10 @@ StatsTable::AddressMaad => { insert_address_maad_stats_rows(connection, &payload.address_maad_rows)? } + StatsTable::AddressConcentration => insert_address_concentration_stats_rows( + connection, + &payload.address_concentration_rows, + )?, } } Ok(()) @@ -3108,7 +3220,44 @@ } } + fn concentration_example() -> AddressConcentrationStatsRow { + AddressConcentrationStatsRow { + dimensions: AddressMaadStatsRow::example().dimensions, + address_side: "source".into(), + measure: "addresses".into(), + result: crate::concentration::compute_sorted([(0u32, 1.0), (1, 1.0)].into_iter(), true), + } + } + #[test] + fn concentration_constraints_reject_invalid_statistics_and_prefix_families() { + let connection = Connection::open_in_memory().unwrap(); + init_stats_tables(&connection).unwrap(); + insert_address_concentration_stats_rows(&connection, &[concentration_example()]).unwrap(); + for change in [ + "hhi = 1.1", + "hhi = NULL", + "top1_share = -0.1", + "top10_share = 0.1", + "entropy_p8 = NULL", + "entropy_p48 = 1", + "weight_total = 0", + "entry_count = 1", + ] { + let error = connection + .execute( + &format!("UPDATE address_concentration_stats SET {change}"), + [], + ) + .unwrap_err(); + assert!( + error.to_string().contains("CHECK constraint failed"), + "{error}" + ); + } + } + + #[test] fn stats_rows_upsert_and_bucket_deletion_cover_every_table() { let connection = Connection::open_in_memory().unwrap(); init_stats_tables(&connection).unwrap(); @@ -3117,6 +3266,7 @@ insert_address_count_stats_rows(&connection, &[AddressCountStatsRow::example()]).unwrap(); insert_port_count_stats_rows(&connection, &[PortCountStatsRow::example()]).unwrap(); insert_address_maad_stats_rows(&connection, &[AddressMaadStatsRow::example()]).unwrap(); + insert_address_concentration_stats_rows(&connection, &[concentration_example()]).unwrap(); assert_eq!( connection .query_row("SELECT flows_tcp FROM traffic_stats", [], |row| row @@ -3191,24 +3341,24 @@ .collect::>>() .unwrap() .join("\n"); + let index_clause = format!("USING INDEX idx_{table}_bucket"); + let bounded_range = + if matches!(table, "address_maad_stats" | "address_concentration_stats") { + vec![ + index_clause.as_str(), + "granularity=?", + "bucket_start>?", + "bucket_start?", + "bucket_start?", - "bucket_start?", - "bucket_start Result<(), VerifyError> { + for table in ["address_maad_stats", "address_concentration_stats"] { + let stored: bool = connection.query_row( + &format!("SELECT EXISTS (SELECT 1 FROM {table} WHERE {INTERNAL_SIDE_MAAD_SQL})"), + [], + |row| row.get(0), + )?; + if stored { + return Err(VerifyError::Incompatible(format!( + "{table} has internal-side rows but datasets skip internal-side MAAD" + ))); - let stored: bool = connection.query_row( - &format!("SELECT EXISTS (SELECT 1 FROM address_maad_stats WHERE {INTERNAL_SIDE_MAAD_SQL})"), - [], - |row| row.get(0), - )?; - if stored { - return Err(VerifyError::Incompatible( - "address_maad_stats has internal-side rows but datasets skip internal-side MAAD".into(), - )); + } } Ok(()) } @@ -694,6 +724,58 @@ Ok(()) } +fn assert_concentration_data( + connection: &Connection, + require_coverage: bool, +) -> Result<(), VerifyError> { + let key = "s.source_id = m.source_id AND s.granularity = m.granularity + AND s.bucket_start = m.bucket_start AND s.bucket_end = m.bucket_end + AND s.ip_version = m.ip_version AND s.src_locality = m.src_locality + AND s.dst_locality = m.dst_locality AND s.address_side = m.address_side + AND s.measure = m.measure"; + let invalid: bool = connection.query_row( + &format!( + "SELECT EXISTS ( + SELECT 1 FROM address_concentration_stats s + LEFT JOIN address_maad_stats m ON {key} + WHERE m.source_id IS NULL OR s.entry_count <> m.total_addrs + OR (s.entry_count < 2) <> (s.hhi IS NULL) + OR (s.measure = 'addresses' AND ( + s.weight_total <> s.entry_count OR + abs(s.hhi - 1.0 / s.entry_count) > 1e-12 + )) + )" + ), + [], + |row| row.get(0), + )?; + if invalid { + return Err(VerifyError::Incompatible( + "address_concentration_stats has an unmatched MAAD set, count, or invalid result" + .into(), + )); + } + if require_coverage { + let missing: bool = connection.query_row( + &format!( + "SELECT EXISTS ( + SELECT 1 FROM address_maad_stats m + WHERE NOT EXISTS (SELECT 1 FROM address_concentration_stats s WHERE {key}) + )" + ), + [], + |row| row.get(0), + )?; + if missing { + return Err(VerifyError::Incompatible( + "address_concentration_stats is missing rows for computed MAAD sets or measures" + .into(), + )); + } + } + Ok(()) +} + fn assert_processed_inputs_complete(connection: &Connection) -> Result<(), VerifyError> { let pending: i64 = connection.query_row( "SELECT COUNT(*) FROM processed_inputs WHERE status != 'processed'", ``` ## tools/netflow-db/tests/concentration_cli.rs Status: added; +279; −0 ```diff @@ -1,0 +1,279 @@ +use std::{ + fs, + path::PathBuf, + process::{Command, Output}, +}; + +use netflow_db::storage::ProductIdentity; +use rusqlite::{Connection, params}; +use tempfile::{TempDir, tempdir}; + +fn run(args: &[&str]) -> Output { + Command::new(env!("CARGO_BIN_EXE_netflow-db")) + .args(args) + .output() + .unwrap() +} + +fn successful(args: &[&str]) { + let output = run(args); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); +} + +fn fixture() -> (TempDir, PathBuf, PathBuf) { + let directory = tempdir().unwrap(); + let csv = directory.path().join("flows.csv"); + fs::write( + &csv, + "received,src,dst,packets,bytes,protocol\n\ + 0,192.0.2.1,198.51.100.1,2,100,TCP\n\ + 10,192.0.2.2,198.51.100.1,3,200,UDP\n\ + 300,192.0.2.2,198.51.100.2,5,300,TCP\n\ + 310,192.0.2.3,198.51.100.2,7,0,TCP\n\ + 0,2001:db8::1,2001:db9::1,2,100,TCP\n\ + 10,2001:db8::2,2001:db9::1,3,200,UDP\n\ + 300,2001:db8::2,2001:db9::2,5,300,TCP\n\ + 310,2001:db8::3,2001:db9::2,7,0,TCP\n", + ) + .unwrap(); + let mapping = directory.path().join("mapping.json"); + fs::write( + &mapping, + serde_json::to_vec(&serde_json::json!({ + "timestamp_format": "unix", "timestamp_timezone": "UTC", + "columns": {"time_received": "received", "src_ip": "src", "dst_ip": "dst", + "packets": "packets", "bytes": "bytes", "protocol": "protocol"}, + "source_id": {"value": "edge"} + })) + .unwrap(), + ) + .unwrap(); + let database = directory.path().join("product.sqlite"); + let config = directory.path().join("pipeline.json"); + fs::write( + &config, + serde_json::to_vec(&serde_json::json!({ + "database_path": database, "timezone": "UTC", "run_maad": true, + "inputs": [{"input_kind": "csv", "path": csv, "mapping_path": mapping}] + })) + .unwrap(), + ) + .unwrap(); + successful(&["pipeline", "--config", config.to_str().unwrap()]); + Connection::open(&database) + .unwrap() + .execute( + "INSERT INTO datasets (id, label, default_start_date, maad_internal_side) + VALUES ('fixture', 'Fixture', '1970-01-01', 0)", + [], + ) + .unwrap(); + (directory, database, config) +} + +#[test] +fn dual_stack_pipeline_stores_concentration_for_every_maad_set_and_rollup() { + let (_directory, database, _config) = fixture(); + successful(&[ + "verify", + database.to_str().unwrap(), + "--require-data", + "--require-maad-data", + "--require-rollup-parity", + ]); + let connection = Connection::open(&database).unwrap(); + let cardinalities: (i64, i64) = connection + .query_row( + "SELECT (SELECT COUNT(*) FROM address_maad_stats), + (SELECT COUNT(*) FROM address_concentration_stats)", + [], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .unwrap(); + assert_eq!(cardinalities.0, cardinalities.1); + for ip in [4, 6] { + for (measure, expected_hhi, expected_top, expected_total, count) in [ + ("addresses", 1.0 / 3.0, 1.0 / 3.0, 3.0, 3), + ("packets", 117.0 / 289.0, 8.0 / 17.0, 17.0, 3), + ("bytes", 26.0 / 36.0, 5.0 / 6.0, 600.0, 2), + ] { + let (hhi, top, total, entries, unused): (f64, f64, f64, i64, Option) = connection + .query_row( + "SELECT hhi, top1_share, weight_total, entry_count, + CASE WHEN ip_version = 4 THEN entropy_p128 ELSE entropy_p8 END + FROM address_concentration_stats WHERE granularity = '10m' AND bucket_start = 0 + AND ip_version = ?1 AND src_locality = 'all' AND dst_locality = 'all' + AND address_side = 'source' AND measure = ?2", + params![ip, measure], + |row| { + Ok(( + row.get(0)?, + row.get(1)?, + row.get(2)?, + row.get(3)?, + row.get(4)?, + )) + }, + ) + .unwrap(); + assert!((hhi - expected_hhi).abs() < 1e-12); + assert!((top - expected_top).abs() < 1e-12); + assert_eq!((total, entries, unused), (expected_total, count, None)); + } + } + let singleton: (i64, f64, Option) = connection.query_row( + "SELECT entry_count, weight_total, hhi FROM address_concentration_stats + WHERE granularity = '5m' AND bucket_start = 300 AND ip_version = 6 + AND src_locality = 'all' AND dst_locality = 'all' AND address_side = 'source' AND measure = 'bytes'", + [], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + ).unwrap(); + assert_eq!(singleton, (1, 300.0, None)); +} + +#[test] +fn verify_requires_concentration_coverage_and_checks_retained_counts() { + let (_directory, database, _config) = fixture(); + let connection = Connection::open(&database).unwrap(); + connection.execute("UPDATE address_concentration_stats SET entry_count = entry_count + 1 WHERE hhi IS NOT NULL AND measure = 'packets'", []).unwrap(); + let result = run(&["verify", database.to_str().unwrap()]); + assert!(!result.status.success()); + assert!(String::from_utf8_lossy(&result.stderr).contains("count")); + connection.execute("UPDATE address_concentration_stats SET entry_count = entry_count - 1 WHERE hhi IS NOT NULL AND measure = 'packets'", []).unwrap(); + connection.execute("DELETE FROM address_concentration_stats WHERE measure = 'bytes' AND granularity = '10m'", []).unwrap(); + let result = run(&["verify", database.to_str().unwrap(), "--require-maad-data"]); + assert!(!result.status.success()); + assert!( + String::from_utf8_lossy(&result.stderr).contains("address_concentration_stats is missing") + ); +} + +#[test] +fn old_product_identity_rejects_the_new_result_contract() { + let (_directory, database, config) = fixture(); + let connection = Connection::open(&database).unwrap(); + let (schema, selection, config_json): (String, String, String) = connection + .query_row( + "SELECT schema_json, selection_json, config_json FROM pipeline_product", + [], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + ) + .unwrap(); + let mut schema: serde_json::Value = serde_json::from_str(&schema).unwrap(); + schema["version"] = 7.into(); + schema["tables"] + .as_array_mut() + .unwrap() + .retain(|table| table["name"] != "address_concentration_stats"); + let selection: serde_json::Value = serde_json::from_str(&selection).unwrap(); + let mut result: serde_json::Value = serde_json::from_str(&config_json).unwrap(); + result["version"] = 6.into(); + result.as_object_mut().unwrap().remove("concentration"); + let old = ProductIdentity::create(&schema, &selection, &result).unwrap(); + connection + .execute( + "UPDATE pipeline_product SET schema_json = ?1, schema_fingerprint = ?2, + config_json = ?3, config_fingerprint = ?4, product_fingerprint = ?5", + params![ + old.schema_json, + old.schema_fingerprint, + old.config_json, + old.config_fingerprint, + old.fingerprint + ], + ) + .unwrap(); + let output = run(&["pipeline", "--config", config.to_str().unwrap()]); + assert!(!output.status.success()); + assert!(String::from_utf8_lossy(&output.stderr).contains("identity mismatch")); +} + +#[test] +fn compare_concentration_uses_maad_tolerance_and_exact_counts() { + let (directory, database, _config) = fixture(); + let candidate = directory.path().join("candidate.sqlite"); + fs::copy(&database, &candidate).unwrap(); + let connection = Connection::open(&candidate).unwrap(); + let compare = || { + let output = run(&[ + "compare", + candidate.to_str().unwrap(), + database.to_str().unwrap(), + "--start", + "0", + "--end", + "600", + "--maad-absolute-tolerance", + "0.000001", + ]); + let report: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); + (output.status.success(), report) + }; + connection.execute("UPDATE address_concentration_stats SET hhi = hhi + 0.0000005 WHERE measure = 'packets' AND hhi IS NOT NULL", []).unwrap(); + assert!(compare().0); + connection.execute("UPDATE address_concentration_stats SET hhi = hhi + 0.001 WHERE measure = 'packets' AND hhi IS NOT NULL", []).unwrap(); + let (success, report) = compare(); + assert!(!success); + assert!( + report["tables"]["address_concentration_stats"]["mismatched_rows"] + .as_i64() + .unwrap() + > 0 + ); + connection.execute("UPDATE address_concentration_stats SET hhi = hhi - 0.0010005 WHERE measure = 'packets' AND hhi IS NOT NULL", []).unwrap(); + connection.execute("UPDATE address_concentration_stats SET entry_count = entry_count + 1 WHERE measure = 'packets' AND hhi IS NOT NULL", []).unwrap(); + assert!(!compare().0); +} + +#[test] +fn compare_skips_reference_internal_concentration_and_rejects_candidate_rows() { + let (directory, candidate, _config) = fixture(); + let reference = directory.path().join("reference.sqlite"); + fs::copy(&candidate, &reference).unwrap(); + let internal_row = "INSERT INTO address_concentration_stats + SELECT source_id, granularity, bucket_start, bucket_end, ip_version, + 'internal', 'external', address_side, measure, weight_total, entry_count, + hhi, top1_share, top10_share, top100_share, entropy_p8, entropy_p16, + entropy_p24, entropy_p32, entropy_p48, entropy_p64, entropy_p128 + FROM address_concentration_stats + WHERE granularity = '5m' AND ip_version = 4 AND bucket_start = 0 + AND src_locality = 'all' AND dst_locality = 'all' + AND address_side = 'source' AND measure = 'addresses'"; + Connection::open(&reference) + .unwrap() + .execute(internal_row, []) + .unwrap(); + let compare = || { + let output = run(&[ + "compare", + candidate.to_str().unwrap(), + reference.to_str().unwrap(), + "--start", + "0", + "--end", + "600", + ]); + let report: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); + (output.status.success(), report) + }; + let (success, report) = compare(); + assert!(success); + assert_eq!( + report["tables"]["address_concentration_stats"]["skipped_reference_rows"], + 1 + ); + let connection = Connection::open(&candidate).unwrap(); + connection.execute(internal_row, []).unwrap(); + let (success, report) = compare(); + assert!(!success); + assert_eq!( + report["tables"]["address_concentration_stats"]["unexpected_candidate_only_rows"], + 1 + ); + let output = run(&["verify", candidate.to_str().unwrap()]); + assert!(!output.status.success()); + assert!(String::from_utf8_lossy(&output.stderr).contains("internal-side")); +} ``` ## tools/netflow-db/tests/ten_minute_rollups.rs Status: modified; +1; −0 ```diff @@ -122,6 +122,7 @@ "address_count_stats", "port_count_stats", "address_maad_stats", + "address_concentration_stats", ] { let rows: i64 = connection .query_row( ```